Rename parallel/ to concurrency/ and split into separate files
Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com> Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -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;
|
||||
}
|
||||
+4
-5
@@ -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';
|
||||
+1
-26
@@ -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]);
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
}
|
||||
}),
|
||||
);
|
||||
}
|
||||
@@ -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]);
|
||||
});
|
||||
});
|
||||
+1
-102
@@ -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<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);
|
||||
}
|
||||
}),
|
||||
);
|
||||
}
|
||||
import { getEnvironmentConcurrency } from './concurrency';
|
||||
|
||||
type WorkerThreadMessage =
|
||||
| {
|
||||
@@ -135,10 +38,6 @@ type WorkerThreadMessage =
|
||||
| {
|
||||
type: 'error';
|
||||
error: ErrorLike;
|
||||
}
|
||||
| {
|
||||
type: 'message';
|
||||
message: unknown;
|
||||
};
|
||||
|
||||
/**
|
||||
@@ -22,5 +22,5 @@
|
||||
|
||||
export * from './git';
|
||||
export * from './monorepo';
|
||||
export * from './parallel';
|
||||
export * from './concurrency';
|
||||
export * from './roles';
|
||||
|
||||
Reference in New Issue
Block a user