Split child and parent wrappers

code-asher · code-asher · commit d55e06936bab · 2020-11-18T12:28:42.000-06:00
I think having them combined and relying on if statements was getting
confusing especially if we want to add additional messages with
different payloads (which will soon be the case).
diff --git a/src/node/entry.ts b/src/node/entry.ts
@@ -19,7 +19,7 @@ import { coderCloudBind } from "./coder-cloud"
 import { commit, version } from "./constants"
 import { register } from "./routes"
 import { humanPath, isFile, open } from "./util"
-import { ipcMain, WrapperProcess } from "./wrapper"
+import { isChild, wrapper } from "./wrapper"
 
 export const runVsCodeCli = (args: DefaultedArgs): void => {
   logger.debug("forking vs code cli...")
@@ -137,7 +137,7 @@ const main = async (args: DefaultedArgs): Promise<void> => {
       logger.info("  - Connected to cloud agent")
     } catch (err) {
       logger.error(err.message)
-      ipcMain.exit(1)
+      wrapper.exit(1)
     }
   }
 
@@ -161,9 +161,9 @@ async function entry(): Promise<void> {
   // There's no need to check flags like --help or to spawn in an existing
   // instance for the child process because these would have already happened in
   // the parent and the child wouldn't have been spawned.
-  if (ipcMain.isChild) {
-    await ipcMain.handshake()
-    ipcMain.preventExit()
+  if (isChild(wrapper)) {
+    await wrapper.handshake()
+    wrapper.preventExit()
     return main(args)
   }
 
@@ -201,11 +201,10 @@ async function entry(): Promise<void> {
     return openInExistingInstance(args, socketPath)
   }
 
-  const wrapper = new WrapperProcess(require("../../package.json").version)
   return wrapper.start()
 }
 
 entry().catch((error) => {
   logger.error(error.message)
-  ipcMain.exit(error)
+  wrapper.exit(error)
 })
diff --git a/src/node/vscode.ts b/src/node/vscode.ts
@@ -8,7 +8,7 @@ import { rootPath } from "./constants"
 import { settings } from "./settings"
 import { SocketProxyProvider } from "./socket"
 import { isFile } from "./util"
-import { ipcMain } from "./wrapper"
+import { wrapper } from "./wrapper"
 
 export class VscodeProvider {
   public readonly serverRootPath: string
@@ -20,7 +20,7 @@ export class VscodeProvider {
   public constructor() {
     this.vsRootPath = path.resolve(rootPath, "lib/vscode")
     this.serverRootPath = path.join(this.vsRootPath, "out/vs/server")
-    ipcMain.onDispose(() => this.dispose())
+    wrapper.onDispose(() => this.dispose())
   }
 
   public async dispose(): Promise<void> {
diff --git a/src/node/wrapper.ts b/src/node/wrapper.ts

-Original file line number
+Diff line change
@@ @@ -1,4 +1,4 @@ @@
 -import { field, logger } from "@coder/logger"
 +import { Logger, field, logger } from "@coder/logger"
 import * as cp from "child_process"
 import * as path from "path"
 import * as rfs from "rotating-file-stream"
   version: string
+}
 -export type Message = RelaunchMessage | HandshakeMessage
 +type Message = RelaunchMessage | HandshakeMessage
 -export class ProcessError extends Error {
 +class ProcessError extends Error {
   public constructor(message: string, public readonly code: number | undefined) {
     super(message)
     this.name = this.constructor.name
+}
 /**
 - * Allows the wrapper and inner processes to communicate.
 + * Wrapper around a process that tries to gracefully exit when a process exits
 + * and provides a way to prevent `process.exit`.
  */
 -export class IpcMain {
 -  private readonly _onMessage = new Emitter<Message>()
 -  public readonly onMessage = this._onMessage.event
 -  private readonly _onDispose = new Emitter<NodeJS.Signals | undefined>()
 +abstract class Process {
 +  /**
 +   * Emit this to trigger a graceful exit.
 +   */
 +  protected readonly _onDispose = new Emitter<NodeJS.Signals | undefined>()
++
 +  /**
 +   * Emitted when the process is about to be disposed.
 +   */
   public readonly onDispose = this._onDispose.event
 -  public readonly processExit: (code?: number) => never = process.exit
 -  public constructor(private readonly parentPid?: number) {
 +  /**
 +   * Uniquely named logger for the process.
 +   */
 +  public abstract logger: Logger
++
 +  public constructor() {
     process.on("SIGINT", () => this._onDispose.emit("SIGINT"))
     process.on("SIGTERM", () => this._onDispose.emit("SIGTERM"))
     process.on("exit", () => this._onDispose.emit(undefined))
       // Remove listeners to avoid possibly triggering disposal again.
       process.removeAllListeners()
 -      // Try waiting for other handlers run first then exit.
 -      logger.debug(`${parentPid ? "inner process" : "wrapper"} ${process.pid} disposing`, field("code", signal))
 +      // Try waiting for other handlers to run first then exit.
 +      this.logger.debug("disposing", field("code", signal))
       wait.then(() => this.exit(0))
       setTimeout(() => this.exit(0), 5000)
     })
+-
 -    // Kill the inner process if the parent dies. This is for the case where the
 -    // parent process is forcefully terminated and cannot clean up.
 -    if (parentPid) {
 -      setInterval(() => {
 -        try {
 -          // process.kill throws an exception if the process doesn't exist.
 -          process.kill(parentPid, 0)
 -        } catch (_) {
 -          // Consider this an error since it should have been able to clean up
 -          // the child process unless it was forcefully killed.
 -          logger.error(`parent process ${parentPid} died`)
 -          this._onDispose.emit(undefined)
 -        }
 -      }, 5000)
 -    }
+  }
   /**
 -   * Ensure we control when the process exits.
 +   * Ensure control over when the process exits.
    */
   public preventExit(): void {
 -    process.exit = function (code?: number) {
 -      logger.warn(`process.exit() was prevented: ${code || "unknown code"}.`)
 -    } as (code?: number) => never
 +    ;(process.exit as any) = (code?: number) => {
 +      this.logger.warn(`process.exit() was prevented: ${code || "unknown code"}.`)
 +    }
+  }
 -  public get isChild(): boolean {
 -    return typeof this.parentPid !== "undefined"
 -  }
 +  private readonly processExit: (code?: number) => never = process.exit
 +  /**
 +   * Will always exit even if normal exit is being prevented.
 +   */
   public exit(error?: number | ProcessError): never {
     if (error && typeof error !== "number") {
       this.processExit(typeof error.code === "number" ? error.code : 1)
     } else {
       this.processExit(error)
+    }
+  }
 +}
 -  public handshake(child?: cp.ChildProcess): Promise<void> {
 -    return new Promise((resolve, reject) => {
 -      const target = child || process
 +/**
 + * Child process that will clean up after itself if the parent goes away and can
 + * perform a handshake with the parent and ask it to relaunch.
 + */
 +class ChildProcess extends Process {
 +  public logger = logger.named(`child:${process.pid}`)
++
 +  public constructor(private readonly parentPid: number) {
 +    super()
++
 +    // Kill the inner process if the parent dies. This is for the case where the
 +    // parent process is forcefully terminated and cannot clean up.
 +    setInterval(() => {
 +      try {
 +        // process.kill throws an exception if the process doesn't exist.
 +        process.kill(this.parentPid, 0)
 +      } catch (_) {
 +        // Consider this an error since it should have been able to clean up
 +        // the child process unless it was forcefully killed.
 +        this.logger.error(`parent process ${parentPid} died`)
 +        this._onDispose.emit(undefined)
 +      }
 +    }, 5000)
 +  }
++
 +  /**
 +   * Initiate the handshake and wait for a response from the parent.
 +   */
 +  public handshake(): Promise<void> {
 +    return new Promise((resolve) => {
       const onMessage = (message: Message): void => {
 -        logger.debug(
 -          `${child ? "wrapper" : "inner process"} ${process.pid} received message from ${
 -            child ? child.pid : this.parentPid
 -          }`,
 -          field("message", message),
 -        )
 +        logger.debug(`received message from ${this.parentPid}`, field("message", message))
         if (message.type === "handshake") {
 -          target.removeListener("message", onMessage)
 -          target.on("message", (msg) => this._onMessage.emit(msg))
 -          // The wrapper responds once the inner process starts the handshake.
 -          if (child) {
 -            if (!target.send) {
 -              throw new Error("child not spawned with IPC")
 -            }
 -            target.send({ type: "handshake" })
 -          }
 +          process.removeListener("message", onMessage)
           resolve()
+        }
+      }
 -      target.on("message", onMessage)
 -      if (child) {
 -        child.once("error", reject)
 -        child.once("exit", (code) => {
 -          reject(new ProcessError(`Unexpected exit with code ${code}`, code !== null ? code : undefined))
 -        })
 -      } else {
 -        // The inner process initiates the handshake.
 -        this.send({ type: "handshake" })
 -      }
 +      // Initiate the handshake and wait for the reply.
 +      process.on("message", onMessage)
 +      this.send({ type: "handshake" })
     })
+  }
 +  /**
 +   * Notify the parent process that it should relaunch the child.
 +   */
   public relaunch(version: string): void {
     this.send({ type: "relaunch", version })
+  }
 +  /**
 +   * Send a message to the parent.
 +   */
   private send(message: Message): void {
     if (!process.send) {
       throw new Error("not spawned with IPC")
+  }
+}
 -/**
 - * Channel for communication between the child and parent processes.
 - */
 -export const ipcMain = new IpcMain(
 -  typeof process.env.CODE_SERVER_PARENT_PID !== "undefined" ? parseInt(process.env.CODE_SERVER_PARENT_PID) : undefined,
 -)
+-
 export interface WrapperOptions {
   maxMemory?: number
   nodeOptions?: string
+}
 /**
 - * Provides a way to wrap a process for the purpose of updating the running
 - * instance.
 + * Parent process wrapper that spawns the child process and performs a handshake
 + * with it. Will relaunch the child if it receives a SIGUSR1 or is asked to by
 + * the child. If the child otherwise exits the parent will also exit.
  */
 -export class WrapperProcess {
 -  private process?: cp.ChildProcess
 +export class ParentProcess extends Process {
 +  public logger = logger.named(`parent:${process.pid}`)
++
 +  private child?: cp.ChildProcess
   private started?: Promise<void>
   private readonly logStdoutStream: rfs.RotatingFileStream
   private readonly logStderrStream: rfs.RotatingFileStream
 +  protected readonly _onChildMessage = new Emitter<Message>()
 +  protected readonly onChildMessage = this._onChildMessage.event
++
   public constructor(private currentVersion: string, private readonly options?: WrapperOptions) {
 +    super()
++
     const opts = {
       size: "10M",
       maxFiles: 10,
+    }
     this.logStdoutStream = rfs.createStream(path.join(paths.data, "coder-logs", "code-server-stdout.log"), opts)
     this.logStderrStream = rfs.createStream(path.join(paths.data, "coder-logs", "code-server-stderr.log"), opts)
 -    ipcMain.onDispose(() => {
 +    this.onDispose(() => {
       this.disposeChild()
     })
 -    ipcMain.onMessage((message) => {
 +    this.onChildMessage((message) => {
       switch (message.type) {
         case "relaunch":
 -          logger.info(`Relaunching: ${this.currentVersion} -> ${message.version}`)
 +          this.logger.info(`Relaunching: ${this.currentVersion} -> ${message.version}`)
           this.currentVersion = message.version
           this.relaunch()
           break
         default:
 -          logger.error(`Unrecognized message ${message}`)
 +          this.logger.error(`Unrecognized message ${message}`)
           break
+      }
     })
+  }
   private disposeChild(): void {
     this.started = undefined
 -    if (this.process) {
 -      this.process.removeAllListeners()
 -      this.process.kill()
 +    if (this.child) {
 +      this.child.removeAllListeners()
 +      this.child.kill()
+    }
+  }
     try {
       await this.start()
     } catch (error) {
 -      logger.error(error.message)
 -      ipcMain.exit(typeof error.code === "number" ? error.code : 1)
 +      this.logger.error(error.message)
 +      this.exit(typeof error.code === "number" ? error.code : 1)
+    }
+  }
   public start(): Promise<void> {
     // If we have a process then we've already bound this.
 -    if (!this.process) {
 +    if (!this.child) {
       process.on("SIGUSR1", async () => {
 -        logger.info("Received SIGUSR1; hotswapping")
 +        this.logger.info("Received SIGUSR1; hotswapping")
         this.relaunch()
       })
+    }
   private async _start(): Promise<void> {
     const child = this.spawn()
 -    this.process = child
 +    this.child = child
     // Log both to stdout and to the log directory.
     if (child.stdout) {
       child.stderr.pipe(process.stderr)
+    }
 -    logger.debug(`spawned inner process ${child.pid}`)
 +    this.logger.debug(`spawned inner process ${child.pid}`)
 -    await ipcMain.handshake(child)
 +    await this.handshake(child)
     child.once("exit", (code) => {
 -      logger.debug(`inner process ${child.pid} exited unexpectedly`)
 -      ipcMain.exit(code || 0)
 +      this.logger.debug(`inner process ${child.pid} exited unexpectedly`)
 +      this.exit(code || 0)
     })
+  }
       stdio: ["ipc"],
     })
+  }
++
 +  /**
 +   * Wait for a handshake from the child then reply.
 +   */
 +  private handshake(child: cp.ChildProcess): Promise<void> {
 +    return new Promise((resolve, reject) => {
 +      const onMessage = (message: Message): void => {
 +        logger.debug(`received message from ${child.pid}`, field("message", message))
 +        if (message.type === "handshake") {
 +          child.removeListener("message", onMessage)
 +          child.on("message", (msg) => this._onChildMessage.emit(msg))
 +          child.send({ type: "handshake" })
 +          resolve()
 +        }
 +      }
 +      child.on("message", onMessage)
 +      child.once("error", reject)
 +      child.once("exit", (code) => {
 +        reject(new ProcessError(`Unexpected exit with code ${code}`, code !== null ? code : undefined))
 +      })
 +    })
 +  }
 +}
++
 +/**
 + * Process wrapper.
 + */
 +export const wrapper =
 +  typeof process.env.CODE_SERVER_PARENT_PID !== "undefined"
 +    ? new ChildProcess(parseInt(process.env.CODE_SERVER_PARENT_PID))
 +    : new ParentProcess(require("../../package.json").version)
++
 +export function isChild(proc: ChildProcess | ParentProcess): proc is ChildProcess {
 +  return proc instanceof ChildProcess
+}
 // It's possible that the pipe has closed (for example if you run code-server
 // --version | head -1). Assume that means we're done.
 if (!process.stdout.isTTY) {
 -  process.stdout.on("error", () => ipcMain.exit())
 +  process.stdout.on("error", () => wrapper.exit())
+}
 // Don't let uncaught exceptions crash the process.
 process.on("uncaughtException", (error) => {
 -  logger.error(`Uncaught exception: ${error.message}`)
 +  wrapper.logger.error(`Uncaught exception: ${error.message}`)
   if (typeof error.stack !== "undefined") {
 -    logger.error(error.stack)
 +    wrapper.logger.error(error.stack)
+  }
 })