Rebuild: a chunk's deadline doubles each time it's hit
At 90 s a chunk that always takes longer would be given up forever (range 155437's worker gave up waiting on its stream). Each deadline a chunk hits doubles its next one (90, 180, 360 ... up to 16x), reset once it comes in. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
d81a202172
commit
dbccec932c
1 file changed
+8
-2
@@ -59,6 +59,8 @@ export class Prefetcher {
|
|||||||
private fetchingNow = 0
|
private fetchingNow = 0
|
||||||
private writingNow = 0
|
private writingNow = 0
|
||||||
private sampler?: NodeJS.Timeout
|
private sampler?: NodeJS.Timeout
|
||||||
|
/** Deadlines a chunk hit, by range and chunk start: each doubles its next deadline, so a slow chunk still comes. */
|
||||||
|
private lateChunks = new Map<string, number>()
|
||||||
/** Requests in flight: what, since when, on which connection (the report names the oldest). */
|
/** Requests in flight: what, since when, on which connection (the report names the oldest). */
|
||||||
private inFlight = new Set<{ rangeId: number; from: number; to: number; since: number; client: XoverClient }>()
|
private inFlight = new Set<{ rangeId: number; from: number; to: number; since: number; client: XoverClient }>()
|
||||||
|
|
||||||
@@ -190,11 +192,14 @@ export class Prefetcher {
|
|||||||
const entry = { rangeId, from: span.from, to: span.to, since: Date.now(), client: c }
|
const entry = { rangeId, from: span.from, to: span.to, since: Date.now(), client: c }
|
||||||
this.inFlight.add(entry)
|
this.inFlight.add(entry)
|
||||||
let late = false
|
let late = false
|
||||||
|
const key = `${rangeId}:${span.from}`
|
||||||
|
const deadlineMs = coordinatorConfig.nntpChunkDeadlineMs * 2 ** Math.min(this.lateChunks.get(key) ?? 0, 4)
|
||||||
const deadline = setTimeout(() => {
|
const deadline = setTimeout(() => {
|
||||||
late = true
|
late = true
|
||||||
this.stats.deadlines++
|
this.stats.deadlines++
|
||||||
|
this.lateChunks.set(key, (this.lateChunks.get(key) ?? 0) + 1)
|
||||||
c.close()
|
c.close()
|
||||||
}, coordinatorConfig.nntpChunkDeadlineMs)
|
}, deadlineMs)
|
||||||
let data: Buffer
|
let data: Buffer
|
||||||
let groupMs = 0
|
let groupMs = 0
|
||||||
try {
|
try {
|
||||||
@@ -204,11 +209,12 @@ export class Prefetcher {
|
|||||||
data = await c.xoverRaw(span.from, span.to)
|
data = await c.xoverRaw(span.from, span.to)
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
if (!late) throw error
|
if (!late) throw error
|
||||||
throw new Error(`Chunk ${span.from}–${span.to} of range ${rangeId} not in after ${coordinatorConfig.nntpChunkDeadlineMs / 1000} s; fetching it again`)
|
throw new Error(`Chunk ${span.from}–${span.to} of range ${rangeId} not in after ${deadlineMs / 1000} s; fetching it again`)
|
||||||
} finally {
|
} finally {
|
||||||
clearTimeout(deadline)
|
clearTimeout(deadline)
|
||||||
this.inFlight.delete(entry)
|
this.inFlight.delete(entry)
|
||||||
}
|
}
|
||||||
|
this.lateChunks.delete(key)
|
||||||
if (!cancelled) {
|
if (!cancelled) {
|
||||||
const t = c.lastTiming
|
const t = c.lastTiming
|
||||||
this.stats.requests++
|
this.stats.requests++
|
||||||
|
|||||||
Reference in new issue
Block a user