diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index 1d93881..7471343 100644 --- a/app/services/prefetcher.ts +++ b/app/services/prefetcher.ts @@ -35,6 +35,8 @@ export class Prefetcher { private stopped = false private spoolBytes = 0 private sizedAt = 0 + /** Fetched since the last report: chunks, articles, ms spent fetching. */ + private stats = { chunks: 0, articles: 0, fetchMs: 0, since: Date.now() } /** One round: refreshes the targets now and then, and puts idle connections to work. */ async tick() { @@ -42,6 +44,7 @@ export class Prefetcher { this.refreshedAt = Date.now() this.targets = await this.readTargets() } + this.report() while (!this.stopped && this.running < NntpPool.size && this.peek()) { this.running++ this.work() @@ -65,10 +68,14 @@ export class Prefetcher { if (!next) return const { target, span } = next try { + const fetchedAt = Date.now() const data = await NntpPool.use(async (client) => { await client.group(target.group) return client.xoverRaw(span.from, span.to) }) + 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 @@ -84,6 +91,20 @@ export class Prefetcher { } } + /** 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 } = this.stats + if (chunks > 0) { + 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, connections busy ${Math.round((fetchMs / (seconds * 1000 * NntpPool.size)) * 100)}%.` + ) + } + this.stats = { chunks: 0, articles: 0, fetchMs: 0, since: Date.now() } + } + private peek(): boolean { return this.targets.some((t) => t.missing.some((s) => !this.claimed.has(`${t.id}:${s.from}`))) } diff --git a/lib/nntp/xover_client.ts b/lib/nntp/xover_client.ts index dcf0f51..0e7dac4 100644 --- a/lib/nntp/xover_client.ts +++ b/lib/nntp/xover_client.ts @@ -1,9 +1,10 @@ import { promisify } from 'node:util' import net from 'node:net' import tls from 'node:tls' -import { deflate, deflateSync, unzipSync } from 'node:zlib' +import { deflate, deflateSync, unzip } from 'node:zlib' const deflateAsync = promisify(deflate) +const unzipAsync = promisify(unzip) export interface XoverClientOptions { host: string @@ -241,7 +242,8 @@ export class XoverClient { const all = this.buffer() const raw = all.subarray(0, all.length - DOT_LINE.length) try { - const inflated = unzipSync(raw) + // On a thread-pool thread: a chunk's ~25 MB of text took the coordinator's event loop ~60 ms. + const inflated = await unzipAsync(raw) this.take(all.length) return { raw: Buffer.from(raw), text: wantText ? inflated.toString('utf8') : '' } } catch {