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/parallel/index.ts b/packages/cli-node/src/concurrency/index.ts similarity index 69% rename from packages/cli-node/src/parallel/index.ts rename to packages/cli-node/src/concurrency/index.ts index f8f9d4d51f..3ff4067a20 100644 --- a/packages/cli-node/src/parallel/index.ts +++ b/packages/cli-node/src/concurrency/index.ts @@ -14,8 +14,7 @@ * limitations under the License. */ -export type { - ConcurrentTasksOptions, - WorkerQueueThreadsOptions, -} from './parallel'; -export { runConcurrentTasks, runWorkerQueueThreads } from './parallel'; +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/parallel/parallel.test.ts b/packages/cli-node/src/concurrency/runConcurrentTasks.test.ts similarity index 83% rename from packages/cli-node/src/parallel/parallel.test.ts rename to packages/cli-node/src/concurrency/runConcurrentTasks.test.ts index b0d3be3769..382a85d297 100644 --- a/packages/cli-node/src/parallel/parallel.test.ts +++ b/packages/cli-node/src/concurrency/runConcurrentTasks.test.ts @@ -14,7 +14,7 @@ * limitations under the License. */ -import { runConcurrentTasks, runWorkerQueueThreads } from './parallel'; +import { runConcurrentTasks } from './runConcurrentTasks'; describe('runConcurrentTasks', () => { afterEach(() => { @@ -142,28 +142,3 @@ describe('runConcurrentTasks', () => { warnSpy.mockRestore(); }); }); - -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/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..20e71531ab --- /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/parallel/parallel.ts b/packages/cli-node/src/concurrency/runWorkerQueueThreads.ts similarity index 65% rename from packages/cli-node/src/parallel/parallel.ts rename to packages/cli-node/src/concurrency/runWorkerQueueThreads.ts index 9d36852351..d1974dba90 100644 --- a/packages/cli-node/src/parallel/parallel.ts +++ b/packages/cli-node/src/concurrency/runWorkerQueueThreads.ts @@ -14,106 +14,9 @@ * limitations under the License. */ -import os from 'node:os'; import { ErrorLike } from '@backstage/errors'; import { Worker } from 'node:worker_threads'; - -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; - -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; -} - -/** - * 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); - } - }), - ); -} +import { getEnvironmentConcurrency } from './concurrency'; type WorkerThreadMessage = | { @@ -135,10 +38,6 @@ type WorkerThreadMessage = | { type: 'error'; error: ErrorLike; - } - | { - type: 'message'; - message: unknown; }; /** diff --git a/packages/cli-node/src/index.ts b/packages/cli-node/src/index.ts index 6d1c519b9e..5540666a05 100644 --- a/packages/cli-node/src/index.ts +++ b/packages/cli-node/src/index.ts @@ -22,5 +22,5 @@ export * from './git'; export * from './monorepo'; -export * from './parallel'; +export * from './concurrency'; export * from './roles';