diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index c33d06c..13a10be 100644 --- a/app/services/prefetcher.ts +++ b/app/services/prefetcher.ts @@ -27,8 +27,9 @@ const SIZE_MS = 30_000 /** * Keeps the news server busy ahead of the workers (REBUILD.md, prefetch). * Every connection takes the next missing chunk: the leased ranges' first - * (a worker waits for them; all connections work on the same range, in - * order), then the next planned ranges, about SPOOL_AHEAD_PER_LEASE per + * (a worker waits for them), then the next planned ranges, with at most + * PREFETCH_PER_RANGE connections on a range and PREFETCH_PER_GROUP on a + * group (Eweka answers faster spread out), about SPOOL_AHEAD_PER_LEASE per * active lease (at least SPOOL_MIN_AHEAD), within SPOOL_MAX_GB. One * connection fills ~26 k overview lines/s, so a range on one connection * kept its worker waiting (LOG.md 2026-10-09). A chunk that takes long is @@ -44,6 +45,9 @@ export class Prefetcher { private targets: Target[] = [] private claimed = new Set() + /** Chunks being fetched per range and per group (spool.perRange, spool.perGroup). */ + private perRange = new Map() + private perGroup = new Map() private running = 0 private refreshedAt = 0 private stopped = false @@ -80,6 +84,8 @@ export class Prefetcher { idle: 0, inFlight: 0, inFlightMin: Infinity, + ranges: 0, + groups: 0, since: Date.now(), } } @@ -154,6 +160,7 @@ export class Prefetcher { } } finally { this.claimed.delete(`${target.id}:${span.from}`) + this.count(target, -1) } } } @@ -224,6 +231,8 @@ export class Prefetcher { s.idle += Math.max(0, this.loops - this.fetchingNow - this.writingNow) s.inFlight += NntpPool.busyCount s.inFlightMin = Math.min(s.inFlightMin, NntpPool.busyCount) + s.ranges += this.perRange.size + s.groups += this.perGroup.size } /** A line a minute: what the connections fetched, and how long a chunk took. */ @@ -244,7 +253,8 @@ export class Prefetcher { const avg = (n: number) => (samples > 0 ? n / samples : 0) const per = (ms: number) => (requests > 0 ? (ms / requests / 1000).toFixed(1) : '-') logger.info( - `Connections: ${avg(st.inFlight).toFixed(1)} of ${NntpPool.size} with a command in flight (min ${st.inFlightMin}); ` + + `Connections: ${avg(st.inFlight).toFixed(1)} of ${NntpPool.size} with a command in flight (min ${st.inFlightMin}), ` + + `on ${avg(st.ranges).toFixed(1)} ranges of ${avg(st.groups).toFixed(1)} groups; ` + `loops ${avg(st.fetching).toFixed(1)} fetching, ${avg(st.writing).toFixed(1)} writing, ${avg(st.idle).toFixed(1)} idle; ` + `a request: GROUP ${per(st.groupMs)} s, XOVER reply ${per(st.replyMs)} s, transfer ${per(st.bodyMs)} s; ` + `${st.silent} of ${requests} transfers silent >5 s; writing ${(st.writeMs / Math.max(1, chunks)).toFixed(0)} ms a chunk.` @@ -254,16 +264,33 @@ export class Prefetcher { } private peek(): boolean { - return this.targets.some((t) => t.missing.some((s) => !this.claimed.has(`${t.id}:${s.from}`))) + return this.targets.some((t) => this.open(t) && t.missing.some((s) => !this.claimed.has(`${t.id}:${s.from}`))) } - /** The first unclaimed missing chunk, in target order. */ + /** Whether the target is under its range's and its group's limit of connections. */ + private open(target: Target) { + const { perRange, perGroup } = coordinatorConfig.spool + return (perRange <= 0 || (this.perRange.get(target.id) ?? 0) < perRange) && (perGroup <= 0 || (this.perGroup.get(target.group) ?? 0) < perGroup) + } + + private count(target: Target, by: number) { + const range = (this.perRange.get(target.id) ?? 0) + by + const group = (this.perGroup.get(target.group) ?? 0) + by + if (range > 0) this.perRange.set(target.id, range) + else this.perRange.delete(target.id) + if (group > 0) this.perGroup.set(target.group, group) + else this.perGroup.delete(target.group) + } + + /** The first unclaimed missing chunk, in target order, of a target under its limits. */ private take(): { target: Target; span: ChunkSpan } | null { for (const target of this.targets) { + if (!this.open(target)) continue const index = target.missing.findIndex((s) => !this.claimed.has(`${target.id}:${s.from}`)) if (index === -1) continue const [span] = target.missing.splice(index, 1) this.claimed.add(`${target.id}:${span.from}`) + this.count(target, 1) return { target, span } } return null diff --git a/config/coordinator.ts b/config/coordinator.ts index 7266571..d9acb4b 100644 --- a/config/coordinator.ts +++ b/config/coordinator.ts @@ -58,9 +58,20 @@ export default { spool: { dir: env.get('SPOOL_DIR', app.tmpPath('spool')), /** Ranges kept fetched ahead, per active lease. */ - aheadPerLease: env.get('SPOOL_AHEAD_PER_LEASE', 2), + aheadPerLease: env.get('SPOOL_AHEAD_PER_LEASE', 4), /** At least this many ranges ahead, so the first leases start at once. */ minAhead: env.get('SPOOL_MIN_AHEAD', 2), maxBytes: env.get('SPOOL_MAX_GB', 10) * 1e9, + /** + * At most this many connections on one range, and on one group, at a + * time (0: no limit): spreads the requests over ranges and groups instead + * of all on the next chunks of the first ranges. Homelab, 40 connections, + * 12 min each, alternated: unlimited (on ~10 ranges) 281k and 167k + * articles/s, 2 per range and 4 per group (on ~22) 319k and 296k, with + * shorter chunk times (11.5 vs 13–24 s; LOG.md 2026-10-09 (18)). Needs + * targets for it: aheadPerLease 4. + */ + perRange: env.get('PREFETCH_PER_RANGE', 2), + perGroup: env.get('PREFETCH_PER_GROUP', 4), }, } diff --git a/start/env.ts b/start/env.ts index 4fb72d2..be506a2 100644 --- a/start/env.ts +++ b/start/env.ts @@ -62,4 +62,6 @@ export default await Env.create(new URL('../', import.meta.url), { SPOOL_AHEAD_PER_LEASE: Env.schema.number.optional(), SPOOL_MIN_AHEAD: Env.schema.number.optional(), SPOOL_MAX_GB: Env.schema.number.optional(), + PREFETCH_PER_RANGE: Env.schema.number.optional(), + PREFETCH_PER_GROUP: Env.schema.number.optional(), })