paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 49 Packages Projects Releases Wiki Activity

Merge pull request #9388 from backstage/rugvip/role-parallel

Browse Source 
cli: refactor parallel helpers + limit parallelism
This commit is contained in:
Patrik Oldsberg
2022-02-07 19:18:15 +01:00
committed by GitHub
parent b5bcc42493 764ee19029
commit 618366b8a7
12 changed files with 327 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/cli/src/commands/app/build.ts
+2 -2
View File
@@ -17,7 +17,7 @@
import fs from 'fs-extra';
import { Command } from 'commander';
import { buildBundle } from '../../lib/bundler';
import { parseParallel, PARALLEL_ENV_VAR } from '../../lib/parallel';
import { getEnvironmentParallelism } from '../../lib/parallel';
import { loadCliConfig } from '../../lib/config';
import { paths } from '../../lib/paths';
@@ -25,7 +25,7 @@ export default async (cmd: Command) => {
const { name } = await fs.readJson(paths.resolveTarget('package.json'));
await buildBundle({
entry: 'src/index',
parallel: parseParallel(process.env[PARALLEL_ENV_VAR]),
parallelism: getEnvironmentParallelism(),
statsJsonEnabled: cmd.stats,
...(await loadCliConfig({
args: cmd.config,
packages/cli/src/commands/backend/bundle.ts
+2 -2
View File
@@ -21,7 +21,7 @@ import tar, { CreateOptions } from 'tar';
import { Command } from 'commander';
import { createDistWorkspace } from '../../lib/packager';
import { paths } from '../../lib/paths';
import { parseParallel, PARALLEL_ENV_VAR } from '../../lib/parallel';
import { getEnvironmentParallelism } from '../../lib/parallel';
import { buildPackage, Output } from '../../lib/builder';
const BUNDLE_FILE = 'bundle.tar.gz';
@@ -40,7 +40,7 @@ export default async (cmd: Command) => {
targetDir: tmpDir,
buildDependencies: Boolean(cmd.buildDependencies),
buildExcludes: [pkg.name],
parallel: parseParallel(process.env[PARALLEL_ENV_VAR]),
parallelism: getEnvironmentParallelism(),
skeleton: SKELETON_FILE,
});
packages/cli/src/commands/build/buildApp.ts
+2 -2
View File
@@ -17,7 +17,7 @@
import fs from 'fs-extra';
import { resolve as resolvePath } from 'path';
import { buildBundle } from '../../lib/bundler';
import { parseParallel, PARALLEL_ENV_VAR } from '../../lib/parallel';
import { getEnvironmentParallelism } from '../../lib/parallel';
import { loadCliConfig } from '../../lib/config';
interface BuildAppOptions {
@@ -32,7 +32,7 @@ export async function buildApp(options: BuildAppOptions) {
await buildBundle({
targetDir,
entry: 'src/index',
parallel: parseParallel(process.env[PARALLEL_ENV_VAR]),
parallelism: getEnvironmentParallelism(),
statsJsonEnabled: writeStats,
...(await loadCliConfig({
args: configPaths,
packages/cli/src/commands/build/buildBackend.ts
+2 -2
View File
@@ -19,7 +19,7 @@ import fs from 'fs-extra';
import { resolve as resolvePath } from 'path';
import tar, { CreateOptions } from 'tar';
import { createDistWorkspace } from '../../lib/packager';
import { parseParallel, PARALLEL_ENV_VAR } from '../../lib/parallel';
import { getEnvironmentParallelism } from '../../lib/parallel';
import { buildPackage, Output } from '../../lib/builder';
const BUNDLE_FILE = 'bundle.tar.gz';
@@ -43,7 +43,7 @@ export async function buildBackend(options: BuildBackendOptions) {
targetDir: tmpDir,
buildDependencies: !skipBuildDependencies,
buildExcludes: [pkg.name],
parallel: parseParallel(process.env[PARALLEL_ENV_VAR]),
parallelism: getEnvironmentParallelism(),
skeleton: SKELETON_FILE,
});
packages/cli/src/commands/repo/build.ts
+27 -25
View File
@@ -20,8 +20,9 @@ import { relative as relativePath } from 'path';
import { buildPackages, getOutputsForRole } from '../../lib/builder';
import { PackageGraph } from '../../lib/monorepo';
import { ExtendedPackage } from '../../lib/monorepo/PackageGraph';
import { runParallelWorkers } from '../../lib/parallel';
import { paths } from '../../lib/paths';
import { getRoleInfo } from '../../lib/role';
import { detectRoleFromPackage } from '../../lib/role';
import { buildApp } from '../build/buildApp';
import { buildBackend } from '../build/buildBackend';
@@ -76,26 +77,30 @@ function createScriptOptionsParser(anyCmd: Command, commandPath: string[]) {
export async function command(cmd: Command): Promise<void> {
const packages = await PackageGraph.listTargetPackages();
const bundledPackages = new Array<ExtendedPackage>();
const apps = new Array<ExtendedPackage>();
const backends = new Array<ExtendedPackage>();
const parseBuildScript = createScriptOptionsParser(cmd, ['script', 'build']);
const options = packages.flatMap(pkg => {
const role = pkg.packageJson.backstage?.role;
const role =
pkg.packageJson.backstage?.role ?? detectRoleFromPackage(pkg.packageJson);
if (!role) {
console.warn(`Ignored ${pkg.packageJson.name} because it has no role`);
return [];
}
if (role === 'app') {
apps.push(pkg);
return [];
} else if (role === 'backend') {
backends.push(pkg);
return [];
}
const outputs = getOutputsForRole(role);
if (outputs.size === 0) {
if (getRoleInfo(role).output.includes('bundle')) {
bundledPackages.push(pkg);
} else {
console.warn(
`Ignored ${pkg.packageJson.name} because it has no output`,
);
}
console.warn(`Ignored ${pkg.packageJson.name} because it has no output`);
return [];
}
@@ -120,13 +125,11 @@ export async function command(cmd: Command): Promise<void> {
await buildPackages(options);
if (cmd.all) {
const apps = bundledPackages.filter(
pkg => pkg.packageJson.backstage?.role === 'app',
);
console.log('Building apps');
await Promise.all(
apps.map(async pkg => {
await runParallelWorkers({
items: apps,
parallelismFactor: 1 / 2,
worker: async pkg => {
const buildOptions = parseBuildScript(pkg.packageJson.scripts?.build);
if (!buildOptions) {
console.warn(
@@ -139,15 +142,14 @@ export async function command(cmd: Command): Promise<void> {
configPaths: (buildOptions.config as string[]) ?? [],
writeStats: Boolean(buildOptions.stats),
});
}),
);
},
});
console.log('Building backends');
const backends = bundledPackages.filter(
pkg => pkg.packageJson.backstage?.role === 'backend',
);
await Promise.all(
backends.map(async pkg => {
await runParallelWorkers({
items: backends,
parallelismFactor: 1 / 2,
worker: async pkg => {
const buildOptions = parseBuildScript(pkg.packageJson.scripts?.build);
if (!buildOptions) {
console.warn(
@@ -159,7 +161,7 @@ export async function command(cmd: Command): Promise<void> {
targetDir: pkg.dir,
skipBuildDependencies: true,
});
}),
);
},
});
}
}
packages/cli/src/commands/versions/bump.test.ts
+11 -11
View File
@@ -127,8 +127,8 @@ describe('bump', () => {
});
expect(logs.filter(Boolean)).toEqual([
'Using default pattern glob @backstage/*',
'Checking for updates of @backstage/theme',
'Checking for updates of @backstage/core',
'Checking for updates of @backstage/theme',
'Checking for updates of @backstage/core-api',
'Some packages are outdated, updating',
'unlocking @backstage/core@^1.0.3 ~> 1.0.6',
@@ -252,19 +252,19 @@ describe('bump', () => {
});
expect(logs.filter(Boolean)).toEqual([
'Using custom pattern glob @{backstage,backstage-extra}/*',
'Checking for updates of @backstage/theme',
'Checking for updates of @backstage-extra/custom-two',
'Checking for updates of @backstage-extra/custom',
'Checking for updates of @backstage/core',
'Checking for updates of @backstage-extra/custom',
'Checking for updates of @backstage-extra/custom-two',
'Checking for updates of @backstage/theme',
'Checking for updates of @backstage/core-api',
'Some packages are outdated, updating',
'unlocking @backstage-extra/custom@^1.0.1 ~> 1.1.0',
'unlocking @backstage/core@^1.0.3 ~> 1.0.6',
'unlocking @backstage-extra/custom@^1.0.1 ~> 1.1.0',
'unlocking @backstage/core-api@^1.0.6 ~> 1.0.7',
'unlocking @backstage/core-api@^1.0.3 ~> 1.0.7',
'bumping @backstage-extra/custom-two in a to ^2.0.0',
'bumping @backstage/theme in b to ^2.0.0',
'bumping @backstage-extra/custom-two in b to ^2.0.0',
'bumping @backstage/theme in b to ^2.0.0',
'Running yarn install to install new versions',
'⚠️ The following packages may have breaking changes:',
' @backstage-extra/custom-two : 1.0.0 ~> 2.0.0',
@@ -348,14 +348,14 @@ describe('bump', () => {
});
expect(logs.filter(Boolean)).toEqual([
'Using default pattern glob @backstage/*',
'Checking for updates of @backstage/theme',
'Checking for updates of @backstage/core',
'Package info not found, ignoring package @backstage/theme',
'Package info not found, ignoring package @backstage/core',
'Checking for updates of @backstage/theme',
'Checking for updates of @backstage/core',
'Package info not found, ignoring package @backstage/theme',
'Package info not found, ignoring package @backstage/core',
'Package info not found, ignoring package @backstage/theme',
'Checking for updates of @backstage/core',
'Checking for updates of @backstage/theme',
'Package info not found, ignoring package @backstage/core',
'Package info not found, ignoring package @backstage/theme',
'All Backstage packages are up to date!',
]);
packages/cli/src/commands/versions/bump.ts
+87 -97
View File
@@ -30,6 +30,7 @@ import {
} from '../../lib/versioning';
import { forbiddenDuplicatesFilter } from './lint';
import { BACKSTAGE_JSON } from '@backstage/cli-common';
import { runParallelWorkers } from '../../lib/parallel';
const DEP_TYPES = [
'dependencies',
@@ -68,67 +69,76 @@ export default async (cmd: Command) => {
const versionBumps = new Map<string, PkgVersionInfo[]>();
// Track package versions that we want to remove from yarn.lock in order to trigger a bump
const unlocked = Array<{ name: string; range: string; target: string }>();
await workerThreads(16, dependencyMap.entries(), async ([name, pkgs]) => {
let target: string;
try {
target = await findTargetVersion(name);
} catch (error) {
if (isError(error) && error.name === 'NotFoundError') {
console.log(`Package info not found, ignoring package ${name}`);
return;
}
throw error;
}
for (const pkg of pkgs) {
if (semver.satisfies(target, pkg.range)) {
if (semver.minVersion(pkg.range)?.version !== target) {
unlocked.push({ name, range: pkg.range, target });
await runParallelWorkers({
parallelismFactor: 4,
items: dependencyMap.entries(),
async worker([name, pkgs]) {
let target: string;
try {
target = await findTargetVersion(name);
} catch (error) {
if (isError(error) && error.name === 'NotFoundError') {
console.log(`Package info not found, ignoring package ${name}`);
return;
}
continue;
throw error;
}
versionBumps.set(
pkg.name,
(versionBumps.get(pkg.name) ?? []).concat({
name,
location: pkg.location,
range: `^${target}`, // TODO(Rugvip): Option to use something else than ^?
target,
}),
);
}
for (const pkg of pkgs) {
if (semver.satisfies(target, pkg.range)) {
if (semver.minVersion(pkg.range)?.version !== target) {
unlocked.push({ name, range: pkg.range, target });
}
continue;
}
versionBumps.set(
pkg.name,
(versionBumps.get(pkg.name) ?? []).concat({
name,
location: pkg.location,
range: `^${target}`, // TODO(Rugvip): Option to use something else than ^?
target,
}),
);
}
},
});
const filter = (name: string) => minimatch(name, pattern);
// Check for updates of transitive backstage dependencies
await workerThreads(16, lockfile.keys(), async name => {
// Only check @backstage packages and friends, we don't want this to do a full update of all deps
if (!filter(name)) {
return;
}
let target: string;
try {
target = await findTargetVersion(name);
} catch (error) {
if (isError(error) && error.name === 'NotFoundError') {
console.log(`Package info not found, ignoring package ${name}`);
await runParallelWorkers({
parallelismFactor: 4,
items: lockfile.keys(),
async worker(name) {
// Only check @backstage packages and friends, we don't want this to do a full update of all deps
if (!filter(name)) {
return;
}
throw error;
}
for (const entry of lockfile.get(name) ?? []) {
// Ignore lockfile entries that don't satisfy the version range, since
// these can't cause the package to be locked to an older version
if (!semver.satisfies(target, entry.range)) {
continue;
let target: string;
try {
target = await findTargetVersion(name);
} catch (error) {
if (isError(error) && error.name === 'NotFoundError') {
console.log(`Package info not found, ignoring package ${name}`);
return;
}
throw error;
}
// Unlock all entries that are within range but on the old version
unlocked.push({ name, range: entry.range, target });
}
for (const entry of lockfile.get(name) ?? []) {
// Ignore lockfile entries that don't satisfy the version range, since
// these can't cause the package to be locked to an older version
if (!semver.satisfies(target, entry.range)) {
continue;
}
// Unlock all entries that are within range but on the old version
unlocked.push({ name, range: entry.range, target });
}
},
});
console.log();
@@ -163,38 +173,42 @@ export default async (cmd: Command) => {
}
const breakingUpdates = new Map<string, { from: string; to: string }>();
await workerThreads(16, versionBumps.entries(), async ([name, deps]) => {
const pkgPath = resolvePath(deps[0].location, 'package.json');
const pkgJson = await fs.readJson(pkgPath);
await runParallelWorkers({
parallelismFactor: 4,
items: versionBumps.entries(),
async worker([name, deps]) {
const pkgPath = resolvePath(deps[0].location, 'package.json');
const pkgJson = await fs.readJson(pkgPath);
for (const dep of deps) {
console.log(
`${chalk.cyan('bumping')} ${dep.name} in ${chalk.cyan(
name,
)} to ${chalk.yellow(dep.range)}`,
);
for (const dep of deps) {
console.log(
`${chalk.cyan('bumping')} ${dep.name} in ${chalk.cyan(
name,
)} to ${chalk.yellow(dep.range)}`,
);
for (const depType of DEP_TYPES) {
if (depType in pkgJson && dep.name in pkgJson[depType]) {
const oldRange = pkgJson[depType][dep.name];
pkgJson[depType][dep.name] = dep.range;
for (const depType of DEP_TYPES) {
if (depType in pkgJson && dep.name in pkgJson[depType]) {
const oldRange = pkgJson[depType][dep.name];
pkgJson[depType][dep.name] = dep.range;
// Check if the update was at least a pre-v1 minor or post-v1 major release
const lockfileEntry = lockfile
.get(dep.name)
?.find(entry => entry.range === oldRange);
if (lockfileEntry) {
const from = lockfileEntry.version;
const to = dep.target;
if (!semver.satisfies(to, `^${from}`)) {
breakingUpdates.set(dep.name, { from, to });
// Check if the update was at least a pre-v1 minor or post-v1 major release
const lockfileEntry = lockfile
.get(dep.name)
?.find(entry => entry.range === oldRange);
if (lockfileEntry) {
const from = lockfileEntry.version;
const to = dep.target;
if (!semver.satisfies(to, `^${from}`)) {
breakingUpdates.set(dep.name, { from, to });
}
}
}
}
}
}
await fs.writeJson(pkgPath, pkgJson, { spaces: 2 });
await fs.writeJson(pkgPath, pkgJson, { spaces: 2 });
},
});
console.log();
@@ -324,27 +338,3 @@ export async function bumpBackstageJsonVersion() {
},
);
}
async function workerThreads<T>(
count: number,
items: IterableIterator<T>,
fn: (item: T) => Promise<void>,
) {
const queue = Array.from(items);
async function pop() {
const item = queue.pop();
if (!item) {
return;
}
await fn(item);
await pop();
}
return Promise.all(
Array(count)
.fill(0)
.map(() => pop()),
);
}
packages/cli/src/lib/builder/packager.ts
+8 -3
View File
@@ -23,6 +23,7 @@ import { makeRollupConfigs } from './config';
import { BuildOptions, Output } from './types';
import { buildTypeDefinitions } from './buildTypeDefinitions';
import { getRoleInfo } from '../role';
import { runParallelWorkers } from '../parallel';
export function formatErrorMessage(error: any) {
let msg = '';
@@ -128,7 +129,7 @@ export const buildPackages = async (options: BuildOptions[]) => {
options.map(({ targetDir }) => fs.remove(resolvePath(targetDir!, 'dist'))),
);
const buildTasks = rollupConfigs.flat().map(rollupBuild);
const buildTasks = rollupConfigs.flat().map(opts => () => rollupBuild(opts));
const typeDefinitionTargetDirs = options
.filter(
@@ -138,10 +139,14 @@ export const buildPackages = async (options: BuildOptions[]) => {
.map(_ => _.targetDir!);
if (typeDefinitionTargetDirs.length > 0) {
buildTasks.push(buildTypeDefinitions(typeDefinitionTargetDirs));
// Make sure this one is started first
buildTasks.unshift(() => buildTypeDefinitions(typeDefinitionTargetDirs));
}
await Promise.all(buildTasks);
await runParallelWorkers({
items: buildTasks,
worker: async task => task(),
});
};
export function getOutputsForRole(role: string): Set<Output> {
packages/cli/src/lib/bundler/types.ts
+3 -4
View File
@@ -16,7 +16,6 @@
import { AppConfig, Config } from '@backstage/config';
import { BundlingPathsOptions } from './paths';
import { ParallelOption } from '../parallel';
import { ConfigSchema } from '@backstage/config-loader';
export type BundlingOptions = {
@@ -25,7 +24,7 @@ export type BundlingOptions = {
frontendConfig: Config;
frontendAppConfigs: AppConfig[];
baseUrl: URL;
parallel?: ParallelOption;
parallelism?: number;
};
export type ServeOptions = BundlingPathsOptions & {
@@ -38,7 +37,7 @@ export type BuildOptions = BundlingPathsOptions & {
// Target directory, defaulting to paths.targetDir
targetDir?: string;
statsJsonEnabled: boolean;
parallel?: ParallelOption;
parallelism?: number;
schema?: ConfigSchema;
frontendConfig: Config;
frontendAppConfigs: AppConfig[];
@@ -47,7 +46,7 @@ export type BuildOptions = BundlingPathsOptions & {
export type BackendBundlingOptions = {
checksEnabled: boolean;
isDev: boolean;
parallel?: ParallelOption;
parallelism?: number;
inspectEnabled: boolean;
inspectBrkEnabled: boolean;
};
packages/cli/src/lib/packager/index.ts
+4 -5
View File
@@ -24,7 +24,6 @@ import { tmpdir } from 'os';
import tar, { CreateOptions } from 'tar';
import { paths } from '../paths';
import { run } from '../run';
import { ParallelOption } from '../parallel';
import {
dependencies as cliDependencies,
devDependencies as cliDevDependencies,
@@ -69,9 +68,9 @@ type Options = {
buildExcludes?: string[];
/**
* Enable (true/false) or control amount of (number) parallelism in some build steps.
* Controls amount of parallelism in some build steps.
*/
parallel?: ParallelOption;
parallelism?: number;
/**
* If set, creates a skeleton tarball that contains all package.json files
@@ -115,8 +114,8 @@ export async function createDistWorkspace(
if (toBuild.length > 0) {
const scopeArgs = toBuild.flatMap(target => ['--scope', target.name]);
const lernaArgs =
options.parallel && Number.isInteger(options.parallel)
? ['--concurrency', options.parallel.toString()]
options.parallelism && Number.isInteger(options.parallelism)
? ['--concurrency', options.parallelism.toString()]
: [];
await run('yarn', ['lerna', ...lernaArgs, 'run', ...scopeArgs, 'build'], {
packages/cli/src/lib/parallel.test.ts
+122 -37
View File
@@ -14,46 +14,131 @@
* limitations under the License.
*/
import { isParallelDefault, parseParallel } from './parallel';
import {
parseParallelismOption,
getEnvironmentParallelism,
runParallelWorkers,
} from './parallel';
describe('parallel', () => {
describe('parseParallel', () => {
it('coerces "false" string to boolean', () => {
expect(parseParallel('false')).toBeFalsy();
});
it('coerces "true" to boolean', () => {
expect(parseParallel('true')).toBeTruthy();
});
it('coerces number string to number', () => {
expect(parseParallel('2')).toBe(2);
});
it.each([[true], [false], [2]])('returns itself for %p', value => {
expect(parseParallel(value as any)).toEqual(value);
});
it.each([[undefined], [null]])('returns true for %p', value => {
expect(parseParallel(value as any)).toBe(true);
});
it.each([['on'], [2.5], ['2.5']])('throws error for %p', value => {
expect(() => parseParallel(value as any)).toThrowError(
`Parallel option value '${value}' is not a boolean or integer`,
);
});
describe('parseParallelismOption', () => {
it('coerces false no parallelism', () => {
expect(parseParallelismOption(false)).toBe(1);
expect(parseParallelismOption('false')).toBe(1);
});
describe('isParallelDefault', () => {
it('returns true if default value', () => {
expect(isParallelDefault(undefined)).toBeTruthy();
expect(isParallelDefault(true)).toBeTruthy();
});
it('coerces true or undefined to default parallelism', () => {
expect(parseParallelismOption(true)).toBe(4);
expect(parseParallelismOption('true')).toBe(4);
expect(parseParallelismOption(undefined)).toBe(4);
expect(parseParallelismOption(null)).toBe(4);
});
it('returns false if not default value', () => {
expect(isParallelDefault(false)).toBeFalsy();
expect(isParallelDefault(2)).toBeFalsy();
expect(isParallelDefault('true' as any)).toBeFalsy();
});
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)).toThrowError(
`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(4);
process.env.BACKSTAGE_CLI_BUILD_PARALLEL = 'false';
expect(getEnvironmentParallelism()).toBe(1);
delete process.env.BACKSTAGE_CLI_BUILD_PARALLEL;
expect(getEnvironmentParallelism()).toBe(4);
});
});
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],
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]);
});
});
packages/cli/src/lib/parallel.ts
+57 -14
View File
@@ -14,30 +14,31 @@
* limitations under the License.
*/
export const DEFAULT_PARALLELISM = 4;
export const PARALLEL_ENV_VAR = 'BACKSTAGE_CLI_BUILD_PARALLEL';
export type ParallelOption = boolean | number | undefined;
export type ParallelismOption = boolean | string | number | null | undefined;
export function isParallelDefault(parallel: ParallelOption) {
return parallel === undefined || parallel === true;
}
export function parseParallel(
parallel: boolean | string | number | undefined,
): ParallelOption {
export function parseParallelismOption(parallel: ParallelismOption): number {
if (parallel === undefined || parallel === null) {
return true;
return DEFAULT_PARALLELISM;
} else if (typeof parallel === 'boolean') {
return parallel;
return parallel ? DEFAULT_PARALLELISM : 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 true;
return parseParallelismOption(true);
} else if (parallel === 'false') {
return false;
} else if (Number.isInteger(parseFloat(parallel.toString()))) {
return Number(parallel);
return parseParallelismOption(false);
}
const parsed = Number(parallel);
if (Number.isInteger(parsed)) {
return parseParallelismOption(parsed);
}
}
@@ -45,3 +46,45 @@ export function parseParallel(
`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);
}
}),
);
}