itns-sidechain/lib/workers/workerpool.js

/*!
 * workerpool.js - worker processes for bcoin
 * Copyright (c) 2014-2015, Fedor Indutny (MIT License)
 * Copyright (c) 2014-2017, Christopher Jeffrey (MIT License).
 * https://github.com/bcoin-org/bcoin
 */

/* eslint no-nested-ternary: "off" */

'use strict';

const assert = require('assert');
const EventEmitter = require('events');
const os = require('os');
const Network = require('../protocol/network');
const consensus = require('../protocol/consensus');
const Child = require('./child');
const jobs = require('./jobs');
const Parser = require('./parser');
const Framer = require('./framer');
const packets = require('./packets');

/**
 * Worker Pool
 * @alias module:workers.WorkerPool
 * @extends EventEmitter
 * @property {Number} size
 * @property {Number} timeout
 * @property {Map} children
 * @property {Number} uid
 */

class WorkerPool extends EventEmitter {
  /**
   * Create a worker pool.
   * @constructor
   * @param {Object} options
   * @param {Number} [options.size=num-cores] - Max pool size.
   * @param {Number} [options.timeout=120000] - Execution timeout.
   */

  constructor(options) {
    super();

    this.enabled = false;
    this.size = getCores();
    this.timeout = 120000;
    this.file = process.env.BCOIN_WORKER_FILE || 'worker.js';

    this.children = new Map();
    this.uid = 0;

    this.set(options);
  }

  /**
   * Set worker pool options.
   * @param {Object} options
   */

  set(options) {
    if (!options)
      return;

    if (options.enabled != null) {
      assert(typeof options.enabled === 'boolean');
      this.enabled = options.enabled;
    }

    if (options.size != null) {
      assert((options.size >>> 0) === options.size);
      assert(options.size > 0);
      this.size = options.size;
    }

    if (options.timeout != null) {
      assert(Number.isSafeInteger(options.timeout));
      assert(options.timeout >= -1);
      this.timeout = options.timeout;
    }

    if (options.file != null) {
      assert(typeof options.file === 'string');
      this.file = options.file;
    }
  }

  /**
   * Open worker pool.
   * @returns {Promise}
   */

  async open() {
    ;
  }

  /**
   * Close worker pool.
   * @returns {Promise}
   */

  async close() {
    this.destroy();
  }

  /**
   * Spawn a new worker.
   * @param {Number} id - Worker ID.
   * @returns {Worker}
   */

  spawn(id) {
    const child = new Worker(this.file);

    child.id = id;

    child.on('error', (err) => {
      this.emit('error', err, child);
    });

    child.on('exit', (code) => {
      this.emit('exit', code, child);

      if (this.children.get(id) === child)
        this.children.delete(id);
    });

    child.on('event', (items) => {
      this.emit('event', items, child);
      this.emit(...items);
    });

    child.on('log', (text) => {
      this.emit('log', text, child);
    });

    this.emit('spawn', child);

    return child;
  }

  /**
   * Allocate a new worker, will not go above `size` option
   * and will automatically load balance the workers.
   * @returns {Worker}
   */

  alloc() {
    const id = this.uid++ % this.size;

    if (!this.children.has(id))
      this.children.set(id, this.spawn(id));

    return this.children.get(id);
  }

  /**
   * Emit an event on the worker side (all workers).
   * @param {String} event
   * @param {...Object} arg
   * @returns {Boolean}
   */

  sendEvent() {
    let result = true;

    for (const child of this.children.values()) {
      if (!child.sendEvent.apply(child, arguments))
        result = false;
    }

    return result;
  }

  /**
   * Destroy all workers.
   */

  destroy() {
    for (const child of this.children.values())
      child.destroy();
  }

  /**
   * Call a method for a worker to execute.
   * @param {Packet} packet
   * @param {Number} timeout
   * @returns {Promise}
   */

  execute(packet, timeout) {
    if (!this.enabled || !Child.hasSupport()) {
      return new Promise((resolve, reject) => {
        setImmediate(() => {
          let result;
          try {
            result = jobs.handle(packet);
          } catch (e) {
            reject(e);
            return;
          }
          resolve(result);
        });
      });
    }

    if (!timeout)
      timeout = this.timeout;

    const child = this.alloc();

    return child.execute(packet, timeout);
  }

  /**
   * Execute the tx check job (default timeout).
   * @method
   * @param {TX} tx
   * @param {CoinView} view
   * @param {VerifyFlags} flags
   * @returns {Promise}
   */

  async check(tx, view, flags) {
    const packet = new packets.CheckPacket(tx, view, flags);
    const result = await this.execute(packet, -1);

    if (result.error)
      throw result.error;

    return null;
  }

