From 471e374533cd6d39c9ab8c5c0b453f0f84c24e59 Mon Sep 17 00:00:00 2001 From: ansuz Date: Tue, 24 Mar 2020 17:43:15 -0400 Subject: [PATCH] compute metadata in the same child process that builds indexes --- lib/commands/metadata.js | 10 +---- lib/historyKeeper.js | 3 +- lib/hk-util.js | 58 ++++++++++++++++++++-------- lib/workers/compute-index.js | 74 +++++++++++++++++++++++------------- 4 files changed, 93 insertions(+), 52 deletions(-) diff --git a/lib/commands/metadata.js b/lib/commands/metadata.js index 802942fcb..1d758564b 100644 --- a/lib/commands/metadata.js +++ b/lib/commands/metadata.js @@ -18,15 +18,7 @@ Data.getMetadataRaw = function (Env, channel /* channelName */, _cb) { } Env.batchMetadata(channel, cb, function (done) { - var ref = {}; - var lineHandler = Meta.createLineHandler(ref, Env.Log.error); - return void Env.msgStore.readChannelMetadata(channel, lineHandler, function (err) { - if (err) { - // stream errors? - return void done(err); - } - done(void 0, ref.meta); - }); + Env.computeMetadata(channel, done); }); }; diff --git a/lib/historyKeeper.js b/lib/historyKeeper.js index 2f6043dfd..c8356b908 100644 --- a/lib/historyKeeper.js +++ b/lib/historyKeeper.js @@ -247,11 +247,10 @@ module.exports.create = function (config, cb) { channelExpirationMs: config.channelExpirationMs, verbose: config.verbose, openFileLimit: config.openFileLimit, - }, w(function (err, computeIndex) { + }, w(function (err) { if (err) { throw new Error(err); } - Env.computeIndex = computeIndex; })); }).nThen(function (w) { // create a task store diff --git a/lib/hk-util.js b/lib/hk-util.js index 9b0c4dd24..513fac04f 100644 --- a/lib/hk-util.js +++ b/lib/hk-util.js @@ -337,7 +337,7 @@ const storeMessage = function (Env, channel, msg, isCp, optionalMessageHash) { * -1 if you didn't find it */ -const getHistoryOffset = (Env, channelName, lastKnownHash, _cb) => { +const getHistoryOffset = (Env, channelName, lastKnownHash, _cb) => { // XXX child process const cb = Util.once(Util.mkAsync(_cb)); const store = Env.store; const Log = Env.Log; @@ -454,7 +454,7 @@ const getHistoryAsync = (Env, channelName, lastKnownHash, beforeHash, handler, c Used by: * GET_HISTORY_RANGE */ -const getOlderHistory = function (Env, channelName, oldestKnownHash, cb) { +const getOlderHistory = function (Env, channelName, oldestKnownHash, cb) { // XXX child process const store = Env.store; const Log = Env.Log; var messageBuffer = []; @@ -833,7 +833,14 @@ HK.initializeIndexWorkers = function (Env, config, _cb) { }); worker.on('message', function (res) { - if (!res || !res.txid) { return; } + if (!res) { return; } + if (!res.txid) { + // !report errors... + if (res.error) { + Env.Log.error(res.error, res.value); + } + return; + } //console.log(res); try { response.handle(res.txid, [res.error, res.value]); @@ -860,20 +867,18 @@ HK.initializeIndexWorkers = function (Env, config, _cb) { }; var workerIndex = 0; - var sendCommand = function (Env, channel, cb) { + var sendCommand = function (msg, _cb) { + var cb = Util.once(Util.mkAsync(_cb)); + workerIndex = (workerIndex + 1) % workers.length; if (workers.length === 0 || typeof(workers[workerIndex].send) !== 'function') { return void cb("NO_WORKERS"); } - Env.store.getWeakLock(channel, function (next) { - const txid = Util.uid(); - response.expect(txid, Util.both(next, cb), 45000); - workers[workerIndex].send({ - txid: txid, - args: channel, - }); - }); + const txid = Util.uid(); + msg.txid = txid; + response.expect(txid, cb, 45000); + workers[workerIndex].send(msg); }; nThen(function (w) { @@ -885,6 +890,30 @@ HK.initializeIndexWorkers = function (Env, config, _cb) { })); }); }).nThen(function () { + Env.computeIndex = function (Env, channel, cb) { + Env.store.getWeakLock(channel, function (next) { + sendCommand({ + channel: channel, + command: 'COMPUTE_INDEX', + }, function (err, index) { + next(); + cb(err, index); + }); + }); + }; + + Env.computeMetadata = function (channel, cb) { + Env.store.getWeakLock(channel, function (next) { + sendCommand({ + channel: channel, + command: 'COMPUTE_METADATA', + }, function (err, metadata) { + next(); + cb(err, metadata); + }); + }); + }; + //console.log("index workers ready"); cb(void 0, sendCommand); }); @@ -925,14 +954,13 @@ HK.initializeValidationWorkers = function (Env) { var nextWorker = 0; const send = function (msg, _cb) { + var cb = Util.once(Util.mkAsync(_cb)); // let's be paranoid about asynchrony and only calling back once.. nextWorker = (nextWorker + 1) % workers.length; if (workers.length === 0 || typeof(workers[nextWorker].send) !== 'function') { - console.error(workers); - throw new Error("INVALID_WORKERS"); + return void cb("INVALID_WORKERS"); } - var cb = Util.once(Util.mkAsync(_cb)); var txid = msg.txid = Util.uid(); // expect a response within 15s diff --git a/lib/workers/compute-index.js b/lib/workers/compute-index.js index f1da2628c..17a292059 100644 --- a/lib/workers/compute-index.js +++ b/lib/workers/compute-index.js @@ -5,6 +5,7 @@ const HK = require("../hk-util"); const Store = require("../storage/file"); const Util = require("../common-util"); const nThen = require("nthen"); +const Meta = require("../metadata"); const Env = {}; @@ -46,7 +47,13 @@ const tryParse = function (Env, str) { * including the initial metadata line, if it exists */ -const computeIndex = function (channelName, cb) { +const computeIndex = function (data, cb) { + if (!data || !data.channel) { + return void cb('E_NO_CHANNEL'); + } + + const channelName = data.channel; + const cpIndex = []; let messageBuf = []; let i = 0; @@ -125,45 +132,60 @@ const computeIndex = function (channelName, cb) { }); }; +const computeMetadata = function (data, cb, errorHandler) { + const ref = {}; + const lineHandler = Meta.createLineHandler(ref, errorHandler); + return void store.readChannelMetadata(data.channel, lineHandler, function (err) { + if (err) { + // stream errors? + return void cb(err); + } + cb(void 0, ref.meta); + }); +}; + +const COMMANDS = { + COMPUTE_INDEX: computeIndex, + COMPUTE_METADATA: computeMetadata, +}; + process.on('message', function (data) { if (!data || !data.txid) { return void process.send({ error:'E_INVAL' }); } - const txid = data.txid; + + const cb = function (err, value) { + if (err) { + return void process.send({ + txid: data.txid, + error: err, + }); + } + process.send({ + txid: data.txid, + value: value, + }); + }; if (!ready) { return void init(data.config, function (err) { - if (err) { - return void process.send({ - txid: txid, - error: err, - }); - } + if (err) { return void cb(err); } ready = true; - process.send({txid: txid,}); + cb(); }); } - const channel = data.args; - if (!channel) { - return void process.send({ - error: 'E_NO_CHANNEL', - }); + const command = COMMANDS[data.command]; + if (typeof(command) !== 'function') { + return void cb("E_BAD_COMMAND"); } - - // computeIndex - computeIndex(channel, function (err, index) { - if (err) { - return void process.send({ - txid: txid, - error: err, - }); - } - return void process.send({ - txid: txid, - value: index, + command(data, cb, function (label, info) { + // for streaming errors + process.send({ + error: label, + value: info, }); }); });