cryptpad/lib/workers/index.js

/* jshint esversion: 6 */
/* global process */
const Util = require("../common-util");
const nThen = require('nthen');
const OS = require("os");
const numCPUs = OS.cpus().length;
const { fork } = require('child_process');
const Workers = module.exports;
const PID = process.pid;

const CRYPTO_PATH = 'lib/workers/crypto-worker';
const DB_PATH = 'lib/workers/db-worker';

Workers.initializeValidationWorkers = function (Env) {
    if (typeof(Env.validateMessage) !== 'undefined') {
        return void console.error("validation workers are already initialized");
    }

    // Create our workers
    const workers = [];
    for (let i = 0; i < numCPUs; i++) {
        workers.push(fork(CRYPTO_PATH));
    }

    const response = Util.response(function (errLabel, info) {
        Env.Log.error('HK_VALIDATE_WORKER__' + errLabel, info);
    });

    var initWorker = function (worker) {
        worker.on('message', function (res) {
            if (!res || !res.txid) { return; }
            response.handle(res.txid, [res.error, res.value]);
        });

        var substituteWorker = Util.once( function () {
            Env.Log.info("SUBSTITUTE_VALIDATION_WORKER", '');
            var idx = workers.indexOf(worker);
            if (idx !== -1) {
                workers.splice(idx, 1);
            }
            // Spawn a new one
            var w = fork(CRYPTO_PATH);
            workers.push(w);
            initWorker(w);
        });

        // Spawn a new process in one ends
        worker.on('exit', substituteWorker);
        worker.on('close', substituteWorker);
        worker.on('error', function (err) {
            substituteWorker();
            Env.Log.error('VALIDATION_WORKER_ERROR', {
                error: err,
            });
        });
    };
    workers.forEach(initWorker);

    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') {
            return void cb("INVALID_WORKERS");
        }

        var txid = msg.txid = Util.uid();

        // expect a response within 45s
        response.expect(txid, cb, 60000);

        // Send the request
        workers[nextWorker].send(msg);
    };

    Env.validateMessage = function (signedMsg, key, cb) {
        send({
            msg: signedMsg,
            key: key,
            command: 'INLINE',
        }, cb);
    };

    Env.checkSignature = function (signedMsg, signature, publicKey, cb) {
        send({
            command: 'DETACHED',
            sig: signature,
            msg: signedMsg,
            key: publicKey,
        }, cb);
    };

    Env.hashChannelList = function (channels, cb) {
        send({
            command: 'HASH_CHANNEL_LIST',
            channels: channels,
        }, cb);
    };
};

