Rebuild: edge:node --processes (one worker per core, supervised)
On the 4-core workstation (shared with coordinator, web server and database): 3 processes indexed 6 boneless ranges (12M headers) in 37 s, ~324k headers/s. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
fa7d4f3394
commit
ead2790515
1 file changed
+43
@@ -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<void>((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<ChildProcess>()
|
||||
let stopping = false
|
||||
let stopped: () => void
|
||||
const done = new Promise<void>((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()
|
||||
|
||||
Reference in new issue
Block a user