gracefully terminate plugin host process without rpc connection

packages/plugin-ext/src/common/rpc-protocol.ts

@@ -61,6 +61,20 @@ export function createProxyIdentifier<T>(identifier: string): ProxyIdentifier<T>
     return new ProxyIdentifier(false, identifier);
 }
 
+export interface ConnectionClosedError extends Error {
+    code: 'RPC_PROTOCOL_CLOSED'
+}
+export namespace ConnectionClosedError {
+    const code: ConnectionClosedError['code'] = 'RPC_PROTOCOL_CLOSED';
+    export function create(message: string = 'connection is closed'): ConnectionClosedError {
+        return Object.assign(new Error(message), { code });
+    }
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    export function is(error: any): error is ConnectionClosedError {
+        return !!error && typeof error === 'object' && 'code' in error && error['code'] === code;
+    }
+}
+
 export class RPCProtocolImpl implements RPCProtocol {
 
     private readonly locals = new Map<string, any>();
@@ -82,7 +96,7 @@ export class RPCProtocolImpl implements RPCProtocol {
         this.toDispose.push(Disposable.create(() => {
             this.proxies.clear();
             for (const reply of this.pendingRPCReplies.values()) {
-                reply.reject(new Error('connection is closed'));
+                reply.reject(ConnectionClosedError.create());
             }
             this.pendingRPCReplies.clear();
         }));
@@ -98,7 +112,7 @@ export class RPCProtocolImpl implements RPCProtocol {
 
     getProxy<T>(proxyId: ProxyIdentifier<T>): T {
         if (this.isDisposed) {
-            throw new Error('connection is closed');
+            throw ConnectionClosedError.create();
         }
         let proxy = this.proxies.get(proxyId.id);
         if (!proxy) {
@@ -110,7 +124,7 @@ export class RPCProtocolImpl implements RPCProtocol {
 
     set<T, R extends T>(identifier: ProxyIdentifier<T>, instance: R): R {
         if (this.isDisposed) {
-            throw new Error('connection is closed');
+            throw ConnectionClosedError.create();
         }
         this.locals.set(identifier.id, instance);
         if (Disposable.is(instance)) {
@@ -135,7 +149,7 @@ export class RPCProtocolImpl implements RPCProtocol {
 
     private remoteCall(proxyId: string, methodName: string, args: any[]): Promise<any> {
         if (this.isDisposed) {
-            return Promise.reject(new Error('connection is closed'));
+            return Promise.reject(ConnectionClosedError.create());
         }
         const cancellationToken: CancellationToken | undefined = args.length && CancellationToken.is(args[args.length - 1]) ? args.pop() : undefined;
         if (cancellationToken && cancellationToken.isCancellationRequested) {
@@ -320,7 +334,7 @@ class RPCMultiplexer implements Disposable {
 
     public send(msg: string): void {
         if (this.toDispose.disposed) {
-            throw new Error('connection is closed');
+            throw ConnectionClosedError.create();
         }
         if (this.messagesToSend.length === 0) {
             if (typeof setImmediate !== 'undefined') {
@@ -460,11 +474,13 @@ function isSerializedObject(obj: any): obj is SerializedObject {
     return obj && obj.$type !== undefined && obj.data !== undefined;
 }
 
-const enum MessageType {
+export const enum MessageType {
     Request = 1,
     Reply = 2,
     ReplyErr = 3,
-    Cancel = 4
+    Cancel = 4,
+    Terminate = 5,
+    Terminated = 6
 }
 
 class CancelMessage {

packages/plugin-ext/src/hosted/node/hosted-plugin-cli-contribution.ts

@@ -18,24 +18,58 @@ import { injectable } from 'inversify';
 import { Argv, Arguments } from 'yargs';
 import { CliContribution } from '@theia/core/lib/node';
 
+let pluginHostTerminateTimeout = 10 * 1000;
+if (process.env.PLUGIN_HOST_TERMINATE_TIMEOUT) {
+    pluginHostTerminateTimeout = Number.parseInt(process.env.PLUGIN_HOST_TERMINATE_TIMEOUT);
+}
+
+let pluginHostStopTimeout = 4 * 1000;
+if (process.env.PLUGIN_HOST_STOP_TIMEOUT) {
+    pluginHostStopTimeout = Number.parseInt(process.env.PLUGIN_HOST_STOP_TIMEOUT);
+}
+
 @injectable()
 export class HostedPluginCliContribution implements CliContribution {
 
     static EXTENSION_TESTS_PATH = 'extensionTestsPath';
+    static PLUGIN_HOST_TERMINATE_TIMEOUT = 'pluginHostTerminateTimeout';
+    static PLUGIN_HOST_STOP_TIMEOUT = 'pluginHostStopTimeout';
 
     protected _extensionTestsPath: string | undefined;
     get extensionTestsPath(): string | undefined {
         return this._extensionTestsPath;
     }
 
+    protected _pluginHostTerminateTimeout = pluginHostTerminateTimeout;
+    get pluginHostTerminateTimeout(): number {
+        return this._pluginHostTerminateTimeout;
+    }
+
+    protected _pluginHostStopTimeout = pluginHostStopTimeout;
+    get pluginHostStopTimeout(): number {
+        return this._pluginHostStopTimeout;
+    }
+
     configure(conf: Argv): void {
         conf.option(HostedPluginCliContribution.EXTENSION_TESTS_PATH, {
             type: 'string'
         });
+        conf.option(HostedPluginCliContribution.PLUGIN_HOST_TERMINATE_TIMEOUT, {
+            type: 'number',
+            default: pluginHostTerminateTimeout,
+            description: 'Timeout in milliseconds to wait for the plugin host process to terminate before killing it. Use 0 for no timeout.'
+        });
+        conf.option(HostedPluginCliContribution.PLUGIN_HOST_STOP_TIMEOUT, {
+            type: 'number',
+            default: pluginHostStopTimeout,
+            description: 'Timeout in milliseconds to wait for the plugin host process to stop internal services. Use 0 for no timeout.'
+        });
     }
 
     setArguments(args: Arguments): void {
         this._extensionTestsPath = args[HostedPluginCliContribution.EXTENSION_TESTS_PATH];
+        this._pluginHostTerminateTimeout = args[HostedPluginCliContribution.PLUGIN_HOST_TERMINATE_TIMEOUT];
+        this._pluginHostStopTimeout = args[HostedPluginCliContribution.PLUGIN_HOST_STOP_TIMEOUT];
     }
 
 }

packages/plugin-ext/src/hosted/node/hosted-plugin-process.ts

@@ -18,13 +18,12 @@ import * as path from 'path';
 import * as cp from 'child_process';
 import { injectable, inject, named } from 'inversify';
 import { ILogger, ConnectionErrorHandler, ContributionProvider, MessageService } from '@theia/core/lib/common';
-import { Emitter } from '@theia/core/lib/common/event';
 import { createIpcEnv } from '@theia/core/lib/node/messaging/ipc-protocol';
 import { HostedPluginClient, ServerPluginRunner, PluginHostEnvironmentVariable, DeployedPlugin } from '../../common/plugin-protocol';
-import { RPCProtocolImpl } from '../../common/rpc-protocol';
-import { MAIN_RPC_CONTEXT } from '../../common/plugin-api-rpc';
+import { MessageType } from '../../common/rpc-protocol';
 import { HostedPluginCliContribution } from './hosted-plugin-cli-contribution';
 import * as psTree from 'ps-tree';
+import { Deferred } from '@theia/core/lib/common/promise-util';
 
 export interface IPCConnectionOptions {
     readonly serverName: string;
@@ -83,7 +82,7 @@ export class HostedPluginProcess implements ServerPluginRunner {
         }
     }
 
-    public terminatePluginServer(): void {
+    async terminatePluginServer(): Promise<void> {
         if (this.childProcess === undefined) {
             return;
         }
@@ -93,34 +92,49 @@ export class HostedPluginProcess implements ServerPluginRunner {
         const cp = this.childProcess;
         this.childProcess = undefined;
 
-        const emitter = new Emitter();
+        const waitForTerminated = new Deferred<void>();
         cp.on('message', message => {
-            emitter.fire(JSON.parse(message));
-        });
-        const rpc = new RPCProtocolImpl({
-            onMessage: emitter.event,
-            send: (m: {}) => {
-                if (cp.send) {
-                    cp.send(JSON.stringify(m));
-                }
+            const msg = JSON.parse(message);
+            if ('type' in msg && msg.type === MessageType.Terminated) {
+                waitForTerminated.resolve();
             }
         });
-        const hostedPluginManager = rpc.getProxy(MAIN_RPC_CONTEXT.HOSTED_PLUGIN_MANAGER_EXT);
-        hostedPluginManager.$stop().then(() => {
-            emitter.dispose();
-            this.killProcessTree(cp.pid);
-        });
+        const stopTimeout = this.cli.pluginHostStopTimeout;
+        cp.send(JSON.stringify({ type: MessageType.Terminate, stopTimeout }));
+
+        const terminateTimeout = this.cli.pluginHostTerminateTimeout;
+        if (terminateTimeout) {
+            await Promise.race([
+                waitForTerminated.promise,
+                new Promise(resolve => setTimeout(resolve, terminateTimeout))
+            ]);
+        } else {
+            await waitForTerminated.promise;
+        }
+
+        this.killProcessTree(cp.pid);
     }
 
     private killProcessTree(parentPid: number): void {
-        psTree(parentPid, (err: Error, childProcesses: Array<psTree.PS>) => {
-            childProcesses.forEach((p: psTree.PS) => {
-                process.kill(parseInt(p.PID));
-            });
-            process.kill(parentPid);
+        psTree(parentPid, (_, childProcesses) => {
+            childProcesses.forEach(childProcess =>
+                this.killProcess(parseInt(childProcess.PID))
+            );
+            this.killProcess(parentPid);
         });
     }
 
+    protected killProcess(pid: number): void {
+        try {
+            process.kill(pid);
+        } catch (e) {
+            if (e && 'code' in e && e.code === 'ESRCH') {
+                return;
+            }
+            this.logger.error(`[${pid}] failed to kill`, e);
+        }
+    }
+
     public runPluginServer(): void {
         if (this.childProcess) {
             this.terminatePluginServer();

packages/plugin-ext/src/hosted/node/plugin-host-rpc.ts

@@ -76,6 +76,10 @@ export class PluginHostRPC {
         );
     }
 
+    async terminate(): Promise<void> {
+        await this.pluginManager.terminate();
+    }
+
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
     initContext(contextPath: string, plugin: Plugin): any {
         const { name, version } = plugin.rawModel;

packages/plugin-ext/src/hosted/node/plugin-host.ts

@@ -15,7 +15,7 @@
  ********************************************************************************/
 
 import { Emitter } from '@theia/core/lib/common/event';
-import { RPCProtocolImpl } from '../../common/rpc-protocol';
+import { RPCProtocolImpl, MessageType, ConnectionClosedError } from '../../common/rpc-protocol';
 import { PluginHostRPC } from './plugin-host-rpc';
 console.log('PLUGIN_HOST(' + process.pid + ') starting instance');
 
@@ -50,6 +50,10 @@ process.on('unhandledRejection', (reason: any, promise: Promise<any>) => {
         if (index >= 0) {
             promise.catch(err => {
                 unhandledPromises.splice(index, 1);
+                if (terminating && (ConnectionClosedError.is(err) || ConnectionClosedError.is(reason))) {
+                    // during termination it is expected that pending rpc requerst are rejected
+                    return;
+                }
                 console.error(`Promise rejection not handled in one second: ${err} , reason: ${reason}`);
                 if (err && err.stack) {
                     console.error(`With stack trace: ${err.stack}`);
@@ -67,19 +71,41 @@ process.on('rejectionHandled', (promise: Promise<any>) => {
     }
 });
 
+let terminating = false;
 const emitter = new Emitter();
 const rpc = new RPCProtocolImpl({
     onMessage: emitter.event,
     send: (m: {}) => {
-        if (process.send) {
+        if (process.send && !terminating) {
             process.send(JSON.stringify(m));
         }
     }
 });
 
-process.on('message', (message: string) => {
+process.on('message', async (message: string) => {
+    if (terminating) {
+        return;
+    }
     try {
-        emitter.fire(JSON.parse(message));
+        const msg = JSON.parse(message);
+        if ('type' in msg && msg.type === MessageType.Terminate) {
+            terminating = true;
+            emitter.dispose();
+            if ('stopTimeout' in msg && typeof msg.stopTimeout === 'number' && msg.stopTimeout) {
+                await Promise.race([
+                    pluginHostRPC.terminate(),
+                    new Promise(resolve => setTimeout(resolve, msg.stopTimeout))
+                ]);
+            } else {
+                await pluginHostRPC.terminate();
+            }
+            rpc.dispose();
+            if (process.send) {
+                process.send(JSON.stringify({ type: MessageType.Terminated }));
+            }
+        } else {
+            emitter.fire(msg);
+        }
     } catch (e) {
         console.error(e);
     }

packages/plugin-ext/src/plugin/command-registry.ts

@@ -85,10 +85,6 @@ export class CommandRegistryImpl implements CommandRegistryExt {
         });
     }
 
-    dispose(): void {
-        throw new Error('Method not implemented.');
-    }
-
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
     $executeCommand<T>(id: string, ...args: any[]): PromiseLike<T | undefined> {
         if (this.handlers.has(id)) {