wrap workers in a function scope and add a validateMessage method to HK's Env

lib/check-signature.js

@@ -5,9 +5,7 @@ const Nacl = require('tweetnacl/nacl-fast');
 // XXX npm "os" and "child_process"
 
 // TODO if this process is using too much CPU, we can use "cluster" to add load balancing to this code
-
+//console.log('New child process', process.pid);
-console.log('New child process', process.pid);
-
 process.on('message', function (data) {
     //console.log('In process', process.pid);
     //console.log(+new Date(), "Message received by subprocess");

lib/historyKeeper.js

@@ -76,6 +76,8 @@ module.exports.create = function (config, cb) {
         domain: config.domain
     };
 
+    HK.initializeValidationWorkers(Env);
+
     (function () {
         var pes = config.premiumUploadSize;
         if (!isNaN(pes) && pes >= Env.maxUploadSize) {

lib/hk-util.js

@@ -924,59 +924,63 @@ HK.onDirectMessage = function (Env, Server, seq, userId, json) {
     * writes messages to the store
 */
 
+HK.initializeValidationWorkers = function (Env) {
+    if (typeof(Env.validateMessage) !== 'undefined') {
+        return void console.error("validation workers are already initialized");
+    }
 
-// Create our workers
+    // Create our workers
-const workers = [];
+    const workers = [];
-for (let i = 0; i < numCPUs; i++) {
+    for (let i = 0; i < numCPUs; i++) {
-    workers.push(fork('lib/check-signature.js'));
+        workers.push(fork('lib/check-signature.js'));
-}
+    }
-var nextWorker = 0;
-
-const response = Util.response();
 
-var initWorker = function (worker) {
+    const response = Util.response();
-    worker.on('message', function (res) {
-        if (!res || !res.txid) { return; }
-        //console.log(+new Date(), "Received verification response");
-        response.handle(res.txid, [res.error]);
-    });
-    // Spawn a new process in one ends
-    worker.on('exit', function () {
-        // XXX make sure it's dead?
-        var idx = workers.indexOf(worker);
-        if (idx !== -1) {
-            workers.splice(idx, 1);
-        }
-        // Spawn a new one
-        var w = fork('lib/check-signature.js');
-        workers.push(w);
-        initWorker(w);
-    });
-};
-workers.forEach(initWorker);
 
+    var initWorker = function (worker) {
+        worker.on('message', function (res) {
+            if (!res || !res.txid) { return; }
+            //console.log(+new Date(), "Received verification response");
+            response.handle(res.txid, [res.error]);
+        });
+        // Spawn a new process in one ends
+        worker.on('exit', function () {
+            // XXX make sure it's dead?
+            var idx = workers.indexOf(worker);
+            if (idx !== -1) {
+                workers.splice(idx, 1);
+            }
+            // Spawn a new one
+            var w = fork('lib/check-signature.js');
+            workers.push(w);
+            initWorker(w);
+        });
+    };
+    workers.forEach(initWorker);
 
-const validateMessage = function (signedMsg, key, _cb) {
+    var nextWorker = 0;
-    // let's be paranoid about asynchrony and only calling back once..
+    Env.validateMessage = function (signedMsg, key, _cb) {
-    var cb = Util.once(Util.mkAsync(_cb));
+        // let's be paranoid about asynchrony and only calling back once..
+        var cb = Util.once(Util.mkAsync(_cb));
 
-    var txid = Util.uid();
+        var txid = Util.uid();
 
-    // expect a response within 15s
+        // expect a response within 15s
-    response.expect(txid, cb, 15000);
+        response.expect(txid, cb, 15000);
 
-    nextWorker = (nextWorker + 1) % workers.length;
+        nextWorker = (nextWorker + 1) % workers.length;
-    if (workers.length === 0 || typeof(workers[nextWorker].send) !== 'function') {
+        if (workers.length === 0 || typeof(workers[nextWorker].send) !== 'function') {
-        console.error(workers);
+            console.error(workers);
-        throw new Error("INVALID_WORKERS");
+            throw new Error("INVALID_WORKERS");
-    }
+        }
 
-    // Send the request
+        // Send the request
-    workers[nextWorker].send({
+        workers[nextWorker].send({
-        txid: txid,
+            txid: txid,
-        msg: signedMsg,
+            msg: signedMsg,
-        key: key,
+            key: key,
-    });
+        });
+    };
 };
 
 HK.onChannelMessage = function (Env, Server, channel, msgStruct) {
@@ -1025,7 +1029,7 @@ HK.onChannelMessage = function (Env, Server, channel, msgStruct) {
 
         // Listen for messages
         //console.log(+new Date(), "Send verification request");
-        validateMessage(signedMsg, metadata.validateKey, w(function (err) {
+        Env.validateMessage(signedMsg, metadata.validateKey, w(function (err) {
             if (err) {
                 // validation can fail in multiple ways
                 if (err === 'FAILED') {