@travetto/worker

Worker

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

Install: @travetto/worker

npm install @travetto/worker

# or

yarn add @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, WorkQueue is provided to support a wide range of use cases. WorkQueue 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/base';
import { Worker, WorkPool, WorkQueue } from '@travetto/worker';

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

  get id(): number | undefined {
    return this.proc.process.pid;
  }

  async destroy(): Promise<void> {
    this.proc.process.kill();
  }

  async execute(path: string): Promise<void> {
    this.active = true;
    try {
      this.proc = ExecUtil.spawn('convert images', [path]);
      await this.proc;
    } catch {
      // Do nothing
    }
    this.active = false;
  }
}

export class ImageCompressor {

  changes: AsyncIterable<unknown>;
  pendingImages = new WorkQueue<string>();

  begin(): void {
    this.changes ??= WorkPool.runStream(() => new ImageProcessor(), this.pendingImages);
  }

  convert(...images: string[]): void {
    this.pendingImages.addAll(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 { ExecutionState } from '@travetto/base';

import { ParentCommChannel } from './comm/parent';
import { Worker } from './pool';

type Simple<V> = (ch: ParentCommChannel<V>) => Promise<unknown | void>;
type Param<V, X> = (ch: ParentCommChannel<V>, input: X) => Promise<unknown | void>;

const empty = async (): Promise<void> => { };

/**
 * Spawned worker
 */
export class WorkUtil {
  /**
   * Create a process channel worker from a given spawn config
   */
  static spawnedWorker<V, X>(
    worker: () => ExecutionState,
    init: Simple<V>,
    execute: Param<V, X>,
    destroy: Simple<V> = empty): Worker<X> {
    const channel = new ParentCommChannel<V>(worker());
    return {
      get id(): number | undefined { return channel.id; },
      get active(): boolean { return channel.active; },
      init: () => init(channel),
      execute: inp => execute(channel, inp),
      async destroy(): Promise<void> {
        await 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 { ExecUtil } from '@travetto/base';
import { WorkPool, WorkUtil } from '@travetto/worker';

export async function main(): Promise<void> {
  await WorkPool.run(
    () => WorkUtil.spawnedWorker<{ data: number }, number>(
      () => ExecUtil.spawn('trv', ['main', '@travetto/worker/doc/spawned.ts']),
      ch => ch.once('ready'), // Wait for child to indicate it is ready
      async (channel, inp) => {
        const res = channel.once('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(`Did not get the double: inp=${inp}, data=${data}`);
        }
      }
    ), [1, 2, 3, 4, 5]);
}

Code: Spawned Worker

import timers from 'node:timers/promises';
import { ChildCommChannel } from '@travetto/worker';

export async function main(): Promise<void> {
  const exec = new ChildCommChannel<{ data: string }>();

  exec.on('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

  for await (const _ of timers.setInterval(5000)) {
    // Keep-alive
  }
}

Terminal: Output

$ trv main doc/spawner.ts

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 }