Skip to content

Encapsulate write queue in separate class #839

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Tyriar merged 2 commits into from
Dec 18, 2025
Merged

Encapsulate write queue in separate class #839

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7c0934e
Encapsulate write queue in separate class
Tyriar Dec 17, 2025
4ab755a
Merge branch 'main' into tyriar/833__refactor
Tyriar Dec 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
166 changes: 94 additions & 72 deletions src/unixTerminal.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ import * as path from 'path';
import * as tty from 'tty';
import { Terminal, DEFAULT_COLS, DEFAULT_ROWS } from './terminal';
import { IProcessEnv, IPtyForkOptions, IPtyOpenOptions } from './interfaces';
import { ArgvOrCommandLine } from './types';
import { ArgvOrCommandLine, IDisposable } from './types';
import { assign, loadNativeModule } from './utils';

const native = loadNativeModule('pty');
Expand All @@ -23,13 +23,6 @@ const DEFAULT_FILE = 'sh';
const DEFAULT_NAME = 'xterm';
const DESTROY_SOCKET_TIMEOUT_MS = 200;

interface IWriteTask {
/** The buffer being written. */
data: Buffer;
/** The current offset of not yet written data. */
offset: number;
}

export class UnixTerminal extends Terminal {
protected _fd: number;
protected _pty: string;
Expand All @@ -43,9 +36,7 @@ export class UnixTerminal extends Terminal {
private _boundClose: boolean = false;
private _emittedClose: boolean = false;

private readonly _writeQueue: IWriteTask[] = [];
private _writeImmediate: NodeJS.Immediate | undefined;
private _encoding?: BufferEncoding = undefined;
private _writeStream: CustomWriteStream;

private _master: net.Socket | undefined;
private _slave: net.Socket | undefined;
Expand Down Expand Up @@ -83,7 +74,6 @@ export class UnixTerminal extends Terminal {
const parsedEnv = this._parseEnv(env);

const encoding = (opt.encoding === undefined ? 'utf8' : opt.encoding);
this._encoding = (encoding || undefined) as BufferEncoding;

const onexit = (code: number, signal: number): void => {
// XXX Sometimes a data event is emitted after exit. Wait til socket is
Expand Down Expand Up @@ -119,6 +109,7 @@ export class UnixTerminal extends Terminal {
if (encoding !== null) {
this._socket.setEncoding(encoding);
}
this._writeStream = new CustomWriteStream(term.fd, (encoding || undefined) as BufferEncoding);

// setup
this._socket.on('error', (err: any) => {
Expand Down Expand Up @@ -175,63 +166,8 @@ export class UnixTerminal extends Terminal {
this._forwardEvents();
}

protected _write(str: string | Buffer): void {
// Writes are put in a queue and processed asynchronously in order to handle
// backpressure from the kernel buffer.
const data = typeof str === 'string'
? Buffer.from(str, this._encoding)
: Buffer.from(str);

if (data.byteLength !== 0) {
this._writeQueue.push({ data, offset: 0 });
if (this._writeQueue.length === 1) {
this._processWriteQueue();
}
}
}

private _processWriteQueue(): void {
this._writeImmediate = undefined;

if (this._writeQueue.length === 0) {
return;
}

const task = this._writeQueue[0];

// Write to the underlying file descriptor and handle it directly, rather
// 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.data, task.offset, (err, written) => {
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());
} else {
// Stop processing immediately on unexpected error and log
this._writeQueue.length = 0;
console.error('Unhandled pty write error', err);
}
return;
}

task.offset += written;
if (task.offset >= task.data.byteLength) {
this._writeQueue.shift();
}

// Since there is more room in the kernel buffer, we can continue to write
// until we hit EAGAIN or exhaust the queue.
//
// Note that old versions of bash, like v3.2 which ships in macOS, appears
// to have a bug in its readline implementation that causes data
// corruption when writes to the pty happens too quickly. Instead of
// trying to workaround that we just accept it so that large pastes are as
// fast as possible.
// Context: https://github.com/microsoft/node-pty/issues/833
this._processWriteQueue();
});
protected _write(data: string | Buffer): void {
this._writeStream.write(data);
}

/* Accessors */
Expand Down Expand Up @@ -307,9 +243,7 @@ export class UnixTerminal extends Terminal {
});

this._socket.destroy();

clearImmediate(this._writeImmediate);
this._writeImmediate = undefined;
this._writeStream.dispose();
}

public kill(signal?: string): void {
Expand Down Expand Up @@ -364,3 +298,91 @@ export class UnixTerminal extends Terminal {
delete env['LINES'];
}
}

interface IWriteTask {
/** The buffer being written. */
buffer: Buffer;
/** The current offset of not yet written data. */
offset: number;
}

/**
* A custom write stream that writes directly to a file descriptor with proper
* handling of backpressure and errors. This avoids some event loop exhaustion
* issues that can occur when using the standard APIs in Node.
*/
class CustomWriteStream implements IDisposable {

private readonly _writeQueue: IWriteTask[] = [];
private _writeImmediate: NodeJS.Immediate | undefined;

constructor(
private readonly _fd: number,
private readonly _encoding: BufferEncoding
) {
}

dispose(): void {
clearImmediate(this._writeImmediate);
this._writeImmediate = undefined;
}

write(data: string | Buffer): void {
// Writes are put in a queue and processed asynchronously in order to handle
// backpressure from the kernel buffer.
const buffer = typeof data === 'string'
? Buffer.from(data, this._encoding)
: Buffer.from(data);

if (buffer.byteLength !== 0) {
this._writeQueue.push({ buffer, offset: 0 });
if (this._writeQueue.length === 1) {
this._processWriteQueue();
}
}
}

private _processWriteQueue(): void {
this._writeImmediate = undefined;

if (this._writeQueue.length === 0) {
return;
}

const task = this._writeQueue[0];

// Write to the underlying file descriptor and handle it directly, rather
// 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) => {
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());
} else {
// Stop processing immediately on unexpected error and log
this._writeQueue.length = 0;
console.error('Unhandled pty write error', err);
}
return;
}

task.offset += written;
if (task.offset >= task.buffer.byteLength) {
this._writeQueue.shift();
}

// Since there is more room in the kernel buffer, we can continue to write
// until we hit EAGAIN or exhaust the queue.
//
// Note that old versions of bash, like v3.2 which ships in macOS, appears
// to have a bug in its readline implementation that causes data
// corruption when writes to the pty happens too quickly. Instead of
// trying to workaround that we just accept it so that large pastes are as
// fast as possible.
// Context: https://github.com/microsoft/node-pty/issues/833
this._processWriteQueue();
});
}
}