Restart db-workers after a given number of tasks

lib/workers/index.js

@@ -52,16 +52,20 @@ Workers.initialize = function (Env, config, _cb) {
         //return Object.keys(workers[index].tasks || {}).length;
     };
 
+//    const WORKER_TASK_LIMIT = 100000;
+    const WORKER_TASK_LIMIT = 100; // XXX
+
     var workerOffset = -1;
     var queue = [];
-    var getAvailableWorkerIndex = function () {
+    var getAvailableWorkerIndex = function (isQueue) {
 //  If there is already a backlog of tasks you can avoid some work
-//  by going to the end of the line
-        if (queue.length) { return -1; }
+//  by going to the end of the line (unless we're trying to
+//  empty the queue)
+        if (queue.length && !isQueue) { return -1; }
 
         var L = workers.length;
         if (L === 0) {
-            Log.error('NO_WORKERS_AVAILABLE', {
+            Log.warn('NO_WORKERS_AVAILABLE', {
                 queue: queue.length,
             });
             return -1;
@@ -93,7 +97,7 @@ Workers.initialize = function (Env, config, _cb) {
     };
 
     var drained = true;
-    var sendCommand = function (msg, _cb, opt) {
+    var sendCommand = function (msg, _cb, opt, isQueue) {
         if (!_cb) {
             return void Log.error('WORKER_COMMAND_MISSING_CB', {
                 msg: msg,
@@ -102,7 +106,7 @@ Workers.initialize = function (Env, config, _cb) {
         }
 
         opt = opt || {};
-        var index = getAvailableWorkerIndex();
+        var index = getAvailableWorkerIndex(isQueue);
 
         var state = workers[index];
         // if there is no worker available:
@@ -114,7 +118,7 @@ Workers.initialize = function (Env, config, _cb) {
             });
             if (drained) {
                 drained = false;
-                Log.error('WORKER_QUEUE_BACKLOG', {
+                Log.warn('WORKER_QUEUE_BACKLOG', {
                     workers: workers.length,
                 });
             }
@@ -134,6 +138,7 @@ Workers.initialize = function (Env, config, _cb) {
             // an upper bound on the amount of parallelism for any given worker.
             // if you run out of slots then the worker locks up.
             delete state.tasks[txid];
+            state.checkTasks();
         })));
 
         if (!msg) {
@@ -164,6 +169,12 @@ Workers.initialize = function (Env, config, _cb) {
             msg._cb = _cb;
             msg._opt = opt;
         });
+
+        state.count++;
+        if (state.count > WORKER_TASK_LIMIT) {
+            // Remove from list and spawn new one
+            if (state.replaceWorker) { state.replaceWorker(); }
+        }
     };
 
     const pluginsResponses = {};
@@ -207,6 +218,8 @@ Workers.initialize = function (Env, config, _cb) {
         if (!res.txid) { return; }
         response.handle(res.txid, [res.error, res.value]);
         delete state.tasks[res.txid];
+        state.checkTasks();
+
         if (!queue.length) {
             if (!drained) {
                 drained = true;
@@ -234,7 +247,7 @@ Workers.initialize = function (Env, config, _cb) {
     to the back because the following msg took its place. OR, in an
     even worse scenario, we cycle through the queue but don't run anything.
 */
-        sendCommand(nextMsg.msg, nextMsg.cb);
+        sendCommand(nextMsg.msg, nextMsg.cb, {}, true);
     };
 
     const initWorker = function (worker, cb) {
@@ -243,13 +256,60 @@ Workers.initialize = function (Env, config, _cb) {
         const state = {
             worker: worker,
             tasks: {},
+            count: Math.floor(Math.random()*(WORKER_TASK_LIMIT/10)),
             pid: worker.pid, // store the child process's id in an easily accessible location
         };
 
+        state.replaceWorker = () => {
+            let index = workers.indexOf(state);
+            if (index === -1) { return; }
+            // Remove old
+            workers.splice(index, 1);
+            // Create new
+            state.complete = true;
+            const w = fork(DB_PATH);
+            Log.debug('WORKER_REPLACE_START', {
+                from: state.worker.pid,
+                to: w.pid
+            });
+            initWorker(w, function (err) {
+                if (err) {
+                    throw new Error(err);
+                }
+            });
+        };
+
+        // If we've reached the limit, kill the worker once
+        // all the tasks are complete or timed out
+        state.checkTasks = () => {
+            // Check limit
+            if (!state.complete || !state.worker) { return; }
+            // Check remaining tasks
+            if (Object.keys(state.tasks).length) { return; }
+            // Kill
+            Log.debug('WORKER_KILL', {
+                worker: state.worker.pid,
+                count: state.count
+            });
+            delete state.worker;
+            worker.kill();
+        }
+
         response.expect(txid, function (err) {
             if (err) { return void cb(err); }
             workers.push(state);
             cb(void 0, state);
+            // We just pushed a new worker, available to receive
+            // a task, so we can empty the queue if necessary
+            if (queue.length) {
+                const nextMsg = queue.shift();
+                if (!nextMsg || !nextMsg.msg) {
+                    return Log.error('WORKER_QUEUE_EMPTY_MESSAGE', {
+                        item: nextMsg,
+                    });
+                }
+                sendCommand(nextMsg.msg, nextMsg.cb, {}, true);
+            }
         }, 15000);
 
         worker.send({
@@ -296,18 +356,21 @@ Workers.initialize = function (Env, config, _cb) {
         });
 
         worker.on('exit', function () {
+            if (!state.worker) { return; } // Manually killed
             substituteWorker();
             Env.Log.error("DB_WORKER_EXIT", {
                 pid: state.pid,
             });
         });
         worker.on('close', function () {
+            if (!state.worker) { return; } // Manually killed
             substituteWorker();
             Env.Log.error("DB_WORKER_CLOSE", {
                 pid: state.pid,
             });
         });
         worker.on('error', function (err) {
+            if (!state.worker) { return; } // Manually killed
             substituteWorker();
             Env.Log.error("DB_WORKER_ERROR", {
                 pid: state.pid,