perf: coalesce terminal output in Main and write it to xterm once per frame
This commit is contained in:
@@ -258,30 +258,215 @@ describe('LocalTerminalSession', () => {
|
|||||||
})
|
})
|
||||||
|
|
||||||
it('pauses at the pending-event boundary and resumes after ACK', async () => {
|
it('pauses at the pending-event boundary and resumes after ACK', async () => {
|
||||||
const pty = new FakePty()
|
vi.useFakeTimers()
|
||||||
const session = await LocalTerminalSession.create({
|
try {
|
||||||
...baseOptions,
|
const pty = new FakePty()
|
||||||
dependencies: sessionDependencies(pty)
|
const session = await LocalTerminalSession.create({
|
||||||
})
|
...baseOptions,
|
||||||
let lastSequence = 0
|
dependencies: sessionDependencies(pty)
|
||||||
session.onEvent((event) => {
|
})
|
||||||
lastSequence = event.sequence
|
let lastSequence = 0
|
||||||
})
|
session.onEvent((event) => {
|
||||||
|
lastSequence = event.sequence
|
||||||
|
})
|
||||||
|
|
||||||
for (
|
for (
|
||||||
let index = 0;
|
let index = 0;
|
||||||
index < TERMINAL_LIMITS.maximumPendingEvents;
|
index < TERMINAL_LIMITS.maximumPendingEvents;
|
||||||
index += 1
|
index += 1
|
||||||
) {
|
) {
|
||||||
pty.emitData('x')
|
pty.emitData('x')
|
||||||
if (pty.pause.mock.calls.length > 0) {
|
// Each flush window yields one event.
|
||||||
break
|
vi.advanceTimersByTime(20)
|
||||||
|
if (pty.pause.mock.calls.length > 0) {
|
||||||
|
break
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
expect(pty.pause).toHaveBeenCalledTimes(1)
|
expect(pty.pause).toHaveBeenCalledTimes(1)
|
||||||
session.acknowledge(lastSequence)
|
expect(session.snapshot().state).toBe('running')
|
||||||
expect(pty.resume).toHaveBeenCalledTimes(1)
|
session.acknowledge(lastSequence)
|
||||||
|
expect(pty.resume).toHaveBeenCalledTimes(1)
|
||||||
|
} finally {
|
||||||
|
vi.useRealTimers()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
it('coalesces bursts into few ordered output events', async () => {
|
||||||
|
vi.useFakeTimers()
|
||||||
|
try {
|
||||||
|
const pty = new FakePty()
|
||||||
|
const session = await LocalTerminalSession.create({
|
||||||
|
...baseOptions,
|
||||||
|
dependencies: sessionDependencies(pty)
|
||||||
|
})
|
||||||
|
const events: TerminalEvent[] = []
|
||||||
|
session.onEvent((event) => events.push(event))
|
||||||
|
const expected: string[] = []
|
||||||
|
for (let index = 0; index < 1_000; index += 1) {
|
||||||
|
const chunk = `line ${index} 你好\r\n`
|
||||||
|
expected.push(chunk)
|
||||||
|
pty.emitData(chunk)
|
||||||
|
if (index % 100 === 0) {
|
||||||
|
vi.advanceTimersByTime(8)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
vi.advanceTimersByTime(20)
|
||||||
|
|
||||||
|
const output = events.filter((event) => event.type === 'output')
|
||||||
|
expect(output.map((event) => event.data).join('')).toBe(expected.join(''))
|
||||||
|
expect(output.length).toBeLessThan(25)
|
||||||
|
const sequences = events.map((event) => event.sequence)
|
||||||
|
expect(sequences).toEqual([...sequences].sort((a, b) => a - b))
|
||||||
|
expect(new Set(sequences).size).toBe(sequences.length)
|
||||||
|
await session.close()
|
||||||
|
} finally {
|
||||||
|
vi.useRealTimers()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
it('flushes buffered output before exit events and stops its timer', async () => {
|
||||||
|
vi.useFakeTimers()
|
||||||
|
try {
|
||||||
|
const pty = new FakePty()
|
||||||
|
const session = await LocalTerminalSession.create({
|
||||||
|
...baseOptions,
|
||||||
|
dependencies: sessionDependencies(pty)
|
||||||
|
})
|
||||||
|
const events: TerminalEvent[] = []
|
||||||
|
session.onEvent((event) => events.push(event))
|
||||||
|
pty.emitData('first ')
|
||||||
|
pty.emitData('trailing output')
|
||||||
|
pty.emitData(' tail')
|
||||||
|
pty.emitExit(0)
|
||||||
|
|
||||||
|
expect(
|
||||||
|
events.slice(2).map((event) =>
|
||||||
|
event.type === 'output' ? event.data : event.type
|
||||||
|
)
|
||||||
|
).toEqual(['first ', 'trailing output tail', 'exit', 'state'])
|
||||||
|
expect(vi.getTimerCount()).toBe(0)
|
||||||
|
vi.advanceTimersByTime(100)
|
||||||
|
expect(events).toHaveLength(6)
|
||||||
|
} finally {
|
||||||
|
vi.useRealTimers()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
it('flushes buffered output before the closing state on kill', async () => {
|
||||||
|
vi.useFakeTimers()
|
||||||
|
try {
|
||||||
|
const pty = new FakePty()
|
||||||
|
const session = await LocalTerminalSession.create({
|
||||||
|
...baseOptions,
|
||||||
|
dependencies: sessionDependencies(pty, {
|
||||||
|
terminate: vi.fn(async () => undefined)
|
||||||
|
})
|
||||||
|
})
|
||||||
|
const events: TerminalEvent[] = []
|
||||||
|
session.onEvent((event) => events.push(event))
|
||||||
|
pty.emitData('a')
|
||||||
|
pty.emitData('b')
|
||||||
|
await session.close()
|
||||||
|
|
||||||
|
expect(
|
||||||
|
events.slice(2).map((event) =>
|
||||||
|
event.type === 'output'
|
||||||
|
? event.data
|
||||||
|
: event.type === 'state'
|
||||||
|
? event.state
|
||||||
|
: event.type
|
||||||
|
)
|
||||||
|
).toEqual(['a', 'b', 'closing', 'exit', 'exited'])
|
||||||
|
expect(vi.getTimerCount()).toBe(0)
|
||||||
|
} finally {
|
||||||
|
vi.useRealTimers()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
it('counts buffered and in-flight bytes for backpressure and keeps them bounded', async () => {
|
||||||
|
vi.useFakeTimers()
|
||||||
|
try {
|
||||||
|
const pty = new FakePty()
|
||||||
|
const session = await LocalTerminalSession.create({
|
||||||
|
...baseOptions,
|
||||||
|
dependencies: sessionDependencies(pty)
|
||||||
|
})
|
||||||
|
const outputs: Extract<TerminalEvent, { type: 'output' }>[] = []
|
||||||
|
session.onEvent((event) => {
|
||||||
|
if (event.type === 'output') {
|
||||||
|
outputs.push(event)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
const block = 'q'.repeat(16 * 1024)
|
||||||
|
let produced = 0
|
||||||
|
// A renderer that never ACKs: output must pause well before the hard
|
||||||
|
// 4 MiB bound and never exceed it.
|
||||||
|
while (pty.pause.mock.calls.length === 0) {
|
||||||
|
pty.emitData(block)
|
||||||
|
produced += block.length
|
||||||
|
expect(produced).toBeLessThan(TERMINAL_LIMITS.maximumBufferedOutputBytes)
|
||||||
|
}
|
||||||
|
expect(session.snapshot().state).toBe('running')
|
||||||
|
const emittedBytes = outputs.reduce((sum, event) => sum + event.data.length, 0)
|
||||||
|
expect(produced - emittedBytes).toBeLessThan(TERMINAL_LIMITS.maximumEventBytes)
|
||||||
|
expect(
|
||||||
|
outputs.every(
|
||||||
|
(event) => Buffer.byteLength(event.data) <= TERMINAL_LIMITS.maximumEventBytes
|
||||||
|
)
|
||||||
|
).toBe(true)
|
||||||
|
|
||||||
|
// In-flight data after pause is absorbed by the headroom.
|
||||||
|
for (let index = 0; index < 16; index += 1) {
|
||||||
|
pty.emitData(block)
|
||||||
|
produced += block.length
|
||||||
|
}
|
||||||
|
vi.advanceTimersByTime(20)
|
||||||
|
expect(session.snapshot().state).toBe('running')
|
||||||
|
|
||||||
|
// A partial ACK that leaves more than half the budget pending must not
|
||||||
|
// resume; ACKing the rest must.
|
||||||
|
session.acknowledge(outputs[0]!.sequence)
|
||||||
|
expect(pty.resume).not.toHaveBeenCalled()
|
||||||
|
session.acknowledge(outputs.at(-1)!.sequence)
|
||||||
|
expect(pty.resume).toHaveBeenCalledTimes(1)
|
||||||
|
expect(
|
||||||
|
outputs.map((event) => event.data).join('').length
|
||||||
|
).toBe(produced)
|
||||||
|
await session.close()
|
||||||
|
} finally {
|
||||||
|
vi.useRealTimers()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
it('fails with output-limit-exceeded when the renderer never ACKs and the PTY ignores pause', async () => {
|
||||||
|
vi.useFakeTimers()
|
||||||
|
try {
|
||||||
|
const pty = new FakePty()
|
||||||
|
const session = await LocalTerminalSession.create({
|
||||||
|
...baseOptions,
|
||||||
|
dependencies: sessionDependencies(pty, {
|
||||||
|
terminate: vi.fn(async () => undefined)
|
||||||
|
})
|
||||||
|
})
|
||||||
|
const block = 'q'.repeat(64 * 1024)
|
||||||
|
for (
|
||||||
|
let produced = 0;
|
||||||
|
produced <= 2 * TERMINAL_LIMITS.maximumBufferedOutputBytes &&
|
||||||
|
session.snapshot().state === 'running';
|
||||||
|
produced += block.length
|
||||||
|
) {
|
||||||
|
pty.emitData(block)
|
||||||
|
}
|
||||||
|
expect(session.snapshot()).toMatchObject({
|
||||||
|
state: 'failed',
|
||||||
|
error: { code: 'output-limit-exceeded' }
|
||||||
|
})
|
||||||
|
expect(vi.getTimerCount()).toBe(0)
|
||||||
|
await session.close()
|
||||||
|
} finally {
|
||||||
|
vi.useRealTimers()
|
||||||
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
it('reports launch failures with stable failed state', async () => {
|
it('reports launch failures with stable failed state', async () => {
|
||||||
|
|||||||
@@ -16,6 +16,10 @@ import {
|
|||||||
} from '../../shared/terminal-contracts'
|
} from '../../shared/terminal-contracts'
|
||||||
import { requestProcessTreeTermination } from '../agent/child-process-termination'
|
import { requestProcessTreeTermination } from '../agent/child-process-termination'
|
||||||
import { buildCredentialFilteredUserEnvironment } from '../agent/process-environment'
|
import { buildCredentialFilteredUserEnvironment } from '../agent/process-environment'
|
||||||
|
import {
|
||||||
|
TerminalOutputCoalescer,
|
||||||
|
type TerminalOutputCoalescerOptions
|
||||||
|
} from './terminal-output-coalescer'
|
||||||
|
|
||||||
export type LocalTerminalTarget = Extract<
|
export type LocalTerminalTarget = Extract<
|
||||||
TerminalTarget,
|
TerminalTarget,
|
||||||
@@ -52,6 +56,7 @@ export type LocalTerminalDependencies = {
|
|||||||
directoryExists?: (path: string) => Promise<boolean>
|
directoryExists?: (path: string) => Promise<boolean>
|
||||||
spawn?: LocalPtySpawn
|
spawn?: LocalPtySpawn
|
||||||
terminate?: (pty: IPty, platform: NodeJS.Platform) => Promise<void>
|
terminate?: (pty: IPty, platform: NodeJS.Platform) => Promise<void>
|
||||||
|
outputCoalescing?: Omit<TerminalOutputCoalescerOptions, 'emit'>
|
||||||
}
|
}
|
||||||
|
|
||||||
export type LocalTerminalSessionOptions = {
|
export type LocalTerminalSessionOptions = {
|
||||||
@@ -72,11 +77,25 @@ type UnsequencedTerminalEvent<T> = T extends TerminalEvent
|
|||||||
type LocalTerminalEvent = UnsequencedTerminalEvent<TerminalEvent>
|
type LocalTerminalEvent = UnsequencedTerminalEvent<TerminalEvent>
|
||||||
|
|
||||||
const CONTROL_EVENT_RESERVE = 2
|
const CONTROL_EVENT_RESERVE = 2
|
||||||
const PAUSE_PENDING_EVENT_COUNT =
|
/** Hard limits: exceeding them fails the session (renderer stopped ACKing). */
|
||||||
|
const MAXIMUM_PENDING_OUTPUT_EVENTS =
|
||||||
TERMINAL_LIMITS.maximumPendingEvents - CONTROL_EVENT_RESERVE
|
TERMINAL_LIMITS.maximumPendingEvents - CONTROL_EVENT_RESERVE
|
||||||
const RESUME_PENDING_EVENT_COUNT = Math.floor(
|
/**
|
||||||
PAUSE_PENDING_EVENT_COUNT / 2
|
* Soft limits: pause the PTY with headroom below the hard limits, because a
|
||||||
|
* paused PTY may still deliver data that was already in flight. Coalesced
|
||||||
|
* events are up to 64 KiB, so the byte limit is normally reached first.
|
||||||
|
*/
|
||||||
|
const PAUSE_PENDING_EVENT_COUNT = Math.floor(
|
||||||
|
(MAXIMUM_PENDING_OUTPUT_EVENTS * 3) / 4
|
||||||
)
|
)
|
||||||
|
const PAUSE_PENDING_OUTPUT_BYTES = Math.floor(
|
||||||
|
(TERMINAL_LIMITS.maximumBufferedOutputBytes * 3) / 4
|
||||||
|
)
|
||||||
|
const RESUME_PENDING_EVENT_COUNT = Math.floor(
|
||||||
|
MAXIMUM_PENDING_OUTPUT_EVENTS / 2
|
||||||
|
)
|
||||||
|
const RESUME_PENDING_OUTPUT_BYTES =
|
||||||
|
TERMINAL_LIMITS.maximumBufferedOutputBytes / 2
|
||||||
const CLOSE_WAIT_MS = 1_500
|
const CLOSE_WAIT_MS = 1_500
|
||||||
const FORCE_WAIT_MS = 250
|
const FORCE_WAIT_MS = 250
|
||||||
|
|
||||||
@@ -248,32 +267,6 @@ export async function resolveLocalTerminalLaunch(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
function splitBoundedOutput(data: string): string[] {
|
|
||||||
const encoded = Buffer.from(data)
|
|
||||||
const chunks: string[] = []
|
|
||||||
let offset = 0
|
|
||||||
while (offset < encoded.byteLength) {
|
|
||||||
let end = Math.min(
|
|
||||||
offset + TERMINAL_LIMITS.maximumEventBytes,
|
|
||||||
encoded.byteLength
|
|
||||||
)
|
|
||||||
if (end < encoded.byteLength) {
|
|
||||||
while (
|
|
||||||
end > offset &&
|
|
||||||
(encoded[end]! & 0xc0) === 0x80
|
|
||||||
) {
|
|
||||||
end -= 1
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if (end === offset) {
|
|
||||||
end = Math.min(offset + 1, encoded.byteLength)
|
|
||||||
}
|
|
||||||
chunks.push(encoded.subarray(offset, end).toString('utf8'))
|
|
||||||
offset = end
|
|
||||||
}
|
|
||||||
return chunks
|
|
||||||
}
|
|
||||||
|
|
||||||
async function waitUntil(
|
async function waitUntil(
|
||||||
predicate: () => boolean,
|
predicate: () => boolean,
|
||||||
milliseconds: number
|
milliseconds: number
|
||||||
@@ -369,12 +362,17 @@ export class LocalTerminalSession {
|
|||||||
private exit: TerminalExit | null = null
|
private exit: TerminalExit | null = null
|
||||||
private error: TerminalError | null = null
|
private error: TerminalError | null = null
|
||||||
private closePromise: Promise<TerminalSnapshot> | undefined
|
private closePromise: Promise<TerminalSnapshot> | undefined
|
||||||
|
private readonly output: TerminalOutputCoalescer
|
||||||
|
|
||||||
private constructor(options: LocalTerminalSessionOptions) {
|
private constructor(options: LocalTerminalSessionOptions) {
|
||||||
this.options = options
|
this.options = options
|
||||||
this.dependencies = options.dependencies ?? {}
|
this.dependencies = options.dependencies ?? {}
|
||||||
this.sessionId = options.sessionId ?? randomUUID()
|
this.sessionId = options.sessionId ?? randomUUID()
|
||||||
this.size = { ...options.size }
|
this.size = { ...options.size }
|
||||||
|
this.output = new TerminalOutputCoalescer({
|
||||||
|
...this.dependencies.outputCoalescing,
|
||||||
|
emit: (data) => this.emitOutput(data)
|
||||||
|
})
|
||||||
this.emit({ type: 'state', state: 'starting' })
|
this.emit({ type: 'state', state: 'starting' })
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -487,8 +485,8 @@ export class LocalTerminalSession {
|
|||||||
if (
|
if (
|
||||||
this.paused &&
|
this.paused &&
|
||||||
this.pendingEvents.length <= RESUME_PENDING_EVENT_COUNT &&
|
this.pendingEvents.length <= RESUME_PENDING_EVENT_COUNT &&
|
||||||
this.pendingOutputBytes <=
|
this.pendingOutputBytes + this.output.bufferedBytes <=
|
||||||
TERMINAL_LIMITS.maximumBufferedOutputBytes / 2
|
RESUME_PENDING_OUTPUT_BYTES
|
||||||
) {
|
) {
|
||||||
this.paused = false
|
this.paused = false
|
||||||
this.pty?.resume()
|
this.pty?.resume()
|
||||||
@@ -545,31 +543,39 @@ export class LocalTerminalSession {
|
|||||||
if (this.state !== 'running' || data.length === 0) {
|
if (this.state !== 'running' || data.length === 0) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
for (const chunk of splitBoundedOutput(data)) {
|
// Bytes count against backpressure as soon as they are buffered, so a slow
|
||||||
if (
|
// renderer pauses the PTY regardless of the coalescing window.
|
||||||
this.pendingEvents.length >= PAUSE_PENDING_EVENT_COUNT ||
|
this.output.push(data)
|
||||||
this.pendingOutputBytes + Buffer.byteLength(chunk) >
|
this.updateBackpressure()
|
||||||
TERMINAL_LIMITS.maximumBufferedOutputBytes
|
}
|
||||||
) {
|
|
||||||
this.fail(
|
/** Called by the coalescer with ordered, bounded (<= 64 KiB) chunks. */
|
||||||
'output-limit-exceeded',
|
private emitOutput(chunk: string): void {
|
||||||
'Terminal output exceeded the bounded Main process buffer'
|
const bytes = Buffer.byteLength(chunk)
|
||||||
)
|
if (
|
||||||
void this.close()
|
this.pendingEvents.length >= MAXIMUM_PENDING_OUTPUT_EVENTS ||
|
||||||
return
|
this.pendingOutputBytes + bytes >
|
||||||
}
|
TERMINAL_LIMITS.maximumBufferedOutputBytes
|
||||||
this.pendingOutputBytes += Buffer.byteLength(chunk)
|
) {
|
||||||
this.emit({ type: 'output', data: chunk })
|
this.fail(
|
||||||
this.updateBackpressure()
|
'output-limit-exceeded',
|
||||||
|
'Terminal output exceeded the bounded Main process buffer'
|
||||||
|
)
|
||||||
|
void this.close()
|
||||||
|
return
|
||||||
}
|
}
|
||||||
|
this.pendingOutputBytes += bytes
|
||||||
|
this.emit({ type: 'output', data: chunk })
|
||||||
|
this.updateBackpressure()
|
||||||
}
|
}
|
||||||
|
|
||||||
private updateBackpressure(): void {
|
private updateBackpressure(): void {
|
||||||
if (
|
if (
|
||||||
!this.paused &&
|
!this.paused &&
|
||||||
|
this.state === 'running' &&
|
||||||
(this.pendingEvents.length >= PAUSE_PENDING_EVENT_COUNT ||
|
(this.pendingEvents.length >= PAUSE_PENDING_EVENT_COUNT ||
|
||||||
this.pendingOutputBytes >=
|
this.pendingOutputBytes + this.output.bufferedBytes >=
|
||||||
TERMINAL_LIMITS.maximumBufferedOutputBytes)
|
PAUSE_PENDING_OUTPUT_BYTES)
|
||||||
) {
|
) {
|
||||||
this.paused = true
|
this.paused = true
|
||||||
this.pty?.pause()
|
this.pty?.pause()
|
||||||
@@ -599,8 +605,10 @@ export class LocalTerminalSession {
|
|||||||
this.exit.signal = 'unknown'
|
this.exit.signal = 'unknown'
|
||||||
}
|
}
|
||||||
this.state = 'exited'
|
this.state = 'exited'
|
||||||
|
// emit() flushes buffered output first, so trailing output precedes exit.
|
||||||
this.emit({ type: 'exit', exit: this.exit })
|
this.emit({ type: 'exit', exit: this.exit })
|
||||||
this.emit({ type: 'state', state: 'exited' })
|
this.emit({ type: 'state', state: 'exited' })
|
||||||
|
this.output.dispose()
|
||||||
}
|
}
|
||||||
|
|
||||||
private fail(
|
private fail(
|
||||||
@@ -618,6 +626,9 @@ export class LocalTerminalSession {
|
|||||||
retryable: code === 'launch-failed'
|
retryable: code === 'launch-failed'
|
||||||
}
|
}
|
||||||
this.state = 'failed'
|
this.state = 'failed'
|
||||||
|
// Failure is terminal: drop buffered output (it is what exceeded the
|
||||||
|
// bound) and cancel the flush timer before announcing the error.
|
||||||
|
this.output.dispose()
|
||||||
this.emit({ type: 'error', error: this.error })
|
this.emit({ type: 'error', error: this.error })
|
||||||
this.emit({ type: 'state', state: 'failed' })
|
this.emit({ type: 'state', state: 'failed' })
|
||||||
}
|
}
|
||||||
@@ -625,6 +636,10 @@ export class LocalTerminalSession {
|
|||||||
private emit(
|
private emit(
|
||||||
event: LocalTerminalEvent
|
event: LocalTerminalEvent
|
||||||
): void {
|
): void {
|
||||||
|
if (event.type !== 'output') {
|
||||||
|
// Control events must follow every byte the PTY produced before them.
|
||||||
|
this.output.flush()
|
||||||
|
}
|
||||||
if (this.lastSequence >= Number.MAX_SAFE_INTEGER) {
|
if (this.lastSequence >= Number.MAX_SAFE_INTEGER) {
|
||||||
this.state = 'failed'
|
this.state = 'failed'
|
||||||
this.error = {
|
this.error = {
|
||||||
|
|||||||
@@ -185,21 +185,83 @@ describe("SSH terminal session", () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
it("emits ordered bounded output and pauses until acknowledged", async () => {
|
it("emits ordered bounded output and pauses until acknowledged", async () => {
|
||||||
const harness = createHarness();
|
vi.useFakeTimers();
|
||||||
const session = await openSession(harness);
|
try {
|
||||||
|
const harness = createHarness();
|
||||||
|
const session = await openSession(harness);
|
||||||
|
|
||||||
for (let index = 0; index < 255; index += 1) {
|
for (let index = 0; index < 255; index += 1) {
|
||||||
harness.channel.emit("data", `chunk-${index}`);
|
harness.channel.emit("data", `chunk-${index}`);
|
||||||
|
// One flush window per chunk keeps one event per chunk.
|
||||||
|
vi.advanceTimersByTime(20);
|
||||||
|
}
|
||||||
|
|
||||||
|
expect(harness.events).toHaveLength(256);
|
||||||
|
expect(harness.events.map((event) => event.sequence)).toEqual(
|
||||||
|
Array.from({ length: 256 }, (_, index) => index + 1),
|
||||||
|
);
|
||||||
|
expect(harness.channel.pause).toHaveBeenCalledOnce();
|
||||||
|
session.acknowledge(256);
|
||||||
|
expect(harness.channel.resume).toHaveBeenCalledOnce();
|
||||||
|
session.close();
|
||||||
|
} finally {
|
||||||
|
vi.useRealTimers();
|
||||||
}
|
}
|
||||||
|
});
|
||||||
|
|
||||||
expect(harness.events).toHaveLength(256);
|
it("coalesces bursts in order and flushes trailing output before exit", async () => {
|
||||||
expect(harness.events.map((event) => event.sequence)).toEqual(
|
vi.useFakeTimers();
|
||||||
Array.from({ length: 256 }, (_, index) => index + 1),
|
try {
|
||||||
);
|
const harness = createHarness();
|
||||||
expect(harness.channel.pause).toHaveBeenCalledOnce();
|
await openSession(harness);
|
||||||
session.acknowledge(256);
|
const expected: string[] = [];
|
||||||
expect(harness.channel.resume).toHaveBeenCalledOnce();
|
for (let index = 0; index < 500; index += 1) {
|
||||||
session.close();
|
const chunk = `chunk-${index};`;
|
||||||
|
expected.push(chunk);
|
||||||
|
harness.channel.emit("data", Buffer.from(chunk));
|
||||||
|
}
|
||||||
|
// Split multi-byte character across two SSH packets.
|
||||||
|
const bytes = Buffer.from("你");
|
||||||
|
harness.channel.emit("data", bytes.subarray(0, 1));
|
||||||
|
harness.channel.emit("data", bytes.subarray(1));
|
||||||
|
expected.push("你");
|
||||||
|
harness.channel.emit("exit", 0);
|
||||||
|
harness.channel.emit("close");
|
||||||
|
|
||||||
|
const output = harness.events.filter((event) => event.type === "output");
|
||||||
|
expect(output.map((event) => event.data).join("")).toBe(expected.join(""));
|
||||||
|
expect(output.length).toBeLessThanOrEqual(2);
|
||||||
|
expect(harness.events.slice(-2).map((event) => event.type)).toEqual([
|
||||||
|
"exit",
|
||||||
|
"state",
|
||||||
|
]);
|
||||||
|
const count = harness.events.length;
|
||||||
|
vi.advanceTimersByTime(100);
|
||||||
|
expect(harness.events).toHaveLength(count);
|
||||||
|
} finally {
|
||||||
|
vi.useRealTimers();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
it("flushes buffered output before closing on kill", async () => {
|
||||||
|
vi.useFakeTimers();
|
||||||
|
try {
|
||||||
|
const harness = createHarness();
|
||||||
|
const session = await openSession(harness);
|
||||||
|
harness.channel.emit("data", "a");
|
||||||
|
harness.channel.emit("data", "b");
|
||||||
|
session.close();
|
||||||
|
expect(
|
||||||
|
harness.events.slice(1).map((event) =>
|
||||||
|
event.type === "output" ? event.data : event.type === "state" ? event.state : event.type,
|
||||||
|
),
|
||||||
|
).toEqual(["a", "b", "closing", "exit", "exited"]);
|
||||||
|
const count = harness.events.length;
|
||||||
|
vi.advanceTimersByTime(100);
|
||||||
|
expect(harness.events).toHaveLength(count);
|
||||||
|
} finally {
|
||||||
|
vi.useRealTimers();
|
||||||
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
it("marks a disconnected SSH transport interrupted immediately and rejects late I/O", async () => {
|
it("marks a disconnected SSH transport interrupted immediately and rejects late I/O", async () => {
|
||||||
|
|||||||
@@ -16,6 +16,10 @@ import {
|
|||||||
type SshConnectionPoolTarget,
|
type SshConnectionPoolTarget,
|
||||||
type SshTerminalConnectionLease,
|
type SshTerminalConnectionLease,
|
||||||
} from "../ssh/ssh-connection-pool";
|
} from "../ssh/ssh-connection-pool";
|
||||||
|
import {
|
||||||
|
TerminalOutputCoalescer,
|
||||||
|
type TerminalOutputCoalescerOptions,
|
||||||
|
} from "./terminal-output-coalescer";
|
||||||
|
|
||||||
const SSH_TERMINAL_TERM = "xterm-256color";
|
const SSH_TERMINAL_TERM = "xterm-256color";
|
||||||
|
|
||||||
@@ -71,29 +75,13 @@ function eventByteLength(event: TerminalEvent): number {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
function splitUtf8(text: string, maximumBytes: number): string[] {
|
const PAUSE_PENDING_BYTES =
|
||||||
const chunks: string[] = [];
|
TERMINAL_LIMITS.maximumBufferedOutputBytes - TERMINAL_LIMITS.maximumEventBytes;
|
||||||
let chunk = "";
|
|
||||||
let chunkBytes = 0;
|
|
||||||
for (const character of text) {
|
|
||||||
const characterBytes = Buffer.byteLength(character);
|
|
||||||
if (chunk && chunkBytes + characterBytes > maximumBytes) {
|
|
||||||
chunks.push(chunk);
|
|
||||||
chunk = "";
|
|
||||||
chunkBytes = 0;
|
|
||||||
}
|
|
||||||
chunk += character;
|
|
||||||
chunkBytes += characterBytes;
|
|
||||||
}
|
|
||||||
if (chunk) {
|
|
||||||
chunks.push(chunk);
|
|
||||||
}
|
|
||||||
return chunks;
|
|
||||||
}
|
|
||||||
|
|
||||||
export async function createSshTerminalSession(
|
export async function createSshTerminalSession(
|
||||||
pool: SshConnectionPool,
|
pool: SshConnectionPool,
|
||||||
options: SshTerminalSessionOptions,
|
options: SshTerminalSessionOptions,
|
||||||
|
coalescing?: Omit<TerminalOutputCoalescerOptions, "emit">,
|
||||||
): Promise<SshTerminalSession> {
|
): Promise<SshTerminalSession> {
|
||||||
const initialSize = terminalSizeSchema.parse(options.size);
|
const initialSize = terminalSizeSchema.parse(options.size);
|
||||||
const workingDirectoryCommand = options.workingDirectory && !options.command
|
const workingDirectoryCommand = options.workingDirectory && !options.command
|
||||||
@@ -128,7 +116,22 @@ export async function createSshTerminalSession(
|
|||||||
const decoder = new StringDecoder("utf8");
|
const decoder = new StringDecoder("utf8");
|
||||||
let running: RunningSession | undefined;
|
let running: RunningSession | undefined;
|
||||||
|
|
||||||
|
const updateBackpressure = (): void => {
|
||||||
|
if (
|
||||||
|
!paused &&
|
||||||
|
(pendingEvents.length >= TERMINAL_LIMITS.maximumPendingEvents ||
|
||||||
|
pendingBytes + output.bufferedBytes >= PAUSE_PENDING_BYTES)
|
||||||
|
) {
|
||||||
|
paused = true;
|
||||||
|
channel.pause();
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
const emit = (event: TerminalEventPayload): void => {
|
const emit = (event: TerminalEventPayload): void => {
|
||||||
|
if (event.type !== "output") {
|
||||||
|
// Control events must follow every byte received before them.
|
||||||
|
output.flush();
|
||||||
|
}
|
||||||
sequence += 1;
|
sequence += 1;
|
||||||
const sequenced = {
|
const sequenced = {
|
||||||
...event,
|
...event,
|
||||||
@@ -139,19 +142,18 @@ export async function createSshTerminalSession(
|
|||||||
pendingEvents.push({ event: sequenced, bytes });
|
pendingEvents.push({ event: sequenced, bytes });
|
||||||
pendingBytes += bytes;
|
pendingBytes += bytes;
|
||||||
options.onEvent(sequenced);
|
options.onEvent(sequenced);
|
||||||
if (
|
updateBackpressure();
|
||||||
!paused &&
|
|
||||||
(pendingEvents.length >= TERMINAL_LIMITS.maximumPendingEvents ||
|
|
||||||
pendingBytes >=
|
|
||||||
TERMINAL_LIMITS.maximumBufferedOutputBytes -
|
|
||||||
TERMINAL_LIMITS.maximumEventBytes)
|
|
||||||
) {
|
|
||||||
paused = true;
|
|
||||||
channel.pause();
|
|
||||||
}
|
|
||||||
};
|
};
|
||||||
|
|
||||||
|
const output = new TerminalOutputCoalescer({
|
||||||
|
...coalescing,
|
||||||
|
emit: (data) => emit({ type: "output", data }),
|
||||||
|
});
|
||||||
|
|
||||||
const release = (): void => {
|
const release = (): void => {
|
||||||
|
// Every terminal path calls release() after its final events; stop the
|
||||||
|
// flush timer and drop anything that can no longer be delivered.
|
||||||
|
output.dispose();
|
||||||
const active = running;
|
const active = running;
|
||||||
running = undefined;
|
running = undefined;
|
||||||
active?.removeDisconnectListener();
|
active?.removeDisconnectListener();
|
||||||
@@ -183,12 +185,8 @@ export async function createSshTerminalSession(
|
|||||||
}
|
}
|
||||||
const text =
|
const text =
|
||||||
typeof chunk === "string" ? chunk : decoder.write(Buffer.from(chunk));
|
typeof chunk === "string" ? chunk : decoder.write(Buffer.from(chunk));
|
||||||
for (const data of splitUtf8(
|
output.push(text);
|
||||||
text,
|
updateBackpressure();
|
||||||
TERMINAL_LIMITS.maximumEventBytes,
|
|
||||||
)) {
|
|
||||||
emit({ type: "output", data });
|
|
||||||
}
|
|
||||||
});
|
});
|
||||||
channel.once(
|
channel.once(
|
||||||
"exit",
|
"exit",
|
||||||
@@ -210,15 +208,8 @@ export async function createSshTerminalSession(
|
|||||||
release();
|
release();
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
const finalText = decoder.end();
|
output.push(decoder.end());
|
||||||
if (finalText) {
|
output.flush();
|
||||||
for (const data of splitUtf8(
|
|
||||||
finalText,
|
|
||||||
TERMINAL_LIMITS.maximumEventBytes,
|
|
||||||
)) {
|
|
||||||
emit({ type: "output", data });
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if (!lease.isUsable()) {
|
if (!lease.isUsable()) {
|
||||||
interrupt();
|
interrupt();
|
||||||
return;
|
return;
|
||||||
@@ -279,9 +270,7 @@ export async function createSshTerminalSession(
|
|||||||
if (
|
if (
|
||||||
paused &&
|
paused &&
|
||||||
pendingEvents.length < TERMINAL_LIMITS.maximumPendingEvents &&
|
pendingEvents.length < TERMINAL_LIMITS.maximumPendingEvents &&
|
||||||
pendingBytes <
|
pendingBytes + output.bufferedBytes < PAUSE_PENDING_BYTES
|
||||||
TERMINAL_LIMITS.maximumBufferedOutputBytes -
|
|
||||||
TERMINAL_LIMITS.maximumEventBytes
|
|
||||||
) {
|
) {
|
||||||
paused = false;
|
paused = false;
|
||||||
channel.resume();
|
channel.resume();
|
||||||
|
|||||||
@@ -0,0 +1,165 @@
|
|||||||
|
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||||
|
import { TERMINAL_LIMITS } from '../../shared/terminal-contracts'
|
||||||
|
import {
|
||||||
|
TERMINAL_OUTPUT_FLUSH_INTERVAL_MS,
|
||||||
|
TerminalOutputCoalescer,
|
||||||
|
splitUtf8Bounded
|
||||||
|
} from './terminal-output-coalescer'
|
||||||
|
|
||||||
|
describe('TerminalOutputCoalescer', () => {
|
||||||
|
beforeEach(() => {
|
||||||
|
vi.useFakeTimers()
|
||||||
|
})
|
||||||
|
afterEach(() => {
|
||||||
|
vi.useRealTimers()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('emits the first chunk immediately and merges a burst into one flush', () => {
|
||||||
|
const emitted: string[] = []
|
||||||
|
const coalescer = new TerminalOutputCoalescer({
|
||||||
|
emit: (data) => emitted.push(data)
|
||||||
|
})
|
||||||
|
|
||||||
|
coalescer.push('$ ')
|
||||||
|
expect(emitted).toEqual(['$ '])
|
||||||
|
for (let index = 0; index < 100; index += 1) {
|
||||||
|
coalescer.push(`${index},`)
|
||||||
|
}
|
||||||
|
expect(emitted).toHaveLength(1)
|
||||||
|
vi.advanceTimersByTime(TERMINAL_OUTPUT_FLUSH_INTERVAL_MS)
|
||||||
|
expect(emitted).toHaveLength(2)
|
||||||
|
expect(emitted.join('')).toBe(
|
||||||
|
'$ ' + Array.from({ length: 100 }, (_, index) => `${index},`).join('')
|
||||||
|
)
|
||||||
|
coalescer.dispose()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('preserves exact order across timer, size-cap and explicit flushes', () => {
|
||||||
|
const emitted: string[] = []
|
||||||
|
const coalescer = new TerminalOutputCoalescer({
|
||||||
|
emit: (data) => emitted.push(data),
|
||||||
|
maximumBatchBytes: 16
|
||||||
|
})
|
||||||
|
const expected: string[] = []
|
||||||
|
for (let index = 0; index < 500; index += 1) {
|
||||||
|
const chunk = index % 7 === 0 ? `你好${index}` : `#${index};`
|
||||||
|
expected.push(chunk)
|
||||||
|
coalescer.push(chunk)
|
||||||
|
if (index % 13 === 0) {
|
||||||
|
vi.advanceTimersByTime(TERMINAL_OUTPUT_FLUSH_INTERVAL_MS)
|
||||||
|
}
|
||||||
|
if (index % 97 === 0) {
|
||||||
|
coalescer.flush()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
coalescer.flush()
|
||||||
|
|
||||||
|
expect(emitted.join('')).toBe(expected.join(''))
|
||||||
|
expect(emitted.every((chunk) => Buffer.byteLength(chunk) <= 16)).toBe(true)
|
||||||
|
expect(coalescer.bufferedBytes).toBe(0)
|
||||||
|
coalescer.dispose()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('emits full batches at the size cap without waiting for the timer', () => {
|
||||||
|
const emitted: string[] = []
|
||||||
|
const coalescer = new TerminalOutputCoalescer({
|
||||||
|
emit: (data) => emitted.push(data)
|
||||||
|
})
|
||||||
|
const cap = TERMINAL_LIMITS.maximumEventBytes
|
||||||
|
coalescer.push('x')
|
||||||
|
const block = 'y'.repeat(4096)
|
||||||
|
for (let index = 0; index < 40; index += 1) {
|
||||||
|
coalescer.push(block)
|
||||||
|
}
|
||||||
|
|
||||||
|
// 160 KiB buffered inside one window: two full 64 KiB batches go out
|
||||||
|
// immediately; the remainder waits for the timer.
|
||||||
|
expect(emitted.slice(1).map((chunk) => Buffer.byteLength(chunk))).toEqual([
|
||||||
|
cap,
|
||||||
|
cap
|
||||||
|
])
|
||||||
|
expect(coalescer.bufferedBytes).toBe(40 * 4096 - 2 * cap)
|
||||||
|
expect(coalescer.bufferedBytes).toBeLessThan(cap)
|
||||||
|
vi.advanceTimersByTime(TERMINAL_OUTPUT_FLUSH_INTERVAL_MS)
|
||||||
|
expect(coalescer.bufferedBytes).toBe(0)
|
||||||
|
expect(emitted.join('')).toBe('x' + block.repeat(40))
|
||||||
|
coalescer.dispose()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('never splits a multi-byte UTF-8 character across batches', () => {
|
||||||
|
const emitted: string[] = []
|
||||||
|
const coalescer = new TerminalOutputCoalescer({
|
||||||
|
emit: (data) => emitted.push(data),
|
||||||
|
maximumBatchBytes: 5
|
||||||
|
})
|
||||||
|
coalescer.push('a')
|
||||||
|
coalescer.push('你好世界🙂')
|
||||||
|
coalescer.flush()
|
||||||
|
expect(emitted.join('')).toBe('a你好世界🙂')
|
||||||
|
for (const chunk of emitted) {
|
||||||
|
expect(Buffer.from(chunk).toString('utf8')).toBe(chunk)
|
||||||
|
expect(chunk).not.toContain('\uFFFD')
|
||||||
|
}
|
||||||
|
coalescer.dispose()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('dispose cancels the timer and drops later input', () => {
|
||||||
|
const emit = vi.fn()
|
||||||
|
const coalescer = new TerminalOutputCoalescer({ emit })
|
||||||
|
coalescer.push('a')
|
||||||
|
coalescer.push('b')
|
||||||
|
coalescer.dispose()
|
||||||
|
expect(vi.getTimerCount()).toBe(0)
|
||||||
|
coalescer.push('c')
|
||||||
|
coalescer.flush()
|
||||||
|
vi.advanceTimersByTime(100)
|
||||||
|
expect(emit.mock.calls).toEqual([['a']])
|
||||||
|
})
|
||||||
|
|
||||||
|
it('stops scheduling once idle', () => {
|
||||||
|
const coalescer = new TerminalOutputCoalescer({ emit: vi.fn() })
|
||||||
|
coalescer.push('a')
|
||||||
|
vi.advanceTimersByTime(TERMINAL_OUTPUT_FLUSH_INTERVAL_MS)
|
||||||
|
expect(vi.getTimerCount()).toBe(0)
|
||||||
|
coalescer.dispose()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('reduces IPC events by orders of magnitude for a 20 MB flood', () => {
|
||||||
|
// Model node-pty delivering ~4 KiB reads; compare the number of output
|
||||||
|
// events (one IPC send each) before and after coalescing.
|
||||||
|
const totalBytes = 20 * 1024 * 1024
|
||||||
|
const readBytes = 4096
|
||||||
|
const read = 'z'.repeat(readBytes)
|
||||||
|
let coalescedEvents = 0
|
||||||
|
let coalescedBytes = 0
|
||||||
|
const coalescer = new TerminalOutputCoalescer({
|
||||||
|
emit: (data) => {
|
||||||
|
coalescedEvents += 1
|
||||||
|
coalescedBytes += data.length
|
||||||
|
}
|
||||||
|
})
|
||||||
|
const reads = totalBytes / readBytes
|
||||||
|
for (let index = 0; index < reads; index += 1) {
|
||||||
|
coalescer.push(read)
|
||||||
|
if (index % 64 === 0) {
|
||||||
|
vi.advanceTimersByTime(1)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
coalescer.flush()
|
||||||
|
coalescer.dispose()
|
||||||
|
|
||||||
|
expect(coalescedBytes).toBe(totalBytes)
|
||||||
|
// Uncoalesced: one event per read (5120). Coalesced: ~64 KiB per event.
|
||||||
|
expect(coalescedEvents).toBeLessThanOrEqual(
|
||||||
|
Math.ceil(totalBytes / TERMINAL_LIMITS.maximumEventBytes) + 100
|
||||||
|
)
|
||||||
|
expect(coalescedEvents).toBeLessThan(reads / 10)
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|
||||||
|
describe('splitUtf8Bounded', () => {
|
||||||
|
it('returns no chunks for empty input and a single chunk when small', () => {
|
||||||
|
expect(splitUtf8Bounded('', 4)).toEqual([])
|
||||||
|
expect(splitUtf8Bounded('abc', 4)).toEqual(['abc'])
|
||||||
|
})
|
||||||
|
})
|
||||||
@@ -0,0 +1,161 @@
|
|||||||
|
import { TERMINAL_LIMITS } from '../../shared/terminal-contracts'
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Minimum spacing between timer-driven flushes. Output arriving while the
|
||||||
|
* coalescer is idle is flushed immediately (leading edge), so interactive
|
||||||
|
* echo gets no added latency; bursts inside the window are merged.
|
||||||
|
*/
|
||||||
|
export const TERMINAL_OUTPUT_FLUSH_INTERVAL_MS = 8
|
||||||
|
|
||||||
|
/** A full batch is flushed without waiting for the timer. */
|
||||||
|
export const TERMINAL_OUTPUT_MAXIMUM_BATCH_BYTES =
|
||||||
|
TERMINAL_LIMITS.maximumEventBytes
|
||||||
|
|
||||||
|
type TimerHandle = ReturnType<typeof setTimeout>
|
||||||
|
|
||||||
|
export type TerminalOutputCoalescerOptions = {
|
||||||
|
/** Receives ordered chunks of at most `maximumBatchBytes` UTF-8 bytes. */
|
||||||
|
emit: (data: string) => void
|
||||||
|
flushIntervalMs?: number
|
||||||
|
maximumBatchBytes?: number
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Splits text into ordered chunks of at most `maximumBytes` UTF-8 bytes without
|
||||||
|
* cutting a multi-byte sequence.
|
||||||
|
*/
|
||||||
|
export function splitUtf8Bounded(
|
||||||
|
data: string,
|
||||||
|
maximumBytes: number
|
||||||
|
): string[] {
|
||||||
|
const encoded = Buffer.from(data)
|
||||||
|
if (encoded.byteLength <= maximumBytes) {
|
||||||
|
return encoded.byteLength === 0 ? [] : [data]
|
||||||
|
}
|
||||||
|
const chunks: string[] = []
|
||||||
|
let offset = 0
|
||||||
|
while (offset < encoded.byteLength) {
|
||||||
|
let end = Math.min(offset + maximumBytes, encoded.byteLength)
|
||||||
|
if (end < encoded.byteLength) {
|
||||||
|
while (end > offset && (encoded[end]! & 0xc0) === 0x80) {
|
||||||
|
end -= 1
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (end === offset) {
|
||||||
|
end = Math.min(offset + 1, encoded.byteLength)
|
||||||
|
}
|
||||||
|
chunks.push(encoded.subarray(offset, end).toString('utf8'))
|
||||||
|
offset = end
|
||||||
|
}
|
||||||
|
return chunks
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Per-terminal output buffer in Main. It never holds more than one batch
|
||||||
|
* (`maximumBatchBytes`), so a slow renderer is handled by the session's
|
||||||
|
* ACK-based backpressure rather than by growing this buffer.
|
||||||
|
*/
|
||||||
|
export class TerminalOutputCoalescer {
|
||||||
|
private readonly emitChunk: (data: string) => void
|
||||||
|
private readonly flushIntervalMs: number
|
||||||
|
private readonly maximumBatchBytes: number
|
||||||
|
private chunks: string[] = []
|
||||||
|
private bytes = 0
|
||||||
|
private timer: TimerHandle | undefined
|
||||||
|
private disposed = false
|
||||||
|
|
||||||
|
constructor(options: TerminalOutputCoalescerOptions) {
|
||||||
|
this.emitChunk = options.emit
|
||||||
|
this.flushIntervalMs =
|
||||||
|
options.flushIntervalMs ?? TERMINAL_OUTPUT_FLUSH_INTERVAL_MS
|
||||||
|
this.maximumBatchBytes =
|
||||||
|
options.maximumBatchBytes ?? TERMINAL_OUTPUT_MAXIMUM_BATCH_BYTES
|
||||||
|
}
|
||||||
|
|
||||||
|
/** UTF-8 bytes received but not yet emitted. */
|
||||||
|
get bufferedBytes(): number {
|
||||||
|
return this.bytes
|
||||||
|
}
|
||||||
|
|
||||||
|
get isDisposed(): boolean {
|
||||||
|
return this.disposed
|
||||||
|
}
|
||||||
|
|
||||||
|
push(data: string): void {
|
||||||
|
if (this.disposed || data.length === 0) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
this.chunks.push(data)
|
||||||
|
this.bytes += Buffer.byteLength(data)
|
||||||
|
if (this.timer === undefined) {
|
||||||
|
this.flush()
|
||||||
|
this.startWindow()
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if (this.bytes >= this.maximumBatchBytes) {
|
||||||
|
this.emitFullBatches()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Emits everything buffered, in order. Safe to call re-entrantly. */
|
||||||
|
flush(): void {
|
||||||
|
if (this.disposed || this.bytes === 0) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
const data = this.take()
|
||||||
|
for (const chunk of splitUtf8Bounded(data, this.maximumBatchBytes)) {
|
||||||
|
if (this.disposed) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
this.emitChunk(chunk)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Drops buffered data and cancels the timer. Further input is ignored. */
|
||||||
|
dispose(): void {
|
||||||
|
this.disposed = true
|
||||||
|
this.chunks = []
|
||||||
|
this.bytes = 0
|
||||||
|
if (this.timer !== undefined) {
|
||||||
|
clearTimeout(this.timer)
|
||||||
|
this.timer = undefined
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private emitFullBatches(): void {
|
||||||
|
const parts = splitUtf8Bounded(this.take(), this.maximumBatchBytes)
|
||||||
|
const last = parts.at(-1)
|
||||||
|
if (last !== undefined && Buffer.byteLength(last) < this.maximumBatchBytes) {
|
||||||
|
parts.pop()
|
||||||
|
this.chunks = [last]
|
||||||
|
this.bytes = Buffer.byteLength(last)
|
||||||
|
}
|
||||||
|
for (const chunk of parts) {
|
||||||
|
if (this.disposed) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
this.emitChunk(chunk)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private take(): string {
|
||||||
|
const data = this.chunks.length === 1 ? this.chunks[0]! : this.chunks.join('')
|
||||||
|
this.chunks = []
|
||||||
|
this.bytes = 0
|
||||||
|
return data
|
||||||
|
}
|
||||||
|
|
||||||
|
private startWindow(): void {
|
||||||
|
if (this.disposed) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
this.timer = setTimeout(() => {
|
||||||
|
this.timer = undefined
|
||||||
|
if (this.bytes > 0) {
|
||||||
|
this.flush()
|
||||||
|
this.startWindow()
|
||||||
|
}
|
||||||
|
}, this.flushIntervalMs)
|
||||||
|
this.timer.unref?.()
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -8,7 +8,9 @@ import type {
|
|||||||
} from '../../shared/terminal-contracts'
|
} from '../../shared/terminal-contracts'
|
||||||
import {
|
import {
|
||||||
TerminalPanel,
|
TerminalPanel,
|
||||||
|
resetTerminalRendererWarningsForTests,
|
||||||
resolveTerminalFontFamily,
|
resolveTerminalFontFamily,
|
||||||
|
warnTerminalRendererFallback,
|
||||||
type TerminalAdapter,
|
type TerminalAdapter,
|
||||||
type TerminalClipboardAdapter,
|
type TerminalClipboardAdapter,
|
||||||
type TerminalEmulator,
|
type TerminalEmulator,
|
||||||
@@ -483,6 +485,120 @@ describe('TerminalPanel', () => {
|
|||||||
)
|
)
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it('batches output into one write per frame and ACKs after the write callback', async () => {
|
||||||
|
const frames = new Set<() => void>()
|
||||||
|
const frameScheduler = (callback: () => void): (() => void) => {
|
||||||
|
frames.add(callback)
|
||||||
|
return () => frames.delete(callback)
|
||||||
|
}
|
||||||
|
const runFrame = (): void => {
|
||||||
|
const current = [...frames]
|
||||||
|
frames.clear()
|
||||||
|
current.forEach((callback) => callback())
|
||||||
|
}
|
||||||
|
const backend = adapterHarness()
|
||||||
|
const emulator = emulatorHarness()
|
||||||
|
const callbacks: Array<() => void> = []
|
||||||
|
emulator.write.mockImplementation((_: string, callback?: () => void) => {
|
||||||
|
if (callback) {
|
||||||
|
callbacks.push(callback)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
render(
|
||||||
|
<TerminalPanel
|
||||||
|
adapter={backend.adapter}
|
||||||
|
frameScheduler={frameScheduler}
|
||||||
|
target={{ type: 'local' }}
|
||||||
|
terminalFactory={emulator.factory}
|
||||||
|
/>
|
||||||
|
)
|
||||||
|
resizeObserver.trigger()
|
||||||
|
await waitFor(() =>
|
||||||
|
expect(backend.adapter.create).toHaveBeenCalledOnce()
|
||||||
|
)
|
||||||
|
|
||||||
|
for (let sequence = 1; sequence <= 5; sequence += 1) {
|
||||||
|
backend.emit({ sessionId, sequence, type: 'output', data: `${sequence};` })
|
||||||
|
}
|
||||||
|
await waitFor(() => expect(frames.size).toBe(1))
|
||||||
|
expect(emulator.write).not.toHaveBeenCalled()
|
||||||
|
runFrame()
|
||||||
|
expect(emulator.write).toHaveBeenCalledOnce()
|
||||||
|
expect(emulator.write).toHaveBeenCalledWith('1;2;3;4;5;', expect.any(Function))
|
||||||
|
expect(backend.adapter.ack).not.toHaveBeenCalled()
|
||||||
|
callbacks.shift()?.()
|
||||||
|
expect(backend.adapter.ack).toHaveBeenCalledOnce()
|
||||||
|
expect(backend.adapter.ack).toHaveBeenCalledWith({ sessionId, sequence: 5 })
|
||||||
|
|
||||||
|
// A control event flushes pending output first and is applied after it.
|
||||||
|
backend.emit({ sessionId, sequence: 6, type: 'output', data: 'bye' })
|
||||||
|
backend.emit({
|
||||||
|
sessionId,
|
||||||
|
sequence: 7,
|
||||||
|
type: 'exit',
|
||||||
|
exit: { exitCode: 0, signal: null }
|
||||||
|
})
|
||||||
|
await waitFor(() =>
|
||||||
|
expect(emulator.write).toHaveBeenLastCalledWith('bye', expect.any(Function))
|
||||||
|
)
|
||||||
|
expect(screen.queryByText('已退出')).not.toBeInTheDocument()
|
||||||
|
callbacks.shift()?.()
|
||||||
|
await waitFor(() =>
|
||||||
|
expect(backend.adapter.ack).toHaveBeenLastCalledWith({ sessionId, sequence: 7 })
|
||||||
|
)
|
||||||
|
expect(backend.adapter.ack).toHaveBeenCalledWith({ sessionId, sequence: 6 })
|
||||||
|
expect(emulator.write).toHaveBeenCalledTimes(2)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('keeps writing output for a hidden terminal tab', async () => {
|
||||||
|
const backend = adapterHarness()
|
||||||
|
const emulator = emulatorHarness()
|
||||||
|
render(
|
||||||
|
<div hidden>
|
||||||
|
<TerminalPanel
|
||||||
|
adapter={backend.adapter}
|
||||||
|
sessionId={sessionId}
|
||||||
|
target={{ type: 'local' }}
|
||||||
|
terminalFactory={emulator.factory}
|
||||||
|
/>
|
||||||
|
</div>
|
||||||
|
)
|
||||||
|
await waitFor(() =>
|
||||||
|
expect(backend.adapter.getSnapshot).toHaveBeenCalled()
|
||||||
|
)
|
||||||
|
await screen.findByText('PowerShell 7')
|
||||||
|
backend.emit({ sessionId, sequence: 1, type: 'output', data: 'a' })
|
||||||
|
backend.emit({ sessionId, sequence: 2, type: 'output', data: 'b' })
|
||||||
|
await waitFor(() =>
|
||||||
|
expect(emulator.write).toHaveBeenCalledWith('ab', expect.any(Function))
|
||||||
|
)
|
||||||
|
await waitFor(() =>
|
||||||
|
expect(backend.adapter.ack).toHaveBeenCalledWith({ sessionId, sequence: 2 })
|
||||||
|
)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('logs a bounded, content-free warning when the WebGL renderer falls back', () => {
|
||||||
|
resetTerminalRendererWarningsForTests()
|
||||||
|
const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
|
||||||
|
try {
|
||||||
|
for (let index = 0; index < 20; index += 1) {
|
||||||
|
warnTerminalRendererFallback(
|
||||||
|
index % 2 === 0 ? 'webgl-load-failed' : 'webgl-context-lost',
|
||||||
|
new TypeError('secret user content')
|
||||||
|
)
|
||||||
|
}
|
||||||
|
expect(warn).toHaveBeenCalledTimes(5)
|
||||||
|
expect(warn).toHaveBeenCalledWith(
|
||||||
|
'[terminal] webgl-load-failed; falling back to the DOM renderer',
|
||||||
|
{ error: 'TypeError' }
|
||||||
|
)
|
||||||
|
expect(JSON.stringify(warn.mock.calls)).not.toContain('secret')
|
||||||
|
} finally {
|
||||||
|
warn.mockRestore()
|
||||||
|
resetTerminalRendererWarningsForTests()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
it('uses the resolved cross-platform monospace font stack', () => {
|
it('uses the resolved cross-platform monospace font stack', () => {
|
||||||
document.documentElement.style.setProperty(
|
document.documentElement.style.setProperty(
|
||||||
'--font-family-mono',
|
'--font-family-mono',
|
||||||
|
|||||||
@@ -28,6 +28,10 @@ import type {
|
|||||||
TerminalTarget
|
TerminalTarget
|
||||||
} from '../../shared/terminal-contracts'
|
} from '../../shared/terminal-contracts'
|
||||||
import './terminal-panel.css'
|
import './terminal-panel.css'
|
||||||
|
import {
|
||||||
|
TerminalOutputBatcher,
|
||||||
|
type TerminalFrameScheduler
|
||||||
|
} from './terminal-output-batcher'
|
||||||
|
|
||||||
export type TerminalAdapter = Omit<DesktopApi['terminal'], 'onEvent'> & {
|
export type TerminalAdapter = Omit<DesktopApi['terminal'], 'onEvent'> & {
|
||||||
subscribe: DesktopApi['terminal']['onEvent']
|
subscribe: DesktopApi['terminal']['onEvent']
|
||||||
@@ -74,6 +78,8 @@ export type TerminalPanelProps = {
|
|||||||
sessionId?: TerminalSessionId
|
sessionId?: TerminalSessionId
|
||||||
title?: string
|
title?: string
|
||||||
terminalFactory?: TerminalFactory
|
terminalFactory?: TerminalFactory
|
||||||
|
/** Test seam for the per-frame output batching scheduler. */
|
||||||
|
frameScheduler?: TerminalFrameScheduler
|
||||||
onSessionChange?: (snapshot: TerminalSnapshot) => void
|
onSessionChange?: (snapshot: TerminalSnapshot) => void
|
||||||
onRename?: (title: string) => void
|
onRename?: (title: string) => void
|
||||||
}
|
}
|
||||||
@@ -105,6 +111,31 @@ export function resolveTerminalFontFamily(): string {
|
|||||||
return configured || fallbackTerminalFontFamily
|
return configured || fallbackTerminalFontFamily
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Diagnostics are bounded per renderer so a flapping GPU cannot spam logs. */
|
||||||
|
const MAXIMUM_RENDERER_WARNINGS = 5
|
||||||
|
let rendererWarningCount = 0
|
||||||
|
|
||||||
|
export function warnTerminalRendererFallback(
|
||||||
|
reason: 'webgl-load-failed' | 'webgl-context-lost',
|
||||||
|
cause?: unknown
|
||||||
|
): void {
|
||||||
|
if (rendererWarningCount >= MAXIMUM_RENDERER_WARNINGS) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
rendererWarningCount += 1
|
||||||
|
// Only the error class is logged: messages could echo environment details.
|
||||||
|
const errorName =
|
||||||
|
cause instanceof Error ? cause.name.slice(0, 64) : undefined
|
||||||
|
console.warn(
|
||||||
|
`[terminal] ${reason}; falling back to the DOM renderer`,
|
||||||
|
errorName ? { error: errorName } : {}
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
export function resetTerminalRendererWarningsForTests(): void {
|
||||||
|
rendererWarningCount = 0
|
||||||
|
}
|
||||||
|
|
||||||
function createDefaultTerminal(): TerminalFactoryResult {
|
function createDefaultTerminal(): TerminalFactoryResult {
|
||||||
const terminal = new Terminal({
|
const terminal = new Terminal({
|
||||||
cursorBlink: true,
|
cursorBlink: true,
|
||||||
@@ -122,14 +153,22 @@ function createDefaultTerminal(): TerminalFactoryResult {
|
|||||||
fitAddon,
|
fitAddon,
|
||||||
searchAddon,
|
searchAddon,
|
||||||
activateRenderer: () => {
|
activateRenderer: () => {
|
||||||
const webglAddon = new WebglAddon()
|
let webglAddon: WebglAddon
|
||||||
|
try {
|
||||||
|
webglAddon = new WebglAddon()
|
||||||
|
} catch (cause) {
|
||||||
|
warnTerminalRendererFallback('webgl-load-failed', cause)
|
||||||
|
return
|
||||||
|
}
|
||||||
const contextLossSubscription = webglAddon.onContextLoss(() => {
|
const contextLossSubscription = webglAddon.onContextLoss(() => {
|
||||||
|
warnTerminalRendererFallback('webgl-context-lost')
|
||||||
contextLossSubscription.dispose()
|
contextLossSubscription.dispose()
|
||||||
webglAddon.dispose()
|
webglAddon.dispose()
|
||||||
})
|
})
|
||||||
try {
|
try {
|
||||||
terminal.loadAddon(webglAddon)
|
terminal.loadAddon(webglAddon)
|
||||||
} catch {
|
} catch (cause) {
|
||||||
|
warnTerminalRendererFallback('webgl-load-failed', cause)
|
||||||
contextLossSubscription.dispose()
|
contextLossSubscription.dispose()
|
||||||
webglAddon.dispose()
|
webglAddon.dispose()
|
||||||
}
|
}
|
||||||
@@ -157,6 +196,7 @@ export function TerminalPanel({
|
|||||||
sessionId,
|
sessionId,
|
||||||
title,
|
title,
|
||||||
terminalFactory = createDefaultTerminal,
|
terminalFactory = createDefaultTerminal,
|
||||||
|
frameScheduler,
|
||||||
onSessionChange,
|
onSessionChange,
|
||||||
onRename
|
onRename
|
||||||
}: TerminalPanelProps): React.JSX.Element {
|
}: TerminalPanelProps): React.JSX.Element {
|
||||||
@@ -176,6 +216,9 @@ export function TerminalPanel({
|
|||||||
const mountedRef = useRef(false)
|
const mountedRef = useRef(false)
|
||||||
const acceptedSequenceRef = useRef(0)
|
const acceptedSequenceRef = useRef(0)
|
||||||
const eventQueueRef = useRef(Promise.resolve())
|
const eventQueueRef = useRef(Promise.resolve())
|
||||||
|
const outputBatcherRef = useRef<TerminalOutputBatcher | undefined>(
|
||||||
|
undefined
|
||||||
|
)
|
||||||
const lastSizeRef = useRef<
|
const lastSizeRef = useRef<
|
||||||
{ cols: number; rows: number } | undefined
|
{ cols: number; rows: number } | undefined
|
||||||
>(undefined)
|
>(undefined)
|
||||||
@@ -213,6 +256,17 @@ export function TerminalPanel({
|
|||||||
}
|
}
|
||||||
}, [])
|
}, [])
|
||||||
|
|
||||||
|
const reportAckFailure = useCallback((reason: unknown): void => {
|
||||||
|
if (mountedRef.current) {
|
||||||
|
setError(
|
||||||
|
failureMessage(
|
||||||
|
reason,
|
||||||
|
translationRef.current('sidebar.terminal.errors.ack')
|
||||||
|
)
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}, [])
|
||||||
|
|
||||||
const consumeEvent = useCallback(
|
const consumeEvent = useCallback(
|
||||||
async (event: TerminalEvent): Promise<void> => {
|
async (event: TerminalEvent): Promise<void> => {
|
||||||
if (
|
if (
|
||||||
@@ -225,33 +279,42 @@ export function TerminalPanel({
|
|||||||
acceptedSequenceRef.current = event.sequence
|
acceptedSequenceRef.current = event.sequence
|
||||||
|
|
||||||
if (event.type === 'output') {
|
if (event.type === 'output') {
|
||||||
await new Promise<void>((resolve) => {
|
// Merged with neighbours and written once per frame; the batcher ACKs
|
||||||
emulatorRef.current?.write(event.data, resolve)
|
// the batch's last sequence after xterm has processed it.
|
||||||
if (!emulatorRef.current) {
|
outputBatcherRef.current?.push(
|
||||||
resolve()
|
event.sessionId,
|
||||||
}
|
event.sequence,
|
||||||
})
|
event.data
|
||||||
} else {
|
)
|
||||||
const current = snapshotRef.current
|
return
|
||||||
if (current) {
|
}
|
||||||
if (event.type === 'state') {
|
// Control events apply only after all earlier output has been written.
|
||||||
publishSnapshot({ ...current, state: event.state })
|
await outputBatcherRef.current?.flush()
|
||||||
} else if (event.type === 'exit') {
|
if (
|
||||||
publishSnapshot({
|
!mountedRef.current ||
|
||||||
...current,
|
event.sessionId !== activeSessionRef.current
|
||||||
state: 'exited',
|
) {
|
||||||
exit: event.exit,
|
return
|
||||||
error: null
|
}
|
||||||
})
|
const current = snapshotRef.current
|
||||||
} else {
|
if (current) {
|
||||||
publishSnapshot({
|
if (event.type === 'state') {
|
||||||
...current,
|
publishSnapshot({ ...current, state: event.state })
|
||||||
state: 'failed',
|
} else if (event.type === 'exit') {
|
||||||
error: event.error,
|
publishSnapshot({
|
||||||
exit: null
|
...current,
|
||||||
})
|
state: 'exited',
|
||||||
setError(event.error.message)
|
exit: event.exit,
|
||||||
}
|
error: null
|
||||||
|
})
|
||||||
|
} else {
|
||||||
|
publishSnapshot({
|
||||||
|
...current,
|
||||||
|
state: 'failed',
|
||||||
|
error: event.error,
|
||||||
|
exit: null
|
||||||
|
})
|
||||||
|
setError(event.error.message)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -262,18 +325,11 @@ export function TerminalPanel({
|
|||||||
sequence: event.sequence
|
sequence: event.sequence
|
||||||
})
|
})
|
||||||
} catch (reason) {
|
} catch (reason) {
|
||||||
if (mountedRef.current) {
|
reportAckFailure(reason)
|
||||||
setError(
|
|
||||||
failureMessage(
|
|
||||||
reason,
|
|
||||||
translationRef.current('sidebar.terminal.errors.ack')
|
|
||||||
)
|
|
||||||
)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
[adapter, publishSnapshot]
|
[adapter, publishSnapshot, reportAckFailure]
|
||||||
)
|
)
|
||||||
|
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
@@ -291,6 +347,19 @@ export function TerminalPanel({
|
|||||||
terminal.open(hostRef.current)
|
terminal.open(hostRef.current)
|
||||||
activateRenderer?.()
|
activateRenderer?.()
|
||||||
}
|
}
|
||||||
|
const outputBatcher = new TerminalOutputBatcher({
|
||||||
|
write: (data, done) => terminal.write(data, done),
|
||||||
|
acknowledge: (ackSessionId, sequence) => {
|
||||||
|
if (!mountedRef.current) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
adapter
|
||||||
|
.ack({ sessionId: ackSessionId, sequence })
|
||||||
|
.catch(reportAckFailure)
|
||||||
|
},
|
||||||
|
...(frameScheduler ? { schedule: frameScheduler } : {})
|
||||||
|
})
|
||||||
|
outputBatcherRef.current = outputBatcher
|
||||||
|
|
||||||
const inputSubscription = terminal.onData((data) => {
|
const inputSubscription = terminal.onData((data) => {
|
||||||
const current = snapshotRef.current
|
const current = snapshotRef.current
|
||||||
@@ -322,12 +391,14 @@ export function TerminalPanel({
|
|||||||
operationGenerationRef.current += 1
|
operationGenerationRef.current += 1
|
||||||
unsubscribe()
|
unsubscribe()
|
||||||
inputSubscription.dispose()
|
inputSubscription.dispose()
|
||||||
|
outputBatcher.dispose()
|
||||||
|
outputBatcherRef.current = undefined
|
||||||
terminal.dispose()
|
terminal.dispose()
|
||||||
emulatorRef.current = undefined
|
emulatorRef.current = undefined
|
||||||
fitAddonRef.current = undefined
|
fitAddonRef.current = undefined
|
||||||
searchAddonRef.current = undefined
|
searchAddonRef.current = undefined
|
||||||
}
|
}
|
||||||
}, [adapter, consumeEvent, terminalFactory])
|
}, [adapter, consumeEvent, frameScheduler, reportAckFailure, terminalFactory])
|
||||||
|
|
||||||
const attachSession = useCallback(
|
const attachSession = useCallback(
|
||||||
async (
|
async (
|
||||||
@@ -548,6 +619,7 @@ export function TerminalPanel({
|
|||||||
if (!mountedRef.current) {
|
if (!mountedRef.current) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
outputBatcherRef.current?.reset()
|
||||||
emulatorRef.current?.clear()
|
emulatorRef.current?.clear()
|
||||||
activeSessionRef.current = undefined
|
activeSessionRef.current = undefined
|
||||||
acceptedSequenceRef.current = 0
|
acceptedSequenceRef.current = 0
|
||||||
|
|||||||
@@ -0,0 +1,164 @@
|
|||||||
|
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||||
|
import {
|
||||||
|
TERMINAL_FRAME_FALLBACK_MS,
|
||||||
|
TerminalOutputBatcher,
|
||||||
|
scheduleTerminalFrame,
|
||||||
|
type TerminalFrameScheduler
|
||||||
|
} from './terminal-output-batcher'
|
||||||
|
|
||||||
|
function manualScheduler(): {
|
||||||
|
schedule: TerminalFrameScheduler
|
||||||
|
runFrame: () => void
|
||||||
|
pendingFrames: () => number
|
||||||
|
} {
|
||||||
|
const callbacks = new Set<() => void>()
|
||||||
|
return {
|
||||||
|
schedule: (callback) => {
|
||||||
|
callbacks.add(callback)
|
||||||
|
return () => callbacks.delete(callback)
|
||||||
|
},
|
||||||
|
runFrame: () => {
|
||||||
|
const current = [...callbacks]
|
||||||
|
callbacks.clear()
|
||||||
|
for (const callback of current) {
|
||||||
|
callback()
|
||||||
|
}
|
||||||
|
},
|
||||||
|
pendingFrames: () => callbacks.size
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
describe('TerminalOutputBatcher', () => {
|
||||||
|
afterEach(() => {
|
||||||
|
vi.useRealTimers()
|
||||||
|
vi.unstubAllGlobals()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('writes once per frame in order and ACKs the last sequence after the write callback', async () => {
|
||||||
|
const frames = manualScheduler()
|
||||||
|
const writes: Array<{ data: string; done: () => void }> = []
|
||||||
|
const acks: number[] = []
|
||||||
|
const batcher = new TerminalOutputBatcher({
|
||||||
|
write: (data, done) => writes.push({ data, done }),
|
||||||
|
acknowledge: (_, sequence) => acks.push(sequence),
|
||||||
|
schedule: frames.schedule
|
||||||
|
})
|
||||||
|
batcher.push('s', 3, 'a')
|
||||||
|
batcher.push('s', 4, 'b')
|
||||||
|
batcher.push('s', 5, 'c')
|
||||||
|
expect(frames.pendingFrames()).toBe(1)
|
||||||
|
expect(writes).toHaveLength(0)
|
||||||
|
|
||||||
|
frames.runFrame()
|
||||||
|
expect(writes.map((write) => write.data)).toEqual(['abc'])
|
||||||
|
expect(acks).toEqual([])
|
||||||
|
writes[0]!.done()
|
||||||
|
expect(acks).toEqual([5])
|
||||||
|
|
||||||
|
batcher.push('s', 6, 'd')
|
||||||
|
frames.runFrame()
|
||||||
|
writes[1]!.done()
|
||||||
|
writes[1]!.done()
|
||||||
|
expect(writes.map((write) => write.data)).toEqual(['abc', 'd'])
|
||||||
|
expect(acks).toEqual([5, 6])
|
||||||
|
batcher.dispose()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('flush writes pending output immediately and waits for the emulator', async () => {
|
||||||
|
const frames = manualScheduler()
|
||||||
|
let release: (() => void) | undefined
|
||||||
|
const batcher = new TerminalOutputBatcher({
|
||||||
|
write: (_, done) => {
|
||||||
|
release = done
|
||||||
|
},
|
||||||
|
acknowledge: vi.fn(),
|
||||||
|
schedule: frames.schedule
|
||||||
|
})
|
||||||
|
batcher.push('s', 1, 'x')
|
||||||
|
let flushed = false
|
||||||
|
const flush = batcher.flush().then(() => {
|
||||||
|
flushed = true
|
||||||
|
})
|
||||||
|
expect(frames.pendingFrames()).toBe(0)
|
||||||
|
await Promise.resolve()
|
||||||
|
expect(flushed).toBe(false)
|
||||||
|
release?.()
|
||||||
|
await flush
|
||||||
|
expect(flushed).toBe(true)
|
||||||
|
batcher.dispose()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('reset drops unwritten output but ACKs it so Main is not left paused', () => {
|
||||||
|
const frames = manualScheduler()
|
||||||
|
const write = vi.fn()
|
||||||
|
const acknowledge = vi.fn()
|
||||||
|
const batcher = new TerminalOutputBatcher({
|
||||||
|
write,
|
||||||
|
acknowledge,
|
||||||
|
schedule: frames.schedule
|
||||||
|
})
|
||||||
|
batcher.push('old', 9, 'stale')
|
||||||
|
batcher.reset()
|
||||||
|
frames.runFrame()
|
||||||
|
expect(write).not.toHaveBeenCalled()
|
||||||
|
expect(acknowledge).toHaveBeenCalledWith('old', 9)
|
||||||
|
batcher.dispose()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('does not ACK after dispose', () => {
|
||||||
|
const frames = manualScheduler()
|
||||||
|
let done: (() => void) | undefined
|
||||||
|
const acknowledge = vi.fn()
|
||||||
|
const batcher = new TerminalOutputBatcher({
|
||||||
|
write: (_, callback) => {
|
||||||
|
done = callback
|
||||||
|
},
|
||||||
|
acknowledge,
|
||||||
|
schedule: frames.schedule
|
||||||
|
})
|
||||||
|
batcher.push('s', 1, 'x')
|
||||||
|
frames.runFrame()
|
||||||
|
batcher.dispose()
|
||||||
|
done?.()
|
||||||
|
expect(acknowledge).not.toHaveBeenCalled()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('keeps flushing hidden terminals via the timer fallback when frames stop', () => {
|
||||||
|
vi.useFakeTimers()
|
||||||
|
// Hidden/minimized windows never run animation frames.
|
||||||
|
vi.stubGlobal('requestAnimationFrame', vi.fn(() => 1))
|
||||||
|
vi.stubGlobal('cancelAnimationFrame', vi.fn())
|
||||||
|
const writes: string[] = []
|
||||||
|
const acks: number[] = []
|
||||||
|
const batcher = new TerminalOutputBatcher({
|
||||||
|
write: (data, done) => {
|
||||||
|
writes.push(data)
|
||||||
|
done()
|
||||||
|
},
|
||||||
|
acknowledge: (_, sequence) => acks.push(sequence)
|
||||||
|
})
|
||||||
|
batcher.push('s', 1, 'hidden ')
|
||||||
|
batcher.push('s', 2, 'output')
|
||||||
|
vi.advanceTimersByTime(TERMINAL_FRAME_FALLBACK_MS)
|
||||||
|
expect(writes).toEqual(['hidden output'])
|
||||||
|
expect(acks).toEqual([2])
|
||||||
|
batcher.dispose()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('scheduleTerminalFrame runs once whichever of frame or timer fires first', () => {
|
||||||
|
vi.useFakeTimers()
|
||||||
|
let frame: (() => void) | undefined
|
||||||
|
vi.stubGlobal('requestAnimationFrame', vi.fn((callback: () => void) => {
|
||||||
|
frame = callback
|
||||||
|
return 7
|
||||||
|
}))
|
||||||
|
const cancelAnimationFrame = vi.fn()
|
||||||
|
vi.stubGlobal('cancelAnimationFrame', cancelAnimationFrame)
|
||||||
|
const callback = vi.fn()
|
||||||
|
scheduleTerminalFrame(callback)
|
||||||
|
frame?.()
|
||||||
|
vi.advanceTimersByTime(TERMINAL_FRAME_FALLBACK_MS * 2)
|
||||||
|
expect(callback).toHaveBeenCalledOnce()
|
||||||
|
expect(vi.getTimerCount()).toBe(0)
|
||||||
|
})
|
||||||
|
})
|
||||||
@@ -0,0 +1,155 @@
|
|||||||
|
/**
|
||||||
|
* Renderer-side terminal output batching (P3): output events are merged and
|
||||||
|
* written to the emulator at most once per animation frame, in sequence order.
|
||||||
|
* Each batch is acknowledged with its highest sequence after the emulator's
|
||||||
|
* write callback, which is what Main's ACK-based backpressure expects.
|
||||||
|
*/
|
||||||
|
|
||||||
|
/** Requests `callback` once; the returned function cancels the request. */
|
||||||
|
export type TerminalFrameScheduler = (callback: () => void) => () => void
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Animation frames stop while the window is minimized or occluded, so a timer
|
||||||
|
* fallback keeps output (and ACKs) flowing for hidden terminals. Without it
|
||||||
|
* Main would pause the PTY until the window became visible again.
|
||||||
|
*/
|
||||||
|
export const TERMINAL_FRAME_FALLBACK_MS = 100
|
||||||
|
|
||||||
|
export const scheduleTerminalFrame: TerminalFrameScheduler = (callback) => {
|
||||||
|
let settled = false
|
||||||
|
let frame: number | undefined
|
||||||
|
const run = (): void => {
|
||||||
|
if (settled) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
settled = true
|
||||||
|
if (frame !== undefined) {
|
||||||
|
cancelAnimationFrame(frame)
|
||||||
|
}
|
||||||
|
clearTimeout(timer)
|
||||||
|
callback()
|
||||||
|
}
|
||||||
|
const timer = setTimeout(run, TERMINAL_FRAME_FALLBACK_MS)
|
||||||
|
if (typeof requestAnimationFrame === 'function') {
|
||||||
|
frame = requestAnimationFrame(run)
|
||||||
|
}
|
||||||
|
return () => {
|
||||||
|
settled = true
|
||||||
|
if (frame !== undefined) {
|
||||||
|
cancelAnimationFrame(frame)
|
||||||
|
}
|
||||||
|
clearTimeout(timer)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export type TerminalOutputBatcherOptions = {
|
||||||
|
/** Writes data to the emulator; `done` must run after it is processed. */
|
||||||
|
write: (data: string, done: () => void) => void
|
||||||
|
/** Acknowledges every output event up to and including `sequence`. */
|
||||||
|
acknowledge: (sessionId: string, sequence: number) => void
|
||||||
|
schedule?: TerminalFrameScheduler
|
||||||
|
}
|
||||||
|
|
||||||
|
export class TerminalOutputBatcher {
|
||||||
|
private readonly writeData: TerminalOutputBatcherOptions['write']
|
||||||
|
private readonly acknowledge: TerminalOutputBatcherOptions['acknowledge']
|
||||||
|
private readonly schedule: TerminalFrameScheduler
|
||||||
|
private pending: string[] = []
|
||||||
|
private pendingSessionId: string | undefined
|
||||||
|
private pendingSequence = 0
|
||||||
|
private cancelFrame: (() => void) | undefined
|
||||||
|
private lastWrite: Promise<void> = Promise.resolve()
|
||||||
|
private disposed = false
|
||||||
|
|
||||||
|
constructor(options: TerminalOutputBatcherOptions) {
|
||||||
|
this.writeData = options.write
|
||||||
|
this.acknowledge = options.acknowledge
|
||||||
|
this.schedule = options.schedule ?? scheduleTerminalFrame
|
||||||
|
}
|
||||||
|
|
||||||
|
get hasPending(): boolean {
|
||||||
|
return this.pending.length > 0
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Queues output; events must be pushed in increasing sequence order. */
|
||||||
|
push(sessionId: string, sequence: number, data: string): void {
|
||||||
|
if (this.disposed) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if (this.pendingSessionId !== undefined && this.pendingSessionId !== sessionId) {
|
||||||
|
// Never mix streams: the previous session's output is obsolete.
|
||||||
|
this.reset()
|
||||||
|
}
|
||||||
|
this.pendingSessionId = sessionId
|
||||||
|
this.pendingSequence = sequence
|
||||||
|
this.pending.push(data)
|
||||||
|
this.cancelFrame ??= this.schedule(() => {
|
||||||
|
this.cancelFrame = undefined
|
||||||
|
void this.flush()
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Writes queued output now and resolves after every write issued so far has
|
||||||
|
* been processed by the emulator (so callers can order control events).
|
||||||
|
*/
|
||||||
|
flush(): Promise<void> {
|
||||||
|
if (this.disposed) {
|
||||||
|
return Promise.resolve()
|
||||||
|
}
|
||||||
|
this.cancelFrame?.()
|
||||||
|
this.cancelFrame = undefined
|
||||||
|
if (this.pending.length > 0) {
|
||||||
|
const data = this.pending.length === 1 ? this.pending[0]! : this.pending.join('')
|
||||||
|
const sessionId = this.pendingSessionId!
|
||||||
|
const sequence = this.pendingSequence
|
||||||
|
this.pending = []
|
||||||
|
this.pendingSessionId = undefined
|
||||||
|
this.lastWrite = new Promise<void>((resolve) => {
|
||||||
|
let finished = false
|
||||||
|
const done = (): void => {
|
||||||
|
if (finished) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
finished = true
|
||||||
|
if (!this.disposed) {
|
||||||
|
this.acknowledge(sessionId, sequence)
|
||||||
|
}
|
||||||
|
resolve()
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
this.writeData(data, done)
|
||||||
|
} catch {
|
||||||
|
done()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
return this.lastWrite
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Drops queued (unwritten) output when the session is replaced. The dropped
|
||||||
|
* events are still acknowledged so Main does not keep them counted against
|
||||||
|
* that session's backpressure window.
|
||||||
|
*/
|
||||||
|
reset(): void {
|
||||||
|
const sessionId = this.pendingSessionId
|
||||||
|
const sequence = this.pendingSequence
|
||||||
|
this.dropPending()
|
||||||
|
if (!this.disposed && sessionId !== undefined) {
|
||||||
|
this.acknowledge(sessionId, sequence)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
dispose(): void {
|
||||||
|
this.disposed = true
|
||||||
|
this.dropPending()
|
||||||
|
}
|
||||||
|
|
||||||
|
private dropPending(): void {
|
||||||
|
this.cancelFrame?.()
|
||||||
|
this.cancelFrame = undefined
|
||||||
|
this.pending = []
|
||||||
|
this.pendingSessionId = undefined
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user