From 872fcca2814f33a6df46cb9b29af1da8cd803cd0 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Sun, 6 Feb 2022 00:36:24 +0100 Subject: [PATCH] cli: refactor parallelism util + new shared worker util Signed-off-by: Patrik Oldsberg --- packages/cli/src/commands/app/build.ts | 4 +- packages/cli/src/commands/backend/bundle.ts | 4 +- packages/cli/src/commands/build/buildApp.ts | 4 +- .../cli/src/commands/build/buildBackend.ts | 4 +- packages/cli/src/commands/versions/bump.ts | 184 +++++++++--------- packages/cli/src/lib/bundler/types.ts | 7 +- packages/cli/src/lib/packager/index.ts | 9 +- packages/cli/src/lib/parallel.test.ts | 159 +++++++++++---- packages/cli/src/lib/parallel.ts | 71 +++++-- 9 files changed, 281 insertions(+), 165 deletions(-) diff --git a/packages/cli/src/commands/app/build.ts b/packages/cli/src/commands/app/build.ts index 36d910fe5b..67b71632c7 100644 --- a/packages/cli/src/commands/app/build.ts +++ b/packages/cli/src/commands/app/build.ts @@ -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, diff --git a/packages/cli/src/commands/backend/bundle.ts b/packages/cli/src/commands/backend/bundle.ts index 338d5cdaa0..ec5fa0a131 100644 --- a/packages/cli/src/commands/backend/bundle.ts +++ b/packages/cli/src/commands/backend/bundle.ts @@ -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, }); diff --git a/packages/cli/src/commands/build/buildApp.ts b/packages/cli/src/commands/build/buildApp.ts index 62d51a0810..b0959b2d6a 100644 --- a/packages/cli/src/commands/build/buildApp.ts +++ b/packages/cli/src/commands/build/buildApp.ts @@ -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, diff --git a/packages/cli/src/commands/build/buildBackend.ts b/packages/cli/src/commands/build/buildBackend.ts index 490cc413a4..e32b7956db 100644 --- a/packages/cli/src/commands/build/buildBackend.ts +++ b/packages/cli/src/commands/build/buildBackend.ts @@ -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, }); diff --git a/packages/cli/src/commands/versions/bump.ts b/packages/cli/src/commands/versions/bump.ts index 2c16f324bc..0d7c1cdfae 100644 --- a/packages/cli/src/commands/versions/bump.ts +++ b/packages/cli/src/commands/versions/bump.ts @@ -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(); // 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(); - 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( - count: number, - items: IterableIterator, - fn: (item: T) => Promise, -) { - 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()), - ); -} diff --git a/packages/cli/src/lib/bundler/types.ts b/packages/cli/src/lib/bundler/types.ts index 14e60c8892..4d6d2e5c9e 100644 --- a/packages/cli/src/lib/bundler/types.ts +++ b/packages/cli/src/lib/bundler/types.ts @@ -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; }; diff --git a/packages/cli/src/lib/packager/index.ts b/packages/cli/src/lib/packager/index.ts index c35e2587b9..40af21bc5e 100644 --- a/packages/cli/src/lib/packager/index.ts +++ b/packages/cli/src/lib/packager/index.ts @@ -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'], { diff --git a/packages/cli/src/lib/parallel.test.ts b/packages/cli/src/lib/parallel.test.ts index c5c33473ff..d851f33f82 100644 --- a/packages/cli/src/lib/parallel.test.ts +++ b/packages/cli/src/lib/parallel.test.ts @@ -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(); + const done = new Array(); + 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(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]); }); }); diff --git a/packages/cli/src/lib/parallel.ts b/packages/cli/src/lib/parallel.ts index f922295a89..840b409078 100644 --- a/packages/cli/src/lib/parallel.ts +++ b/packages/cli/src/lib/parallel.ts @@ -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 = { + /** + * Decides the number of parallel workers by multiplying + * this with the configured parallelism, which defaults to 4 + */ + parallelismFactor: number; + parallelismSetting?: ParallelismOption; + items: Iterable; + worker: (item: TItem) => Promise; +}; + +export async function runParallelWorkers( + options: ParallelWorkerOptions, +) { + const { parallelismFactor, parallelismSetting, items, worker } = options; + const parallelism = parallelismSetting + ? parseParallelismOption(parallelismSetting) + : getEnvironmentParallelism(); + + const iterator = items[Symbol.iterator](); + + async function pop() { + const el = iterator.next(); + if (el.done) { + return; + } + + await worker(el.value); + await pop(); + } + + return Promise.all( + Array(Math.max(Math.floor(parallelismFactor * parallelism), 1)) + .fill(0) + .map(() => pop()), + ); +}