diff --git a/spec/background-jobs/worker-pooled-distribution-spec.js b/spec/background-jobs/worker-pooled-distribution-spec.js new file mode 100644 index 00000000..1f2a3394 --- /dev/null +++ b/spec/background-jobs/worker-pooled-distribution-spec.js @@ -0,0 +1,93 @@ +// @ts-check + +import BackgroundJobsWorker from "../../src/background-jobs/worker.js" +import {describe, expect, it} from "../../src/testing/test.js" + +/** + * @param {string} id + * @returns {import("../../src/background-jobs/types.js").BackgroundJobPayload & {id: string}} + */ +function fakePayload(id) { + return /** @type {import("../../src/background-jobs/types.js").BackgroundJobPayload & {id: string}} */ (/** @type {unknown} */ ({id})) +} + +/** + * Adds a pooled child to a worker without forking a real process, so pooled child + * selection can be exercised directly. + * @param {BackgroundJobsWorker} worker - Worker under test. + * @param {{inflight?: number, lastDispatchSeq?: number, retiring?: boolean}} [args] - Seed state. + * @returns {import("node:child_process").ChildProcess} - The fake child. + */ +function addFakePooledChild(worker, {inflight = 0, lastDispatchSeq = 0, retiring = false} = {}) { + const child = /** @type {import("node:child_process").ChildProcess} */ (/** @type {unknown} */ ({send() {}})) + /** @type {Map} */ + const inflightMap = new Map() + + for (let index = 0; index < inflight; index++) { + inflightMap.set(`seed-${worker.pooledChildren.size}-${index}`, {payload: fakePayload(`seed-${index}`)}) + } + + worker.pooledChildren.add(child) + worker.pooledChildStates.set(child, {createdAtMs: 0, inflight: inflightMap, jobsRun: 0, lastDispatchSeq, retiring}) + + return child +} + +/** + * @param {BackgroundJobsWorker} worker - Worker under test. + * @param {import("node:child_process").ChildProcess} child - Pooled child. + * @returns {number} - Its in-flight job count. + */ +function inflightSize(worker, child) { + const state = worker.pooledChildStates.get(child) + + if (!state) throw new Error("Missing pooled child state") + + return state.inflight.size +} + +describe("Background jobs - pooled distribution", () => { + it("spreads jobs evenly across children instead of first-fit packing the earliest", () => { + const worker = new BackgroundJobsWorker({pooledRunnerConcurrency: 4, pooledRunnerCount: 3}) + const first = addFakePooledChild(worker) + const second = addFakePooledChild(worker) + const third = addFakePooledChild(worker) + + for (let index = 0; index < 6; index++) worker._runPooledJob(fakePayload(`job-${index}`)) + + // Round-robin: two each. First-fit would have packed [4, 2, 0]. + expect([inflightSize(worker, first), inflightSize(worker, second), inflightSize(worker, third)]).toEqual([2, 2, 2]) + }) + + it("does not burst a freshly added child to catch up to already-loaded children", () => { + const worker = new BackgroundJobsWorker({pooledRunnerConcurrency: 10, pooledRunnerCount: 3}) + + worker._pooledDispatchSeq = 2 + + const loadedA = addFakePooledChild(worker, {inflight: 2, lastDispatchSeq: 1}) + const loadedB = addFakePooledChild(worker, {inflight: 2, lastDispatchSeq: 2}) + const fresh = addFakePooledChild(worker, {inflight: 0, lastDispatchSeq: 0}) + + for (let index = 0; index < 3; index++) worker._runPooledJob(fakePayload(`job-${index}`)) + + // The fresh child takes one job as its turn comes up, then the rotation moves on to + // the others — not a burst of all three to "catch up" (which naive least-loaded, + // always picking the lowest in-flight, would do). + expect(inflightSize(worker, fresh)).toEqual(1) + expect(inflightSize(worker, loadedA)).toEqual(3) + expect(inflightSize(worker, loadedB)).toEqual(3) + }) + + it("skips retiring children and lazily spawns only when every non-retiring child is full", () => { + const worker = new BackgroundJobsWorker({pooledRunnerConcurrency: 2, pooledRunnerCount: 4}) + const retiring = addFakePooledChild(worker, {retiring: true}) + const full = addFakePooledChild(worker, {inflight: 2}) + + // No non-retiring child has a free slot, so selection returns undefined (the caller + // then spawns a new child). + expect(worker._selectPooledChild()).toEqual(undefined) + // The retiring child is never selected even though it has open slots. + expect(inflightSize(worker, retiring)).toEqual(0) + expect(inflightSize(worker, full)).toEqual(2) + }) +}) diff --git a/src/background-jobs/worker.js b/src/background-jobs/worker.js index 9abaee91..88aec312 100644 --- a/src/background-jobs/worker.js +++ b/src/background-jobs/worker.js @@ -188,8 +188,11 @@ export default class BackgroundJobsWorker { this.inflightPooledJobs = new Set() /** @type {Set} */ this.pooledChildren = new Set() - /** @type {Map void, timeoutTimer?: ReturnType | null}>, retiring: boolean, settling?: boolean, timeoutSigkillTimer?: ReturnType | null}>} */ + /** @type {Map void, timeoutTimer?: ReturnType | null}>, lastDispatchSeq: number, retiring: boolean, settling?: boolean, timeoutSigkillTimer?: ReturnType | null}>} */ this.pooledChildStates = new Map() + // Monotonic dispatch counter for round-robin child selection: each dispatch stamps + // the chosen child, and selection prefers the child dispatched least recently. + this._pooledDispatchSeq = 0 } /** @@ -643,18 +646,13 @@ export default class BackgroundJobsWorker { * @returns {Promise} - Resolves after the durable report. */ _runPooledJob(payload) { - let child - for (const candidate of this.pooledChildren) { - const state = this.pooledChildStates.get(candidate) - if (state && !state.retiring && state.inflight.size < this.pooledRunnerConcurrency) { - child = candidate - break - } - } - if (!child) child = this._createPooledChild() + const child = this._selectPooledChild() || this._createPooledChild() const state = this.pooledChildStates.get(child) if (!state) throw new Error("Pooled runner state missing") + // Stamp the round-robin cursor so the next dispatch prefers a different child. + state.lastDispatchSeq = ++this._pooledDispatchSeq + return new Promise((resolve) => { const timeoutTimer = this._armPooledJobTimeout({child, jobId: payload.id}) @@ -667,6 +665,36 @@ export default class BackgroundJobsWorker { }) } + /** + * Selects a pooled child to run the next job, or undefined when every non-retiring + * child is already full (the caller then lazily spawns one). Among children with a + * free concurrency slot, picks the one dispatched least recently — a round-robin that + * spreads jobs (notably multi-minute RunBuildJobs, each pinning a tenant connection + * for its whole run) evenly across children instead of first-fit packing the earliest + * one until it is full. A freshly spawned or replacement child therefore takes its + * fair share one job at a time as its turn comes up, rather than absorbing a burst to + * "catch up" to the others. + * @returns {import("node:child_process").ChildProcess | undefined} - The chosen child, or undefined when all non-retiring children are full. + */ + _selectPooledChild() { + /** @type {import("node:child_process").ChildProcess | undefined} */ + let selected + let selectedSeq = Infinity + + for (const child of this.pooledChildren) { + const state = this.pooledChildStates.get(child) + + if (!state || state.retiring || state.inflight.size >= this.pooledRunnerConcurrency) continue + + if (state.lastDispatchSeq < selectedSeq) { + selected = child + selectedSeq = state.lastDispatchSeq + } + } + + return selected + } + /** * Arms a per-job wall-clock backstop for a pooled job. A pooled child hosts many * concurrent jobs, so a single genuinely-hung job would otherwise pin its @@ -733,7 +761,7 @@ export default class BackgroundJobsWorker { }) this.pooledChildren.add(child) this.inflightProcessChildren.add(child) - this.pooledChildStates.set(child, {createdAtMs: Date.now(), jobsRun: 0, inflight: new Map(), retiring: false}) + this.pooledChildStates.set(child, {createdAtMs: Date.now(), jobsRun: 0, inflight: new Map(), lastDispatchSeq: 0, retiring: false}) child.on("message", (message) => this._handlePooledChildMessage({child, message})) child.once("exit", (code, signal) => this._handlePooledChildFailure({child, error: new Error(`Pooled background job runner exited: code=${code} signal=${signal || "none"}`)})) child.once("error", (error) => this._handlePooledChildFailure({child, error}))