Added support for a "master" producer, in case certain jobs should only be scheduled by a single producer

* Local backend automatically elects the Producer as master
* Redis backend acquires and refreshes a distributed mutex to track the current master

Author: Jesper Svennevid

lib/backends/local.js

@@ -6,6 +6,7 @@ function ProducerBackend(frontend) {
     this.pending = {};
     this.active = {};
     this.queue = [];
+    this.isMaster = false;
 }
 
 _.extend(ProducerBackend.prototype, {
@@ -71,6 +72,15 @@ _.extend(ProducerBackend.prototype, {
             });
         }
         this.processQueue();
+    },
+
+    acquireMaster: function (callback) {
+        this.isMaster = true;
+        callback(true);
+    },
+
+    releaseMaster: function () {
+        this.isMaster = false
     }
 });
 

lib/backends/redis.js

@@ -2,6 +2,27 @@ var _ = require('underscore'),
     cluster = require('cluster'),
     redis = require('redis');
 
+var redis_lock = "" +
+    "local result = redis.call('set', KEYS[1], ARGV[1], 'nx', 'ex', ARGV[2])\n" +
+    "if result then\n" +
+    " return 1\n" +
+    "else\n" +
+    " if redis.call('get', KEYS[1]) == ARGV[1] then\n" +
+    "  redis.call('expire', KEYS[1], ARGV[2])\n" +
+    "  return 2\n" +
+    " else\n" +
+    "  return 0\n" +
+    " end\n" +
+    "end";
+
+var redis_unlock = "" +
+    "if redis.call('get', KEYS[1]) == ARGV[1] then\n" +
+    "then\n" +
+    " return redis.call('del', KEYS[1])\n" +
+    "else\n" +
+    " return 0\n" +
+    "end";
+
 function ProducerBackend(frontend, options) {
     options = _.clone(options) || {};
 
@@ -17,6 +38,7 @@ function ProducerBackend(frontend, options) {
     this.active = {};
     this.queue = [];
     this.producerId = -1;
+    this.isMaster = false;
     this.choked = false;
 
     var producerClient = this.producerClient = redis.createClient(options.port, options.host);
@@ -134,6 +156,35 @@ _.extend(ProducerBackend.prototype, {
             });
         }
         this.processQueue();
+    },
+
+    acquireMaster: function (timeout, callback) {
+        if (this.producerId < 0) {
+            callback(false);
+            return;
+        }
+
+        var masterLock = this.options.prefix + "master";
+        this.producerClient.eval(redis_lock, 1, masterLock, this.producerId, Math.floor(timeout / 1000), _.bind(function (err, response) {
+            if (err) {
+                callback(false);
+                return;
+            }
+
+            this.isMaster = response > 0;
+            callback(this.isMaster);
+        }, this));
+    },
+
+    releaseMaster: function () {
+        this.isMaster = false;
+
+        if (this.producerId < 0) {
+            return;
+        }
+
+        var masterLock = this.options.prefix + "master";
+        this.producerClient.eval(redis_unlock, 1, masterLock, this.producerId);
     }
 });
 

lib/workqueue.js

@@ -10,44 +10,56 @@ function Producer(options) {
         type: 'local',
         workers: require('os').cpus().length,
         timeout: 10000,
+        masterTimeout: 10000,
         redis: {},
         local: {}
     });
 
-    this.index = 0;
     this.options = options;
-    this.debugPort = 0;
-    this.debugMode = _.some(process.execArgv, _.bind(function (s) {
-        var offset = s.indexOf('--debug');
-        if (offset !== -1) {
-            var valueOffset = s.indexOf('=', offset);
-            if (valueOffset !== -1) {
-                this.debugPort = parseInt(s.slice(valueOffset+1));
+
+    this.initialize();
+}
+
+_.extend(Producer.prototype, {
+    initialize: function () {
+        this.index = 0;
+        this.debugPort = 0;
+        this.debugMode = _.some(process.execArgv, _.bind(function (s) {
+            var offset = s.indexOf('--debug');
+            if (offset !== -1) {
+                var valueOffset = s.indexOf('=', offset);
+                if (valueOffset !== -1) {
+                    this.debugPort = parseInt(s.slice(valueOffset+1));
+                }
+                return true;
             }
-            return true;
+            return false;
+        }, this));
+
+        if (this.debugMode) {
+            cluster.setupMaster({
+                execArgv: process.execArgv.filter(function (s) {
+                    return s.indexOf('--debug') === -1;
+                })
+            });
         }
-        return false;
-    }, this));
-
-    if (this.debugMode) {
-        cluster.setupMaster({
-            execArgv: process.execArgv.filter(function (s) {
-                return s.indexOf('--debug') === -1;
-            })
-        });
-    }
 
-    switch (options.type) {
-        default:case 'local': this.backend = new local_backend.ProducerBackend(this, options.local); break;
-        case 'redis': this.backend = new redis_backend.ProducerBackend(this, options.redis); break;
-    }
+        switch (this.options.type) {
+            default:case 'local': this.backend = new local_backend.ProducerBackend(this, this.options.local); break;
+            case 'redis': this.backend = new redis_backend.ProducerBackend(this, this.options.redis); break;
+        }
 
-    for (var i = 0; i < options.workers; ++i) {
-        this.startWorker();
-    }
-}
+        for (var i = 0; i < this.options.workers; ++i) {
+            this.startWorker();
+        }
+
+        this.acquireMaster();
+    },
+
+    destroy: function () {
+        this.releaseMaster();
+    },
 
-_.extend(Producer.prototype, {
     post: function (type, options, callback) {
         var message = {
             payload: {
@@ -94,6 +106,23 @@ _.extend(Producer.prototype, {
             clearTimeout(message.timeout);
         }
         message.callback(response.error);
+    },
+
+    acquireMaster: function () {
+        this.backend.acquireMaster(this.options.masterTimeout * 2, _.bind(function (success) {
+            this.masterTimer = setTimeout(_.bind(this.acquireMaster, this), this.options.masterTimeout + Math.floor(Math.random() * this.options.masterTimeout * 0.5));
+        }, this));
+    },
+
+    releaseMaster: function () {
+        clearTimeout(this.masterTimer);
+        delete this.masterTimer;
+
+        this.backend.releaseMaster();
+    },
+
+    isMaster: function () {
+        return this.backend.isMaster;
     }
 });