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

cli: refactor parallelism util + new shared worker util

Browse Source 
Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>
This commit is contained in:
Patrik Oldsberg
2022-02-06 00:36:24 +01:00
parent 9d56dabe65
commit 872fcca281
9 changed files with 281 additions and 165 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/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/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
*/
parallelismFactor: number;
parallelismSetting?: ParallelismOption;
items: Iterable<TItem>;
worker: (item: TItem) => Promise<void>;
};
export async function runParallelWorkers<TItem>(
options: ParallelWorkerOptions<TItem>,
) {
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()),
);
}