@travetto/worker

Worker

Process management utilties, with a focus on inter-process communication

Install: @travetto/worker

npm install @travetto/worker

This module provides the necessary primitives for handling dependent workers. A worker can be an individual actor or could be a pool of workers. Node provides ipc (inter-process communication) functionality out of the box. This module builds upon that by providing enhanced event management, richer process management, as well as constructs for orchestrating a conversation between two processes.

Execution Pools

With respect to managing multiple executions, WorkPool is provided to allow for concurrent operation, and processing of jobs concurrently. To manage the flow of jobs, there are various InputSource implementation that allow for a wide range of use cases.

The only provided InputSource is the IterableInputSource which supports all Iterable and Iterator sources. Additionally, the module provides DynamicAsyncIterator which allows for manual control of iteration, which is useful for event driven work loads.

Below is a pool that will convert images on demand, while queuing as needed.

Code: Image processing queue, with a fixed batch/pool size

import { ExecUtil, ExecutionState } from '@travetto/boot';
import { Worker, WorkPool, IterableInputSource, DynamicAsyncIterator } from '@travetto/worker';

class ImageProcessor implements Worker<string> {
  active = false;
  proc: ExecutionState;

  get id() {
    return this.proc.process.pid;
  }

  async destroy() {
    this.proc.process.kill();
  }

  async execute(path: string) {
    this.active = true;
    try {
      this.proc = ExecUtil.spawn('convert images', [path]);
      await this.proc;
    } catch (e) {

    }
    this.active = false;
  }
}

export class ImageCompressor extends WorkPool<string, ImageProcessor> {

  pendingImages = new DynamicAsyncIterator<string>();

  constructor() {
    super(async () => new ImageProcessor());
  }

  begin() {
    this.process(new IterableInputSource(this.pendingImages));
  }

  convert(...images: string[]) {
    this.pendingImages.add(images);
  }
}

Once a pool is constructed, it can be shutdown by calling the .shutdown() method, and awaiting the result.

IPC Support

Within the comm package, there is support for two primary communication elements: ChildCommChannel and ParentCommChannel. Usually ParentCommChannel indicates it is the owner of the sub process. ChildCommChannel indicates that it has been created/spawned/forked by the parent and will communicate back to it's parent. This generally means that a ParentCommChannel can be destroyed (i.e. killing the subprocess) where a ChildCommChannel can only exit the process, but the channel cannot be destroyed.

IPC as a Worker

A common pattern is to want to model a sub process as a worker, to be a valid candidate in a WorkPool. The WorkUtil class provides a utility to facilitate this desire.

Code: Spawned Worker

import { ExecUtil, ExecutionOptions, } from '@travetto/boot';
import { ParentCommChannel } from './comm/parent';
import { Worker } from './pool';

/**
 * Spawned worker
 */
export class WorkUtil {
  /**
   * Create a process channel worker from a given spawn config
   */
  static spawnedWorker<X>(
    command: string,
    { args, opts, handlers }: {
      args?: string[];
      opts?: ExecutionOptions;
      handlers: {
        init?: (ch: ParentCommChannel) => Promise<any>;
        execute: (ch: ParentCommChannel, input: X) => Promise<any>;
        destroy?: (ch: ParentCommChannel) => Promise<any>;
      };
    }
  ): Worker<X> {
    const channel = new ParentCommChannel(
      ExecUtil.fork(command, args, {
        ...(opts ?? {})
      })
    );
    return {
      get id() { return channel.id; },
      get active() { return channel.active; },
      init: handlers.init ? handlers.init.bind(handlers, channel) : undefined,
      execute: handlers.execute.bind(handlers, channel),
      async destroy() {
        if (handlers.destroy) {
          await handlers.destroy(channel);
        }
        await channel.destroy();
      },
    };
  }
}

When creating your work, via process spawning, you will need to provide the script (and any other features you would like in SpawnConfig). Additionally you must, at a minimum, provide functionality to run whenever an input element is up for grabs in the input source. This method will be provided the communication channel (ParentCommChannel) and the input value. A simple example could look like:

Code: Spawning Pool

import { WorkPool, WorkUtil, IterableInputSource } from '@travetto/worker';
import { FsUtil } from '@travetto/boot';

const pool = new WorkPool(() =>
  WorkUtil.spawnedWorker<string>(FsUtil.resolveUnix(__dirname, 'spawned.js'), {
    handlers: {
      async init(channel) {
        return channel.listenOnce('ready'); // Wait for child to indicate it is ready
      },
      async execute(channel, inp) {
        const res = channel.listenOnce('response'); //  Register response listener
        channel.send('request', { data: inp }); // Send request

        const { data } = await res; // Get answer
        console.log('Request complete', { input: inp, output: data });

        if (!(inp + inp === data)) {
          // Ensure the answer is double the input
          throw new Error(`Didn't get the double`);
        }
      }
    }
  })
);

if (process.argv.pop() === 'top') {
  pool.process(new IterableInputSource([1, 2, 3, 4, 5])).then(x => pool.shutdown());
}

Code: Spawned Worker

require('@travetto/boot/register');
require('@travetto/base').PhaseManager.init().then(async () => {
  const { ChildCommChannel } = require('..');

  const exec = new ChildCommChannel();

  exec.listenFor('request', data => {
    exec.send('response', { data: (data.data + data.data) }); // When data is received, return double
  });

  exec.send('ready'); // Indicate the child is ready to receive requests

  const heartbeat = () => setTimeout(heartbeat, 5000); // Keep-alive
  heartbeat();
});

Terminal: Output

$ node -r @travetto/boot/register ./doc/spawner.ts top

Request complete { input: 1, output: 2 }
Request complete { input: 2, output: 4 }
Request complete { input: 3, output: 6 }
Request complete { input: 4, output: 8 }
Request complete { input: 5, output: 10 }