streaming: wall-clock flush so low-rate emissions appear immediately
Worker batching used to wait for 64KB / 4096 newlines before
flushing. A program like
(let loop ((n 0))
(when (zero? (modulo n 10000))
(display "Emitted at iteration: ") (display n) (newline))
(loop (+ n 1)))
emits ~25 bytes every 10K iterations, so the user saw nothing for
several seconds — fox reported "this used to work and doesn't seem
to anymore" because the first emission was buried in the batch
buffer.
Added a 100ms wall-clock check alongside the existing size /
newline thresholds. The first chunk to arrive triggers a flush
(lastFlushTime starts at 0), and low-rate streams cap at ~10
flushes/sec. High-rate streams still hit the byte / newline ceilings
first so chunk batching for tight (display) loops is unaffected.
Verified headlessly: the infinite-emitter program now shows the
first emission within a frame, subsequent ones streaming live as
the loop iterates.
This commit is contained in:
parent
63808f72a5
commit
f66124b0bc
4 changed files with 44 additions and 4 deletions
|
|
@ -55,13 +55,22 @@ self.onmessage = async (e) => {
|
||||||
// calls.
|
// calls.
|
||||||
const CHUNK_FLUSH_BYTES = 65536;
|
const CHUNK_FLUSH_BYTES = 65536;
|
||||||
const CHUNK_FLUSH_NEWLINES = 4096;
|
const CHUNK_FLUSH_NEWLINES = 4096;
|
||||||
|
// Wall-clock flush — guarantees the first emission lands on the UI
|
||||||
|
// immediately and caps low-rate streams at ~10 flushes/sec. Without
|
||||||
|
// this, a (let loop ...) that emits every 10K iterations would
|
||||||
|
// accumulate 30 bytes per emit, take 2000+ emits to hit the 64KB
|
||||||
|
// threshold, and the user sees nothing for seconds. lastFlushTime
|
||||||
|
// starts at 0 so the first chunk to land always triggers a flush.
|
||||||
|
const FLUSH_INTERVAL_MS = 100;
|
||||||
let chunkBuffer = "";
|
let chunkBuffer = "";
|
||||||
let chunkNewlines = 0;
|
let chunkNewlines = 0;
|
||||||
|
let lastFlushTime = 0;
|
||||||
function flushChunkBuffer() {
|
function flushChunkBuffer() {
|
||||||
if (!chunkBuffer) return;
|
if (!chunkBuffer) return;
|
||||||
self.postMessage({ kind: "chunk-batch", runId, tier, text: chunkBuffer });
|
self.postMessage({ kind: "chunk-batch", runId, tier, text: chunkBuffer });
|
||||||
chunkBuffer = "";
|
chunkBuffer = "";
|
||||||
chunkNewlines = 0;
|
chunkNewlines = 0;
|
||||||
|
lastFlushTime = Date.now();
|
||||||
}
|
}
|
||||||
try {
|
try {
|
||||||
const output = await evalOnTier(
|
const output = await evalOnTier(
|
||||||
|
|
@ -77,7 +86,8 @@ self.onmessage = async (e) => {
|
||||||
if (chunk.charCodeAt(i) === 10) chunkNewlines++;
|
if (chunk.charCodeAt(i) === 10) chunkNewlines++;
|
||||||
}
|
}
|
||||||
if (chunkBuffer.length >= CHUNK_FLUSH_BYTES
|
if (chunkBuffer.length >= CHUNK_FLUSH_BYTES
|
||||||
|| chunkNewlines >= CHUNK_FLUSH_NEWLINES) {
|
|| chunkNewlines >= CHUNK_FLUSH_NEWLINES
|
||||||
|
|| (Date.now() - lastFlushTime) >= FLUSH_INTERVAL_MS) {
|
||||||
flushChunkBuffer();
|
flushChunkBuffer();
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
|
|
||||||
|
|
@ -55,13 +55,22 @@ self.onmessage = async (e) => {
|
||||||
// calls.
|
// calls.
|
||||||
const CHUNK_FLUSH_BYTES = 65536;
|
const CHUNK_FLUSH_BYTES = 65536;
|
||||||
const CHUNK_FLUSH_NEWLINES = 4096;
|
const CHUNK_FLUSH_NEWLINES = 4096;
|
||||||
|
// Wall-clock flush — guarantees the first emission lands on the UI
|
||||||
|
// immediately and caps low-rate streams at ~10 flushes/sec. Without
|
||||||
|
// this, a (let loop ...) that emits every 10K iterations would
|
||||||
|
// accumulate 30 bytes per emit, take 2000+ emits to hit the 64KB
|
||||||
|
// threshold, and the user sees nothing for seconds. lastFlushTime
|
||||||
|
// starts at 0 so the first chunk to land always triggers a flush.
|
||||||
|
const FLUSH_INTERVAL_MS = 100;
|
||||||
let chunkBuffer = "";
|
let chunkBuffer = "";
|
||||||
let chunkNewlines = 0;
|
let chunkNewlines = 0;
|
||||||
|
let lastFlushTime = 0;
|
||||||
function flushChunkBuffer() {
|
function flushChunkBuffer() {
|
||||||
if (!chunkBuffer) return;
|
if (!chunkBuffer) return;
|
||||||
self.postMessage({ kind: "chunk-batch", runId, tier, text: chunkBuffer });
|
self.postMessage({ kind: "chunk-batch", runId, tier, text: chunkBuffer });
|
||||||
chunkBuffer = "";
|
chunkBuffer = "";
|
||||||
chunkNewlines = 0;
|
chunkNewlines = 0;
|
||||||
|
lastFlushTime = Date.now();
|
||||||
}
|
}
|
||||||
try {
|
try {
|
||||||
const output = await evalOnTier(
|
const output = await evalOnTier(
|
||||||
|
|
@ -77,7 +86,8 @@ self.onmessage = async (e) => {
|
||||||
if (chunk.charCodeAt(i) === 10) chunkNewlines++;
|
if (chunk.charCodeAt(i) === 10) chunkNewlines++;
|
||||||
}
|
}
|
||||||
if (chunkBuffer.length >= CHUNK_FLUSH_BYTES
|
if (chunkBuffer.length >= CHUNK_FLUSH_BYTES
|
||||||
|| chunkNewlines >= CHUNK_FLUSH_NEWLINES) {
|
|| chunkNewlines >= CHUNK_FLUSH_NEWLINES
|
||||||
|
|| (Date.now() - lastFlushTime) >= FLUSH_INTERVAL_MS) {
|
||||||
flushChunkBuffer();
|
flushChunkBuffer();
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
|
|
||||||
|
|
@ -55,13 +55,22 @@ self.onmessage = async (e) => {
|
||||||
// calls.
|
// calls.
|
||||||
const CHUNK_FLUSH_BYTES = 65536;
|
const CHUNK_FLUSH_BYTES = 65536;
|
||||||
const CHUNK_FLUSH_NEWLINES = 4096;
|
const CHUNK_FLUSH_NEWLINES = 4096;
|
||||||
|
// Wall-clock flush — guarantees the first emission lands on the UI
|
||||||
|
// immediately and caps low-rate streams at ~10 flushes/sec. Without
|
||||||
|
// this, a (let loop ...) that emits every 10K iterations would
|
||||||
|
// accumulate 30 bytes per emit, take 2000+ emits to hit the 64KB
|
||||||
|
// threshold, and the user sees nothing for seconds. lastFlushTime
|
||||||
|
// starts at 0 so the first chunk to land always triggers a flush.
|
||||||
|
const FLUSH_INTERVAL_MS = 100;
|
||||||
let chunkBuffer = "";
|
let chunkBuffer = "";
|
||||||
let chunkNewlines = 0;
|
let chunkNewlines = 0;
|
||||||
|
let lastFlushTime = 0;
|
||||||
function flushChunkBuffer() {
|
function flushChunkBuffer() {
|
||||||
if (!chunkBuffer) return;
|
if (!chunkBuffer) return;
|
||||||
self.postMessage({ kind: "chunk-batch", runId, tier, text: chunkBuffer });
|
self.postMessage({ kind: "chunk-batch", runId, tier, text: chunkBuffer });
|
||||||
chunkBuffer = "";
|
chunkBuffer = "";
|
||||||
chunkNewlines = 0;
|
chunkNewlines = 0;
|
||||||
|
lastFlushTime = Date.now();
|
||||||
}
|
}
|
||||||
try {
|
try {
|
||||||
const output = await evalOnTier(
|
const output = await evalOnTier(
|
||||||
|
|
@ -77,7 +86,8 @@ self.onmessage = async (e) => {
|
||||||
if (chunk.charCodeAt(i) === 10) chunkNewlines++;
|
if (chunk.charCodeAt(i) === 10) chunkNewlines++;
|
||||||
}
|
}
|
||||||
if (chunkBuffer.length >= CHUNK_FLUSH_BYTES
|
if (chunkBuffer.length >= CHUNK_FLUSH_BYTES
|
||||||
|| chunkNewlines >= CHUNK_FLUSH_NEWLINES) {
|
|| chunkNewlines >= CHUNK_FLUSH_NEWLINES
|
||||||
|
|| (Date.now() - lastFlushTime) >= FLUSH_INTERVAL_MS) {
|
||||||
flushChunkBuffer();
|
flushChunkBuffer();
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
|
|
||||||
|
|
@ -55,13 +55,22 @@ self.onmessage = async (e) => {
|
||||||
// calls.
|
// calls.
|
||||||
const CHUNK_FLUSH_BYTES = 65536;
|
const CHUNK_FLUSH_BYTES = 65536;
|
||||||
const CHUNK_FLUSH_NEWLINES = 4096;
|
const CHUNK_FLUSH_NEWLINES = 4096;
|
||||||
|
// Wall-clock flush — guarantees the first emission lands on the UI
|
||||||
|
// immediately and caps low-rate streams at ~10 flushes/sec. Without
|
||||||
|
// this, a (let loop ...) that emits every 10K iterations would
|
||||||
|
// accumulate 30 bytes per emit, take 2000+ emits to hit the 64KB
|
||||||
|
// threshold, and the user sees nothing for seconds. lastFlushTime
|
||||||
|
// starts at 0 so the first chunk to land always triggers a flush.
|
||||||
|
const FLUSH_INTERVAL_MS = 100;
|
||||||
let chunkBuffer = "";
|
let chunkBuffer = "";
|
||||||
let chunkNewlines = 0;
|
let chunkNewlines = 0;
|
||||||
|
let lastFlushTime = 0;
|
||||||
function flushChunkBuffer() {
|
function flushChunkBuffer() {
|
||||||
if (!chunkBuffer) return;
|
if (!chunkBuffer) return;
|
||||||
self.postMessage({ kind: "chunk-batch", runId, tier, text: chunkBuffer });
|
self.postMessage({ kind: "chunk-batch", runId, tier, text: chunkBuffer });
|
||||||
chunkBuffer = "";
|
chunkBuffer = "";
|
||||||
chunkNewlines = 0;
|
chunkNewlines = 0;
|
||||||
|
lastFlushTime = Date.now();
|
||||||
}
|
}
|
||||||
try {
|
try {
|
||||||
const output = await evalOnTier(
|
const output = await evalOnTier(
|
||||||
|
|
@ -77,7 +86,8 @@ self.onmessage = async (e) => {
|
||||||
if (chunk.charCodeAt(i) === 10) chunkNewlines++;
|
if (chunk.charCodeAt(i) === 10) chunkNewlines++;
|
||||||
}
|
}
|
||||||
if (chunkBuffer.length >= CHUNK_FLUSH_BYTES
|
if (chunkBuffer.length >= CHUNK_FLUSH_BYTES
|
||||||
|| chunkNewlines >= CHUNK_FLUSH_NEWLINES) {
|
|| chunkNewlines >= CHUNK_FLUSH_NEWLINES
|
||||||
|
|| (Date.now() - lastFlushTime) >= FLUSH_INTERVAL_MS) {
|
||||||
flushChunkBuffer();
|
flushChunkBuffer();
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue