diff --git a/app/services/nntp_pool.ts b/app/services/nntp_pool.ts index faebf22..631f361 100644 --- a/app/services/nntp_pool.ts +++ b/app/services/nntp_pool.ts @@ -32,6 +32,11 @@ class NntpPool { return this.acquire().then((client) => this.run(client, work)) } + /** Connections with a command in flight (the prefetcher's, the planner's). */ + get busyCount() { + return this.busy.size + } + /** How long ago the connection was opened, in ms. */ ageOf(client: XoverClient): number { return Date.now() - (this.openedAt.get(client) ?? Date.now()) diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index 01324a9..c33d06c 100644 --- a/app/services/prefetcher.ts +++ b/app/services/prefetcher.ts @@ -51,6 +51,10 @@ export class Prefetcher { private sizedAt = 0 /** Fetched since the last report: chunks, articles, ms spent fetching. */ private stats = this.freshStats() + /** Loops in a fetch or a spool write right now. */ + private fetchingNow = 0 + private writingNow = 0 + private sampler?: NodeJS.Timeout private freshStats() { return { @@ -61,6 +65,21 @@ export class Prefetcher { hedgeWon: 0, /** First requests per connection-age bucket, and how many of them were slow. */ byAge: AGE_BUCKETS.map(() => ({ requests: 0, slow: 0 })), + /** Spool writes and the range's update, ms. */ + writeMs: 0, + /** Per answered request, ms: GROUP, XOVER until its status line, the transfer; transfers with a silence over 5 s. */ + requests: 0, + groupMs: 0, + replyMs: 0, + bodyMs: 0, + silent: 0, + /** Once a second: loops fetching, writing, not running; pool connections with a command in flight. */ + samples: 0, + fetching: 0, + writing: 0, + idle: 0, + inFlight: 0, + inFlightMin: Infinity, since: Date.now(), } } @@ -72,6 +91,7 @@ export class Prefetcher { /** One round: refreshes the targets now and then, and puts idle connections to work. */ async tick() { + this.sampler ??= setInterval(() => this.sample(), 1000).unref() if (Date.now() - this.refreshedAt >= REFRESH_MS) { this.refreshedAt = Date.now() this.targets = await this.readTargets() @@ -87,6 +107,7 @@ export class Prefetcher { stop() { this.stopped = true + clearInterval(this.sampler) } get active() { @@ -101,23 +122,36 @@ export class Prefetcher { const { target, span } = next try { const fetchedAt = Date.now() + this.fetchingNow++ const fetch = async (client: Parameters[0]>[0]) => { await client.group(target.group) return client.xoverRaw(span.from, span.to) } - const data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await this.fetchHedged(target.group, span) + let data: Buffer + try { + data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await this.fetchHedged(target.group, span) + } finally { + this.fetchingNow-- + } this.stats.chunks++ this.stats.articles += span.to - span.from + 1 this.stats.fetchMs += Date.now() - fetchedAt - const existed = await Spool.has(target.id, span) - await Spool.write(target.id, span, data) - if (existed) continue - this.spoolBytes += data.length - const { rows } = await db.rawQuery( - `UPDATE ranges SET spooled_chunks = spooled_chunks + 1 WHERE id = ? RETURNING spooled_chunks`, - [target.id] - ) - if (Number(rows[0]?.spooled_chunks) >= target.total) await db.from('ranges').where('id', target.id).update({ spooled_at: new Date() }) + const writtenAt = Date.now() + this.writingNow++ + try { + const existed = await Spool.has(target.id, span) + await Spool.write(target.id, span, data) + if (existed) continue + this.spoolBytes += data.length + const { rows } = await db.rawQuery( + `UPDATE ranges SET spooled_chunks = spooled_chunks + 1 WHERE id = ? RETURNING spooled_chunks`, + [target.id] + ) + if (Number(rows[0]?.spooled_chunks) >= target.total) await db.from('ranges').where('id', target.id).update({ spooled_at: new Date() }) + } finally { + this.writingNow-- + this.stats.writeMs += Date.now() - writtenAt + } } finally { this.claimed.delete(`${target.id}:${span.from}`) } @@ -142,8 +176,19 @@ export class Prefetcher { if (cancelled) return Buffer.alloc(0) client = c if (first) age = NntpPool.ageOf(c) + const groupAt = performance.now() await c.group(group) - return c.xoverRaw(span.from, span.to) + const groupMs = performance.now() - groupAt + const data = await c.xoverRaw(span.from, span.to) + if (!cancelled) { + const t = c.lastTiming + this.stats.requests++ + this.stats.groupMs += groupMs + this.stats.replyMs += t.replyMs + this.stats.bodyMs += t.bodyMs + if (t.longestGapMs > 5000) this.stats.silent++ + } + return data }) if (!done) return null done.catch(() => {}) @@ -171,20 +216,39 @@ export class Prefetcher { return result.value } + private sample() { + const s = this.stats + s.samples++ + s.fetching += this.fetchingNow + s.writing += this.writingNow + s.idle += Math.max(0, this.loops - this.fetchingNow - this.writingNow) + s.inFlight += NntpPool.busyCount + s.inFlightMin = Math.min(s.inFlightMin, NntpPool.busyCount) + } + /** A line a minute: what the connections fetched, and how long a chunk took. */ private report() { const seconds = (Date.now() - this.stats.since) / 1000 if (seconds < 60) return - const { chunks, articles, fetchMs, hedged, hedgeWon, byAge } = this.stats + const { chunks, articles, fetchMs, hedged, hedgeWon, byAge, samples, requests } = this.stats if (chunks > 0) { const ages = byAge .map((b, i) => `${i === 0 ? '<' : `${AGE_BUCKETS[i - 1]}–`}${AGE_BUCKETS[i] === Infinity ? '' : AGE_BUCKETS[i]} min ${b.slow}/${b.requests}`.replace('– min', '+ min')) .join(', ') logger.info( `Prefetched ${chunks} chunks in ${seconds.toFixed(0)} s: ${Math.round(articles / seconds).toLocaleString('en')} articles/s, ` + - `${(fetchMs / chunks / 1000).toFixed(1)} s a chunk, loops busy ${Math.round((fetchMs / (seconds * 1000 * this.loops)) * 100)}%; ` + + `${(fetchMs / chunks / 1000).toFixed(1)} s a chunk; ` + `hedged ${hedged} (second won ${hedgeWon}); slow by connection age: ${ages}.` ) + const st = this.stats + 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}); ` + + `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.` + ) } this.stats = this.freshStats() } diff --git a/lib/nntp/xover_client.ts b/lib/nntp/xover_client.ts index de61169..c985aac 100644 --- a/lib/nntp/xover_client.ts +++ b/lib/nntp/xover_client.ts @@ -40,12 +40,26 @@ export class XoverClient { private wake?: () => void private failure?: Error compressed = false + /** + * The last xoverRaw's timing, ms: until the status line came (the server + * thinking), from there to the end (the transfer), and the longest silence + * during the transfer. + */ + lastTiming = { replyMs: 0, bodyMs: 0, longestGapMs: 0 } + private measuring = false + private lastDataAt = 0 + private longestGap = 0 private constructor( private sock: net.Socket, private timeoutMs: number ) { sock.on('data', (chunk: Buffer) => { + if (this.measuring) { + const now = performance.now() + this.longestGap = Math.max(this.longestGap, now - this.lastDataAt) + this.lastDataAt = now + } this.chunks.push(chunk) this.length += chunk.length this.wake?.() @@ -119,12 +133,22 @@ export class XoverClient { async xoverRaw(from: number, to: number): Promise { return this.timed(async () => { + const sentAt = performance.now() const reply = await this.command(`XOVER ${from}-${to}`) + const repliedAt = performance.now() if (reply.startsWith('420') || reply.startsWith('423')) return deflateSync(Buffer.alloc(0)) await this.expect(reply, ['224']) - if (reply.includes('COMPRESS=GZIP')) return (await this.compressedRaw(false)).raw - // Plain replies (the relay's default): compressed off the event loop, at the fastest level. - return deflateAsync(Buffer.from(await this.plainBody()), { level: 1 }) + this.measuring = true + this.lastDataAt = repliedAt + this.longestGap = 0 + try { + if (reply.includes('COMPRESS=GZIP')) return (await this.compressedRaw(false)).raw + // Plain replies (the relay's default): compressed off the event loop, at the fastest level. + return deflateAsync(Buffer.from(await this.plainBody()), { level: 1 }) + } finally { + this.measuring = false + this.lastTiming = { replyMs: repliedAt - sentAt, bodyMs: performance.now() - repliedAt, longestGapMs: this.longestGap } + } }) }