mirror of
https://github.com/stablyai/orca.git
synced 2026-09-21 16:02:20 +00:00
perf(terminal): split dense SGR batches without gating on parse
This commit is contained in:
Neil
@@ -7,7 +7,6 @@ import { clearForegroundRelease, isEntryDrainable } from './pane-terminal-foregr
|
||||
import { hasHighPriorityBacklog, hasQueuedChunks } from './pane-terminal-output-queue-backlog'
|
||||
import {
|
||||
BACKGROUND_DRAIN_INTERVAL_MS,
|
||||
canDrainQueueEntry,
|
||||
DRAIN_TIME_BUDGET_MS,
|
||||
HIGH_PRIORITY_DRAIN_INTERVAL_MS,
|
||||
HIGH_PRIORITY_MAX_WRITES_PER_DRAIN,
|
||||
@@ -25,7 +24,7 @@ import { writeQueuedChunk } from './pane-terminal-output-pipeline'
|
||||
|
||||
function hasDrainableBacklog(): boolean {
|
||||
for (const entry of queuedByTerminal.values()) {
|
||||
if (isEntryDrainable(entry) && canDrainQueueEntry(entry)) {
|
||||
if (isEntryDrainable(entry)) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
@@ -40,9 +39,6 @@ function takeNextDrainableEntry(): QueueEntry | null {
|
||||
if (!isEntryDrainable(entry)) {
|
||||
continue
|
||||
}
|
||||
if (!canDrainQueueEntry(entry)) {
|
||||
continue
|
||||
}
|
||||
// Why: active/foreground output should be chosen first, not left in insertion order behind older background terminals.
|
||||
if (entry.highPriority) {
|
||||
queuedByTerminal.delete(entry.terminal)
|
||||
@@ -60,9 +56,6 @@ function takeNextDrainableEntry(): QueueEntry | null {
|
||||
if (!isEntryDrainable(entry)) {
|
||||
continue
|
||||
}
|
||||
if (!canDrainQueueEntry(entry)) {
|
||||
continue
|
||||
}
|
||||
queuedByTerminal.delete(entry.terminal)
|
||||
return entry
|
||||
}
|
||||
@@ -108,11 +101,6 @@ export function drainQueuedOutputImpl(): void {
|
||||
entry.highPriority = false
|
||||
clearForegroundRelease(entry)
|
||||
}
|
||||
// Dense SGR batches are parser paced. A completion callback schedules the
|
||||
// next drain after xterm has finished this batch.
|
||||
if (entry.denseSgr) {
|
||||
break
|
||||
}
|
||||
// Why: xterm parsing and DOM work share the renderer thread with input; keep draining cooperative so WSL/agent output can't pin the UI.
|
||||
if (writes > 0 && getDrainNow() - startedAt >= DRAIN_TIME_BUDGET_MS) {
|
||||
break
|
||||
|
||||
@@ -22,15 +22,12 @@ import {
|
||||
import { discardDetachedQueueEntry, hasQueuedChunks } from './pane-terminal-output-queue-backlog'
|
||||
import { clearForegroundRelease, isEntryDrainable } from './pane-terminal-foreground-queue-state'
|
||||
import {
|
||||
BACKGROUND_CHUNK_CHARS,
|
||||
canDrainQueueEntry,
|
||||
discardTerminalOutput,
|
||||
fireQueuedAckCredits,
|
||||
queuedByTerminal,
|
||||
requestRegisteredTerminalBacklogRecovery,
|
||||
resolveQueueEntryChunkLimit,
|
||||
scheduleDrain,
|
||||
reserveDenseSgrBatch,
|
||||
type TerminalOutputTarget
|
||||
} from './pane-terminal-output-queue-registry'
|
||||
|
||||
@@ -43,8 +40,6 @@ export function flushTerminalOutputImpl(
|
||||
if (!entry) {
|
||||
return
|
||||
}
|
||||
// Why: a budget-free flush is an ordering barrier (replay paint, shutdown capture, parse settle) whose caller writes straight to xterm next, so it must submit every queued byte; only budgeted callers tolerate dense pacing.
|
||||
const explicitFullDrain = options?.maxChars === undefined
|
||||
queuedByTerminal.delete(terminal)
|
||||
if (isTerminalWritePipelineCertifiedDead(terminal)) {
|
||||
discardDetachedQueueEntry(entry)
|
||||
@@ -55,11 +50,6 @@ export function flushTerminalOutputImpl(
|
||||
queuedByTerminal.set(terminal, entry)
|
||||
return
|
||||
}
|
||||
if (!explicitFullDrain && !canDrainQueueEntry(entry)) {
|
||||
queuedByTerminal.set(terminal, entry)
|
||||
scheduleDrain(0)
|
||||
return
|
||||
}
|
||||
if (entry.backgroundBacklogDropped && requestRegisteredTerminalBacklogRecovery(terminal)) {
|
||||
fireQueuedAckCredits(entry)
|
||||
entry.chunks.length = 0
|
||||
@@ -72,17 +62,13 @@ export function flushTerminalOutputImpl(
|
||||
}
|
||||
|
||||
let flushedChars = 0
|
||||
const chunkLimit = resolveQueueEntryChunkLimit(entry)
|
||||
let queuedWrite = takeQueuedChunk(entry, explicitFullDrain ? BACKGROUND_CHUNK_CHARS : chunkLimit)
|
||||
let queuedWrite = takeQueuedChunk(entry, resolveQueueEntryChunkLimit(entry))
|
||||
while (queuedWrite) {
|
||||
flushedChars += queuedWrite.data.length
|
||||
if (debugEnabled) {
|
||||
debugState.flushWriteCount++
|
||||
}
|
||||
const ackCreditsParsed = registerTerminalOutputAckCredits(terminal, queuedWrite.ackCredits)
|
||||
// Why not reserved on a full drain: this path ignores the pacing gate, so a reservation only strands the counter and blocks the terminal's next dense entry until xterm parses every batch.
|
||||
const denseSgrRelease =
|
||||
!explicitFullDrain && entry.denseSgr ? reserveDenseSgrBatch(terminal) : undefined
|
||||
armTerminalWriteStallWatch(terminal, {
|
||||
onCertifiedDead: () => discardTerminalOutput(terminal)
|
||||
})
|
||||
@@ -102,27 +88,16 @@ export function flushTerminalOutputImpl(
|
||||
terminal,
|
||||
queuedWrite.onParsed,
|
||||
ackCreditsParsed,
|
||||
undefined,
|
||||
denseSgrRelease
|
||||
undefined
|
||||
),
|
||||
onWriteFailure: composeWriteFailureCallback(
|
||||
terminal,
|
||||
ackCreditsParsed,
|
||||
denseSgrRelease
|
||||
)
|
||||
onWriteFailure: composeWriteFailureCallback(terminal, ackCreditsParsed)
|
||||
}
|
||||
)
|
||||
: writeBackgroundTerminalChunk(
|
||||
terminal,
|
||||
queuedWrite.data,
|
||||
composeParsedCallback(
|
||||
terminal,
|
||||
queuedWrite.onParsed,
|
||||
ackCreditsParsed,
|
||||
undefined,
|
||||
denseSgrRelease
|
||||
),
|
||||
composeWriteFailureCallback(terminal, ackCreditsParsed, denseSgrRelease)
|
||||
composeParsedCallback(terminal, queuedWrite.onParsed, ackCreditsParsed, undefined),
|
||||
composeWriteFailureCallback(terminal, ackCreditsParsed)
|
||||
)
|
||||
if (!writeAccepted) {
|
||||
fireQueuedAckCredits(entry)
|
||||
@@ -136,9 +111,6 @@ export function flushTerminalOutputImpl(
|
||||
if (ackCreditsParsed) {
|
||||
runGuardedWriteCompletionStep('flush-abort-ack-credits', ackCreditsParsed)
|
||||
}
|
||||
if (denseSgrRelease) {
|
||||
runGuardedWriteCompletionStep('flush-abort-dense-release', denseSgrRelease)
|
||||
}
|
||||
fireQueuedAckCredits(entry)
|
||||
clearForegroundRelease(entry)
|
||||
recordQueueDebugPressure()
|
||||
@@ -147,10 +119,7 @@ export function flushTerminalOutputImpl(
|
||||
if (options?.maxChars !== undefined && flushedChars >= options.maxChars) {
|
||||
break
|
||||
}
|
||||
if (!explicitFullDrain && entry.denseSgr) {
|
||||
break
|
||||
}
|
||||
queuedWrite = takeQueuedChunk(entry, BACKGROUND_CHUNK_CHARS)
|
||||
queuedWrite = takeQueuedChunk(entry, resolveQueueEntryChunkLimit(entry))
|
||||
}
|
||||
if (hasQueuedChunks(entry)) {
|
||||
entry.highPriority = true
|
||||
|
||||
@@ -16,7 +16,6 @@ import {
|
||||
discardTerminalOutput,
|
||||
fireQueuedAckCredits,
|
||||
queuedByTerminal,
|
||||
reserveDenseSgrBatch,
|
||||
resolveQueueEntryChunkLimit,
|
||||
scheduleDrain,
|
||||
type QueueEntry,
|
||||
@@ -73,21 +72,17 @@ export function composeParsedCallback(
|
||||
terminal: TerminalOutputTarget,
|
||||
onParsed: TerminalOutputParsedCallback | undefined,
|
||||
ackCreditsParsed: (() => void) | undefined,
|
||||
pacer: (() => void) | undefined,
|
||||
denseSgrRelease: (() => void) | undefined = undefined
|
||||
pacer: (() => void) | undefined
|
||||
): TerminalOutputParsedCallback {
|
||||
// Why always non-undefined: the callback doubles as the pipeline-health settle signal — with none, the stall watch could never settle, forcing a probe round-trip per healthy idle pane.
|
||||
return () => {
|
||||
try {
|
||||
onParsed?.()
|
||||
} finally {
|
||||
// Why guarded per step: one throwing step would skip every later one, and a skipped dense release pins inFlight so the pane never drains again.
|
||||
// Why guarded per step: one throwing step would skip every later one.
|
||||
if (ackCreditsParsed) {
|
||||
runGuardedWriteCompletionStep('parsed-ack-credits', ackCreditsParsed)
|
||||
}
|
||||
if (denseSgrRelease) {
|
||||
runGuardedWriteCompletionStep('parsed-dense-release', denseSgrRelease)
|
||||
}
|
||||
if (pacer) {
|
||||
runGuardedWriteCompletionStep('parsed-pacer', pacer)
|
||||
}
|
||||
@@ -100,8 +95,7 @@ export function composeParsedCallback(
|
||||
|
||||
export function composeWriteFailureCallback(
|
||||
terminal: TerminalOutputTarget,
|
||||
ackCreditsParsed: (() => void) | undefined,
|
||||
denseSgrRelease: (() => void) | undefined = undefined
|
||||
ackCreditsParsed: (() => void) | undefined
|
||||
): () => void {
|
||||
return () => {
|
||||
try {
|
||||
@@ -109,9 +103,6 @@ export function composeWriteFailureCallback(
|
||||
if (ackCreditsParsed) {
|
||||
runGuardedWriteCompletionStep('write-failure-ack-credits', ackCreditsParsed)
|
||||
}
|
||||
if (denseSgrRelease) {
|
||||
runGuardedWriteCompletionStep('write-failure-dense-release', denseSgrRelease)
|
||||
}
|
||||
} finally {
|
||||
// Why: a synchronous rejection proves undeliverability but nothing about parse progress; recover without extending replay guards.
|
||||
failTerminalWriteStallWatch(terminal)
|
||||
@@ -131,7 +122,6 @@ export function writeQueuedChunk(entry: QueueEntry): 'foreground' | 'background'
|
||||
return null
|
||||
}
|
||||
const pacer = entry.highPriority ? makeParseClockPacer() : undefined
|
||||
const denseSgrRelease = entry.denseSgr ? reserveDenseSgrBatch(entry.terminal) : undefined
|
||||
const ackCreditsParsed = registerTerminalOutputAckCredits(entry.terminal, queuedWrite.ackCredits)
|
||||
// Why armed BEFORE the write: a wedged WriteBuffer (issue #2836) or disposed xterm (6.1.0-beta.287) never runs the parsed callback, so the watch must be live first to catch it.
|
||||
armTerminalWriteStallWatch(entry.terminal, {
|
||||
@@ -153,27 +143,16 @@ export function writeQueuedChunk(entry: QueueEntry): 'foreground' | 'background'
|
||||
entry.terminal,
|
||||
queuedWrite.onParsed,
|
||||
ackCreditsParsed,
|
||||
pacer,
|
||||
denseSgrRelease
|
||||
pacer
|
||||
),
|
||||
onWriteFailure: composeWriteFailureCallback(
|
||||
entry.terminal,
|
||||
ackCreditsParsed,
|
||||
denseSgrRelease
|
||||
)
|
||||
onWriteFailure: composeWriteFailureCallback(entry.terminal, ackCreditsParsed)
|
||||
}
|
||||
)
|
||||
: writeBackgroundTerminalChunk(
|
||||
entry.terminal,
|
||||
queuedWrite.data,
|
||||
composeParsedCallback(
|
||||
entry.terminal,
|
||||
queuedWrite.onParsed,
|
||||
ackCreditsParsed,
|
||||
pacer,
|
||||
denseSgrRelease
|
||||
),
|
||||
composeWriteFailureCallback(entry.terminal, ackCreditsParsed, denseSgrRelease)
|
||||
composeParsedCallback(entry.terminal, queuedWrite.onParsed, ackCreditsParsed, pacer),
|
||||
composeWriteFailureCallback(entry.terminal, ackCreditsParsed)
|
||||
)
|
||||
if (!writeAccepted) {
|
||||
// Why: the failure callback credited the submitted chunk; credit and abandon the detached tail so the drain can't retry a certified-dead xterm.
|
||||
@@ -191,9 +170,6 @@ export function writeQueuedChunk(entry: QueueEntry): 'foreground' | 'background'
|
||||
if (ackCreditsParsed) {
|
||||
runGuardedWriteCompletionStep('drain-abort-ack-credits', ackCreditsParsed)
|
||||
}
|
||||
if (denseSgrRelease) {
|
||||
runGuardedWriteCompletionStep('drain-abort-dense-release', denseSgrRelease)
|
||||
}
|
||||
fireQueuedAckCredits(entry)
|
||||
entry.chunks.length = 0
|
||||
entry.chunkIndex = 0
|
||||
|
||||
@@ -17,7 +17,6 @@ import {
|
||||
MAX_BACKGROUND_QUEUE_CHUNKS,
|
||||
fireQueuedAckCredits,
|
||||
getTerminalOutputMaxQueueChars,
|
||||
canDrainQueueEntry,
|
||||
queuedByTerminal,
|
||||
type QueueEntry,
|
||||
type TerminalOutputBeforeWrite
|
||||
@@ -96,7 +95,6 @@ export function hasHighPriorityBacklog(): boolean {
|
||||
for (const entry of queuedByTerminal.values()) {
|
||||
if (
|
||||
isEntryDrainable(entry) &&
|
||||
canDrainQueueEntry(entry) &&
|
||||
(entry.highPriority || entry.queuedChars > LARGE_BACKLOG_CHARS)
|
||||
) {
|
||||
return true
|
||||
|
||||
@@ -134,51 +134,6 @@ let useMessageChannelDrain = typeof MessageChannel !== 'undefined' && !isVitestE
|
||||
let drainChannel: MessageChannel | null = null
|
||||
// Why indirect: the drain loop lives downstream of this module, so it registers itself here rather than being imported back into the queue state it operates on.
|
||||
let runDrain: (() => void) | null = null
|
||||
type DenseSgrPacingState = {
|
||||
generation: number
|
||||
inFlight: number
|
||||
}
|
||||
|
||||
const denseSgrPacingByTerminal = new WeakMap<TerminalOutputTarget, DenseSgrPacingState>()
|
||||
|
||||
export function canDrainQueueEntry(entry: QueueEntry): boolean {
|
||||
return !entry.denseSgr || (denseSgrPacingByTerminal.get(entry.terminal)?.inFlight ?? 0) === 0
|
||||
}
|
||||
|
||||
export function reserveDenseSgrBatch(terminal: TerminalOutputTarget): () => void {
|
||||
const state = denseSgrPacingByTerminal.get(terminal) ?? { generation: 0, inFlight: 0 }
|
||||
denseSgrPacingByTerminal.set(terminal, state)
|
||||
state.inFlight += 1
|
||||
const generation = state.generation
|
||||
let released = false
|
||||
return () => {
|
||||
if (released) {
|
||||
return
|
||||
}
|
||||
released = true
|
||||
const current = denseSgrPacingByTerminal.get(terminal)
|
||||
// A disposed xterm can invoke an old callback after its queue is cleared
|
||||
// and the same object is reused. Do not let that callback release a new
|
||||
// generation's reservation.
|
||||
if (!current || current.generation !== generation) {
|
||||
return
|
||||
}
|
||||
current.inFlight = Math.max(0, current.inFlight - 1)
|
||||
scheduleDrain(0)
|
||||
}
|
||||
}
|
||||
|
||||
export function clearDenseSgrPacing(terminal: TerminalOutputTarget): void {
|
||||
const state = denseSgrPacingByTerminal.get(terminal)
|
||||
if (state) {
|
||||
state.generation += 1
|
||||
state.inFlight = 0
|
||||
return
|
||||
}
|
||||
// Retain a tombstone generation so a callback reserved before the first
|
||||
// clear cannot match a reservation made after it.
|
||||
denseSgrPacingByTerminal.set(terminal, { generation: 1, inFlight: 0 })
|
||||
}
|
||||
|
||||
// Why re-classified per batch rather than latched at enqueue: density changes mid-stream — a plain banner ahead of a TUI would miss the pacing entirely, and a dense header ahead of a long plain tail would pin that tail at the dense budget, which is slower than no pacing at all.
|
||||
export function resolveQueueEntryChunkLimit(entry: QueueEntry): number {
|
||||
@@ -316,7 +271,6 @@ export function discardTerminalOutput(terminal: TerminalOutputTarget): void {
|
||||
}
|
||||
discardInFlightTerminalOutputAckCredits(terminal)
|
||||
queuedByTerminal.delete(terminal)
|
||||
clearDenseSgrPacing(terminal)
|
||||
discardForegroundRenderSettle(terminal)
|
||||
// Why: cancel the watch without masquerading as parse progress; replay guards use real completions to tell slow from wedged.
|
||||
cancelTerminalWriteStallWatch(terminal)
|
||||
|
||||
@@ -335,16 +335,10 @@ describe('pane terminal output scheduler', () => {
|
||||
expect(terminal.write).toHaveBeenCalledTimes(6)
|
||||
})
|
||||
|
||||
it('paces dense SGR output to one parser batch and preserves bytes', async () => {
|
||||
it('splits dense SGR output into 4 KiB parser batches and preserves bytes', async () => {
|
||||
vi.useFakeTimers()
|
||||
const { writeTerminalOutput } = await loadScheduler()
|
||||
const { queuedByTerminal, writeTerminalOutput } = await loadScheduler()
|
||||
const terminal = createTerminal()
|
||||
const parsed: (() => void)[] = []
|
||||
terminal.write.mockImplementation((_data: string, callback?: () => void) => {
|
||||
if (callback) {
|
||||
parsed.push(callback)
|
||||
}
|
||||
})
|
||||
const input = Array.from(
|
||||
{ length: 1_200 },
|
||||
(_, index) => `\x1b[${30 + (index % 8)}mX\x1b[0m`
|
||||
@@ -352,32 +346,22 @@ describe('pane terminal output scheduler', () => {
|
||||
|
||||
writeTerminalOutput(terminal, input, { foreground: false })
|
||||
vi.advanceTimersByTime(50)
|
||||
|
||||
expect(terminal.write).toHaveBeenCalledTimes(1)
|
||||
expect(parsed).toHaveLength(1)
|
||||
const written: string[] = [terminal.write.mock.calls[0]?.[0] ?? '']
|
||||
while (parsed.length > 0) {
|
||||
parsed.shift()?.()
|
||||
vi.advanceTimersByTime(0)
|
||||
const next = terminal.write.mock.calls[written.length]?.[0]
|
||||
if (next !== undefined) {
|
||||
written.push(next)
|
||||
}
|
||||
while (queuedByTerminal.has(terminal)) {
|
||||
vi.advanceTimersByTime(16)
|
||||
}
|
||||
|
||||
const written = terminal.write.mock.calls.map(([data]) => data)
|
||||
expect(written.length).toBeGreaterThan(1)
|
||||
for (const data of written) {
|
||||
expect(data.length).toBeLessThanOrEqual(4 * 1024)
|
||||
}
|
||||
expect(written.join('')).toBe(input)
|
||||
})
|
||||
|
||||
it('classifies dense SGR split across sub-batch deliveries', async () => {
|
||||
vi.useFakeTimers()
|
||||
const { writeTerminalOutput } = await loadScheduler()
|
||||
const { queuedByTerminal, writeTerminalOutput } = await loadScheduler()
|
||||
const terminal = createTerminal()
|
||||
const parsed: (() => void)[] = []
|
||||
terminal.write.mockImplementation((_data: string, callback?: () => void) => {
|
||||
if (callback) {
|
||||
parsed.push(callback)
|
||||
}
|
||||
})
|
||||
const input = Array.from(
|
||||
{ length: 1_200 },
|
||||
(_, index) => `\x1b[${30 + (index % 8)}mX\x1b[0m`
|
||||
@@ -390,179 +374,40 @@ describe('pane terminal output scheduler', () => {
|
||||
}
|
||||
vi.advanceTimersByTime(50)
|
||||
|
||||
expect(terminal.write).toHaveBeenCalledTimes(1)
|
||||
expect(terminal.write.mock.calls[0]?.[0]).toHaveLength(4 * 1024)
|
||||
while (parsed.length > 0) {
|
||||
parsed.shift()?.()
|
||||
vi.advanceTimersByTime(0)
|
||||
while (queuedByTerminal.has(terminal)) {
|
||||
vi.advanceTimersByTime(16)
|
||||
}
|
||||
|
||||
expect(terminal.write.mock.calls.map(([data]) => data).join('')).toBe(input)
|
||||
expect(terminal.write.mock.calls.length).toBeGreaterThan(1)
|
||||
})
|
||||
|
||||
it('ignores a stale dense parse release after the terminal queue is discarded', async () => {
|
||||
vi.useFakeTimers()
|
||||
const { discardTerminalOutput, writeTerminalOutput } = await loadScheduler()
|
||||
const terminal = createTerminal()
|
||||
const parsed: (() => void)[] = []
|
||||
terminal.write.mockImplementation((_data: string, callback?: () => void) => {
|
||||
if (callback) {
|
||||
parsed.push(callback)
|
||||
}
|
||||
})
|
||||
const dense = Array.from(
|
||||
{ length: 1_200 },
|
||||
(_, index) => `\x1b[${30 + (index % 8)}mX\x1b[0m`
|
||||
).join('')
|
||||
|
||||
writeTerminalOutput(terminal, dense, { foreground: false })
|
||||
vi.advanceTimersByTime(50)
|
||||
expect(terminal.write).toHaveBeenCalledTimes(1)
|
||||
const staleRelease = parsed.shift()
|
||||
expect(staleRelease).toBeDefined()
|
||||
|
||||
discardTerminalOutput(terminal)
|
||||
writeTerminalOutput(terminal, dense, { foreground: false })
|
||||
vi.advanceTimersByTime(50)
|
||||
expect(terminal.write).toHaveBeenCalledTimes(2)
|
||||
|
||||
staleRelease?.()
|
||||
vi.advanceTimersByTime(0)
|
||||
// The old callback must not release the replacement generation's batch.
|
||||
expect(terminal.write).toHaveBeenCalledTimes(2)
|
||||
|
||||
parsed.shift()?.()
|
||||
vi.advanceTimersByTime(0)
|
||||
expect(terminal.write).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
|
||||
it('keeps foreground bytes behind a dense batch retained by an explicit flush', async () => {
|
||||
it('keeps foreground bytes behind dense output still queued after a drain', async () => {
|
||||
vi.useFakeTimers()
|
||||
const { writeTerminalOutput } = await loadScheduler()
|
||||
const terminal = createTerminal()
|
||||
const parsed: (() => void)[] = []
|
||||
terminal.write.mockImplementation((_data: string, callback?: () => void) => {
|
||||
if (callback) {
|
||||
parsed.push(callback)
|
||||
}
|
||||
})
|
||||
const dense = Array.from(
|
||||
{ length: 600 },
|
||||
{ length: 1_300 },
|
||||
(_, index) => `\x1b[${30 + (index % 8)}mX\x1b[0m`
|
||||
).join('')
|
||||
const input = dense.slice(0, 4 * 1024 + 256)
|
||||
|
||||
writeTerminalOutput(terminal, input, { foreground: false })
|
||||
writeTerminalOutput(terminal, dense, { foreground: false })
|
||||
vi.advanceTimersByTime(50)
|
||||
expect(terminal.write).toHaveBeenCalledTimes(1)
|
||||
// One background tick submits MAX_WRITES_PER_DRAIN batches; the rest stays queued.
|
||||
expect(terminal.write).toHaveBeenCalledTimes(2)
|
||||
writeTerminalOutput(terminal, 'echo', { foreground: true })
|
||||
|
||||
// The foreground write's budget-free flush submits the retained dense tail first.
|
||||
expect(terminal.write.mock.calls.map(([data]) => data)).toEqual([
|
||||
input.slice(0, 4 * 1024),
|
||||
input.slice(4 * 1024),
|
||||
'echo'
|
||||
])
|
||||
// The foreground write's flush submits the queued dense tail first.
|
||||
const written = terminal.write.mock.calls.map(([data]) => data)
|
||||
expect(written.at(-1)).toBe('echo')
|
||||
expect(written.join('')).toBe(`${dense}echo`)
|
||||
})
|
||||
|
||||
it('drains a dense entry completely when the flush carries no char budget', async () => {
|
||||
it('guards a throwing parsed ack credit so the terminal keeps draining', async () => {
|
||||
vi.useFakeTimers()
|
||||
const { flushTerminalOutput, queuedByTerminal, writeTerminalOutput } = await loadScheduler()
|
||||
const { queuedByTerminal, writeTerminalOutput } = await loadScheduler()
|
||||
const terminal = createTerminal()
|
||||
const parsed: (() => void)[] = []
|
||||
terminal.write.mockImplementation((_data: string, callback?: () => void) => {
|
||||
if (callback) {
|
||||
parsed.push(callback)
|
||||
}
|
||||
})
|
||||
const dense = Array.from(
|
||||
{ length: 1_300 },
|
||||
(_, index) => `\x1b[${30 + (index % 8)}mX\x1b[0m`
|
||||
).join('')
|
||||
|
||||
writeTerminalOutput(terminal, dense, { foreground: false })
|
||||
vi.advanceTimersByTime(50)
|
||||
expect(terminal.write).toHaveBeenCalledTimes(1)
|
||||
|
||||
// The replay/shutdown-capture callers write straight to xterm next, so the
|
||||
// flush must leave nothing queued behind the batch still in flight.
|
||||
flushTerminalOutput(terminal)
|
||||
|
||||
expect(queuedByTerminal.has(terminal)).toBe(false)
|
||||
expect(terminal.write.mock.calls.map(([data]) => data).join('')).toBe(dense)
|
||||
})
|
||||
|
||||
it('does not reserve pacing slots for the batches a budget-free flush drains', async () => {
|
||||
vi.useFakeTimers()
|
||||
const { flushTerminalOutput, writeTerminalOutput } = await loadScheduler()
|
||||
const terminal = createTerminal()
|
||||
const parsed: (() => void)[] = []
|
||||
terminal.write.mockImplementation((_data: string, callback?: () => void) => {
|
||||
if (callback) {
|
||||
parsed.push(callback)
|
||||
}
|
||||
})
|
||||
const denseChunk = (count: number): string =>
|
||||
Array.from({ length: count }, (_, index) => `\x1b[${30 + (index % 8)}mX\x1b[0m`).join('')
|
||||
|
||||
writeTerminalOutput(terminal, denseChunk(1_300), { foreground: false })
|
||||
vi.advanceTimersByTime(50)
|
||||
flushTerminalOutput(terminal)
|
||||
|
||||
// Release the paced batch the drain submitted; only the flush's own writes
|
||||
// stay pending, modelling xterm not having parsed them yet.
|
||||
parsed.shift()?.()
|
||||
const pendingFromFlush = parsed.length
|
||||
const writesFromFlush = terminal.write.mock.calls.length
|
||||
writeTerminalOutput(terminal, denseChunk(1_300), { foreground: false })
|
||||
for (let tick = 0; tick < 4; tick += 1) {
|
||||
vi.advanceTimersByTime(50)
|
||||
while (parsed.length > pendingFromFlush) {
|
||||
parsed.pop()?.()
|
||||
}
|
||||
}
|
||||
|
||||
// Slots stranded by the flush would gate every batch after the first, which
|
||||
// the front of the queue clears before it is classified dense.
|
||||
expect(terminal.write.mock.calls.length - writesFromFlush).toBeGreaterThan(1)
|
||||
})
|
||||
|
||||
it('keeps pacing a dense entry when the flush carries a char budget', async () => {
|
||||
vi.useFakeTimers()
|
||||
const { flushTerminalOutput, queuedByTerminal, writeTerminalOutput } = await loadScheduler()
|
||||
const terminal = createTerminal()
|
||||
const parsed: (() => void)[] = []
|
||||
terminal.write.mockImplementation((_data: string, callback?: () => void) => {
|
||||
if (callback) {
|
||||
parsed.push(callback)
|
||||
}
|
||||
})
|
||||
const dense = Array.from(
|
||||
{ length: 1_300 },
|
||||
(_, index) => `\x1b[${30 + (index % 8)}mX\x1b[0m`
|
||||
).join('')
|
||||
|
||||
writeTerminalOutput(terminal, dense, { foreground: false })
|
||||
vi.advanceTimersByTime(50)
|
||||
expect(terminal.write).toHaveBeenCalledTimes(1)
|
||||
|
||||
flushTerminalOutput(terminal, { maxChars: 64 * 1024 })
|
||||
|
||||
expect(terminal.write).toHaveBeenCalledTimes(1)
|
||||
expect(queuedByTerminal.has(terminal)).toBe(true)
|
||||
})
|
||||
|
||||
it('releases the dense pacing slot when a parsed ack credit throws', async () => {
|
||||
vi.useFakeTimers()
|
||||
const { writeTerminalOutput } = await loadScheduler()
|
||||
const terminal = createTerminal()
|
||||
const parsed: (() => void)[] = []
|
||||
terminal.write.mockImplementation((_data: string, callback?: () => void) => {
|
||||
if (callback) {
|
||||
parsed.push(callback)
|
||||
}
|
||||
})
|
||||
const dense = Array.from(
|
||||
{ length: 300 },
|
||||
(_, index) => `\x1b[${30 + (index % 8)}mX\x1b[0m`
|
||||
@@ -576,25 +421,21 @@ describe('pane terminal output scheduler', () => {
|
||||
})
|
||||
writeTerminalOutput(terminal, dense, { foreground: false })
|
||||
vi.advanceTimersByTime(50)
|
||||
expect(terminal.write).toHaveBeenCalledTimes(1)
|
||||
while (queuedByTerminal.has(terminal)) {
|
||||
vi.advanceTimersByTime(16)
|
||||
}
|
||||
|
||||
parsed.shift()?.()
|
||||
vi.advanceTimersByTime(0)
|
||||
|
||||
// A throwing credit must not skip the dense release, or inFlight stays
|
||||
// pinned and the terminal never drains again.
|
||||
expect(terminal.write).toHaveBeenCalledTimes(2)
|
||||
expect(terminal.write.mock.calls.map(([data]) => data).join('')).toBe(`${dense}${dense}`)
|
||||
expect(mocks.recordRendererCrashBreadcrumb).toHaveBeenCalledWith(
|
||||
'terminal_write_completion_error',
|
||||
expect.objectContaining({ context: 'parsed-ack-credits' })
|
||||
)
|
||||
})
|
||||
|
||||
it('starts pacing once a plain banner gives way to dense output', async () => {
|
||||
it('shrinks the batch once a plain banner gives way to dense output', async () => {
|
||||
vi.useFakeTimers()
|
||||
const { writeTerminalOutput } = await loadScheduler()
|
||||
const terminal = createTerminal()
|
||||
terminal.write.mockImplementation(() => {})
|
||||
const banner = 'banner\r\n'.repeat(640)
|
||||
const dense = Array.from(
|
||||
{ length: 2_000 },
|
||||
@@ -607,21 +448,15 @@ describe('pane terminal output scheduler', () => {
|
||||
|
||||
const written = terminal.write.mock.calls.map(([data]) => data)
|
||||
expect(written[0]).toHaveLength(16 * 1024)
|
||||
// Once the dense body reaches the front the budget must drop, or this is
|
||||
// the 128 KiB parser burst the pacing exists to bound.
|
||||
// Once the dense body reaches the front the batch must shrink, or this is
|
||||
// the 128 KiB parser burst the split exists to bound.
|
||||
expect(written[1]).toHaveLength(4 * 1024)
|
||||
})
|
||||
|
||||
it('stops pacing once a dense header gives way to a plain tail', async () => {
|
||||
it('restores the full batch once a dense header gives way to a plain tail', async () => {
|
||||
vi.useFakeTimers()
|
||||
const { writeTerminalOutput } = await loadScheduler()
|
||||
const { queuedByTerminal, writeTerminalOutput } = await loadScheduler()
|
||||
const terminal = createTerminal()
|
||||
const parsed: (() => void)[] = []
|
||||
terminal.write.mockImplementation((_data: string, callback?: () => void) => {
|
||||
if (callback) {
|
||||
parsed.push(callback)
|
||||
}
|
||||
})
|
||||
const dense = Array.from(
|
||||
{ length: 500 },
|
||||
(_, index) => `\x1b[${30 + (index % 8)}mX\x1b[0m`
|
||||
@@ -631,21 +466,16 @@ describe('pane terminal output scheduler', () => {
|
||||
writeTerminalOutput(terminal, dense, { foreground: false })
|
||||
writeTerminalOutput(terminal, tail, { foreground: false })
|
||||
vi.advanceTimersByTime(50)
|
||||
expect(terminal.write.mock.calls[0]?.[0]).toHaveLength(4 * 1024)
|
||||
expect(terminal.write).toHaveBeenCalledTimes(1)
|
||||
|
||||
parsed.shift()?.()
|
||||
vi.advanceTimersByTime(0)
|
||||
|
||||
// A latched verdict would pin the plain tail at 4 KiB per parse
|
||||
// round-trip, which is slower than not pacing at all.
|
||||
expect(terminal.write.mock.calls[1]?.[0]).toHaveLength(16 * 1024)
|
||||
|
||||
while (parsed.length > 0 || terminal.write.mock.calls.length < 2) {
|
||||
parsed.shift()?.()
|
||||
while (queuedByTerminal.has(terminal)) {
|
||||
vi.advanceTimersByTime(16)
|
||||
}
|
||||
expect(terminal.write.mock.calls.map(([data]) => data).join('')).toBe(`${dense}${tail}`)
|
||||
|
||||
const written = terminal.write.mock.calls.map(([data]) => data)
|
||||
expect(written[0]).toHaveLength(4 * 1024)
|
||||
// A latched verdict would pin the plain tail at 4 KiB per batch, which is
|
||||
// slower than not splitting at all.
|
||||
expect(written[1]).toHaveLength(16 * 1024)
|
||||
expect(written.join('')).toBe(`${dense}${tail}`)
|
||||
})
|
||||
|
||||
it('writes a latency-sensitive foreground redraw whole, even when it is dense', async () => {
|
||||
|
||||
Reference in New Issue
Block a user