Prober: where the lanes' time goes, in the minute report
Fetching, waiting for candidates, idle, inspecting; claims and flushes. The link had room (222 of 1000 Mbit/s with the coordinator at 239k headers/s, one test connection ~84 Mbit/s meanwhile) while the prober did ~5/s. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
358f5a8571
commit
123ed81782
1 file changed
+29
-3
+29
-3
@@ -45,7 +45,12 @@ export class Prober {
|
||||
private outcomes: Outcome[] = []
|
||||
private refilling: Promise<void> | null = null
|
||||
private stopped = false
|
||||
private stats = { articles: 0, bytes: 0, done: 0, missing: 0, error: 0, cached: 0, since: Date.now() }
|
||||
private stats = Prober.freshStats()
|
||||
|
||||
/** Per minute: what was read, and where the lanes' time went (ms summed over lanes; flushes ms on their own). */
|
||||
private static freshStats() {
|
||||
return { articles: 0, bytes: 0, done: 0, missing: 0, error: 0, cached: 0, waitMs: 0, idleMs: 0, fetchMs: 0, inspectMs: 0, flushMs: 0, claimMs: 0, since: Date.now() }
|
||||
}
|
||||
|
||||
async run() {
|
||||
// A previous run's claims: one prober at a time.
|
||||
@@ -56,7 +61,9 @@ export class Prober {
|
||||
const flusher = (async () => {
|
||||
while (!this.stopped || this.outcomes.length > 0) {
|
||||
await sleep(FLUSH_MS)
|
||||
const flushed = Date.now()
|
||||
await this.flush().catch((error) => logger.error(`Probing: storing results failed: ${error.message}`))
|
||||
this.stats.flushMs += Date.now() - flushed
|
||||
this.report()
|
||||
}
|
||||
})()
|
||||
@@ -76,12 +83,15 @@ export class Prober {
|
||||
let client: XoverClient | null = null
|
||||
while (!this.stopped) {
|
||||
const batch: Candidate[] = []
|
||||
const waited = Date.now()
|
||||
while (batch.length < coordinatorConfig.probe.pipeline) {
|
||||
const candidate = await this.next()
|
||||
if (!candidate) break
|
||||
batch.push(candidate)
|
||||
}
|
||||
this.stats.waitMs += Date.now() - waited
|
||||
if (batch.length === 0) {
|
||||
this.stats.idleMs += 30_000
|
||||
await sleep(30_000)
|
||||
continue
|
||||
}
|
||||
@@ -95,7 +105,10 @@ export class Prober {
|
||||
compress: false,
|
||||
timeoutMs: coordinatorConfig.nntpTimeoutMs,
|
||||
})
|
||||
const fetched = Date.now()
|
||||
const bodies = await client.bodies(batch.map((c) => c.messageId))
|
||||
this.stats.fetchMs += Date.now() - fetched
|
||||
const inspected = Date.now()
|
||||
for (const [i, candidate] of batch.entries()) {
|
||||
const body = bodies[i]
|
||||
this.stats.articles++
|
||||
@@ -113,6 +126,7 @@ export class Prober {
|
||||
this.outcomes.push({ candidate, status: 'error' })
|
||||
}
|
||||
}
|
||||
this.stats.inspectMs += Date.now() - inspected
|
||||
} catch (error: any) {
|
||||
logger.warn(`Probing ${batch[0].messageId} and ${batch.length - 1} more: ${error.message}`)
|
||||
for (const candidate of batch) this.outcomes.push({ candidate, status: 'waiting' })
|
||||
@@ -139,6 +153,15 @@ export class Prober {
|
||||
* where most take 20–60 ms (a.b.zogg 2016: 7 articles/s on 6 connections).
|
||||
*/
|
||||
private async claim() {
|
||||
const started = Date.now()
|
||||
try {
|
||||
await this.claimTiers()
|
||||
} finally {
|
||||
this.stats.claimMs += Date.now() - started
|
||||
}
|
||||
}
|
||||
|
||||
private async claimTiers() {
|
||||
for (const { kind, unreadable } of TIERS) {
|
||||
const want = coordinatorConfig.probe.batch - this.queue.length
|
||||
if (want <= 0) return
|
||||
@@ -260,8 +283,11 @@ export class Prober {
|
||||
const { articles, bytes, done, missing, error } = this.stats
|
||||
logger.info(
|
||||
`Probed ${articles} articles in ${seconds.toFixed(0)} s: ${(articles / seconds).toFixed(1)}/s, ${((bytes * 8) / seconds / 1e6).toFixed(0)} Mbit/s; ` +
|
||||
`${done} inspected, ${missing} missing, ${error} not yEnc; ${this.stats.cached} from the probe cache.`
|
||||
`${done} inspected, ${missing} missing, ${error} not yEnc; ${this.stats.cached} from the probe cache. ` +
|
||||
`Lanes: fetching ${(this.stats.fetchMs / 1000).toFixed(0)} s, waiting for candidates ${(this.stats.waitMs / 1000).toFixed(0)} s, ` +
|
||||
`idle ${(this.stats.idleMs / 1000).toFixed(0)} s, inspecting ${(this.stats.inspectMs / 1000).toFixed(1)} s; ` +
|
||||
`claims ${(this.stats.claimMs / 1000).toFixed(1)} s, flushes ${(this.stats.flushMs / 1000).toFixed(1)} s.`
|
||||
)
|
||||
this.stats = { articles: 0, bytes: 0, done: 0, missing: 0, error: 0, cached: 0, since: Date.now() }
|
||||
this.stats = Prober.freshStats()
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user