Workers.initializeIndexWorkers = function (Env, config, _cb) {
    var cb = Util.once(Util.mkAsync(_cb));

    const workers = [];

    const response = Util.response(function (errLabel, info) {
        Env.Log.error('HK_DB_WORKER__' + errLabel, info);
    });

    const Log = Env.Log;
    const handleLog = function (level, label, info) {
        if (typeof(Log[level]) !== 'function') { return; }
        Log[level](label, info);
    };

    var isWorker = function (value) {
        return value && value.worker && typeof(value.worker.send) === 'function';
    };

    // pick ids that aren't already in use...
    const guid = function () {
        var id = Util.uid();
        return response.expected(id)? guid(): id;
    };

    var workerIndex = 0;
    var sendCommand = function (msg, _cb) {
        var cb = Util.once(Util.mkAsync(_cb));

        workerIndex = (workerIndex + 1) % workers.length;
        if (!isWorker(workers[workerIndex])) {
            return void cb("NO_WORKERS");
        }

        var state = workers[workerIndex];

        // XXX insert a queue here to prevent timeouts

        const txid = guid();
        msg.txid = txid;
        msg.pid = PID;

        // track which worker is doing which jobs
        state.tasks[txid] = msg;
        response.expect(txid, function (err, value) {
            // clean up when you get a response
            delete state[txid];
            cb(err, value);
        }, 60000);
        state.worker.send(msg);
    };

    const initWorker = function (worker, cb) {
        const txid = guid();

        const state = {
            worker: worker,
            tasks: {},
        };

        response.expect(txid, function (err) {
            if (err) { return void cb(err); }
            workers.push(state);
            cb(void 0, state);
        }, 15000);

        worker.send({
            pid: PID,
            txid: txid,
            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]);
        });

        var substituteWorker = Util.once(function () {
            Env.Log.info("SUBSTITUTE_DB_WORKER", '');
            var idx = workers.indexOf(state);
            if (idx !== -1) {
                workers.splice(idx, 1);
            }

            Object.keys(state.tasks).forEach(function (txid) {
                const cb = response.expectation(txid);
                if (typeof(cb) !== 'function') { return; }
                const task = state.tasks[txid];
                if (!task && task.msg) { return; }
                response.clear(txid);
                Log.info('DB_WORKER_RESEND', task.msg);
                sendCommand(task.msg, cb);
            });

            var w = fork(DB_PATH);
            initWorker(w, function (err, state) {
                if (err) {
                    throw new Error(err);
                }
                workers.push(state);
            });
        });

        worker.on('exit', substituteWorker);
        worker.on('close', substituteWorker);
        worker.on('error', function (err) {
            substituteWorker();
            Env.Log.error("DB_WORKER_ERROR", {
                error: err,
            });
        });
    };

    nThen(function (w) {
        OS.cpus().forEach(function () {
            initWorker(fork(DB_PATH), w(function (err) {
                if (!err) { return; }
                w.abort();
                return void cb(err);
            }));
        });
    }).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);
                });
            });
        };

        Env.getOlderHistory = function (channel, oldestKnownHash, cb) {
            Env.store.getWeakLock(channel, function (next) {
                sendCommand({
                    channel: channel,
                    command: "GET_OLDER_HISTORY",
                    hash: oldestKnownHash,
                }, Util.both(next, cb));
            });
        };

        Env.getPinState = function (safeKey, cb) {
            Env.pinStore.getWeakLock(safeKey, function (next) {
                sendCommand({
                    key: safeKey,
                    command: 'GET_PIN_STATE',
                }, Util.both(next, cb));
            });
        };

        Env.getFileSize = function (channel, cb) {
            sendCommand({
                command: 'GET_FILE_SIZE',
                 channel: channel,
            }, cb);
        };

        Env.getDeletedPads = function (channels, cb) {
            sendCommand({
                command: "GET_DELETED_PADS",
                channels: channels,
            }, cb);
        };

        Env.getTotalSize = function (channels, cb) {
            // we could take out locks for all of these channels,
            // but it's OK if the size is slightly off
            sendCommand({
                command: 'GET_TOTAL_SIZE',
                channels: channels,
            }, cb);
        };

        Env.getMultipleFileSize = function (channels, cb) {
            sendCommand({
                command: "GET_MULTIPLE_FILE_SIZE",
                channels: channels,
            }, cb);
        };

        Env.getHashOffset = function (channel, hash, cb) {
            Env.store.getWeakLock(channel, function (next) {
                sendCommand({
                    command: 'GET_HASH_OFFSET',
                    channel: channel,
                    hash: hash,
                }, Util.both(next, cb));
            });
        };

        Env.removeOwnedBlob = function (blobId, safeKey, cb) {
            sendCommand({
                command: 'REMOVE_OWNED_BLOB',
                blobId: blobId,
                safeKey: safeKey,
            }, cb);
        };

        Env.runTasks = function (cb) {
            sendCommand({
                command: 'RUN_TASKS',
            }, cb);
        };

        cb(void 0);
    });
};

Workers.initialize = function (Env, config, cb) {
    Workers.initializeValidationWorkers(Env);
    Workers.initializeIndexWorkers(Env, config, cb);
};