Merge pull request #32956 from backstage/rugvip/cli-node-parallel-helpers

cli-node: move parallel worker utilities from cli

.changeset/cli-internal-refactor.md

@@ -2,4 +2,4 @@
 '@backstage/cli': patch
 ---
 
-Internal refactor to improve module independence.
+Internal refactor to use new concurrency utilities from `@backstage/cli-node`.

.changeset/cli-node-parallel-helpers.md

@@ -0,0 +1,5 @@
+---
+'@backstage/cli-node': patch
+---
+
+Added `runConcurrentTasks` and `runWorkerQueueThreads` utilities, moved from the `@backstage/cli` internal code.

packages/cli-node/report.api.md

@@ -86,6 +86,13 @@ export interface BackstagePackageJson {
   version: string;
 }
 
+// @public
+export type ConcurrentTasksOptions<TItem> = {
+  concurrencyFactor?: number;
+  items: Iterable<TItem>;
+  worker: (item: TItem) => Promise<void>;
+};
+
 // @public
 export class GitUtils {
   static listChangedFiles(ref: string): Promise<string[]>;
@@ -200,4 +207,27 @@ export class PackageRoles {
   static getRoleFromPackage(pkgJson: unknown): PackageRole | undefined;
   static getRoleInfo(role: string): PackageRoleInfo;
 }
+
+// @public
+export function runConcurrentTasks<TItem>(
+  options: ConcurrentTasksOptions<TItem>,
+): Promise<void>;
+
+// @public
+export function runWorkerQueueThreads<TItem, TResult, TContext>(
+  options: WorkerQueueThreadsOptions<TItem, TResult, TContext>,
+): Promise<{
+  results: TResult[];
+}>;
+
+// @public
+export type WorkerQueueThreadsOptions<TItem, TResult, TContext> = {
+  items: Iterable<TItem>;
+  workerFactory: (
+    context: TContext,
+  ) =>
+    | ((item: TItem) => Promise<TResult>)
+    | Promise<(item: TItem) => Promise<TResult>>;
+  context?: TContext;
+};
 ```

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

@@ -0,0 +1,70 @@
+/*
+ * 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';
+
+const defaultConcurrency = Math.max(Math.ceil(os.cpus().length / 2), 1);
+
+const CONCURRENCY_ENV_VAR = 'BACKSTAGE_CLI_CONCURRENCY';
+const DEPRECATED_CONCURRENCY_ENV_VAR = 'BACKSTAGE_CLI_BUILD_PARALLEL';
+
+type ConcurrencyOption = boolean | string | number | null | undefined;
+
+function parseConcurrencyOption(value: ConcurrencyOption): number {
+  if (value === undefined || value === null) {
+    return defaultConcurrency;
+  } else if (typeof value === 'boolean') {
+    return value ? defaultConcurrency : 1;
+  } else if (typeof value === 'number' && Number.isInteger(value)) {
+    if (value < 1) {
+      return 1;
+    }
+    return value;
+  } else if (typeof value === 'string') {
+    if (value === 'true') {
+      return parseConcurrencyOption(true);
+    } else if (value === 'false') {
+      return parseConcurrencyOption(false);
+    }
+    const parsed = Number(value);
+    if (Number.isInteger(parsed)) {
+      return parseConcurrencyOption(parsed);
+    }
+  }
+
+  throw Error(
+    `Concurrency option value '${value}' is not a boolean or integer`,
+  );
+}
+
+let hasWarnedDeprecation = false;
+
+/** @internal */
+export function getEnvironmentConcurrency() {
+  if (process.env[CONCURRENCY_ENV_VAR] !== undefined) {
+    return parseConcurrencyOption(process.env[CONCURRENCY_ENV_VAR]);
+  }
+  if (process.env[DEPRECATED_CONCURRENCY_ENV_VAR] !== undefined) {
+    if (!hasWarnedDeprecation) {
+      hasWarnedDeprecation = true;
+      console.warn(
+        `The ${DEPRECATED_CONCURRENCY_ENV_VAR} environment variable is deprecated, use ${CONCURRENCY_ENV_VAR} instead`,
+      );
+    }
+    return parseConcurrencyOption(process.env[DEPRECATED_CONCURRENCY_ENV_VAR]);
+  }
+  return defaultConcurrency;
+}

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

@@ -0,0 +1,20 @@
+/*
+ * 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 { ConcurrentTasksOptions } from './runConcurrentTasks';
+export { runConcurrentTasks } from './runConcurrentTasks';
+export type { WorkerQueueThreadsOptions } from './runWorkerQueueThreads';
+export { runWorkerQueueThreads } from './runWorkerQueueThreads';

packages/cli-node/src/concurrency/runConcurrentTasks.test.ts

@@ -0,0 +1,144 @@
+/*
+ * 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 { runConcurrentTasks } from './runConcurrentTasks';
+
+describe('runConcurrentTasks', () => {
+  afterEach(() => {
+    delete process.env.BACKSTAGE_CLI_CONCURRENCY;
+  });
+
+  it('executes work in parallel', async () => {
+    const started = new Array<number>();
+    const done = new Array<number>();
+    const waiting = new Array<() => void>();
+
+    process.env.BACKSTAGE_CLI_CONCURRENCY = '4';
+    const work = runConcurrentTasks({
+      items: [0, 1, 2, 3, 4],
+      concurrencyFactor: 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 = runConcurrentTasks({
+      items: [0, 1, 2, 3, 4],
+      concurrencyFactor: 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]);
+  });
+
+  it('returns void', async () => {
+    const result = await runConcurrentTasks({
+      items: [1, 2, 3],
+      worker: async () => {},
+    });
+    expect(result).toBeUndefined();
+  });
+
+  it('defaults to environment concurrency', async () => {
+    const started = new Array<number>();
+    process.env.BACKSTAGE_CLI_CONCURRENCY = '2';
+
+    await runConcurrentTasks({
+      items: [0, 1],
+      worker: async item => {
+        started.push(item);
+      },
+    });
+
+    expect(started).toEqual([0, 1]);
+  });
+
+  it('supports the deprecated BACKSTAGE_CLI_BUILD_PARALLEL with a warning', async () => {
+    const warnSpy = jest.spyOn(console, 'warn').mockImplementation();
+
+    process.env.BACKSTAGE_CLI_BUILD_PARALLEL = '2';
+    await runConcurrentTasks({
+      items: [0, 1],
+      worker: async () => {},
+    });
+
+    expect(warnSpy).toHaveBeenCalledWith(
+      'The BACKSTAGE_CLI_BUILD_PARALLEL environment variable is deprecated, use BACKSTAGE_CLI_CONCURRENCY instead',
+    );
+
+    delete process.env.BACKSTAGE_CLI_BUILD_PARALLEL;
+    warnSpy.mockRestore();
+  });
+});

packages/cli-node/src/concurrency/runConcurrentTasks.ts

@@ -0,0 +1,62 @@
+/*
+ * 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 { getEnvironmentConcurrency } from './concurrency';
+
+/**
+ * Options for {@link runConcurrentTasks}.
+ *
+ * @public
+ */
+export type ConcurrentTasksOptions<TItem> = {
+  /**
+   * Decides the number of concurrent workers by multiplying
+   * this with the configured concurrency.
+   *
+   * Defaults to 1.
+   */
+  concurrencyFactor?: number;
+  items: Iterable<TItem>;
+  worker: (item: TItem) => Promise<void>;
+};
+
+/**
+ * Runs items through a worker function concurrently across multiple async workers.
+ *
+ * @public
+ */
+export async function runConcurrentTasks<TItem>(
+  options: ConcurrentTasksOptions<TItem>,
+): Promise<void> {
+  const { concurrencyFactor = 1, items, worker } = options;
+  const concurrency = getEnvironmentConcurrency();
+
+  const sharedIterator = items[Symbol.iterator]();
+  const sharedIterable = {
+    [Symbol.iterator]: () => sharedIterator,
+  };
+
+  const workerCount = Math.max(Math.floor(concurrencyFactor * concurrency), 1);
+  await Promise.all(
+    Array(workerCount)
+      .fill(0)
+      .map(async () => {
+        for (const value of sharedIterable) {
+          await worker(value);
+        }
+      }),
+  );
+}

packages/cli-node/src/concurrency/runWorkerQueueThreads.test.ts

@@ -0,0 +1,42 @@
+/*
+ * 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 { runWorkerQueueThreads } from './runWorkerQueueThreads';
+
+describe('runWorkerQueueThreads', () => {
+  it('should execute work in parallel', async () => {
+    const sharedData = new SharedArrayBuffer(10);
+    const sharedView = new Uint8Array(sharedData);
+
+    const { results } = await runWorkerQueueThreads({
+      context: sharedData,
+      items: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9],
+      workerFactory: (data: SharedArrayBuffer) => {
+        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]);
+  });
+});

packages/cli-node/src/concurrency/runWorkerQueueThreads.ts

@@ -0,0 +1,175 @@
+/*
+ * 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 { ErrorLike } from '@backstage/errors';
+import { Worker } from 'node:worker_threads';
+import { getEnvironmentConcurrency } from './concurrency';
+
+type WorkerThreadMessage =
+  | {
+      type: 'done';
+    }
+  | {
+      type: 'item';
+      index: number;
+      item: unknown;
+    }
+  | {
+      type: 'start';
+    }
+  | {
+      type: 'result';
+      index: number;
+      result: unknown;
+    }
+  | {
+      type: 'error';
+      error: ErrorLike;
+    };
+
+/**
+ * Options for {@link runWorkerQueueThreads}.
+ *
+ * @public
+ */
+export type WorkerQueueThreadsOptions<TItem, TResult, TContext> = {
+  /** 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 `context` 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: (
+    context: TContext,
+  ) =>
+    | ((item: TItem) => Promise<TResult>)
+    | Promise<(item: TItem) => Promise<TResult>>;
+  /** Context data supplied to each worker factory */
+  context?: TContext;
+};
+
+/**
+ * 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, TContext>(
+  options: WorkerQueueThreadsOptions<TItem, TResult, TContext>,
+): Promise<{ results: TResult[] }> {
+  const items = Array.from(options.items);
+  const workerFactory = options.workerFactory;
+  const workerData = options.context;
+  const threadCount = Math.min(getEnvironmentConcurrency(), items.length);
+
+  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 }),
+    );
+}

packages/cli-node/src/index.ts

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

packages/cli/src/lib/parallel.test.ts

@@ -1,215 +0,0 @@
-/*
- * 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/src/lib/parallel.ts

@@ -1,342 +0,0 @@
-/*
- * 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';
-
-export type ParallelismOption = boolean | string | number | null | undefined;
-
-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`,
-  );
-}
-
-export function getEnvironmentParallelism() {
-  return parseParallelismOption(process.env[PARALLEL_ENV_VAR]);
-}
-
-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>;
-};
-
-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;
-    };
-
-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`.
- */
-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 }),
-    );
-}
-
-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.
- */
-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 });
-    },
-  );
-}

packages/cli/src/modules/build/commands/repo/build.ts

@@ -23,8 +23,8 @@ import {
   BackstagePackage,
   PackageGraph,
   PackageRoles,
+  runConcurrentTasks,
 } from '@backstage/cli-node';
-import { runParallelWorkers } from '../../../../lib/parallel';
 import { buildFrontend } from '../../lib/buildFrontend';
 import { buildBackend } from '../../lib/buildBackend';
 import { createScriptOptionsParser } from '../../../../lib/optionsParser';
@@ -100,9 +100,9 @@ export async function command(opts: OptionValues, cmd: Command): Promise<void> {
 
   if (opts.all) {
     console.log('Building apps');
-    await runParallelWorkers({
+    await runConcurrentTasks({
       items: apps,
-      parallelismFactor: 1 / 2,
+      concurrencyFactor: 1 / 2,
       worker: async pkg => {
         const buildOptions = parseBuildScript(pkg.packageJson.scripts?.build);
         if (!buildOptions) {
@@ -121,9 +121,9 @@ export async function command(opts: OptionValues, cmd: Command): Promise<void> {
     });
 
     console.log('Building backends');
-    await runParallelWorkers({
+    await runConcurrentTasks({
       items: backends,
-      parallelismFactor: 1 / 2,
+      concurrencyFactor: 1 / 2,
       worker: async pkg => {
         const buildOptions = parseBuildScript(pkg.packageJson.scripts?.build);
         if (!buildOptions) {

packages/cli/src/modules/build/lib/buildBackend.ts

@@ -19,7 +19,6 @@ import fs from 'fs-extra';
 import { resolve as resolvePath } from 'node:path';
 import * as tar from 'tar';
 import { createDistWorkspace } from './packager';
-import { getEnvironmentParallelism } from '../../../lib/parallel';
 import { buildPackage, Output } from './builder';
 import { PackageGraph } from '@backstage/cli-node';
 
@@ -53,7 +52,6 @@ export async function buildBackend(options: BuildBackendOptions) {
       configPaths,
       buildDependencies: !skipBuildDependencies,
       buildExcludes: [pkg.name],
-      parallelism: getEnvironmentParallelism(),
       skeleton: SKELETON_FILE,
       minify,
     });

packages/cli/src/modules/build/lib/buildFrontend.ts

@@ -17,9 +17,8 @@
 import fs from 'fs-extra';
 import { resolve as resolvePath } from 'node:path';
 import { buildBundle, getModuleFederationRemoteOptions } from './bundler';
-import { getEnvironmentParallelism } from '../../../lib/parallel';
-import { loadCliConfig } from '../../config/lib/config';
 import { BackstagePackageJson } from '@backstage/cli-node';
+import { loadCliConfig } from '../../config/lib/config';
 
 interface BuildAppOptions {
   targetDir: string;
@@ -37,7 +36,6 @@ export async function buildFrontend(options: BuildAppOptions) {
   await buildBundle({
     targetDir,
     entry: 'src/index',
-    parallelism: getEnvironmentParallelism(),
     statsJsonEnabled: writeStats,
     moduleFederationRemote: options.isModuleFederationRemote
       ? await getModuleFederationRemoteOptions(

packages/cli/src/modules/build/lib/builder/packager.ts

@@ -21,8 +21,7 @@ import { relative as relativePath, resolve as resolvePath } from 'node:path';
 import { paths } from '../../../../lib/paths';
 import { makeRollupConfigs } from './config';
 import { BuildOptions, Output } from './types';
-import { PackageRoles } from '@backstage/cli-node';
-import { runParallelWorkers } from '../../../../lib/parallel';
+import { PackageRoles, runConcurrentTasks } from '@backstage/cli-node';
 
 export function formatErrorMessage(error: any) {
   let msg = '';
@@ -127,7 +126,7 @@ export const buildPackages = async (options: BuildOptions[]) => {
 
   const buildTasks = rollupConfigs.flat().map(opts => () => rollupBuild(opts));
 
-  await runParallelWorkers({
+  await runConcurrentTasks({
     items: buildTasks,
     worker: async task => task(),
   });

packages/cli/src/modules/build/lib/bundler/types.ts

@@ -36,7 +36,6 @@ export type BundlingOptions = {
   isDev: boolean;
   frontendConfig: Config;
   getFrontendAppConfigs(): AppConfig[];
-  parallelism?: number;
   additionalEntryPoints?: string[];
   // Path to append to the detected public path, e.g. '/public'
   publicSubPath?: string;
@@ -63,7 +62,6 @@ export type BuildOptions = BundlingPathsOptions & {
   // Target directory, defaulting to paths.targetDir
   targetDir?: string;
   statsJsonEnabled: boolean;
-  parallelism?: number;
   schema?: ConfigSchema;
   frontendConfig: Config;
   frontendAppConfigs: AppConfig[];
@@ -75,7 +73,6 @@ export type BuildOptions = BundlingPathsOptions & {
 export type BackendBundlingOptions = {
   checksEnabled: boolean;
   isDev: boolean;
-  parallelism?: number;
   inspectEnabled: boolean;
   inspectBrkEnabled: boolean;
   require?: string;

packages/cli/src/modules/build/lib/packager/createDistWorkspace.ts

@@ -41,8 +41,8 @@ import {
   PackageRoles,
   PackageGraph,
   PackageGraphNode,
+  runConcurrentTasks,
 } from '@backstage/cli-node';
-import { runParallelWorkers } from '../../../../lib/parallel';
 import { createTypeDistProject } from '../../../../lib/typeDistProject';
 
 // These packages aren't safe to pack in parallel since the CLI depends on them
@@ -86,11 +86,6 @@ type Options = {
    */
   buildExcludes?: string[];
 
-  /**
-   * Controls amount of parallelism in some build steps.
-   */
-  parallelism?: number;
-
   /**
    * If set, creates a skeleton tarball that contains all package.json files
    * with the same structure as the workspace dir.
@@ -225,7 +220,7 @@ export async function createDistWorkspace(
     await buildPackages(standardBuilds);
 
     if (customBuild.length > 0) {
-      await runParallelWorkers({
+      await runConcurrentTasks({
         items: customBuild,
         worker: async ({ name, dir, args }) => {
           await run(['yarn', 'run', 'build', ...(args || [])], {
@@ -368,7 +363,7 @@ async function moveToDistWorkspace(
   }
 
   // Repacking in parallel is much faster and safe for all packages outside of the Backstage repo
-  await runParallelWorkers({
+  await runConcurrentTasks({
     items: safePackages.map((target, index) => ({ target, index })),
     worker: async ({ target, index }) => {
       await pack(target, `temp-package-${index}.tgz`);

packages/cli/src/modules/lint/commands/repo/lint.ts

@@ -23,9 +23,9 @@ import {
   PackageGraph,
   BackstagePackageJson,
   Lockfile,
+  runWorkerQueueThreads,
 } from '@backstage/cli-node';
 import { paths } from '../../../../lib/paths';
-import { runWorkerQueueThreads } from '../../../../lib/parallel';
 import { createScriptOptionsParser } from '../../../../lib/optionsParser';
 import { SuccessCache } from '../../../../lib/cache/SuccessCache';
 
@@ -105,9 +105,9 @@ export async function command(opts: OptionValues, cmd: Command): Promise<void> {
     }),
   );
 
-  const resultsList = await runWorkerQueueThreads({
+  const { results: resultsList } = await runWorkerQueueThreads({
     items: items.filter(item => item.lintOptions), // Filter out packages without lint script
-    workerData: {
+    context: {
       fix: Boolean(opts.fix),
       format: opts.format as string | undefined,
       shouldCache: Boolean(cacheContext),

packages/cli/src/modules/migrate/commands/versions/bump.ts

@@ -33,7 +33,7 @@ import {
   mapDependencies,
   YarnInfoInspectData,
 } from '../../../../lib/versioning';
-import { runParallelWorkers } from '../../../../lib/parallel';
+import { runConcurrentTasks } from '@backstage/cli-node';
 import {
   getManifestByReleaseLine,
   getManifestByVersion,
@@ -145,8 +145,8 @@ export default async (opts: OptionValues) => {
   // Next check with the package registry to see which dependency ranges we need to bump
   const versionBumps = new Map<string, PkgVersionInfo[]>();
 
-  await runParallelWorkers({
-    parallelismFactor: 4,
+  await runConcurrentTasks({
+    concurrencyFactor: 4,
     items: dependencyMap.entries(),
     async worker([name, pkgs]) {
       let target: string;
@@ -182,8 +182,8 @@ export default async (opts: OptionValues) => {
     console.log();
 
     const breakingUpdates = new Map<string, { from: string; to: string }>();
-    await runParallelWorkers({
-      parallelismFactor: 4,
+    await runConcurrentTasks({
+      concurrencyFactor: 4,
       items: versionBumps.entries(),
       async worker([name, deps]) {
         const pkgPath = resolvePath(deps[0].location, 'package.json');