honeymoose
/
angular-cn
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(ngcc): create new entry-point for cluster workers (#36637) 

PR Close #36637
This commit is contained in:
Pete Bacon Darwin 2020-04-15 16:35:09 +01:00 committed by Matias Niemelä
parent 7e5e60b757
commit 443f5eee85
7 changed files with 232 additions and 247 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
packages/compiler-cli/ngcc/src/execution/cluster/executor.ts
View File

@ -5,11 +5,7 @@
* Use of this source code is governed by an MIT-style license that can be
* found in the LICENSE file at https://angular.io/license
*/
/// <reference types="node" />
import * as cluster from 'cluster';
import {FileSystem} from '../../../../src/ngtsc/file_system';
import {AsyncLocker} from '../../locking/async_locker';
import {Logger} from '../../logging/logger';
import {PackageJsonUpdater} from '../../writing/package_json_updater';
@ -17,8 +13,6 @@ import {AnalyzeEntryPointsFn, CreateCompileFn, Executor} from '../api';
import {CreateTaskCompletedCallback} from '../tasks/api';
import {ClusterMaster} from './master';
import {ClusterWorker} from './worker';
/**
* An `Executor` that processes tasks in parallel (on multiple processes) and completes
@ -26,26 +20,19 @@ import {ClusterWorker} from './worker';
*/
export class ClusterExecutor implements Executor {
constructor(
private workerCount: number, private logger: Logger,
private workerCount: number, private fileSystem: FileSystem, private logger: Logger,
private pkgJsonUpdater: PackageJsonUpdater, private lockFile: AsyncLocker,
private createTaskCompletedCallback: CreateTaskCompletedCallback) {}
async execute(analyzeEntryPoints: AnalyzeEntryPointsFn, createCompileFn: CreateCompileFn):
async execute(analyzeEntryPoints: AnalyzeEntryPointsFn, _createCompileFn: CreateCompileFn):
Promise<void> {
if (cluster.isMaster) {
// This process is the cluster master.
return this.lockFile.lock(() => {
this.logger.debug(`Running ngcc on ${this.constructor.name} (using ${
this.workerCount} worker processes).`);
this.logger.debug(
`Running ngcc on ${this.constructor.name} (using ${this.workerCount} worker processes).`);
const master = new ClusterMaster(
this.workerCount, this.logger, this.pkgJsonUpdater, analyzeEntryPoints,
this.workerCount, this.fileSystem, this.logger, this.pkgJsonUpdater, analyzeEntryPoints,
this.createTaskCompletedCallback);
return master.run();
});
} else {
// This process is a cluster worker.
const worker = new ClusterWorker(this.logger, createCompileFn);
return worker.run();
}
}
}

9
packages/compiler-cli/ngcc/src/execution/cluster/master.ts
View File

@ -10,7 +10,7 @@
import * as cluster from 'cluster';
import {resolve} from '../../../../src/ngtsc/file_system';
import {FileSystem} from '../../../../src/ngtsc/file_system';
import {Logger} from '../../logging/logger';
import {PackageJsonUpdater} from '../../writing/package_json_updater';
import {AnalyzeEntryPointsFn} from '../api';
@ -33,13 +33,16 @@ export class ClusterMaster {
private onTaskCompleted: TaskCompletedCallback;
constructor(
private maxWorkerCount: number, private logger: Logger,
private maxWorkerCount: number, private fileSystem: FileSystem, private logger: Logger,
private pkgJsonUpdater: PackageJsonUpdater, analyzeEntryPoints: AnalyzeEntryPointsFn,
createTaskCompletedCallback: CreateTaskCompletedCallback) {
if (!cluster.isMaster) {
throw new Error('Tried to instantiate `ClusterMaster` on a worker process.');
}
// Set the worker entry-point
cluster.setupMaster({exec: this.fileSystem.resolve(__dirname, 'worker.js')});
this.taskQueue = analyzeEntryPoints();
this.onTaskCompleted = createTaskCompletedCallback(this.taskQueue);
}
@ -227,7 +230,7 @@ export class ClusterMaster {
JSON.stringify(msg));
}
const expectedPackageJsonPath = resolve(task.entryPoint.path, 'package.json');
const expectedPackageJsonPath = this.fileSystem.resolve(task.entryPoint.path, 'package.json');
const parsedPackageJson = task.entryPoint.packageJson;
if (expectedPackageJsonPath !== msg.packageJsonPath) {

82
packages/compiler-cli/ngcc/src/execution/cluster/worker.ts
View File

@ -5,45 +5,90 @@
* Use of this source code is governed by an MIT-style license that can be
* found in the LICENSE file at https://angular.io/license
*/
/// <reference types="node" />
import * as cluster from 'cluster';
import {Logger} from '../../logging/logger';
import {CompileFn, CreateCompileFn} from '../api';
import {absoluteFrom, CachedFileSystem, getFileSystem, NodeJSFileSystem, setFileSystem} from '../../../../src/ngtsc/file_system';
import {readConfiguration} from '../../../../src/perform_compile';
import {parseCommandLineOptions} from '../../command_line_options';
import {ConsoleLogger} from '../../logging/console_logger';
import {Logger, LogLevel} from '../../logging/logger';
import {getPathMappingsFromTsConfig} from '../../utils';
import {DirectPackageJsonUpdater} from '../../writing/package_json_updater';
import {CreateCompileFn} from '../api';
import {getCreateCompileFn} from '../create_compile_function';
import {stringifyTask} from '../tasks/utils';
import {MessageToWorker} from './api';
import {ClusterPackageJsonUpdater} from './package_json_updater';
import {sendMessageToMaster} from './utils';
// Cluster worker entry point
if (require.main === module) {
process.title = 'ngcc (worker)';
setFileSystem(new CachedFileSystem(new NodeJSFileSystem()));
let {
basePath,
targetEntryPointPath,
createNewEntryPointFormats = false,
logger = new ConsoleLogger(LogLevel.info),
pathMappings,
errorOnFailedEntryPoint = false,
enableI18nLegacyMessageIdFormat = true,
tsConfigPath
} = parseCommandLineOptions(process.argv.slice(2));
(async () => {
try {
if (!!targetEntryPointPath) {
// targetEntryPointPath forces us to error if an entry-point fails.
errorOnFailedEntryPoint = true;
}
/**
* A cluster worker is responsible for processing one task (i.e. one format property for a specific
* entry-point) at a time and reporting results back to the cluster master.
*/
export class ClusterWorker {
private compile: CompileFn;
const fileSystem = getFileSystem();
const absBasePath = absoluteFrom(basePath);
const projectPath = fileSystem.dirname(absBasePath);
const tsConfig =
tsConfigPath !== null ? readConfiguration(tsConfigPath || projectPath) : null;
constructor(private logger: Logger, createCompileFn: CreateCompileFn) {
if (pathMappings === undefined) {
pathMappings = getPathMappingsFromTsConfig(tsConfig, projectPath);
}
const pkgJsonUpdater =
new ClusterPackageJsonUpdater(new DirectPackageJsonUpdater(fileSystem));
// The function for creating the `compile()` function.
const createCompileFn = getCreateCompileFn(
fileSystem, logger, pkgJsonUpdater, createNewEntryPointFormats, errorOnFailedEntryPoint,
enableI18nLegacyMessageIdFormat, tsConfig, pathMappings);
await startWorker(logger, createCompileFn);
process.exitCode = 0;
} catch (e) {
console.error(e.stack || e.message);
process.exitCode = 1;
}
})();
}
export async function startWorker(logger: Logger, createCompileFn: CreateCompileFn): Promise<void> {
if (cluster.isMaster) {
throw new Error('Tried to instantiate `ClusterWorker` on the master process.');
throw new Error('Tried to run cluster worker on the master process.');
}
this.compile = createCompileFn(
(_task, outcome, message) =>
sendMessageToMaster({type: 'task-completed', outcome, message}));
}
const compile = createCompileFn(
(_task, outcome, message) => sendMessageToMaster({type: 'task-completed', outcome, message}));
run(): Promise<void> {
// Listen for `ProcessTaskMessage`s and process tasks.
cluster.worker.on('message', (msg: MessageToWorker) => {
try {
switch (msg.type) {
case 'process-task':
this.logger.debug(
logger.debug(
`[Worker #${cluster.worker.id}] Processing task: ${stringifyTask(msg.task)}`);
return this.compile(msg.task);
return compile(msg.task);
default:
throw new Error(
`[Worker #${cluster.worker.id}] Invalid message received: ${JSON.stringify(msg)}`);
@ -59,4 +104,3 @@ export class ClusterWorker {
// Return a promise that is never resolved.
return new Promise(() => undefined);
}
}

2
packages/compiler-cli/ngcc/src/main.ts
View File

@ -178,7 +178,7 @@ function getExecutor(
// Execute in parallel. Use up to 8 CPU cores for workers, always reserving one for master.
const workerCount = Math.min(8, os.cpus().length - 1);
return new ClusterExecutor(
workerCount, logger, pkgJsonUpdater, locker, createTaskCompletedCallback);
workerCount, fileSystem, logger, pkgJsonUpdater, locker, createTaskCompletedCallback);
} else {
// Execute serially, on a single thread (async).
return new SingleProcessExecutorAsync(logger, locker, createTaskCompletedCallback);

1
packages/compiler-cli/ngcc/test/BUILD.bazel
View File

@ -29,6 +29,7 @@ ts_library(
"@npm//magic-string",
"@npm//sourcemap-codec",
"@npm//typescript",
"@npm//yargs",
],
)

49
packages/compiler-cli/ngcc/test/execution/cluster/executor_spec.ts
View File

@ -8,23 +8,22 @@
/// <reference types="node" />
import {getFileSystem} from '@angular/compiler-cli/src/ngtsc/file_system';
import * as cluster from 'cluster';
import {MockFileSystemNative} from '../../../../src/ngtsc/file_system/testing';
import {MockFileSystemNative, runInEachFileSystem} from '../../../../src/ngtsc/file_system/testing';
import {ClusterExecutor} from '../../../src/execution/cluster/executor';
import {ClusterMaster} from '../../../src/execution/cluster/master';
import {ClusterWorker} from '../../../src/execution/cluster/worker';
import {AsyncLocker} from '../../../src/locking/async_locker';
import {PackageJsonUpdater} from '../../../src/writing/package_json_updater';
import {MockLockFile} from '../../helpers/mock_lock_file';
import {MockLogger} from '../../helpers/mock_logger';
import {mockProperty} from '../../helpers/spy_utils';
runInEachFileSystem(() => {
describe('ClusterExecutor', () => {
const runAsClusterMaster = mockProperty(cluster, 'isMaster');
let masterRunSpy: jasmine.Spy;
let workerRunSpy: jasmine.Spy;
let mockLogger: MockLogger;
let lockFileLog: string[];
let mockLockFile: MockLockFile;
@ -35,8 +34,6 @@ describe('ClusterExecutor', () => {
beforeEach(() => {
masterRunSpy = spyOn(ClusterMaster.prototype, 'run')
.and.returnValue(Promise.resolve('CusterMaster#run()' as any));
workerRunSpy = spyOn(ClusterWorker.prototype, 'run')
.and.returnValue(Promise.resolve('CusterWorker#run()' as any));
createTaskCompletedCallback = jasmine.createSpy('createTaskCompletedCallback');
mockLogger = new MockLogger();
@ -44,11 +41,11 @@ describe('ClusterExecutor', () => {
mockLockFile = new MockLockFile(new MockFileSystemNative(), lockFileLog);
locker = new AsyncLocker(mockLockFile, mockLogger, 200, 2);
executor = new ClusterExecutor(
42, mockLogger, null as unknown as PackageJsonUpdater, locker, createTaskCompletedCallback);
42, getFileSystem(), mockLogger, null as unknown as PackageJsonUpdater, locker,
createTaskCompletedCallback);
});
describe('execute()', () => {
describe('(on cluster master)', () => {
beforeEach(() => runAsClusterMaster(true));
it('should log debug info about the executor', async () => {
@ -68,7 +65,6 @@ describe('ClusterExecutor', () => {
.toBe('CusterMaster#run()' as any);
expect(masterRunSpy).toHaveBeenCalledWith();
expect(workerRunSpy).not.toHaveBeenCalled();
expect(analyzeEntryPointsSpy).toHaveBeenCalledWith();
expect(createCompilerFnSpy).not.toHaveBeenCalled();
@ -102,7 +98,7 @@ describe('ClusterExecutor', () => {
});
executor = new ClusterExecutor(
42, mockLogger, null as unknown as PackageJsonUpdater, locker,
42, getFileSystem(), mockLogger, null as unknown as PackageJsonUpdater, locker,
createTaskCompletedCallback);
let error = '';
try {
@ -122,7 +118,7 @@ describe('ClusterExecutor', () => {
});
executor = new ClusterExecutor(
42, mockLogger, null as unknown as PackageJsonUpdater, locker,
42, getFileSystem(), mockLogger, null as unknown as PackageJsonUpdater, locker,
createTaskCompletedCallback);
let error = '';
try {
@ -135,36 +131,5 @@ describe('ClusterExecutor', () => {
expect(masterRunSpy).toHaveBeenCalled();
});
});
describe('(on cluster worker)', () => {
beforeEach(() => runAsClusterMaster(false));
it('should not log debug info about the executor', async () => {
const anyFn: () => any = () => undefined;
await executor.execute(anyFn, anyFn);
expect(mockLogger.logs.debug).toEqual([]);
});
it('should delegate to `ClusterWorker#run()`', async () => {
const analyzeEntryPointsSpy = jasmine.createSpy('analyzeEntryPoints');
const createCompilerFnSpy = jasmine.createSpy('createCompilerFn');
expect(await executor.execute(analyzeEntryPointsSpy, createCompilerFnSpy))
.toBe('CusterWorker#run()' as any);
expect(masterRunSpy).not.toHaveBeenCalledWith();
expect(workerRunSpy).toHaveBeenCalled();
expect(analyzeEntryPointsSpy).not.toHaveBeenCalled();
expect(createCompilerFnSpy).toHaveBeenCalledWith(jasmine.any(Function));
});
it('should not call LockFile.write() or LockFile.remove()', async () => {
const anyFn: () => any = () => undefined;
await executor.execute(anyFn, anyFn);
expect(lockFileLog).toEqual([]);
});
});
});
});

53
packages/compiler-cli/ngcc/test/execution/cluster/worker_spec.ts
View File

@ -11,13 +11,13 @@
import * as cluster from 'cluster';
import {EventEmitter} from 'events';
import {ClusterWorker} from '../../../src/execution/cluster/worker';
import {startWorker} from '../../../src/execution/cluster/worker';
import {Task, TaskCompletedCallback, TaskProcessingOutcome} from '../../../src/execution/tasks/api';
import {MockLogger} from '../../helpers/mock_logger';
import {mockProperty} from '../../helpers/spy_utils';
describe('ClusterWorker', () => {
describe('startWorker()', () => {
const runAsClusterMaster = mockProperty(cluster, 'isMaster');
const mockProcessSend = mockProperty(process, 'send');
let processSendSpy: jasmine.Spy;
@ -35,27 +35,33 @@ describe('ClusterWorker', () => {
mockLogger = new MockLogger();
});
describe('constructor()', () => {
describe('(on cluster master)', () => {
beforeEach(() => runAsClusterMaster(true));
it('should throw an error', () => {
expect(() => new ClusterWorker(mockLogger, createCompileFnSpy))
.toThrowError('Tried to instantiate `ClusterWorker` on the master process.');
it('should throw an error', async () => {
await expectAsync(startWorker(mockLogger, createCompileFnSpy))
.toBeRejectedWithError('Tried to run cluster worker on the master process.');
expect(createCompileFnSpy).not.toHaveBeenCalled();
});
});
describe('(on cluster worker)', () => {
beforeEach(() => runAsClusterMaster(false));
// The `cluster.worker` property is normally `undefined` on the master process and set to the
// current `cluster.worker` on worker processes.
const mockClusterWorker = mockProperty(cluster, 'worker');
beforeEach(() => {
runAsClusterMaster(false);
mockClusterWorker(Object.assign(new EventEmitter(), {id: 42}) as cluster.Worker);
});
it('should create the `compileFn()`', () => {
new ClusterWorker(mockLogger, createCompileFnSpy);
startWorker(mockLogger, createCompileFnSpy);
expect(createCompileFnSpy).toHaveBeenCalledWith(jasmine.any(Function));
});
it('should set up `compileFn()` to send `task-completed` messages to master', () => {
new ClusterWorker(mockLogger, createCompileFnSpy);
startWorker(mockLogger, createCompileFnSpy);
const onTaskCompleted: TaskCompletedCallback = createCompileFnSpy.calls.argsFor(0)[0];
onTaskCompleted(null as any, TaskProcessingOutcome.Processed, null);
@ -74,29 +80,9 @@ describe('ClusterWorker', () => {
message: 'error message',
});
});
});
});
describe('run()', () => {
describe(
'(on cluster master)',
() => {/* No tests needed, becasue the constructor would have thrown. */});
describe('(on cluster worker)', () => {
// The `cluster.worker` property is normally `undefined` on the master process and set to the
// current `cluster.Worker` on worker processes.
const mockClusterWorker = mockProperty(cluster, 'worker');
let worker: ClusterWorker;
beforeEach(() => {
runAsClusterMaster(false);
mockClusterWorker(Object.assign(new EventEmitter(), {id: 42}) as cluster.Worker);
worker = new ClusterWorker(mockLogger, createCompileFnSpy);
});
it('should return a promise (that is never resolved)', done => {
const promise = worker.run();
const promise = startWorker(mockLogger, createCompileFnSpy);
expect(promise).toEqual(jasmine.any(Promise));
@ -116,7 +102,7 @@ describe('ClusterWorker', () => {
processDts: true,
} as unknown as Task;
worker.run();
startWorker(mockLogger, createCompileFnSpy);
cluster.worker.emit('message', {type: 'process-task', task: mockTask});
expect(compileFnSpy).toHaveBeenCalledWith(mockTask);
@ -139,7 +125,7 @@ describe('ClusterWorker', () => {
throw err;
});
worker.run();
startWorker(mockLogger, createCompileFnSpy);
err = 'Error string.';
cluster.worker.emit('message', {type: 'process-task', task: mockTask});
@ -151,7 +137,7 @@ describe('ClusterWorker', () => {
});
it('should throw, when an unknown message type is received', () => {
worker.run();
startWorker(mockLogger, createCompileFnSpy);
cluster.worker.emit('message', {type: 'unknown', foo: 'bar'});
expect(compileFnSpy).not.toHaveBeenCalled();
@ -163,4 +149,3 @@ describe('ClusterWorker', () => {
});
});
});
});