diff --git a/.changeset/cli-internal-refactor.md b/.changeset/cli-internal-refactor.md index 97957a2cd2..c3f8f38540 100644 --- a/.changeset/cli-internal-refactor.md +++ b/.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`. diff --git a/.changeset/cli-node-parallel-helpers.md b/.changeset/cli-node-parallel-helpers.md new file mode 100644 index 0000000000..5ad7ad7f4d --- /dev/null +++ b/.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. diff --git a/packages/cli-node/report.api.md b/packages/cli-node/report.api.md index 55d2443643..b3d8fe5641 100644 --- a/packages/cli-node/report.api.md +++ b/packages/cli-node/report.api.md @@ -86,6 +86,13 @@ export interface BackstagePackageJson { version: string; } +// @public +export type ConcurrentTasksOptions = { + concurrencyFactor?: number; + items: Iterable; + worker: (item: TItem) => Promise; +}; + // @public export class GitUtils { static listChangedFiles(ref: string): Promise; @@ -200,4 +207,27 @@ export class PackageRoles { static getRoleFromPackage(pkgJson: unknown): PackageRole | undefined; static getRoleInfo(role: string): PackageRoleInfo; } + +// @public +export function runConcurrentTasks( + options: ConcurrentTasksOptions, +): Promise; + +// @public +export function runWorkerQueueThreads( + options: WorkerQueueThreadsOptions, +): Promise<{ + results: TResult[]; +}>; + +// @public +export type WorkerQueueThreadsOptions = { + items: Iterable; + workerFactory: ( + context: TContext, + ) => + | ((item: TItem) => Promise) + | Promise<(item: TItem) => Promise>; + context?: TContext; +}; ``` diff --git a/packages/cli-node/src/concurrency/concurrency.ts b/packages/cli-node/src/concurrency/concurrency.ts new file mode 100644 index 0000000000..fa46f132a0 --- /dev/null +++ b/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; +} diff --git a/packages/cli-node/src/concurrency/index.ts b/packages/cli-node/src/concurrency/index.ts new file mode 100644 index 0000000000..3ff4067a20 --- /dev/null +++ b/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'; diff --git a/packages/cli-node/src/concurrency/runConcurrentTasks.test.ts b/packages/cli-node/src/concurrency/runConcurrentTasks.test.ts new file mode 100644 index 0000000000..382a85d297 --- /dev/null +++ b/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(); + const done = new Array(); + 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(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(); + const done = new Array(); + 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(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(); + 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(); + }); +}); diff --git a/packages/cli-node/src/concurrency/runConcurrentTasks.ts b/packages/cli-node/src/concurrency/runConcurrentTasks.ts new file mode 100644 index 0000000000..1695f25403 --- /dev/null +++ b/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 = { + /** + * Decides the number of concurrent workers by multiplying + * this with the configured concurrency. + * + * Defaults to 1. + */ + concurrencyFactor?: number; + items: Iterable; + worker: (item: TItem) => Promise; +}; + +/** + * Runs items through a worker function concurrently across multiple async workers. + * + * @public + */ +export async function runConcurrentTasks( + options: ConcurrentTasksOptions, +): Promise { + 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); + } + }), + ); +} diff --git a/packages/cli-node/src/concurrency/runWorkerQueueThreads.test.ts b/packages/cli-node/src/concurrency/runWorkerQueueThreads.test.ts new file mode 100644 index 0000000000..5dd2cda438 --- /dev/null +++ b/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]); + }); +}); diff --git a/packages/cli-node/src/concurrency/runWorkerQueueThreads.ts b/packages/cli-node/src/concurrency/runWorkerQueueThreads.ts new file mode 100644 index 0000000000..2d2a3b37e4 --- /dev/null +++ b/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 = { + /** The items to process */ + items: Iterable; + /** + * 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) + | Promise<(item: TItem) => Promise>; + /** 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( + options: WorkerQueueThreadsOptions, +): 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(); + let itemIndex = 0; + + await Promise.all( + Array(threadCount) + .fill(0) + .map(async () => { + const thread = new Worker(`(${workerQueueThread})(${workerFactory})`, { + eval: true, + workerData, + }); + + return new Promise((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>, +) { + 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 }), + ); +} diff --git a/packages/cli-node/src/index.ts b/packages/cli-node/src/index.ts index 1d3f3ec2ed..5540666a05 100644 --- a/packages/cli-node/src/index.ts +++ b/packages/cli-node/src/index.ts @@ -22,4 +22,5 @@ export * from './git'; export * from './monorepo'; +export * from './concurrency'; export * from './roles'; diff --git a/packages/cli/src/lib/parallel.test.ts b/packages/cli/src/lib/parallel.test.ts deleted file mode 100644 index 389e22102b..0000000000 --- a/packages/cli/src/lib/parallel.test.ts +++ /dev/null @@ -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(); - const done = new Array(); - 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(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(); - const done = new Array(); - 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(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(); - - 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']); - }); -}); diff --git a/packages/cli/src/lib/parallel.ts b/packages/cli/src/lib/parallel.ts deleted file mode 100644 index a4ca5dd13d..0000000000 --- a/packages/cli/src/lib/parallel.ts +++ /dev/null @@ -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 = { - /** - * 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; - worker: (item: TItem) => Promise; -}; - -export async function runParallelWorkers( - options: ParallelWorkerOptions, -) { - 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 = { - /** The items to process */ - items: Iterable; - /** - * 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) - | Promise<(item: TItem) => Promise>; - /** 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( - options: WorkerQueueThreadsOptions, -): Promise { - 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(); - let itemIndex = 0; - - await Promise.all( - Array(threadCount) - .fill(0) - .map(async () => { - const thread = new Worker(`(${workerQueueThread})(${workerFactory})`, { - eval: true, - workerData, - }); - - return new Promise((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>, -) { - 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 = { - /** - * 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; - /** 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( - options: WorkerThreadsOptions, -): Promise { - 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((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, -) { - 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 }); - }, - ); -} diff --git a/packages/cli/src/modules/build/commands/repo/build.ts b/packages/cli/src/modules/build/commands/repo/build.ts index e64fe9eff0..6c99ebb0af 100644 --- a/packages/cli/src/modules/build/commands/repo/build.ts +++ b/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 { 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 { }); 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) { diff --git a/packages/cli/src/modules/build/lib/buildBackend.ts b/packages/cli/src/modules/build/lib/buildBackend.ts index f377d068bd..bf2ebf16ba 100644 --- a/packages/cli/src/modules/build/lib/buildBackend.ts +++ b/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, }); diff --git a/packages/cli/src/modules/build/lib/buildFrontend.ts b/packages/cli/src/modules/build/lib/buildFrontend.ts index 7b0955a519..6a5785a5b9 100644 --- a/packages/cli/src/modules/build/lib/buildFrontend.ts +++ b/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( diff --git a/packages/cli/src/modules/build/lib/builder/packager.ts b/packages/cli/src/modules/build/lib/builder/packager.ts index 12017b694c..77b1e2b134 100644 --- a/packages/cli/src/modules/build/lib/builder/packager.ts +++ b/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(), }); diff --git a/packages/cli/src/modules/build/lib/bundler/types.ts b/packages/cli/src/modules/build/lib/bundler/types.ts index a9a4b0f51f..d8960bc03e 100644 --- a/packages/cli/src/modules/build/lib/bundler/types.ts +++ b/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; diff --git a/packages/cli/src/modules/build/lib/packager/createDistWorkspace.ts b/packages/cli/src/modules/build/lib/packager/createDistWorkspace.ts index 0f096a3127..296355e2dd 100644 --- a/packages/cli/src/modules/build/lib/packager/createDistWorkspace.ts +++ b/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`); diff --git a/packages/cli/src/modules/lint/commands/repo/lint.ts b/packages/cli/src/modules/lint/commands/repo/lint.ts index 458ccf1a63..98a100fe3e 100644 --- a/packages/cli/src/modules/lint/commands/repo/lint.ts +++ b/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 { }), ); - 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), diff --git a/packages/cli/src/modules/migrate/commands/versions/bump.ts b/packages/cli/src/modules/migrate/commands/versions/bump.ts index ab65f8c604..a0a73dd78e 100644 --- a/packages/cli/src/modules/migrate/commands/versions/bump.ts +++ b/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(); - 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(); - await runParallelWorkers({ - parallelismFactor: 4, + await runConcurrentTasks({ + concurrencyFactor: 4, items: versionBumps.entries(), async worker([name, deps]) { const pkgPath = resolvePath(deps[0].location, 'package.json');