OCram85/code-server
Archived
1
0
Fork 0
Code Pull Requests 2 Activity

Merge pull request #2334 from cdr/wrappers

Browse Source 
Separate process wrappers and pass arguments
This commit is contained in:
Asher 2020-11-19 10:28:54 -06:00 committed by GitHub
commit 4380356e0c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 242 additions and 191 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
src/node/entry.ts
View File

@ -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)
}
}
@ -154,19 +154,22 @@ const main = async (args: DefaultedArgs): Promise<void> => {
}
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. We also get the
// arguments from the parent so we don't have to parse twice and to account
// for environment manipulation (like how PASSWORD gets removed to avoid
// leaking to child processes).
if (isChild(wrapper)) {
const args = await wrapper.handshake()
wrapper.preventExit()
return main(args)
}
const cliArgs = parse(process.argv.slice(2))
const configArgs = await readConfigFile(cliArgs.config)
const args = await setDefaults(cliArgs, configArgs)
// 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()
return main(args)
}
if (args.help) {
console.log("code-server", version, commit)
console.log("")
@ -201,11 +204,10 @@ async function entry(): Promise<void> {
return openInExistingInstance(args, socketPath)
}
const wrapper = new WrapperProcess(require("../../package.json").version)
return wrapper.start()
return wrapper.start(args)
}
entry().catch((error) => {
logger.error(error.message)
ipcMain.exit(error)
wrapper.exit(error)
})

68
src/node/vscode.ts
View File

