Settle every 5 min in rounds of 200k, not ~3M at once hourly; range blocks in a loop of their own
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
ce43099188
commit
ec2e5a60be
2 files changed
+28
-9
No files matched your search
@@ -18,9 +18,10 @@ const BATCH = 50_000
|
||||
* reads par2 sets for those). Phase 2 picks its work from settled releases.
|
||||
*/
|
||||
class Settler {
|
||||
async run(): Promise<number> {
|
||||
/** Up to `batches` batches of BATCH (all there are without). */
|
||||
async run(batches = Infinity): Promise<number> {
|
||||
let settled = 0
|
||||
while (true) {
|
||||
for (let batch = 0; batch < batches; batch++) {
|
||||
// Their probe candidates become claimable (Prober.claim), the unreadable ones first.
|
||||
const { rows } = await db.rawQuery(
|
||||
`WITH settled AS (
|
||||
@@ -39,6 +40,7 @@ class Settler {
|
||||
settled += n
|
||||
if (n < BATCH) return settled
|
||||
}
|
||||
return settled
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -14,6 +14,13 @@ const PLAN_EVERY_MS = 60 * 60 * 1000
|
||||
const MERGE_EVERY_MS = 60 * 1000
|
||||
const MERGE_BUDGET_MS = 30 * 1000
|
||||
const STOP_WAIT_MS = 15 * 1000
|
||||
/**
|
||||
* Settling in rounds of at most SETTLE_BATCHES batches every few minutes: once an hour, ~3M releases rewrote their
|
||||
* rows and index entries in one go (2.65M in 30 min), the NVMe saturated and the batch stores queued on their locks
|
||||
* (LOG.md 2026-10-16 (67)).
|
||||
*/
|
||||
const SETTLE_EVERY_MS = 5 * 60 * 1000
|
||||
const SETTLE_BATCHES = 4
|
||||
/** range_blocks (release_final) after ranges finish. */
|
||||
const BLOCKS_EVERY_MS = 5 * 60 * 1000
|
||||
|
||||
@@ -59,22 +66,32 @@ export default class CoordinatorRun extends BaseCommand {
|
||||
await sleep(TICK_MS)
|
||||
}
|
||||
})()
|
||||
// range_blocks in a loop of its own: a settling or merging round doesn't hold release_final back.
|
||||
const blocks = (async () => {
|
||||
while (main && !stopping) {
|
||||
try {
|
||||
await RangeBlocks.rebuild()
|
||||
} catch (error: any) {
|
||||
this.logger.error(`Range blocks failed: ${error.message}`)
|
||||
}
|
||||
for (let waited = 0; waited < BLOCKS_EVERY_MS && !stopping; waited += TICK_MS) await sleep(TICK_MS)
|
||||
}
|
||||
})()
|
||||
// History for new groups hourly; the tail every few minutes.
|
||||
let plannedAt = 0
|
||||
let tailedAt = 0
|
||||
let mergedAt = 0
|
||||
let blocksAt = 0
|
||||
let settledAt = 0
|
||||
while (!stopping) {
|
||||
try {
|
||||
if (main && Date.now() - blocksAt >= BLOCKS_EVERY_MS) {
|
||||
blocksAt = Date.now()
|
||||
await RangeBlocks.rebuild()
|
||||
}
|
||||
if (main && Date.now() - plannedAt >= PLAN_EVERY_MS) {
|
||||
plannedAt = Date.now()
|
||||
const { groups, ranges } = await Planner.planAll()
|
||||
if (ranges > 0) this.logger.info(`Planned ${ranges} ranges over ${groups} groups.`)
|
||||
const settled = await Settler.run()
|
||||
}
|
||||
if (main && Date.now() - settledAt >= SETTLE_EVERY_MS) {
|
||||
settledAt = Date.now()
|
||||
const settled = await Settler.run(SETTLE_BATCHES)
|
||||
if (settled > 0) this.logger.info(`Settled ${settled} releases.`)
|
||||
}
|
||||
if (main && Date.now() - tailedAt >= coordinatorConfig.tail.everySeconds * 1000) {
|
||||
@@ -106,7 +123,7 @@ export default class CoordinatorRun extends BaseCommand {
|
||||
}
|
||||
await sleep(TICK_MS)
|
||||
}
|
||||
await prefetching
|
||||
await Promise.all([prefetching, blocks])
|
||||
// The chunks in flight, for at most STOP_WAIT_MS (a connection the server holds back can take minutes); then closed.
|
||||
const waitUntil = Date.now() + STOP_WAIT_MS
|
||||
while (prefetcher.active > 0 && Date.now() < waitUntil) await sleep(200)
|
||||
|
||||
Reference in new issue
Block a user