MaashkinoozheandClaude Opus 5.5
d7d3bf5b2b
Each user gets an own read-ahead queue, served in turn, so two people reading at once no longer cancel each other's chapter. The report rounds phase times and measures the quiet time before a slow storage probe correctly. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
79 lines
3.4 KiB
JavaScript
79 lines
3.4 KiB
JavaScript
'use strict';
|
|
/*
|
|
* Read-ahead for network libraries: when a chapter is opened, its pages (then the next chapter's) are read
|
|
* once in page order, a few at a time, so the OS page cache holds them before the browser asks. On a NAS
|
|
* with spinning disks a cold page costs 10-100 ms (seconds while a disk wakes up); a cached one well under
|
|
* 5 ms. The data is discarded; only the cache effect matters.
|
|
*
|
|
* One queue per reader (`key`, the user): opening another chapter replaces that reader's queue only, and
|
|
* the queues take turns so two people reading at once both get their pages ahead.
|
|
*/
|
|
const fs = require('fs');
|
|
|
|
class ReadAhead {
|
|
constructor({ concurrency = 2, remember = 4000, onBatch = () => {} } = {}) {
|
|
this.concurrency = concurrency; this.remember = remember; this.onBatch = onBatch;
|
|
this.batches = new Map(); this.turn = []; this.active = 0; this.recent = new Map(); this.buffers = [];
|
|
}
|
|
|
|
// `files` in reading order; `tag` names the batch in the trace.
|
|
warm(files, tag, key = '') {
|
|
const old = this.batches.get(key);
|
|
if (old) this.finish(key, old, true);
|
|
const queue = files.filter(f => !this.recent.has(f));
|
|
const batch = { tag, queue, files: queue.length, skipped: files.length - queue.length, bytes: 0, done: 0, active: 0, t0: performance.now() };
|
|
this.batches.set(key, batch);
|
|
if (!this.turn.includes(key)) this.turn.unshift(key); // a fresh request goes next
|
|
if (!queue.length) this.finish(key, batch, false);
|
|
this.pump();
|
|
}
|
|
// Runs after the reader's current batch (e.g. the next chapter) unless a newer warm() replaces it first.
|
|
then(files, tag, key = '') {
|
|
const batch = this.batches.get(key);
|
|
if (!batch) this.warm(files, tag, key); else batch.next = () => this.warm(files, tag, key);
|
|
}
|
|
|
|
pump() {
|
|
while (this.active < this.concurrency) {
|
|
const key = this.turn.find(k => this.batches.get(k)?.queue.length);
|
|
if (key === undefined) return;
|
|
this.turn.push(...this.turn.splice(this.turn.indexOf(key), 1)); // round robin between readers
|
|
const batch = this.batches.get(key), file = batch.queue.shift();
|
|
this.active++; batch.active++;
|
|
this.readFile(file).then(bytes => { this.note(file); batch.bytes += bytes; batch.done++; }, () => {}).finally(() => {
|
|
this.active--; batch.active--;
|
|
if (this.batches.get(key) === batch && !batch.queue.length && !batch.active) this.finish(key, batch, false);
|
|
this.pump();
|
|
});
|
|
}
|
|
}
|
|
finish(key, b, replaced) {
|
|
if (this.batches.get(key) === b) this.batches.delete(key);
|
|
if (!this.batches.has(key)) this.turn = this.turn.filter(k => k !== key);
|
|
b.queue = [];
|
|
this.onBatch({ tag: b.tag, files: b.files, done: b.done, skipped: b.skipped, bytes: b.bytes,
|
|
ms: Math.round(performance.now() - b.t0), replaced });
|
|
if (!replaced && b.next) b.next();
|
|
}
|
|
note(file) {
|
|
this.recent.delete(file); this.recent.set(file, 1);
|
|
while (this.recent.size > this.remember) this.recent.delete(this.recent.keys().next().value);
|
|
}
|
|
|
|
async readFile(file) {
|
|
const buf = this.buffers.pop() || Buffer.allocUnsafe(256 * 1024);
|
|
const fh = await fs.promises.open(file, 'r');
|
|
let total = 0;
|
|
try {
|
|
for (;;) {
|
|
const { bytesRead } = await fh.read(buf, 0, buf.length, total);
|
|
if (!bytesRead) break;
|
|
total += bytesRead;
|
|
}
|
|
} finally { await fh.close(); this.buffers.push(buf); }
|
|
return total;
|
|
}
|
|
}
|
|
|
|
module.exports = { ReadAhead };
|