<file_sep># worker-union

Every Node.JS developer is aware of the single-threaded nature of this platform
where everything except your Javascript code runs in parallel..
But what if I tell you that it has become possible to run JS in parallel as well?

The worker_threads module has been added to node.js v10.5.0,
which allows javascript to be executed in separate threads within a single Node.JS process.
Yes, no spawn, exec, fork, cluster and other parodies on multithreading!

Worker-union allows you to take advantage of worker_threads,
while eliminating the need to manually manage messaging,
the number of threads, load balancing between them, and so on.

Just write logic, the rest will be done by promise-powered worker-union 🤓

Let’s jump into it?

## Requirements

Since module 'worker_threads' was added only in __Node.JS v10.5.0__,
you cannot use 'worker-union' in earlier versions.

Also, when you start your application, you will need to use `--experimental-worker` flag
(ex. `node --experimental-worker index.js`)

## Install

```bash
npm i -S worker-union
```

## Workerify

You can save yourself from the boilerplate at all, using the `workerify`!
Just pass on your function and get an asynchronous wrapper.
The code of the function itself will be executed in the Worker Pool.

```javascript
const workerify = require('worker-union/workerify');

const parseJSON = value => JSON.parse(value);

const asyncParseJSON = workerify(parseJSON);

asyncParseJSON('{"foo":"bar"}').then(result => result.foo); // -> bar
```

It is important to know that your function must be serializable.
You can easily verify this using `func.toString()` method. If you want to workerify the native Node.JS function,
for example _JSON.parse()_ (when you call `func.toString()` you see `[native code]`), then wrap it as in the example above.

By default, _worker-union_ will allocate for your function the number of threads
equal to the number of cores of your processor. If you want to tune the WorkerPool settings by yourself,
pass them as second argument to `workerify` function:

```javascript
const asyncFn = workerify(fn, {
  count: 4,
});
```

## Basic usage of WorkerPool

In this example, work with the worker on the __"request-response"__ principle will be demonstrated.

__Project structure:__

```
. your-project-directory
└── node_modules
└── index.js
└── worker.js
```

__index.js:__
```javascript
const path = require('path');
const WorkerPool = require('worker-union');

const pool = new WorkerPool({
  path: path.resolve('worker.js'), // absolute path to worker source code
  count: 1,
});

pool.start();

pool.send('ping').then(response => console.log(response));
```

__worker.js__
```javascript
const Worker = require('worker-union/worker');

new Worker((message) => {
  console.log('Received message:', message);

  return 'pong';
});
```

__Result:__
```bash
node --experimental-worker index.js
# -> pong
```

## Another WorkerPool example

And in this example it will be shown that the worker can send messages to the main thread __without requests__.

__Project structure:__

```
. your-project-directory
└── node_modules
└── index.js
└── worker.js
```

__index.js:__
```javascript
const path = require('path');
const WorkerPool = require('worker-union');

const pool = new WorkerPool({
  path: path.resolve('worker.js'),
  count: 1,
  eventHandler: message => console.log(`Recieved message from worker: ${message}`),
});

pool.start();

// we doesn't send any data here
```

__worker.js__
```javascript
const Worker = require('worker-union/worker');

const worker = new Worker();

setInterval(() => worker.emit('Hello!'), 1000); // send 'Hello' to main thread every second
```

__Result:__
```bash
node --experimental-worker index.js
# -> Received message from worker: Hello!
# -> Received message from worker: Hello!
# -> Received message from worker: Hello!
# -> Received message from worker: Hello!
```

## Advanced features

A description of all the features of the library, such as automatic thread allolocation, shared-memory state etc., will be ready soon. Stay tuned!

## License

MIT.
<file_sep>const workerThreads = require('worker_threads');

