parcel-bundler/lib/WorkerFarm.js

'use strict';

function _asyncToGenerator(fn) { return function () { var gen = fn.apply(this, arguments); return new Promise(function (resolve, reject) { function step(key, arg) { try { var info = gen[key](arg); var value = info.value; } catch (error) { reject(error); return; } if (info.done) { resolve(value); } else { return Promise.resolve(value).then(function (value) { step("next", value); }, function (err) { step("throw", err); }); } } return step("next"); }); }; }

var _require = require('events');

const EventEmitter = _require.EventEmitter;

const os = require('os');
const Farm = require('worker-farm/lib/farm');
const promisify = require('./utils/promisify');
const logger = require('./Logger');

let shared = null;

class WorkerFarm extends Farm {
  constructor(options) {
    let opts = {
      maxConcurrentWorkers: getNumWorkers()
    };

    let workerPath = parseInt(process.versions.node, 10) < 8 ? require.resolve('../lib/worker') : require.resolve('../src/worker');

    super(opts, workerPath);

    this.localWorker = this.promisifyWorker(require('./worker'));
    this.remoteWorker = this.promisifyWorker(this.setup(['init', 'run']));

    this.started = false;
    this.warmWorkers = 0;
    this.init(options);
  }

  init(options) {
    this.localWorker.init(options);
    this.initRemoteWorkers(options);
  }

  promisifyWorker(worker) {
    let res = {};

    for (let key in worker) {
      res[key] = promisify(worker[key].bind(worker));
    }

    return res;
  }

  initRemoteWorkers(options) {
    var _this = this;

    return _asyncToGenerator(function* () {
      _this.started = false;
      _this.warmWorkers = 0;

      let promises = [];
      for (let i = 0; i < _this.options.maxConcurrentWorkers; i++) {
        promises.push(_this.remoteWorker.init(options));
      }

      yield Promise.all(promises);
      if (_this.options.maxConcurrentWorkers > 0) {
        _this.started = true;
      }
    })();
  }

  receive(data) {
    if (data.event) {
      this.emit(data.event, ...data.args);
    } else if (data.type === 'logger') {
      if (this.shouldUseRemoteWorkers()) {
        logger.handleMessage(data);
      }
    } else if (this.children[data.child]) {
      super.receive(data);
    }
  }

  shouldUseRemoteWorkers() {
    return this.started && this.warmWorkers >= this.activeChildren;
  }

  run(...args) {
    var _this2 = this;

    return _asyncToGenerator(function* () {
      // Child process workers are slow to start (~600ms).
      // While we're waiting, just run on the main thread.
      // This significantly speeds up startup time.
      if (_this2.shouldUseRemoteWorkers()) {
        return _this2.remoteWorker.run(...args, false);
      } else {
        // Workers have started, but are not warmed up yet.
        // Send the job to a remote worker in the background,
        // but use the result from the local worker - it will be faster.
        if (_this2.started) {
          _this2.remoteWorker.run(...args, true).then(function () {
            _this2.warmWorkers++;
          }, function () {
            // ignore error
          });
        }

        return _this2.localWorker.run(...args, false);
      }
    })();
  }

  end() {
    // Force kill all children
    this.ending = true;
    for (let child in this.children) {
      this.stopChild(child);
    }

    this.ending = false;
    shared = null;
  }

  static getShared(options) {
    if (!shared) {
      shared = new WorkerFarm(options);
    } else {
      shared.init(options);
    }

    return shared;
  }
}

for (let key in EventEmitter.prototype) {
  WorkerFarm.prototype[key] = EventEmitter.prototype[key];
}

function getNumWorkers() {
  if (process.env.PARCEL_WORKERS) {
    return parseInt(process.env.PARCEL_WORKERS, 10);
  }

  let cores;
  try {
    cores = require('physical-cpu-count');
  } catch (err) {
    cores = os.cpus().length;
  }
  return cores || 1;
}

module.exports = WorkerFarm;