  /**
   * Execute the tx signing job (default timeout).
   * @method
   * @param {MTX} tx
   * @param {KeyRing[]} ring
   * @param {SighashType} type
   * @returns {Promise}
   */

  async sign(tx, ring, type) {
    let rings = ring;

    if (!Array.isArray(rings))
      rings = [rings];

    const packet = new packets.SignPacket(tx, rings, type);
    const result = await this.execute(packet, -1);

    result.inject(tx);

    return result.total;
  }

  /**
   * Execute the tx input check job (default timeout).
   * @method
   * @param {TX} tx
   * @param {Number} index
   * @param {Coin|Output} coin
   * @param {VerifyFlags} flags
   * @returns {Promise}
   */

  async checkInput(tx, index, coin, flags) {
    const packet = new packets.CheckInputPacket(tx, index, coin, flags);
    const result = await this.execute(packet, -1);

    if (result.error)
      throw result.error;

    return null;
  }

  /**
   * Execute the tx input signing job (default timeout).
   * @method
   * @param {MTX} tx
   * @param {Number} index
   * @param {Coin|Output} coin
   * @param {KeyRing} ring
   * @param {SighashType} type
   * @returns {Promise}
   */

  async signInput(tx, index, coin, ring, type) {
    const packet = new packets.SignInputPacket(tx, index, coin, ring, type);
    const result = await this.execute(packet, -1);
    result.inject(tx);
    return result.value;
  }

  /**
   * Execute the secp256k1 verify job (no timeout).
   * @method
   * @param {Buffer} msg
   * @param {Buffer} sig - DER formatted.
   * @param {Buffer} key
   * @returns {Promise}
   */

  async ecVerify(msg, sig, key) {
    const packet = new packets.ECVerifyPacket(msg, sig, key);
    const result = await this.execute(packet, -1);
    return result.value;
  }

  /**
   * Execute the secp256k1 signing job (no timeout).
   * @method
   * @param {Buffer} msg
   * @param {Buffer} key
   * @returns {Promise}
   */

  async ecSign(msg, key) {
    const packet = new packets.ECSignPacket(msg, key);
    const result = await this.execute(packet, -1);
    return result.sig;
  }

  /**
   * Execute the mining job (no timeout).
   * @method
   * @param {Buffer} hdr
   * @param {Buffer} target
   * @param {Number} rounds
   * @param {Object} params
   * @returns {Promise} - Returns {Number}.
   */

  async mine(hdr, target, rounds, params) {
    const packet = new packets.MinePacket(hdr, target, rounds, params);
    const {nonce, sol} = await this.execute(packet, -1);
    nonce.copy(hdr, consensus.NONCE_POS);
    return [nonce, sol];
  }

  /**
   * Execute scrypt job (no timeout).
   * @method
   * @param {Buffer} passwd
   * @param {Buffer} salt
   * @param {Number} N
   * @param {Number} r
   * @param {Number} p
   * @param {Number} len
   * @returns {Promise}
   */

  async scrypt(passwd, salt, N, r, p, len) {
    const packet = new packets.ScryptPacket(passwd, salt, N, r, p, len);
    const result = await this.execute(packet, -1);
    return result.key;
  }
}

/**
 * Worker
 * @alias module:workers.Worker
 * @extends EventEmitter
 */

class Worker extends EventEmitter {
  /**
   * Create a worker.
   * @constructor
   * @param {String} file
   */

  constructor(file) {
    super();

    this.id = -1;
    this.framer = new Framer();
    this.parser = new Parser();
    this.pending = new Map();

    this.child = new Child(file);

    this.init();
  }

  /**
   * Initialize worker. Bind to events.
   * @private
   */

  init() {
    this.child.on('data', (data) => {
      this.parser.feed(data);
    });

    this.child.on('exit', (code, signal) => {
      this.emit('exit', code, signal);
    });

    this.child.on('error', (err) => {
      this.emit('error', err);
    });

    this.parser.on('error', (err) => {
      this.emit('error', err);
    });

    this.parser.on('packet', (packet) => {
      this.emit('packet', packet);
    });

    this.listen();
  }

  /**
   * Listen for packets.
   * @private
   */

  listen() {
    this.on('exit', (code, signal) => {
      this.killJobs();
    });

    this.on('error', (err) => {
      this.killJobs();
    });

    this.on('packet', (packet) => {
      try {
        this.handlePacket(packet);
      } catch (e) {
        this.emit('error', e);
      }
    });

    this.sendEnv({
      BCOIN_WORKER_NETWORK: Network.type,
      BCOIN_WORKER_ISTTY: process.stdout
        ? (process.stdout.isTTY ? '1' : '0')
        : '0'
    });
  }