if (!workerThreads.isMainThread) {
  const {
    workerData: {
      bufferAccessMap,
      stateDataView,
      initialData,
    },
    parentPort,
    threadId: workerId,
  } = workerThreads;

  module.exports = class Worker {
    constructor(handler) {
      this.initialData = initialData;
      this.id = workerId;

      this.state = new Proxy({}, {
        get: (target, key) => Boolean(stateDataView.getUint8(bufferAccessMap.get(key))),
        set: (target, key, value) => stateDataView.setUint8(bufferAccessMap.get(key), value ? 1 : 0),
      });

      if (handler !== undefined) {
        this.setMessageHandler(handler);
      }
    }

    static getWorkerId() {
      return workerId;
    }

    emit(data) {
      parentPort.postMessage({
        isEvent: true,
        workerId,
        data,
      });
    }

    setMessageHandler(fn) {
      parentPort.on('message', async ({ messageId, data }) => {
        const resolve = response => parentPort.postMessage({
          failed: false,
          data: response,
          messageId,
          workerId,
        });

        const reject = error => parentPort.postMessage({
          failed: true,
          data: error,
          messageId,
          workerId,
        });

        try {
          const result = await fn(data);

          resolve(result);
        } catch ({ message, stack }) {
          // errors are not clonable objects, so we do this
          reject({ message, stack });
        }
      });
    }
  };
} else {
  throw new Error('This module should only be used in your worker file. You cannot use it in the main thread.');
}
<file_sep>const path = require('path');
const WorkerPool = require('../WorkerPool');

const workerPath = path.join(__dirname, 'worker.js');

/* eslint-disable no-param-reassign */
module.exports = (func, options = {}) => {
  delete options.initialData;
  delete options.path;

  const pool = new WorkerPool({
    initialData: func.toString(),
    path: workerPath,
    ...options,
  });

  pool.start();

  return (...args) => pool.send(args);
};
<file_sep>module.exports = require('./src/workerify');
<file_sep>const os = require('os');
const { Worker } = require('worker_threads');
const generateRandomString = require('./utils/generateRandomString');

const cpuCoresCount = os.cpus().length;

