Add MasterRequest cancellation - fixes
MODIFY
packages/@romejs/cli/cli.ts
packages/@romejs/cli/cli.ts
@@ -313,7 +313,7 @@ export default async function cli() {
}
const res = await client.query({
- command,
+ commandName: command,
commandFlags,
args,
requestFlags,
@@ -364,6 +364,11 @@ export default async function cli() {
break;
}
+ case 'CANCELLED': {
+ process.exit(0);
+ break;
+ }
+
case 'SUCCESS': {
process.exit(0);
break;

MODIFY
packages/@romejs/core/client/Client.ts
packages/@romejs/core/client/Client.ts
@@ -77,6 +77,7 @@ export default class Client {
constructor(opts: ClientOptions) {
this.options = opts;
this.userConfig = loadUserConfig();
+ this.queryCounter = 0;
this.flags = {
...DEFAULT_CLIENT_FLAGS,
@@ -118,6 +119,7 @@ export default class Client {
});
}
+ queryCounter: number;
userConfig: UserConfig;
options: ClientOptions;
flags: ClientFlags;
@@ -303,9 +305,7 @@ export default class Client {
);
// Collect CPU profile
-
// Callback will be called later once it has been collected
-
// Initial async work is just connecting to the processes and setting up handlers
let profileEvents: Array<TraceEvent> = [];
await this.profile(
@@ -375,7 +375,7 @@ export default class Client {
if (bridgeStatus !== undefined) {
const status = await this.query({
silent: true,
- command: 'status',
+ commandName: 'status',
});
if (status.type === 'SUCCESS') {
writer.append(
@@ -405,6 +405,32 @@ export default class Client {
return res;
}
+ cancellableQuery(
+ query: PartialMasterQueryRequest,
+ type?: ClientRequestType,
+ ): {
+ promise: Promise<MasterQueryResponse>;
+ cancel: () => Promise<void>;
+ } {
+ const cancelToken = String(this.queryCounter++);
+
+ return {
+ promise: this.query(
+ {
+ ...query,
+ cancelToken,
+ },
+ type,
+ ),
+ cancel: async () => {
+ const status = this.getBridge();
+ if (status !== undefined) {
+ await status.bridge.cancelQuery.call(cancelToken);
+ }
+ },
+ };
+ }
+
getBridge(): undefined | BridgeStatus {
return this.bridgeStatus;
}

MODIFY
packages/@romejs/core/client/ClientRequest.ts
packages/@romejs/core/client/ClientRequest.ts
@@ -61,7 +61,7 @@ export default class ClientRequest {
}
async initCommand(): Promise<MasterQueryResponse> {
- const localCommand = localCommands.get(this.query.command);
+ const localCommand = localCommands.get(this.query.commandName);
if (this.type === 'master' || localCommand === undefined) {
return this.initFromMaster();

MODIFY
packages/@romejs/core/client/commands/lsp.ts
packages/@romejs/core/client/commands/lsp.ts
@@ -46,7 +46,7 @@ export default createLocalCommand({
await req.client.query(
{
- command: 'lsp',
+ commandName: 'lsp',
},
'master',
);

MODIFY
packages/@romejs/core/client/commands/restart.ts
packages/@romejs/core/client/commands/restart.ts
@@ -19,12 +19,12 @@ export default createLocalCommand({
},
async callback(req: ClientRequest) {
const stopped = await req.client.query({
- command: 'stop',
+ commandName: 'stop',
});
if (stopped.type === 'SUCCESS' && stopped.data === true) {
const started = await req.client.query({
- command: 'start',
+ commandName: 'start',
});
return started.type === 'SUCCESS' && started.data === true;
} else {

MODIFY
packages/@romejs/core/client/commands/run.ts
packages/@romejs/core/client/commands/run.ts
@@ -38,7 +38,7 @@ export default createLocalCommand({
const res = await req.client.query(
{
- command: 'run',
+ commandName: 'run',
args: req.query.args,
},
'master',

MODIFY
packages/@romejs/core/client/commands/status.ts
packages/@romejs/core/client/commands/status.ts
@@ -23,7 +23,7 @@ export default createLocalCommand({
if (bridge) {
const status = await req.client.query(
{
- command: 'status',
+ commandName: 'status',
},
'master',
);

MODIFY
packages/@romejs/core/client/commands/stop.ts
packages/@romejs/core/client/commands/stop.ts
@@ -24,7 +24,7 @@ export default createLocalCommand({
if (bridge) {
const stop = await req.client.query(
{
- command: 'stop',
+ commandName: 'stop',
},
'master',
);

MODIFY
packages/@romejs/core/client/review.ts
packages/@romejs/core/client/review.ts
@@ -149,14 +149,14 @@ async function check(
// Execute action
const actionRes = await client.query(
{
- command: action.command,
+ commandName: action.command,
args: action.args,
commandFlags: action.commandFlags,
requestFlags,
},
'master',
);
- if (actionRes.type === 'ERROR' || actionRes.type === 'INVALID_REQUEST') {
+ if (actionRes.type !== 'DIAGNOSTICS' && actionRes.type !== 'SUCCESS') {
return actionRes;
}

MODIFY
packages/@romejs/core/common/bridges/MasterBridge.ts
packages/@romejs/core/common/bridges/MasterBridge.ts
@@ -26,14 +26,15 @@ export type MasterQueryRequest = {
silent: boolean;
noData: boolean;
terminateWhenIdle: boolean;
+ cancelToken?: string;
};
export type PartialMasterQueryRequest = Partial<Omit<
MasterQueryRequest,
- 'requestFlags'
+ 'requestFlags' | 'commandName'
>> & {
requestFlags?: Partial<ClientRequestFlags>;
- command: string;
+ commandName: string;
};
export type MasterQueryResponseSuccess = {
@@ -63,10 +64,15 @@ export type MasterQueryResponseInvalid = {
showHelp: boolean;
};
+export type MasterQueryResponseCancelled = {
+ type: 'CANCELLED';
+};
+
export type MasterQueryResponse =
| MasterQueryResponseInvalid
| MasterQueryResponseSuccess
| MasterQueryResponseError
+ | MasterQueryResponseCancelled
| MasterQueryResponseDiagnostics;
export type ProfilingStartData = {
@@ -141,6 +147,11 @@ export default class MasterBridge extends Bridge {
direction: 'server<-client',
});
+ cancelQuery = this.createEvent<string, void>({
+ name: 'cancel',
+ direction: 'server<-client',
+ });
+
profilingGetWorkers = this.createEvent<void, Array<number>>({
name: 'profiling.getWorkers',
direction: 'server<-client',

MODIFY
packages/@romejs/core/master/Master.ts
packages/@romejs/core/master/Master.ts
@@ -30,6 +30,7 @@ import {ConsumePath, consume} from '@romejs/consume';
import {Event, EventSubscription} from '@romejs/events';
import MasterRequest, {
EMPTY_SUCCESS_RESPONSE,
+ MasterRequestCancelled,
MasterRequestInvalid,
} from './MasterRequest';
import ProjectManager from './project/ProjectManager';
@@ -74,6 +75,7 @@ export type MasterClient = {
bridge: MasterBridge;
flags: ClientFlags;
version: string;
+ requestsInFlight: Set<MasterRequest>;
};
export type MasterOptions = {
@@ -395,8 +397,14 @@ export default class Master {
}
async end() {
+ // Cancel all queries in flight
+ for (const client of this.connectedClients) {
+ for (const req of client.requestsInFlight) {
+ req.cancel();
+ }
+ }
+
// We should remove everything that has an external dependency like a socket or process
- // TODO terminate all queries in flight
await this.endEvent.callOptional();
this.workerManager.end();
this.memoryFs.unwatchAll();
@@ -406,7 +414,6 @@ export default class Master {
let profiler: undefined | Profiler;
// If we aren't a dedicated process then we should only expect a single connection
-
// and when that ends. End the Master.
if (this.options.dedicated === false) {
bridge.endEvent.subscribe(() => {
@@ -466,6 +473,14 @@ export default class Master {
return await this.handleRequest(client, request);
});
+ bridge.cancelQuery.subscribe(async (token) => {
+ for (const req of client.requestsInFlight) {
+ if (req.query.cancelToken === token) {
+ req.cancel();
+ }
+ }
+ });
+
await this.clientStartEvent.callOptional(client);
}
@@ -554,12 +569,12 @@ export default class Master {
reporter,
flags,
version,
+ requestsInFlight: new Set(),
};
this.connectedClients.add(client);
// When enableWorkerLogs is called we setup subscriptions to the worker logs
-
// Logs are never transported from workers to the master unless there is a subscription
let subscribedWorkers = false;
bridge.enableWorkerLogs.subscribe(() => {
@@ -594,6 +609,11 @@ export default class Master {
});
bridge.endEvent.subscribe(() => {
+ // Cancel any requests still in flight
+ for (const req of client.requestsInFlight) {
+ req.cancel();
+ }
+
this.connectedClients.delete(client);
this.connectedClientsListeningForLogs.delete(client);
this.connectedReporters.removeStream(errStream);
@@ -648,7 +668,7 @@ export default class Master {
};
const query: MasterQueryRequest = {
- commandName: partialQuery.command,
+ commandName: partialQuery.commandName,
args: partialQuery.args === undefined ? [] : partialQuery.args,
noData: partialQuery.noData === true,
requestFlags,
@@ -657,6 +677,7 @@ export default class Master {
commandFlags: partialQuery.commandFlags === undefined
? {}
: partialQuery.commandFlags,
+ cancelToken: partialQuery.cancelToken,
};
const {bridge} = client;
@@ -868,6 +889,10 @@ export default class Master {
message: err.message,
stack: err.stack,
};
+ } else if (err instanceof MasterRequestCancelled) {
+ return {
+ type: 'CANCELLED',
+ };
} else if (err instanceof MasterRequestInvalid) {
return {
type: 'INVALID_REQUEST',

MODIFY
packages/@romejs/core/master/MasterRequest.ts
packages/@romejs/core/master/MasterRequest.ts
@@ -144,12 +144,22 @@ function hash(val: JSONObject): string {
: crypto.createHash('sha256').update(JSON.stringify(val)).digest('hex');
}
+export class MasterRequestCancelled extends Error {
+ constructor() {
+ super(
+ 'MasterRequest has been cancelled. This error is meant to be seen by Master',
+ );
+ }
+}
+
export default class MasterRequest {
constructor(opts: MasterRequestOptions) {
this.query = opts.query;
this.master = opts.master;
this.bridge = opts.client.bridge;
this.reporter = opts.client.reporter;
+ this.cancelled = false;
+ this.toredown = false;
this.markerEvent = new Event({
name: 'MasterRequest.marker',
onError: this.master.onFatalErrorBound,
@@ -163,11 +173,11 @@ export default class MasterRequest {
this.id = requestIdCounter++;
this.markers = [];
this.start = Date.now();
-
this.normalizedCommandFlags = {
flags: {},
defaultFlags: {},
};
+ this.client.requestsInFlight.add(this);
}
start: number;
@@ -181,6 +191,8 @@ export default class MasterRequest {
endEvent: Event<undefined | MasterQueryResponse, void>;
normalizedCommandFlags: NormalizedCommandFlags;
markers: Array<MasterMarker>;
+ cancelled: boolean;
+ toredown: boolean;
async init() {
if (this.query.requestFlags.collectMarkers) {
@@ -192,9 +204,29 @@ export default class MasterRequest {
await this.master.handleRequestStart(this);
}
+ checkCancelled() {
+ if (this.cancelled) {
+ throw new MasterRequestCancelled();
+ }
+ }
+
+ cancel() {
+ this.cancelled = true;
+ this.teardown({
+ type: 'CANCELLED',
+ });
+ }
+
teardown(
res: undefined | MasterQueryResponse,
): undefined | MasterQueryResponse {
+ if (this.toredown) {
+ return;
+ }
+
+ this.toredown = true;
+ this.client.requestsInFlight.delete(this);
+
// Output timing information
if (this.query.requestFlags.timing) {
const end = Date.now();
@@ -457,6 +489,8 @@ export default class MasterRequest {
projects: Set<ProjectDefinition>;
resolvedArgs: ResolvedArgs;
}> {
+ this.checkCancelled();
+
const projects: Set<ProjectDefinition> = new Set();
const rawArgs = overrideArgs === undefined ? this.query.args : overrideArgs;
const resolvedArgs: ResolvedArgs = [];
@@ -517,6 +551,8 @@ export default class MasterRequest {
initial: boolean,
) => Promise<void>,
): Promise<EventSubscription> {
+ this.checkCancelled();
+
// Everything needs to be relative to this
const {resolvedArgs} = await this.resolveFilesFromArgs();
@@ -586,12 +622,19 @@ export default class MasterRequest {
);
const fileChangeEvent = this.master.fileChangeEvent.subscribe(onChange);
+ this.endEvent.subscribe(() => {
+ evictSubscription.unsubscribe();
+ fileChangeEvent.unsubscribe();
+ });
+
return mergeEventSubscriptions([evictSubscription, fileChangeEvent]);
}
async getFilesFromArgs(
opts: MasterRequestGetFilesOptions = {},
): Promise<MasterRequestGetFilesResult> {
+ this.checkCancelled();
+
const {master} = this;
const {configCategory, ignoreProjectIgnore} = opts;
const {projects, resolvedArgs} = await this.resolveFilesFromArgs(opts.args);
@@ -788,6 +831,8 @@ export default class MasterRequest {
path: AbsoluteFilePath,
content: string,
): Promise<void> {
+ this.checkCancelled();
+
await this.wrapRequestDiagnostic(
'updateBuffer',
path,
@@ -800,6 +845,8 @@ export default class MasterRequest {
path: AbsoluteFilePath,
opts: WorkerParseOptions,
): Promise<Program> {
+ this.checkCancelled();
+
return this.wrapRequestDiagnostic(
'parse',
path,
@@ -811,6 +858,8 @@ export default class MasterRequest {
path: AbsoluteFilePath,
optionsWithoutModSigs: Omit<WorkerLintOptions, 'prefetchedModuleSignatures'>,
): Promise<WorkerLintResult> {
+ this.checkCancelled();
+
const {cache} = this.master;
const cacheEntry = await cache.get(path);
@@ -852,6 +901,8 @@ export default class MasterRequest {
path: AbsoluteFilePath,
parseOptions: WorkerParseOptions,
): Promise<undefined | WorkerFormatResult> {
+ this.checkCancelled();
+
return await this.wrapRequestDiagnostic(
'format',
path,
@@ -865,6 +916,8 @@ export default class MasterRequest {
options: WorkerCompilerOptions,
parseOptions: WorkerParseOptions,
): Promise<WorkerCompileResult> {
+ this.checkCancelled();
+
const {cache} = this.master;
// Create a cache key comprised of the stage and hash of the options
@@ -917,6 +970,8 @@ export default class MasterRequest {
path: AbsoluteFilePath,
parseOptions: WorkerParseOptions,
): Promise<WorkerAnalyzeDependencyResult> {
+ this.checkCancelled();
+
const {cache} = this.master;
const cacheEntry = await cache.get(path);
@@ -949,6 +1004,8 @@ export default class MasterRequest {
path: AbsoluteFilePath,
parseOptions: WorkerParseOptions,
): Promise<ModuleSignature> {
+ this.checkCancelled();
+
const {cache} = this.master;
const cacheEntry = await cache.get(path);
@@ -973,6 +1030,8 @@ export default class MasterRequest {
async maybePrefetchModuleSignatures(
path: AbsoluteFilePath,
): Promise<PrefetchedModuleSignatures> {
+ this.checkCancelled();
+
const {projectManager} = this.master;
const prefetchedModuleSignatures: PrefetchedModuleSignatures = {};

MODIFY
packages/@romejs/events/Event.ts
packages/@romejs/events/Event.ts
@@ -190,16 +190,14 @@ export default class Event<Param, Ret = void> {
if (this.subscriptions.has(callback)) {
this.subscriptions.delete(callback);
this.onSubscriptionChange();
- return undefined;
+ return;
}
// If this callback was the root subscription, then set it to the next one
if (callback === this.rootSubscription) {
this.rootSubscription = Array.from(this.subscriptions)[0];
this.onSubscriptionChange();
- return undefined;
+ return;
}
-
- throw new Error('Not a current subscription');
}
}

Add MasterRequest cancellation - fixes #324