From c53675c9d5e4803850806b943679e19a276666ce Mon Sep 17 00:00:00 2001
From: ansuz <ansuz@transitiontech.ca>
Date: Mon, 6 Apr 2020 10:28:26 -0400
Subject: [PATCH] WIP worker rpc call queue

---
 lib/workers/index.js | 117 +++++++++++++++++++++++++++++++++++--------
 1 file changed, 95 insertions(+), 22 deletions(-)

diff --git a/lib/workers/index.js b/lib/workers/index.js
index 7d8c1dcc7..847081104 100644
--- a/lib/workers/index.js
+++ b/lib/workers/index.js
@@ -124,18 +124,58 @@ Workers.initializeIndexWorkers = function (Env, config, _cb) {
         return response.expected(id)? guid(): id;
     };
 
-    var workerIndex = 0;
+    var workerOffset = -1;
+    var getAvailableWorkerIndex = function () {
+        var L = workers.length;
+        if (L === 0) {
+            console.log("no workers available");
+            return -1;
+        }
+
+        // cycle through the workers once
+        // start from a different offset each time
+        // return -1 if none are available
+
+        workerOffset = (workerOffset + 1) % L;
+
+        var temp;
+        for (let i = 0; i < L; i++) {
+            temp = (workerOffset + i) % L;
+            if (workers[temp] && workers[temp].count > 0) {
+                return temp;
+            }
+        }
+        return -1;
+    };
+
+    var queue = [];
+    var MAX_JOBS = 32; //1; //8;
+
     var sendCommand = function (msg, _cb) {
-        var cb = Util.once(Util.mkAsync(_cb));
+        var index = getAvailableWorkerIndex();
+
+        var state = workers[index];
+        // if there is no worker available:
+        if (!isWorker(state)) {
+            console.log("queueing for later");
+            // queue the message for when one becomes available
+            queue.push({
+                msg: msg,
+                cb: _cb,
+            });
+            return;
+            //return void cb("NO_WORKERS");
+        } else {
+            console.log("worker #%s handling %s messages currently", index, MAX_JOBS + 1 - state.count);
 
-        workerIndex = (workerIndex + 1) % workers.length;
-        if (!isWorker(workers[workerIndex])) {
-            return void cb("NO_WORKERS");
         }
 
-        var state = workers[workerIndex];
+        console.log("%s queued messages", queue.length);
 
-        // XXX insert a queue here to prevent timeouts
+        console.log("[%s]\n", msg.command);
+        //console.log(msg);
+
+        var cb = Util.once(Util.mkAsync(_cb));
 
         const txid = guid();
         msg.txid = txid;
@@ -143,20 +183,67 @@ Workers.initializeIndexWorkers = function (Env, config, _cb) {
 
         // track which worker is doing which jobs
         state.tasks[txid] = msg;
+        state.count--;
+
+        if (state.count < 0) {
+            console.log(state);
+            throw new Error("too many jobs"); // XXX
+        }
+
         response.expect(txid, function (err, value) {
             // clean up when you get a response
             delete state[txid];
+            state.count++;
             cb(err, value);
         }, 60000);
         state.worker.send(msg);
     };
 
+    var backlogged;
+    var handleResponse = function (res) {
+        if (!res) { return; }
+        // handle log messages before checking if it was addressed to your PID
+        // it might still be useful to know what happened inside an orphaned worker
+        if (res.log) {
+            return void handleLog(res.log, res.label, res.info);
+        }
+        // but don't bother handling things addressed to other processes
+        // since it's basically guaranteed not to work
+        if (res.pid !== PID) {
+            return void Log.error("WRONG_PID", res);
+        }
+
+        setTimeout(function () {
+            response.handle(res.txid, [res.error, res.value]);
+
+            if (!queue.length) {
+                if (backlogged) {
+                    backlogged = false;
+                    console.log("queue has been drained");
+                }
+                return;
+            } else {
+                backlogged = true;
+                console.log(queue, queue.length);
+            }
+
+            console.log("taking queued message");
+
+            // XXX take a request from the queue
+            var nextMsg = queue.shift();
+            sendCommand(nextMsg.msg, nextMsg.cb); // XXX doesn't feel right
+            console.log("%s queued messages remaining", queue.length);
+
+        }, (Math.floor(Math.random() * 150) * 10));
+    };
+
     const initWorker = function (worker, cb) {
         const txid = guid();
 
         const state = {
             worker: worker,
             tasks: {},
+            count: MAX_JOBS, //1, // XXX
         };
 
         response.expect(txid, function (err) {
@@ -171,21 +258,7 @@ Workers.initializeIndexWorkers = function (Env, config, _cb) {
             config: config,
         });
 
-        worker.on('message', function (res) {
-            if (!res) { return; }
-            // handle log messages before checking if it was addressed to your PID
-            // it might still be useful to know what happened inside an orphaned worker
-            if (res.log) {
-                return void handleLog(res.log, res.label, res.info);
-            }
-            // but don't bother handling things addressed to other processes
-            // since it's basically guaranteed not to work
-            if (res.pid !== PID) {
-                return void Log.error("WRONG_PID", res);
-            }
-
-            response.handle(res.txid, [res.error, res.value]);
-        });
+        worker.on('message', handleResponse);
 
         var substituteWorker = Util.once(function () {
             Env.Log.info("SUBSTITUTE_DB_WORKER", '');