From 298cb7033b14ad59088675aae2358c1ad260e5b8 Mon Sep 17 00:00:00 2001 From: David Entzat Date: Tue, 18 Aug 2026 16:23:02 +0200 Subject: [PATCH] fix(unix): pace the EAGAIN write retry so a stalled reader can't saturate the thread CustomWriteStream retries an EAGAIN write with setImmediate, which re-attempts within microseconds. A pty whose reader has stopped draining keeps that branch failing, so the retry becomes an unbounded busy-loop on the thread it runs on. In an embedder multiplexing many PTYs onto one process, every terminal in that process freezes while the machine itself looks healthy: observed in production as ~143,000 EAGAIN/s and a pinned core. This does not undo #833. Measured against an actively draining reader at 4/32/64 MB, EAGAIN never fires at any volume -- a merely slow reader backpressures through the normal completion path, and the kernel only returns EAGAIN once the buffer is full and stays full. The paced branch therefore only executes when throughput is already zero, and the fast path is untouched. Also fixes a pre-existing disposal race that the delay widens: an fs.write already in flight when dispose() runs still invokes its callback, and on EAGAIN that callback re-arms the retry against a closed fd. Track disposal explicitly and check it before scheduling or processing further writes. Both behaviours are covered by tests that fail without this change: 3244 retry attempts per 100ms becomes ~20, and the post-dispose write is caught. Co-Authored-By: Claude Opus 5 (1M context) --- src/unixTerminal.test.ts | 61 ++++++++++++++++++++++++++++++++++++++++ src/unixTerminal.ts | 36 +++++++++++++++++++----- 2 files changed, 90 insertions(+), 7 deletions(-) diff --git a/src/unixTerminal.test.ts b/src/unixTerminal.test.ts index a666e91aa..df8542ab0 100644 --- a/src/unixTerminal.test.ts +++ b/src/unixTerminal.test.ts @@ -124,6 +124,67 @@ if (process.platform !== 'win32') { term.destroy(); }); }); + describe('EAGAIN backpressure', () => { + let realWrite: typeof fs.write; + + beforeEach(() => { + realWrite = fs.write; + }); + + afterEach(() => { + (fs as any).write = realWrite; + }); + + it('should pace retries when the fd keeps returning EAGAIN', (done) => { + const term = new UnixTerminal('node', [ '-e', 'setInterval(() => {}, 1000)' ]); + let attempts = 0; + (fs as any).write = (fd: number, buf: Buffer, offset: number, cb: Function): void => { + attempts++; + const err: any = new Error('EAGAIN'); + err.code = 'EAGAIN'; + process.nextTick(() => cb(err)); + }; + + term.write('trigger'); + setTimeout(() => { + (fs as any).write = realWrite; + term.destroy(); + // At 5ms pacing this is ~20 attempts in 100ms. An immediate re-attempt + // produces tens of thousands, which is the busy-loop being fixed. + assert.ok(attempts > 0, 'expected the EAGAIN path to be exercised'); + assert.ok(attempts < 100, `expected paced retries, got ${attempts} attempts in 100ms`); + done(); + }, 100); + }); + + it('should not write after dispose while a write is in flight', (done) => { + const term = new UnixTerminal('node', [ '-e', 'setInterval(() => {}, 1000)' ]); + let attempts = 0; + let release: (() => void) | undefined; + (fs as any).write = (fd: number, buf: Buffer, offset: number, cb: Function): void => { + attempts++; + const err: any = new Error('EAGAIN'); + err.code = 'EAGAIN'; + // Hold the completion so it lands *after* dispose, as a real in-flight + // write on the libuv threadpool would. + release = () => cb(err); + }; + + term.write('trigger'); + setTimeout(() => { + const before = attempts; + term.destroy(); + release!(); + setTimeout(() => { + (fs as any).write = realWrite; + assert.strictEqual(attempts, before, + 'a write was issued after dispose; the in-flight completion re-armed the retry'); + done(); + }, 50); + }, 20); + }); + }); + describe('signals in parent and child', () => { it('SIGINT - custom in parent and child', done => { // this test is cumbersome - we have to run it in a sub process to diff --git a/src/unixTerminal.ts b/src/unixTerminal.ts index 2776d501e..aa6bf49e8 100644 --- a/src/unixTerminal.ts +++ b/src/unixTerminal.ts @@ -308,6 +308,13 @@ interface IWriteTask { offset: number; } +/** + * How long to wait before re-attempting a write that returned EAGAIN. Large enough to + * stop the retry from monopolising the thread while a reader is stalled, small enough + * to be invisible to interactive writes. + */ +const EAGAIN_RETRY_DELAY_MS = 5; + /** * A custom write stream that writes directly to a file descriptor with proper * handling of backpressure and errors. This avoids some event loop exhaustion @@ -316,7 +323,8 @@ interface IWriteTask { class CustomWriteStream implements IDisposable { private readonly _writeQueue: IWriteTask[] = []; - private _writeImmediate: NodeJS.Immediate | undefined; + private _writeRetry: NodeJS.Timeout | undefined; + private _isDisposed: boolean = false; constructor( private readonly _fd: number, @@ -325,8 +333,9 @@ class CustomWriteStream implements IDisposable { } dispose(): void { - clearImmediate(this._writeImmediate); - this._writeImmediate = undefined; + this._isDisposed = true; + clearTimeout(this._writeRetry); + this._writeRetry = undefined; } write(data: string | Buffer): void { @@ -345,7 +354,11 @@ class CustomWriteStream implements IDisposable { } private _processWriteQueue(): void { - this._writeImmediate = undefined; + if (this._isDisposed) { + return; + } + + this._writeRetry = undefined; if (this._writeQueue.length === 0) { return; @@ -357,11 +370,20 @@ class CustomWriteStream implements IDisposable { // than using the `net.Socket`/`tty.WriteStream` wrappers which swallow and // mask errors like EAGAIN and can cause the thread to block indefinitely. fs.write(this._fd, task.buffer, task.offset, (err, written) => { + // The fd may have been closed while this write was in flight. Anything below + // would either re-arm the retry or write again against a dead descriptor. + if (this._isDisposed) { + return; + } + if (err) { if ('code' in err && err.code === 'EAGAIN') { - // `setImmediate` is used to yield to the event loop and re-attempt - // the write later. - this._writeImmediate = setImmediate(() => this._processWriteQueue()); + // Re-attempt the write later. A delay rather than `setImmediate`: EAGAIN here + // means the reader has stopped draining the pty, so the branch keeps failing + // and an immediate re-attempt turns it into a busy-loop that saturates the + // thread. A reader that is merely slow applies backpressure without reaching + // this branch at all, so the delay does not throttle normal writes. + this._writeRetry = setTimeout(() => this._processWriteQueue(), EAGAIN_RETRY_DELAY_MS); } else { // Stop processing immediately on unexpected error and log this._writeQueue.length = 0;