Prober: claim ahead of the lanes, not when they run dry
Test and deploy / deploy (push) Successful in 1m34s
Test and deploy / test (push) Successful in 3m6s

The next claim started at half a batch of four pipeline rounds a lane,
3-6 s of work, while a claim takes 1.5-4 s and more; at 20-24 lanes the
lanes waited for candidates up to 19% of their time. queueLevels: the
claim starts below eight rounds of every lane's pipeline (~13 s) and
fills the queue to twice that. Taken candidates go back to waiting at
start-up as before, so a bigger queue costs nothing on a restart.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Daan MeijerandClaude Opus 5.5 committed 2026-10-28 14:52:48 +01:00
1 parent 68b480c97e
commit 0ccb850ea4
2 files changed
+38 -3

No files matched your search

+22 -3
View File
@@ -49,6 +49,20 @@ function jsonb(value: unknown): string {
return JSON.stringify(value, (_key, v) => (typeof v === 'string' ? v.replaceAll('\0', '') : v))
}
/** Rounds of every lane's pipeline still queued when the next claim starts. */
const LOW_ROUNDS = 8
/**
* The candidate queue's levels: the next claim starts below `low`, LOW_ROUNDS rounds of every lane's pipeline (at
* ~0.2 s an article and a pipeline of 8, ~13 s of work), and fills it to `target`, twice that. A claim takes 1.5–4 s
* and more (random slices of an 11 M-row table); refilling at half of four rounds left 3–6 s, and at 20–24 lanes the
* lanes waited for candidates up to 19% of their time (LOG.md 2026-10-28 (120)). At least `minBatch` either way.
*/
export function queueLevels(lanes: number, pipeline: number, minBatch: number): { low: number; target: number } {
const low = Math.max(minBatch, lanes * pipeline * LOW_ROUNDS)
return { low, target: 2 * low }
}
/** Slices a claim takes from random points of the queue. */
const SLICES = 20
const REPORT_MS = 60_000
@@ -209,13 +223,18 @@ export class Prober {
}
}
/** Candidates claimed at a time: a few rounds of every lane's pipeline (a claim takes seconds; 400 lasted ~2 s at 37 lanes). */
/** The queue's low-water mark and the level a claim fills it to (queueLevels). */
private get levels() {
return queueLevels(this.running, coordinatorConfig.probe.pipeline, coordinatorConfig.probe.batch)
}
/** Claims to the queue's target; the claim after starts when it's below its low-water mark (queueLevels). */
private get batch() {
return Math.max(coordinatorConfig.probe.batch, this.running * coordinatorConfig.probe.pipeline * 4)
return this.levels.target
}
private async next(): Promise<Candidate | undefined> {
if (this.queue.length < this.batch / 2) {
if (this.queue.length < this.levels.low) {
this.refilling ??= this.claim().finally(() => (this.refilling = null))
if (this.queue.length === 0) await this.refilling
}
@@ -0,0 +1,16 @@
import { test } from '@japa/runner'
import { queueLevels } from '#services/prober'
test.group('Prober / queue levels', () => {
test('the next claim starts with eight pipeline rounds of every lane still queued, and fills to twice that', ({
assert,
}) => {
assert.deepEqual(queueLevels(22, 8, 400), { low: 1408, target: 2816 })
assert.deepEqual(queueLevels(7, 8, 400), { low: 448, target: 896 })
})
test('few lanes: at least the configured batch', ({ assert }) => {
assert.deepEqual(queueLevels(2, 8, 400), { low: 400, target: 800 })
assert.deepEqual(queueLevels(0, 8, 400), { low: 400, target: 800 })
})
})