Merge remote-tracking branch 'origin/scaling' into scaling2
commit
67dd4a608c
|
@ -0,0 +1,45 @@
|
|||
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);
|
||||
|
||||
process.on('message', function (data) {
|
||||
console.log('In process', process.pid)
|
||||
console.log(+new Date(), "Message received by subprocess");
|
||||
if (!data || !data.key || !data.msg || !data.txid) {
|
||||
process.send({
|
||||
error:'E_INVAL'
|
||||
});
|
||||
return;
|
||||
}
|
||||
const txid = data.txid;
|
||||
|
||||
const signedMsg = Nacl.util.decodeBase64(data.msg);
|
||||
var validateKey;
|
||||
try {
|
||||
validateKey = Nacl.util.decodeBase64(data.key);
|
||||
} catch (e) {
|
||||
process.send({
|
||||
txid: txid,
|
||||
error:'E_BADKEY'
|
||||
});
|
||||
return;
|
||||
}
|
||||
// validate the message
|
||||
const validated = Nacl.sign.open(signedMsg, validateKey);
|
||||
if (!validated) {
|
||||
process.send({
|
||||
txid: txid,
|
||||
error:'FAILED'
|
||||
});
|
||||
return;
|
||||
}
|
||||
console.log(+new Date(), "Verification done in the subprocess");
|
||||
process.send({
|
||||
txid: txid,
|
||||
success: true
|
||||
});
|
||||
});
|
|
@ -3,6 +3,7 @@
|
|||
const Core = module.exports;
|
||||
const Util = require("../common-util");
|
||||
const escapeKeyCharacters = Util.escapeKeyCharacters;
|
||||
const { fork } = require('child_process');
|
||||
|
||||
/* Use Nacl for checking signatures of messages */
|
||||
const Nacl = require("tweetnacl/nacl-fast");
|
||||
|
|
|
@ -6,6 +6,8 @@ const nThen = require('nthen');
|
|||
const Util = require("./common-util");
|
||||
const MetaRPC = require("./commands/metadata");
|
||||
const Nacl = require('tweetnacl/nacl-fast');
|
||||
const { fork } = require('child_process');
|
||||
const numCPUs = require('os').cpus().length;
|
||||
|
||||
const now = function () { return (new Date()).getTime(); };
|
||||
const ONE_DAY = 1000 * 60 * 60 * 24; // one day in milliseconds
|
||||
|
@ -921,7 +923,36 @@ HK.onDirectMessage = function (Env, Server, seq, userId, json) {
|
|||
* adds timestamps to incoming messages
|
||||
* writes messages to the store
|
||||
*/
|
||||
|
||||
|
||||
const onChecked = Util.mkEvent();
|
||||
// Create our workers
|
||||
const workers = [];
|
||||
for (let i = 0; i < numCPUs; i++) {
|
||||
workers.push(fork('lib/check-signature.js'));
|
||||
}
|
||||
var nextWorker = 0;
|
||||
var initWorker = function (worker) {
|
||||
worker.on('message', function (res) {
|
||||
onChecked.fire(res);
|
||||
});
|
||||
// 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);
|
||||
|
||||
HK.onChannelMessage = function (Env, Server, channel, msgStruct) {
|
||||
console.log(+new Date(), "onChannelMessage");
|
||||
const Log = Env.Log;
|
||||
|
||||
// TODO our usage of 'channel' here looks prone to errors
|
||||
|
@ -962,20 +993,38 @@ HK.onChannelMessage = function (Env, Server, channel, msgStruct) {
|
|||
let signedMsg = (isCp) ? msgStruct[4].replace(CHECKPOINT_PATTERN, '') : msgStruct[4];
|
||||
// convert the message from a base64 string into a Uint8Array
|
||||
|
||||
// FIXME this can fail and the client won't notice
|
||||
signedMsg = Nacl.util.decodeBase64(signedMsg);
|
||||
const txid = Util.uid();
|
||||
const next = w();
|
||||
|
||||
// FIXME this can blow up
|
||||
// TODO check that that won't cause any problems other than not being able to append...
|
||||
const validateKey = Nacl.util.decodeBase64(metadata.validateKey);
|
||||
// validate the message
|
||||
const validated = Nacl.sign.open(signedMsg, validateKey);
|
||||
if (!validated) {
|
||||
// don't go any further if the message fails validation
|
||||
w.abort();
|
||||
Log.info("HK_SIGNED_MESSAGE_REJECTED", 'Channel '+channel.id);
|
||||
return;
|
||||
}
|
||||
// Listen for messages
|
||||
const onCheck = function (res) {
|
||||
if (!res) { return; }
|
||||
if (res.txid !== txid) { return; }
|
||||
// Wev'e received an answer, remove this handler
|
||||
console.log(+new Date(), "Received verification response");
|
||||
onChecked.unreg(onCheck);
|
||||
if (res.error) {
|
||||
// don't go any further if the message fails validation
|
||||
if (res.error === 'FAILED') {
|
||||
// Signature doesn't match the public key
|
||||
Log.info("HK_SIGNED_MESSAGE_REJECTED", 'Channel '+channel.id);
|
||||
}
|
||||
w.abort();
|
||||
return;
|
||||
}
|
||||
// Success, continue to next block
|
||||
next();
|
||||
};
|
||||
onChecked.reg(onCheck);
|
||||
|
||||
console.log(+new Date(), "Send verification request");
|
||||
// Send the request
|
||||
workers[nextWorker].send({
|
||||
txid: txid,
|
||||
msg: signedMsg,
|
||||
key: metadata.validateKey
|
||||
});
|
||||
if (++nextWorker > numCPUs) { nextWorker = 0; }
|
||||
}).nThen(function () {
|
||||
// do checkpoint stuff...
|
||||
|
||||
|
@ -1002,7 +1051,9 @@ HK.onChannelMessage = function (Env, Server, channel, msgStruct) {
|
|||
msgStruct.push(now());
|
||||
|
||||
// storeMessage
|
||||
console.log(+new Date(), "Storing message");
|
||||
storeMessage(Env, channel, JSON.stringify(msgStruct), isCp, getHash(msgStruct[4], Log));
|
||||
console.log(+new Date(), "Message stored");
|
||||
});
|
||||
};
|
||||
|
||||
|
|
Loading…
Reference in New Issue