module.exports = class WorkerPool {
  constructor({
    path,
    state = {},
    autoAllocation = false,
    minCount = 2,
    maxCount = cpuCoresCount,
    count = cpuCoresCount,
    messagesLimit = 10,
    eventHandler = () => {},
    initialData,
  }) {
    this.workerFilePath = path;
    this.initialData = initialData;

    this.autoAllocation = autoAllocation;
    this.messagesLimit = messagesLimit;
    this.defaultWorkersCount = count;
    this.minWorkersCount = minCount;
    this.maxWorkersCount = autoAllocation ? maxCount : count;

    this.workerMap = new Map();
    this.messageMap = new Map(); // messageId -> { resolve, reject }

    this.eventHandler = eventHandler;

    this.initializeState(state);
  }

  start() {
    const workerLoopLimit = this.autoAllocation ? this.minWorkersCount : this.defaultWorkersCount;

    for (let i = 0; i < workerLoopLimit; i++) {
      this.spawnWorker();
    }

    if (this.autoAllocation) setInterval(this.manageWorkersCount.bind(this), 1000);
  }

  reload() {
    for (const [workerId] of this.workerMap) {
      this.killWorker(workerId);
    }

    this.start();
  }

  initializeState(state) {
    const stateBuffer = new SharedArrayBuffer(Object.keys(state).length * 4);

    this.stateDataView = new DataView(stateBuffer);

    this.bufferAccessMap = Object.keys(state).reduce((map, key, i) => map.set(key, i), new Map());

    this.state = Object.keys(state).reduce((acc, key) => Object.defineProperty(acc, key, {
      get: () => Boolean(this.stateDataView.getUint8(this.bufferAccessMap.get(key))),
      set: value => this.stateDataView.setUint8(this.bufferAccessMap.get(key), value ? 1 : 0),
    }), {});
  }

  get currentWorkersCount() {
    return this.workerMap.size;
  }

  get overageMessageAmountPerWorker() {
    let totalMessagesCount = 0;

    for (const { messagesCount } of this.workerMap.values()) {
      totalMessagesCount += messagesCount;
    }

    return totalMessagesCount / this.currentWorkersCount;
  }

  manageWorkersCount() {
    const workers = this.currentWorkersCount;
    const messages = this.overageMessageAmountPerWorker;
    const min = this.minWorkersCount;
    const max = this.maxWorkersCount;

    if (workers === min) {
      if (messages > this.messagesLimit) {
        this.spawnWorker();
      }
    } else if (workers > min && workers < max) {
      if (messages > this.messagesLimit) {
        this.spawnWorker();

        return;
      }

      this.killWorker();
    } else if (workers === max) {
      if (messages <= this.messagesLimit) {
        this.killWorker();
      }
    }
  }

  spawnWorker() {
    if (this.currentWorkersCount >= this.maxWorkersCount) {
      console.warn('Maximum number of workers reached');

      return;
    }

    const worker = new Worker(this.workerFilePath, {
      workerData: {
        bufferAccessMap: this.bufferAccessMap,
        stateDataView: this.stateDataView,
        initialData: this.initialData,
      },
    });

    worker.on('message', this.handleMessage.bind(this));

    this.workerMap.set(worker.threadId, {
      isTerminating: false,
      messagesCount: 0,
      worker,
    });
  }

  killWorker(workerId) {
    if (this.currentWorkersCount === this.minWorkersCount) {
      console.warn('Minimum number of workers reached');

      return;
    }

    if (workerId !== undefined) {
      this.workerMap.get(workerId).worker.terminate();
      this.workerMap.delete(workerId);

      return;
    }

    const targetWorkerId = this.getLeastLoadedWorkerId();

    this.workerMap.get(targetWorkerId).isTerminating = true;
  }

  handleMessage({
    isEvent = false,
    workerId,
    messageId,
    data,
    failed,
  }) {
    if (isEvent) {
      this.eventHandler(data);
      return;
    }

    const { resolve, reject } = this.messageMap.get(messageId);

    if (failed) {
      // errors are not clonable objects, so we do this
      const e = new Error();
      Object.assign(e, data);
      reject(e);
    } else {
      resolve(data);
    }

    const workerData = this.workerMap.get(workerId);

    workerData.messagesCount--;

    if (workerData.isTerminating && workerData.messagesCount === 0) {
      this.killWorker(workerId);
    }
  }

  getLeastLoadedWorkerId() {
    let leastCount = Infinity;
    let chosenWorkerId;

    for (const [workerId, { messagesCount, isTerminating }] of this.workerMap) {
      if (isTerminating && messagesCount === 0) {
        this.killWorker(workerId);

        continue;
      }

      if (leastCount > messagesCount) {
        leastCount = messagesCount;

        chosenWorkerId = workerId;
      }
    }

    return chosenWorkerId || this.workerMap.keys().next().value;
  }

  send(data, forcedWorkerId) {
    const promiseMethods = {};

    const promise = new Promise((resolve, reject) => {
      promiseMethods.resolve = resolve;
      promiseMethods.reject = reject;
    });

    const messageId = generateRandomString();
    const workerId = forcedWorkerId || this.getLeastLoadedWorkerId();

    this.messageMap.set(messageId, promiseMethods);
    const workerData = this.workerMap.get(workerId);

    workerData.worker.postMessage({ messageId, data });
    workerData.messagesCount++;

    return { workerId, promise };
  }

  async broadcast(data, sequentially = false) {
    let promises = [];

    for (const [workerId] of this.workerMap) {
      const { promise } = this.send(data, workerId);

      if (sequentially) await promise;

      promises.push(promise);
    }

    return Promise.all(promises);
  }
};
<file_sep>const Worker = require('../Worker');

const worker = new Worker();

// eslint-disable-next-line no-eval
const func = eval(worker.initialData);

worker.setMessageHandler(async (args, resolve, reject) => {
  try {
    resolve(await func(...args));
  } catch (e) {
    reject(e.message);
  }
});
