diff --git a/commands/edge_node.ts b/commands/edge_node.ts index 3ecd144..9150b52 100644 --- a/commands/edge_node.ts +++ b/commands/edge_node.ts @@ -1,6 +1,8 @@ import { BaseCommand, flags } from '@adonisjs/core/ace' import type { CommandOptions } from '@adonisjs/core/types/ace' import { setTimeout as sleep } from 'node:timers/promises' +import { spawn, type ChildProcess } from 'node:child_process' +import { availableParallelism } from 'node:os' import { promisify } from 'node:util' import { gzipSync, unzip } from 'node:zlib' import { RangeIndexer, decodeFrames, isKeepalive, toWire, type EdgeBatch, type Lease } from '#lib/edge/index' @@ -29,12 +31,18 @@ export default class EdgeNode extends BaseCommand { @flags.number({ description: 'Stop after this many ranges (default: keep going)' }) declare ranges?: number + @flags.number({ description: 'Worker processes (one core each; default: EDGE_PROCESSES or 1, 0 = one per core)' }) + declare processes?: number + private stopping = false async run() { const url = process.env.COORDINATOR_URL const token = process.env.NODE_TOKEN if (!url || !token) throw new Error('COORDINATOR_URL and NODE_TOKEN are needed') + const requested = this.processes ?? Number(process.env.EDGE_PROCESSES ?? 1) + const processes = requested === 0 ? availableParallelism() : requested + if (processes > 1 && !process.env.EDGE_CHILD) return this.supervise(processes) let stopped: () => void const done = new Promise((resolve) => (stopped = resolve)) this.app.terminating(async () => { @@ -65,6 +73,41 @@ export default class EdgeNode extends BaseCommand { await this.terminate() } + /** + * One worker per process (a worker is single-threaded): starts `count` + * children, restarts one that exits unexpectedly, and passes SIGTERM on + * (each gives its range back). + */ + private async supervise(count: number) { + this.logger.info(`Starting ${count} worker processes.`) + const children = new Set() + let stopping = false + let stopped: () => void + const done = new Promise((resolve) => (stopped = resolve)) + const start = (slot: number) => { + const args = [process.argv[1], 'edge:node', `--batch-chunks=${this.batchChunks}`, ...(this.ranges ? [`--ranges=${this.ranges}`] : [])] + const child = spawn(process.execPath, [...process.execArgv, ...args], { env: { ...process.env, EDGE_CHILD: String(slot) }, stdio: 'inherit' }) + children.add(child) + child.on('exit', (code) => { + children.delete(child) + if (stopping || (this.ranges && code === 0)) { + if (children.size === 0) stopped() + return + } + this.logger.warning(`Worker ${slot} exited (${code}); starting it again in 10 s.`) + setTimeout(() => !stopping && start(slot), 10_000) + }) + } + for (let slot = 1; slot <= count; slot++) start(slot) + this.app.terminating(async () => { + stopping = true + for (const child of children) child.kill('SIGTERM') + await done + }) + await done + await this.terminate() + } + private async runRange(api: Api, lease: Lease) { const { range, rules } = lease const started = Date.now()