diff --git a/src/unixTerminal.test.ts b/src/unixTerminal.test.ts index a666e91a..df8542ab 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 2776d501..aa6bf49e 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;