Move parallel worker utilities to @backstage/cli-node

Moves `runParallelWorkers`, `runWorkerQueueThreads`, `runWorkerThreads`,
`parseParallelismOption`, and `getEnvironmentParallelism` from the CLI
internal lib to the shared `@backstage/cli-node` package.

This is part of the ongoing effort to make CLI modules independent of
each other and the shared lib code.

Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>

packages/cli-node/report.api.md

@@ -86,6 +86,9 @@ export interface BackstagePackageJson {
   version: string;
 }
 
+// @public
+export function getEnvironmentParallelism(): number;
+
 // @public
 export class GitUtils {
   static listChangedFiles(ref: string): Promise<string[]>;
@@ -200,4 +203,56 @@ export class PackageRoles {
   static getRoleFromPackage(pkgJson: unknown): PackageRole | undefined;
   static getRoleInfo(role: string): PackageRoleInfo;
 }
+
+// @public
+export type ParallelismOption = boolean | string | number | null | undefined;
+
+// @public
+export type ParallelWorkerOptions<TItem> = {
+  parallelismFactor?: number;
+  parallelismSetting?: ParallelismOption;
+  items: Iterable<TItem>;
+  worker: (item: TItem) => Promise<void>;
+};
+
+// @public
+export function parseParallelismOption(parallel: ParallelismOption): number;
+
+// @public
+export function runParallelWorkers<TItem>(
+  options: ParallelWorkerOptions<TItem>,
+): Promise<void[]>;
+
+// @public
+export function runWorkerQueueThreads<TItem, TResult, TData>(
+  options: WorkerQueueThreadsOptions<TItem, TResult, TData>,
+): Promise<TResult[]>;
+
+// @public
+export function runWorkerThreads<TResult, TData, TMessage>(
+  options: WorkerThreadsOptions<TResult, TData, TMessage>,
+): Promise<TResult[]>;
+
+// @public
+export type WorkerQueueThreadsOptions<TItem, TResult, TData> = {
+  items: Iterable<TItem>;
+  workerFactory: (
+    data: TData,
+  ) =>
+    | ((item: TItem) => Promise<TResult>)
+    | Promise<(item: TItem) => Promise<TResult>>;
+  workerData?: TData;
+  threadCount?: number;
+};
+
+// @public
+export type WorkerThreadsOptions<TResult, TData, TMessage> = {
+  worker: (
+    data: TData,
+    sendMessage: (message: TMessage) => void,
+  ) => Promise<TResult>;
+  workerData?: TData;
+  threadCount?: number;
+  onMessage?: (message: TMessage) => void;
+};
 ```

packages/cli-node/src/index.ts

@@ -22,4 +22,5 @@
 
 export * from './git';
 export * from './monorepo';
+export * from './parallel';
 export * from './roles';

packages/cli-node/src/parallel/index.ts

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2020 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+export type { ParallelismOption, ParallelWorkerOptions } from './parallel';
+export {
+  parseParallelismOption,
+  getEnvironmentParallelism,
+  runParallelWorkers,
+  runWorkerQueueThreads,
+  runWorkerThreads,
+} from './parallel';
+export type {
+  WorkerQueueThreadsOptions,
+  WorkerThreadsOptions,
+} from './parallel';

packages/cli-node/src/parallel/parallel.test.ts

@@ -0,0 +1,215 @@
+/*
+ * Copyright 2020 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import os from 'node:os';
+import {
+  parseParallelismOption,
+  getEnvironmentParallelism,
+  runParallelWorkers,
+  runWorkerQueueThreads,
+  runWorkerThreads,
+} from './parallel';
+
+const defaultParallelism = Math.ceil(os.cpus().length / 2);
+
+describe('parseParallelismOption', () => {
+  it('coerces false no parallelism', () => {
+    expect(parseParallelismOption(false)).toBe(1);
+    expect(parseParallelismOption('false')).toBe(1);
+  });
+
+  it('coerces true or undefined to default parallelism', () => {
+    expect(parseParallelismOption(true)).toBe(defaultParallelism);
+    expect(parseParallelismOption('true')).toBe(defaultParallelism);
+    expect(parseParallelismOption(undefined)).toBe(defaultParallelism);
+    expect(parseParallelismOption(null)).toBe(defaultParallelism);
+  });
+
+  it('coerces number string to number', () => {
+    expect(parseParallelismOption('2')).toBe(2);
+  });
+
+  it.each([['on'], [2.5], ['2.5']])('throws error for %p', value => {
+    expect(() => parseParallelismOption(value as any)).toThrow(
+      `Parallel option value '${value}' is not a boolean or integer`,
+    );
+  });
+});
+
+describe('getEnvironmentParallelism', () => {
+  it('reads the parallelism setting from the environment', () => {
+    process.env.BACKSTAGE_CLI_BUILD_PARALLEL = '2';
+    expect(getEnvironmentParallelism()).toBe(2);
+
+    process.env.BACKSTAGE_CLI_BUILD_PARALLEL = 'true';
+    expect(getEnvironmentParallelism()).toBe(defaultParallelism);
+
+    process.env.BACKSTAGE_CLI_BUILD_PARALLEL = 'false';
+    expect(getEnvironmentParallelism()).toBe(1);
+
+    delete process.env.BACKSTAGE_CLI_BUILD_PARALLEL;
+    expect(getEnvironmentParallelism()).toBe(defaultParallelism);
+  });
+});
+
+describe('runParallelWorkers', () => {
+  it('executes work in parallel', async () => {
+    const started = new Array<number>();
+    const done = new Array<number>();
+    const waiting = new Array<() => void>();
+
+    const work = runParallelWorkers({
+      items: [0, 1, 2, 3, 4],
+      parallelismSetting: 4,
+      parallelismFactor: 0.5, // 2 at a time
+      worker: async item => {
+        started.push(item);
+        await new Promise<void>(resolve => {
+          waiting[item] = resolve;
+        });
+        done.push(item);
+      },
+    });
+
+    await new Promise(resolve => setTimeout(resolve));
+    expect(started).toEqual([0, 1]);
+    expect(done).toEqual([]);
+    waiting[0]();
+
+    await new Promise(resolve => setTimeout(resolve));
+    expect(started).toEqual([0, 1, 2]);
+    expect(done).toEqual([0]);
+    waiting[1]();
+    waiting[2]();
+
+    await new Promise(resolve => setTimeout(resolve));
+    expect(started).toEqual([0, 1, 2, 3, 4]);
+    expect(done).toEqual([0, 1, 2]);
+    waiting[3]();
+    waiting[4]();
+
+    await work;
+    expect(done).toEqual([0, 1, 2, 3, 4]);
+  });
+
+  it('executes work sequentially', async () => {
+    const started = new Array<number>();
+    const done = new Array<number>();
+    const waiting = new Array<() => void>();
+
+    const work = runParallelWorkers({
+      items: [0, 1, 2, 3, 4],
+      parallelismFactor: 0, // 1 at a time
+      worker: async item => {
+        started.push(item);
+        await new Promise<void>(resolve => {
+          waiting[item] = resolve;
+        });
+        done.push(item);
+      },
+    });
+
+    await new Promise(resolve => setTimeout(resolve));
+    expect(started).toEqual([0]);
+    expect(done).toEqual([]);
+    waiting[0]();
+
+    await new Promise(resolve => setTimeout(resolve));
+    expect(started).toEqual([0, 1]);
+    expect(done).toEqual([0]);
+    waiting[1]();
+
+    await new Promise(resolve => setTimeout(resolve));
+    expect(started).toEqual([0, 1, 2]);
+    waiting[2]();
+
+    await new Promise(resolve => setTimeout(resolve));
+    expect(started).toEqual([0, 1, 2, 3]);
+    waiting[3]();
+
+    await new Promise(resolve => setTimeout(resolve));
+    expect(started).toEqual([0, 1, 2, 3, 4]);
+    waiting[4]();
+
+    await work;
+    expect(done).toEqual([0, 1, 2, 3, 4]);
+  });
+});
+
+describe('runWorkerQueueThreads', () => {
+  it('should execute work in parallel', async () => {
+    const sharedData = new SharedArrayBuffer(10);
+    const sharedView = new Uint8Array(sharedData);
+
+    const results = await runWorkerQueueThreads({
+      threadCount: 4,
+      workerData: sharedData,
+      items: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9],
+      workerFactory: data => {
+        const view = new Uint8Array(data);
+
+        return async (i: number) => {
+          view[i] = 10 + i;
+          return 20 + i;
+        };
+      },
+    });
+
+    expect(Array.from(sharedView)).toEqual([
+      10, 11, 12, 13, 14, 15, 16, 17, 18, 19,
+    ]);
+    expect(results).toEqual([20, 21, 22, 23, 24, 25, 26, 27, 28, 29]);
+  });
+});
+
+describe('runWorkerThreads', () => {
+  it('should run a single thread without items', async () => {
+    const [result] = await runWorkerThreads({
+      threadCount: 1,
+      workerData: 'foo',
+      worker: async data => `${data}bar`,
+    });
+
+    expect(result).toBe('foobar');
+  });
+
+  it('should run multiple threads without items', async () => {
+    const results = await runWorkerThreads({
+      threadCount: 4,
+      worker: async () => 'foo',
+    });
+
+    expect(results).toEqual(['foo', 'foo', 'foo', 'foo']);
+  });
+
+  it('should send messages', async () => {
+    const messages = new Array<string>();
+
+    await runWorkerThreads({
+      threadCount: 2,
+      worker: async (_data, sendMessage) => {
+        sendMessage('a');
+        await new Promise(resolve => setTimeout(resolve, 10));
+        sendMessage('b');
+        await new Promise(resolve => setTimeout(resolve, 10));
+        sendMessage('c');
+      },
+      onMessage: (message: string) => messages.push(message),
+    });
+
+    expect(messages.sort()).toEqual(['a', 'a', 'b', 'b', 'c', 'c']);
+  });
+});

packages/cli-node/src/parallel/parallel.ts

@@ -0,0 +1,384 @@
+/*
+ * Copyright 2020 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import os from 'node:os';
+import { ErrorLike } from '@backstage/errors';
+import { Worker } from 'node:worker_threads';
+
+const defaultParallelism = Math.ceil(os.cpus().length / 2);
+
+const PARALLEL_ENV_VAR = 'BACKSTAGE_CLI_BUILD_PARALLEL';
+
+/**
+ * Options for configuring parallelism. Can be a boolean, string, number, null, or undefined.
+ * - Boolean: true uses default parallelism (half of CPUs), false uses 1
+ * - Number: explicit worker count
+ * - String: parsed as boolean or integer (e.g. "true", "4")
+ *
+ * @public
+ */
+export type ParallelismOption = boolean | string | number | null | undefined;
+
+/**
+ * Parses a parallelism option value into a concrete worker count.
+ *
+ * @public
+ */
+export function parseParallelismOption(parallel: ParallelismOption): number {
+  if (parallel === undefined || parallel === null) {
+    return defaultParallelism;
+  } else if (typeof parallel === 'boolean') {
+    return parallel ? defaultParallelism : 1;
+  } else if (typeof parallel === 'number' && Number.isInteger(parallel)) {
+    if (parallel < 1) {
+      return 1;
+    }
+    return parallel;
+  } else if (typeof parallel === 'string') {
+    if (parallel === 'true') {
+      return parseParallelismOption(true);
+    } else if (parallel === 'false') {
+      return parseParallelismOption(false);
+    }
+    const parsed = Number(parallel);
+    if (Number.isInteger(parsed)) {
+      return parseParallelismOption(parsed);
+    }
+  }
+
+  throw Error(
+    `Parallel option value '${parallel}' is not a boolean or integer`,
+  );
+}
+
+/**
+ * Returns the parallelism value from the BACKSTAGE_CLI_BUILD_PARALLEL environment variable.
+ *
+ * @public
+ */
+export function getEnvironmentParallelism() {
+  return parseParallelismOption(process.env[PARALLEL_ENV_VAR]);
+}
+
+/**
+ * Options for runParallelWorkers.
+ *
+ * @public
+ */
+export type ParallelWorkerOptions<TItem> = {
+  /**
+   * Decides the number of parallel workers by multiplying
+   * this with the configured parallelism, which defaults to 4.
+   *
+   * Defaults to 1.
+   */
+  parallelismFactor?: number;
+  parallelismSetting?: ParallelismOption;
+  items: Iterable<TItem>;
+  worker: (item: TItem) => Promise<void>;
+};
+
+/**
+ * Runs items through a worker function in parallel across multiple async workers.
+ *
+ * @public
+ */
+export async function runParallelWorkers<TItem>(
+  options: ParallelWorkerOptions<TItem>,
+) {
+  const { parallelismFactor = 1, parallelismSetting, items, worker } = options;
+  const parallelism = parallelismSetting
+    ? parseParallelismOption(parallelismSetting)
+    : getEnvironmentParallelism();
+
+  const sharedIterator = items[Symbol.iterator]();
+  const sharedIterable = {
+    [Symbol.iterator]: () => sharedIterator,
+  };
+
+  const workerCount = Math.max(Math.floor(parallelismFactor * parallelism), 1);
+  return Promise.all(
+    Array(workerCount)
+      .fill(0)
+      .map(async () => {
+        for (const value of sharedIterable) {
+          await worker(value);
+        }
+      }),
+  );
+}
+
+type WorkerThreadMessage =
+  | {
+      type: 'done';
+    }
+  | {
+      type: 'item';
+      index: number;
+      item: unknown;
+    }
+  | {
+      type: 'start';
+    }
+  | {
+      type: 'result';
+      index: number;
+      result: unknown;
+    }
+  | {
+      type: 'error';
+      error: ErrorLike;
+    }
+  | {
+      type: 'message';
+      message: unknown;
+    };
+
+/**
+ * Options for runWorkerQueueThreads.
+ *
+ * @public
+ */
+export type WorkerQueueThreadsOptions<TItem, TResult, TData> = {
+  /** The items to process */
+  items: Iterable<TItem>;
+  /**
+   * A function that will be called within each worker thread at startup,
+   * which should return the worker function that will be called for each item.
+   *
+   * This function must be defined as an arrow function or using the
+   * function keyword, and must be entirely self contained, not referencing
+   * any variables outside of its scope. This is because the function source
+   * is stringified and evaluated in the worker thread.
+   *
+   * To pass data to the worker, use the `workerData` option and `items`, but
+   * note that they are both copied by value into the worker thread, except for
+   * types that are explicitly shareable across threads, such as `SharedArrayBuffer`.
+   */
+  workerFactory: (
+    data: TData,
+  ) =>
+    | ((item: TItem) => Promise<TResult>)
+    | Promise<(item: TItem) => Promise<TResult>>;
+  /** Data supplied to each worker factory */
+  workerData?: TData;
+  /** Number of threads, defaults to half of the number of available CPUs */
+  threadCount?: number;
+};
+
+/**
+ * Spawns one or more worker threads using the `worker_threads` module.
+ * Each thread processes one item at a time from the provided `options.items`.
+ *
+ * @public
+ */
+export async function runWorkerQueueThreads<TItem, TResult, TData>(
+  options: WorkerQueueThreadsOptions<TItem, TResult, TData>,
+): Promise<TResult[]> {
+  const items = Array.from(options.items);
+  const {
+    workerFactory,
+    workerData,
+    threadCount = Math.min(getEnvironmentParallelism(), items.length),
+  } = options;
+
+  const iterator = items[Symbol.iterator]();
+  const results = new Array<TResult>();
+  let itemIndex = 0;
+
+  await Promise.all(
+    Array(threadCount)
+      .fill(0)
+      .map(async () => {
+        const thread = new Worker(`(${workerQueueThread})(${workerFactory})`, {
+          eval: true,
+          workerData,
+        });
+
+        return new Promise<void>((resolve, reject) => {
+          thread.on('message', (message: WorkerThreadMessage) => {
+            if (message.type === 'start' || message.type === 'result') {
+              if (message.type === 'result') {
+                results[message.index] = message.result as TResult;
+              }
+              const { value, done } = iterator.next();
+              if (done) {
+                thread.postMessage({ type: 'done' });
+              } else {
+                thread.postMessage({
+                  type: 'item',
+                  index: itemIndex,
+                  item: value,
+                });
+                itemIndex += 1;
+              }
+            } else if (message.type === 'error') {
+              const error = new Error(message.error.message);
+              error.name = message.error.name;
+              error.stack = message.error.stack;
+              reject(error);
+            }
+          });
+
+          thread.on('error', reject);
+          thread.on('exit', (code: number) => {
+            if (code !== 0) {
+              reject(new Error(`Worker thread exited with code ${code}`));
+            } else {
+              resolve();
+            }
+          });
+        });
+      }),
+  );
+
+  return results;
+}
+
+/* istanbul ignore next */
+function workerQueueThread(
+  workerFuncFactory: (
+    data: unknown,
+  ) => Promise<(item: unknown) => Promise<unknown>>,
+) {
+  const { parentPort, workerData } = require('node:worker_threads');
+
+  Promise.resolve()
+    .then(() => workerFuncFactory(workerData))
+    .then(
+      workerFunc => {
+        parentPort.on('message', async (message: WorkerThreadMessage) => {
+          if (message.type === 'done') {
+            parentPort.close();
+            return;
+          }
+          if (message.type === 'item') {
+            try {
+              const result = await workerFunc(message.item);
+              parentPort.postMessage({
+                type: 'result',
+                index: message.index,
+                result,
+              });
+            } catch (error) {
+              parentPort.postMessage({ type: 'error', error });
+            }
+          }
+        });
+
+        parentPort.postMessage({ type: 'start' });
+      },
+      error => parentPort.postMessage({ type: 'error', error }),
+    );
+}
+
+/**
+ * Options for runWorkerThreads.
+ *
+ * @public
+ */
+export type WorkerThreadsOptions<TResult, TData, TMessage> = {
+  /**
+   * A function that is called by each worker thread to produce a result.
+   *
+   * This function must be defined as an arrow function or using the
+   * function keyword, and must be entirely self contained, not referencing
+   * any variables outside of its scope. This is because the function source
+   * is stringified and evaluated in the worker thread.
+   *
+   * To pass data to the worker, use the `workerData` option, but
+   * note that they are both copied by value into the worker thread, except for
+   * types that are explicitly shareable across threads, such as `SharedArrayBuffer`.
+   */
+  worker: (
+    data: TData,
+    sendMessage: (message: TMessage) => void,
+  ) => Promise<TResult>;
+  /** Data supplied to each worker */
+  workerData?: TData;
+  /** Number of threads, defaults to 1 */
+  threadCount?: number;
+  /** An optional handler for messages posted from the worker thread */
+  onMessage?: (message: TMessage) => void;
+};
+
+/**
+ * Spawns one or more worker threads using the `worker_threads` module.
+ *
+ * @public
+ */
+export async function runWorkerThreads<TResult, TData, TMessage>(
+  options: WorkerThreadsOptions<TResult, TData, TMessage>,
+): Promise<TResult[]> {
+  const { worker, workerData, threadCount = 1, onMessage } = options;
+
+  return Promise.all(
+    Array(threadCount)
+      .fill(0)
+      .map(async () => {
+        const thread = new Worker(`(${workerThread})(${worker})`, {
+          eval: true,
+          workerData,
+        });
+
+        return new Promise<TResult>((resolve, reject) => {
+          thread.on('message', (message: WorkerThreadMessage) => {
+            if (message.type === 'result') {
+              resolve(message.result as TResult);
+            } else if (message.type === 'error') {
+              reject(message.error);
+            } else if (message.type === 'message') {
+              onMessage?.(message.message as TMessage);
+            }
+          });
+
+          thread.on('error', reject);
+          thread.on('exit', (code: number) => {
+            reject(
+              new Error(`Unexpected worker thread exit with code ${code}`),
+            );
+          });
+        });
+      }),
+  );
+}
+
+/* istanbul ignore next */
+function workerThread(
+  workerFunc: (
+    data: unknown,
+    sendMessage: (message: unknown) => void,
+  ) => Promise<unknown>,
+) {
+  const { parentPort, workerData } = require('node:worker_threads');
+
+  const sendMessage = (message: unknown) => {
+    parentPort.postMessage({ type: 'message', message });
+  };
+
+  workerFunc(workerData, sendMessage).then(
+    result => {
+      parentPort.postMessage({
+        type: 'result',
+        index: 0,
+        result,
+      });
+    },
+    error => {
+      parentPort.postMessage({ type: 'error', error });
+    },
+  );
+}