  /**
   * Handle packet.
   * @private
   * @param {Packet} packet
   */

  handlePacket(packet) {
    switch (packet.cmd) {
      case packets.types.EVENT:
        this.emit('event', packet.items);
        this.emit(...packet.items);
        break;
      case packets.types.LOG:
        this.emit('log', packet.text);
        break;
      case packets.types.ERROR:
        this.emit('error', packet.error);
        break;
      case packets.types.ERRORRESULT:
        this.rejectJob(packet.id, packet.error);
        break;
      default:
        this.resolveJob(packet.id, packet);
        break;
    }
  }

  /**
   * Send data to worker.
   * @param {Buffer} data
   * @returns {Boolean}
   */

  write(data) {
    return this.child.write(data);
  }

  /**
   * Frame and send a packet.
   * @param {Packet} packet
   * @returns {Boolean}
   */

  send(packet) {
    return this.write(this.framer.packet(packet));
  }

  /**
   * Send environment.
   * @param {Object} env
   * @returns {Boolean}
   */

  sendEnv(env) {
    return this.send(new packets.EnvPacket(env));
  }

  /**
   * Emit an event on the worker side.
   * @param {String} event
   * @param {...Object} arg
   * @returns {Boolean}
   */

  sendEvent(...items) {
    return this.send(new packets.EventPacket(items));
  }

  /**
   * Destroy the worker.
   */

  destroy() {
    return this.child.destroy();
  }

  /**
   * Call a method for a worker to execute.
   * @param {Packet} packet
   * @param {Number} timeout
   * @returns {Promise}
   */

  execute(packet, timeout) {
    return new Promise((resolve, reject) => {
      this._execute(packet, timeout, resolve, reject);
    });
  }

  /**
   * Call a method for a worker to execute.
   * @private
   * @param {Packet} packet
   * @param {Number} timeout
   * @param {Function} resolve
   * @param {Function} reject
   * the worker method specifies.
   */

  _execute(packet, timeout, resolve, reject) {
    const job = new PendingJob(this, packet.id, resolve, reject);

    assert(!this.pending.has(packet.id), 'ID overflow.');

    this.pending.set(packet.id, job);

    job.start(timeout);

    this.send(packet);
  }

  /**
   * Resolve a job.
   * @param {Number} id
   * @param {Packet} result
   */

  resolveJob(id, result) {
    const job = this.pending.get(id);

    if (!job)
      throw new Error(`Job ${id} is not in progress.`);

    job.resolve(result);
  }

  /**
   * Reject a job.
   * @param {Number} id
   * @param {Error} err
   */

  rejectJob(id, err) {
    const job = this.pending.get(id);

    if (!job)
      throw new Error(`Job ${id} is not in progress.`);

    job.reject(err);
  }

  /**
   * Kill all jobs associated with worker.
   */

  killJobs() {
    for (const job of this.pending.values())
      job.destroy();
  }
}

/**
 * Pending Job
 * @ignore
 */

class PendingJob {
  /**
   * Create a pending job.
   * @constructor
   * @param {Worker} worker
   * @param {Number} id
   * @param {Function} resolve
   * @param {Function} reject
   */

  constructor(worker, id, resolve, reject) {
    this.worker = worker;
    this.id = id;
    this.job = { resolve, reject };
    this.timer = null;
  }

  /**
   * Start the timer.
   * @param {Number} timeout
   */

  start(timeout) {
    if (!timeout || timeout <= 0)
      return;

    this.timer = setTimeout(() => {
      this.reject(new Error('Worker timed out.'));
    }, timeout);
  }

  /**
   * Destroy the job with an error.
   */

  destroy() {
    this.reject(new Error('Job was destroyed.'));
  }

  /**
   * Cleanup job state.
   * @returns {Job}
   */

  cleanup() {
    const job = this.job;

    assert(job, 'Already finished.');

    this.job = null;

    if (this.timer != null) {
      clearTimeout(this.timer);
      this.timer = null;
    }

    assert(this.worker.pending.has(this.id));
    this.worker.pending.delete(this.id);

    return job;
  }

  /**
   * Complete job with result.
   * @param {Object} result
   */

  resolve(result) {
    const job = this.cleanup();
    job.resolve(result);
  }

  /**
   * Complete job with error.
   * @param {Error} err
   */

  reject(err) {
    const job = this.cleanup();
    job.reject(err);
  }
}

/*
 * Helpers
 */

function getCores() {
  return Math.max(2, os.cpus().length);
}

/*
 * Expose
 */

module.exports = WorkerPool;