diff --git a/lib/workers/index.js b/lib/workers/index.js index 5bdbd5d7a..2fe5be544 100644 --- a/lib/workers/index.js +++ b/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,