diff --git a/commands/SeedNode.ts b/commands/SeedNode.ts index 41bad90..025134b 100644 --- a/commands/SeedNode.ts +++ b/commands/SeedNode.ts @@ -1,4 +1,6 @@ import { gzipSync, unzipSync } from 'node:zlib' +import { spawn, type ChildProcess } from 'node:child_process' +import { availableParallelism } from 'node:os' import { setTimeout as sleep } from 'node:timers/promises' import { BaseCommand, flags } from '@adonisjs/core/ace' import { CommandOptions } from '@adonisjs/core/types/ace' @@ -78,10 +80,19 @@ export default class SeedNode extends BaseCommand { }) declare fetch: string + @flags.number({ + description: + 'Processes the ranges and inspections are spread over; one process uses one core (default: SEED_PROCESSES or the number of cores)', + default: process.env.SEED_PROCESSES ? Number(process.env.SEED_PROCESSES) : availableParallelism(), + }) + declare processes: number + @flags.boolean({ description: 'Stop when there is nothing to lease instead of waiting' }) declare once: boolean private stopping = false + /** In a worker process: its number, in front of its log lines. */ + private readonly child = process.env.SEED_PROCESS ?? '' public async run() { const base = (this.url ?? process.env.SEED_URL ?? '').replace(/\/+$/, '') @@ -91,15 +102,19 @@ export default class SeedNode extends BaseCommand { this.exitCode = 1 return } + // JavaScript runs on one core per process: several slots are spread over worker processes. + const processes = Math.max(1, Math.min(this.processes || 1, this.ranges + this.inspectors)) + if (!this.child && processes > 1) return this.supervise(base, token, processes) + const stop = () => { - if (!this.stopping) this.logger.info('Stopping: giving back the ranges in progress…') + if (!this.stopping && !this.child) this.logger.info('Stopping: giving back the ranges in progress…') this.stopping = true } process.on('SIGTERM', stop) process.on('SIGINT', stop) const api = new SeedApi(base, token, (message) => this.logger.warning(message)) - this.logger.info( + if (!this.child) this.logger.info( `Seeding node for ${base}: ${this.ranges} range(s) at once, ${this.fetch === 'direct' ? 'own NNTP connections' : 'headers through the seed API'}` + (this.inspectors > 0 ? `; ${this.inspectors} inspection(s) at once, articles through the seed API.` : '.') ) @@ -107,11 +122,62 @@ export default class SeedNode extends BaseCommand { ...Array.from({ length: this.ranges }, (_, i) => this.work(api, i + 1)), ...Array.from({ length: this.inspectors }, (_, i) => this.inspect(api, i + 1)), ]) + if (!this.child) this.logger.info('Seeding node stopped.') + } + + /** + * Runs the slots in `processes` worker processes (this command again, with + * SEED_PROCESS set), each with its share of the ranges and inspections. A + * worker that exits while the node runs is restarted; SIGTERM/SIGINT are + * passed on, and the node stops when all workers have given their work back. + */ + private async supervise(base: string, token: string, processes: number) { + // Slots dealt out in turn (ranges first, then inspections), so every process has work. + const share = (from: number, to: number, i: number) => + Array.from({ length: to - from }, (_, k) => from + k).filter((k) => k % processes === i).length + this.logger.info( + `Seeding node for ${base}: ${this.ranges} range(s) and ${this.inspectors} inspection(s) at once, over ${processes} processes, ` + + `${this.fetch === 'direct' ? 'own NNTP connections' : 'articles and headers through the seed API'}.` + ) + const children = new Set() + const start = (n: number): Promise => + new Promise((resolve) => { + const args = [ + ...process.execArgv, + process.argv[1], + 'seed:node', + `--ranges=${share(0, this.ranges, n - 1)}`, + `--inspectors=${share(this.ranges, this.ranges + this.inspectors, n - 1)}`, + `--fetch=${this.fetch}`, + `--chunk=${this.chunk}`, + `--submit-every=${this.submitEvery}`, + ...(this.once ? ['--once'] : []), + ] + const child = spawn(process.execPath, args, { + env: { ...process.env, SEED_URL: base, SEED_TOKEN: token, SEED_PROCESS: String(n) }, + stdio: 'inherit', + }) + children.add(child) + child.on('exit', (code, signal) => { + children.delete(child) + if (this.stopping || this.once || code === 0) return resolve() + this.logger.warning(`Process ${n} exited (${signal ?? code}); restarting in 10 s`) + setTimeout(() => start(n).then(resolve), 10_000) + }) + }) + const stop = (signal: NodeJS.Signals) => { + if (!this.stopping) this.logger.info('Stopping: giving back the ranges and inspections in progress…') + this.stopping = true + for (const child of children) child.kill(signal) + } + process.on('SIGTERM', () => stop('SIGTERM')) + process.on('SIGINT', () => stop('SIGINT')) + await Promise.all(Array.from({ length: processes }, (_, i) => start(i + 1))) this.logger.info('Seeding node stopped.') } private async work(api: SeedApi, worker: number) { - const log = (message: string) => this.logger.info(`[${worker}] ${message}`) + const log = (message: string) => this.logger.info(`[${this.child && `${this.child}/`}${worker}] ${message}`) // The NNTP connection is opened for a lease and kept while leases keep coming; with // nothing to lease it's closed, so an idle node holds no connections. let client: XoverClient | undefined @@ -162,7 +228,7 @@ export default class SeedNode extends BaseCommand { * indexer applies them; a job left unanswered goes back after 10 minutes. */ private async inspect(api: SeedApi, slot: number) { - const log = (message: string) => this.logger.info(`[inspect ${slot}] ${message}`) + const log = (message: string) => this.logger.info(`[inspect ${this.child && `${this.child}/`}${slot}] ${message}`) const fetchArticle = (id: string) => api.get(`/api/v1/seed/articles?id=${encodeURIComponent(id)}`, 4) while (!this.stopping) { let jobs: InspectJob[] = [] diff --git a/seed-node/README.md b/seed-node/README.md index 4b6fe32..84dc10e 100644 --- a/seed-node/README.md +++ b/seed-node/README.md @@ -64,6 +64,7 @@ for the same name replaces the token; `--disable` revokes it. | `SEED_TOKEN` | the token from step 1 | | `SEED_RANGES` | ranges seeded at once. Start with 2–4; `0` for no seeding. | | `SEED_INSPECTORS` | inspections and archive walks at once (default `0`). Try 2–4 on a machine with spare cores. | +| `SEED_PROCESSES` | optional: how many processes the ranges and inspections are spread over, one CPU core each. Default: the number of cores (never more than ranges + inspections). Lower it to leave cores free. | | `SEED_FETCH=direct` and `NNTP_*` | optional: seed with this node's own NNTP account (see the commented lines). `NNTP_TLS_SERVERNAME` only through a tunnel: the news server's real hostname, for the TLS certificate check. | Nothing else: the database, Redis and web server settings the indexer diff --git a/seed-node/node.env.example b/seed-node/node.env.example index 5c0c285..c612640 100644 --- a/seed-node/node.env.example +++ b/seed-node/node.env.example @@ -6,6 +6,8 @@ SEED_TOKEN=sn_replace_me SEED_RANGES=4 # Content inspections and archive walks at once (CPU for decoding and parsing; articles come through the seed API). SEED_INSPECTORS=0 +# Ranges and inspections are spread over this many processes, one core each (default: the number of cores). +# SEED_PROCESSES=4 # Optional: SEED_FETCH=direct seeds with this node's own NNTP account instead of through the seed API. # SEED_FETCH=direct