@ -1,4 +1,4 @@
import { field, logger } from "@coder/logger"
import { logger } from "@coder/logger"
import * as cp from "child_process"
import * as net from "net"
import * as path from "path"
@ -8,19 +8,18 @@ import { rootPath } from "./constants"
import { settings } from "./settings"
import { SocketProxyProvider } from "./socket"
import { isFile } from "./util"
import { ipcMain } from "./wrapper"
import { onMessage, wrapper } from "./wrapper"
export class VscodeProvider {
public readonly serverRootPath: string
public readonly vsRootPath: string
private _vscode?: Promise<cp.ChildProcess>
private timeoutInterval = 10000 // 10s, matches VS Code's timeouts.
private readonly socketProvider = new SocketProxyProvider()
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> {
@ -69,10 +68,13 @@ export class VscodeProvider {
vscode,
)
const message = await this.onMessage(vscode, (message): message is ipc.OptionsMessage => {
// There can be parallel initializations so wait for the right ID.
return message.type === "options" && message.id === id
})
const message = await onMessage<ipc.VscodeMessage, ipc.OptionsMessage>(
vscode,
(message): message is ipc.OptionsMessage => {
// There can be parallel initializations so wait for the right ID.
return message.type === "options" && message.id === id
},
)
return message.options
}
@ -104,61 +106,13 @@ export class VscodeProvider {
dispose()
})
this._vscode = this.onMessage(vscode, (message): message is ipc.ReadyMessage => {
this._vscode = onMessage<ipc.VscodeMessage, ipc.ReadyMessage>(vscode, (message): message is ipc.ReadyMessage => {
return message.type === "ready"
}).then(() => vscode)
return this._vscode
}
/**
* Listen to a single message from a process. Reject if the process errors,
* exits, or times out.
*
* `fn` is a function that determines whether the message is the one we're
* waiting for.
*/
private onMessage<T extends ipc.VscodeMessage>(
proc: cp.ChildProcess,
fn: (message: ipc.VscodeMessage) => message is T,
): Promise<T> {
return new Promise((resolve, reject) => {
const cleanup = () => {
proc.off("error", onError)
proc.off("exit", onExit)
proc.off("message", onMessage)
clearTimeout(timeout)
}
const timeout = setTimeout(() => {
cleanup()
reject(new Error("timed out"))
}, this.timeoutInterval)
const onError = (error: Error) => {
cleanup()
reject(error)
}
const onExit = (code: number | null) => {
cleanup()
reject(new Error(`VS Code exited unexpectedly with code ${code}`))
}
const onMessage = (message: ipc.VscodeMessage) => {
logger.trace("got message from vscode", field("message", message))
if (fn(message)) {
cleanup()
resolve(message)
}
}
proc.on("message", onMessage)
proc.on("error", onError)
proc.on("exit", onExit)
})
}
/**
* VS Code expects a raw socket. It will handle all the web socket frames.
*/

335
src/node/wrapper.ts
View File

@ -1,11 +1,70 @@
import { field, logger } from "@coder/logger"
import { field, Logger, logger } from "@coder/logger"
import * as cp from "child_process"
import * as path from "path"
import * as rfs from "rotating-file-stream"
import { Emitter } from "../common/emitter"
import { DefaultedArgs } from "./cli"
import { paths } from "./util"
interface HandshakeMessage {
const timeoutInterval = 10000 // 10s, matches VS Code's timeouts.
/**
* Listen to a single message from a process. Reject if the process errors,
* exits, or times out.
*
* `fn` is a function that determines whether the message is the one we're
* waiting for.
*/
export function onMessage<M, T extends M>(
proc: cp.ChildProcess | NodeJS.Process,
fn: (message: M) => message is T,
customLogger?: Logger,
): Promise<T> {
return new Promise((resolve, reject) => {
const cleanup = () => {
proc.off("error", onError)
proc.off("exit", onExit)
proc.off("message", onMessage)
clearTimeout(timeout)
}
const timeout = setTimeout(() => {
cleanup()
reject(new Error("timed out"))
}, timeoutInterval)
const onError = (error: Error) => {
cleanup()
reject(error)
}
const onExit = (code: number) => {
cleanup()
reject(new Error(`exited unexpectedly with code ${code}`))
}
const onMessage = (message: M) => {
;(customLogger || logger).trace("got message", field("message", message))
if (fn(message)) {
cleanup()
resolve(message)
}
}
proc.on("message", onMessage)
// NodeJS.Process doesn't have `error` but binding anyway shouldn't break
// anything. It does have `exit` but the types aren't working.
;(proc as cp.ChildProcess).on("error", onError)
;(proc as cp.ChildProcess).on("exit", onExit)
})
}
interface ParentHandshakeMessage {
type: "handshake"
args: DefaultedArgs
}
interface ChildHandshakeMessage {
type: "handshake"
}
@ -14,9 +73,10 @@ interface RelaunchMessage {
version: string
}
export type Message = RelaunchMessage | HandshakeMessage
type ChildMessage = RelaunchMessage | ChildHandshakeMessage
type ParentMessage = ParentHandshakeMessage
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
@ -25,16 +85,26 @@ export class ProcessError extends Error {
}
/**
* 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>()
public readonly onDispose = this._onDispose.event
public readonly processExit: (code?: number) => never = process.exit
abstract class Process {
/**
* Emit this to trigger a graceful exit.
*/
protected readonly _onDispose = new Emitter<NodeJS.Signals | undefined>()
public constructor(private readonly parentPid?: number) {
/**
* Emitted when the process is about to be disposed.
*/
public readonly onDispose = this._onDispose.event
/**
* 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))
@ -43,42 +113,27 @@ export class IpcMain {
// 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)
@ -86,48 +141,59 @@ export class IpcMain {
this.processExit(error)
}
}
}
public handshake(child?: cp.ChildProcess): Promise<void> {
return new Promise((resolve, reject) => {
const target = child || process
const onMessage = (message: Message): void => {
logger.debug(
`${child ? "wrapper" : "inner process"} ${process.pid} received message from ${
child ? child.pid : 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" })
}
resolve()
}
/**
* 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)
}
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" })
}
})
}, 5000)
}
/**
* Initiate the handshake and wait for a response from the parent.
*/
public async handshake(): Promise<DefaultedArgs> {
this.send({ type: "handshake" })
const message = await onMessage<ParentMessage, ParentHandshakeMessage>(
process,
(message): message is ParentHandshakeMessage => {
return message.type === "handshake"
},
this.logger,
)
return message.args
}
/**
* Notify the parent process that it should relaunch the child.
*/
public relaunch(version: string): void {
this.send({ type: "relaunch", version })
}
private send(message: Message): void {
/**
* Send a message to the parent.
*/
private send(message: ChildMessage): void {
if (!process.send) {
throw new Error("not spawned with IPC")
}
@ -136,28 +202,31 @@ export class IpcMain {
}
/**
* Channel for communication between the child and parent processes.
* 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 const ipcMain = new IpcMain(
typeof process.env.CODE_SERVER_PARENT_PID !== "undefined" ? parseInt(process.env.CODE_SERVER_PARENT_PID) : undefined,
)
export class ParentProcess extends Process {
public logger = logger.named(`parent:${process.pid}`)
export interface WrapperOptions {
maxMemory?: number
nodeOptions?: string
}
/**
* Provides a way to wrap a process for the purpose of updating the running
* instance.
*/
export class WrapperProcess {
private process?: cp.ChildProcess
private child?: cp.ChildProcess
private started?: Promise<void>
private readonly logStdoutStream: rfs.RotatingFileStream
private readonly logStderrStream: rfs.RotatingFileStream
public constructor(private currentVersion: string, private readonly options?: WrapperOptions) {
protected readonly _onChildMessage = new Emitter<ChildMessage>()
protected readonly onChildMessage = this._onChildMessage.event
private args?: DefaultedArgs
public constructor(private currentVersion: string) {
super()
process.on("SIGUSR1", async () => {
this.logger.info("Received SIGUSR1; hotswapping")
this.relaunch()
})
const opts = {
size: "10M",
maxFiles: 10,
@ -165,19 +234,19 @@ export class WrapperProcess {
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
}
})
@ -185,30 +254,26 @@ export class WrapperProcess {
private disposeChild(): void {
this.started = undefined
if (this.process) {
this.process.removeAllListeners()
this.process.kill()
if (this.child) {
this.child.removeAllListeners()
this.child.kill()
}
}
private async relaunch(): Promise<void> {
this.disposeChild()
try {
await this.start()
this.started = this._start()
await this.started
} 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) {
process.on("SIGUSR1", async () => {
logger.info("Received SIGUSR1; hotswapping")
this.relaunch()
})
}
public start(args: DefaultedArgs): Promise<void> {
// Store for relaunches.
this.args = args
if (!this.started) {
this.started = this._start()
}
@ -217,7 +282,7 @@ export class WrapperProcess {
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) {
@ -229,45 +294,75 @@ export class WrapperProcess {
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)
})
}
private spawn(): cp.ChildProcess {
// Flags to pass along to the Node binary.
let nodeOptions = `${process.env.NODE_OPTIONS || ""} ${(this.options && this.options.nodeOptions) || ""}`
if (!/max_old_space_size=(\d+)/g.exec(nodeOptions)) {
nodeOptions += ` --max_old_space_size=${(this.options && this.options.maxMemory) || 2048}`
}
// Use spawn (instead of fork) to use the new binary in case it was updated.
return cp.spawn(process.argv[0], process.argv.slice(1), {
env: {
...process.env,
CODE_SERVER_PARENT_PID: process.pid.toString(),
NODE_OPTIONS: nodeOptions,
NODE_OPTIONS: `--max-old-space-size=2048 ${process.env.NODE_OPTIONS || ""}`,
},
stdio: ["ipc"],
})
}
/**
* Wait for a handshake from the child then reply.
*/
private async handshake(child: cp.ChildProcess): Promise<void> {
if (!this.args) {
throw new Error("started without args")
}
await onMessage<ChildMessage, ChildHandshakeMessage>(
child,
(message): message is ChildHandshakeMessage => {
return message.type === "handshake"
},
this.logger,
)
this.send(child, { type: "handshake", args: this.args })
}
/**
* Send a message to the child.
*/
private send(child: cp.ChildProcess, message: ParentMessage): void {
child.send(message)
}
}
/**
* 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)
}
})