From bc1048062dda8d64020309c425a5bb8094c37490 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rg=20Breitbart?= Date: Wed, 20 Jan 2021 20:05:14 +0100 Subject: [PATCH] early version of async support for CSI and ESC handlers --- src/common/CoreTerminal.ts | 2 +- src/common/InputHandler.ts | 66 +++++++++++--- src/common/input/WriteBuffer.ts | 65 +++++++++++-- src/common/parser/EscapeSequenceParser.ts | 106 +++++++++++++++++++++- src/common/parser/Types.d.ts | 2 +- test/benchmark/Terminal.benchmark.ts | 36 ++++++-- 6 files changed, 245 insertions(+), 32 deletions(-) diff --git a/src/common/CoreTerminal.ts b/src/common/CoreTerminal.ts index 2f636349..11e71bad 100644 --- a/src/common/CoreTerminal.ts +++ b/src/common/CoreTerminal.ts @@ -109,7 +109,7 @@ export abstract class CoreTerminal extends Disposable implements ICoreTerminal { this.register(this.optionsService.onOptionChange(key => this._updateOptions(key))); // Setup WriteBuffer - this._writeBuffer = new WriteBuffer(data => this._inputHandler.parse(data)); + this._writeBuffer = new WriteBuffer((data, promiseResult) => this._inputHandler.parse(data, promiseResult)); } public dispose(): void { diff --git a/src/common/InputHandler.ts b/src/common/InputHandler.ts index 626e1fe2..32428c77 100644 --- a/src/common/InputHandler.ts +++ b/src/common/InputHandler.ts @@ -333,7 +333,7 @@ export class InputHandler extends Disposable implements IInputHandler { this._parser.setCsiHandler({prefix: '?', final: 'h'}, params => this.setModePrivate(params)); this._parser.setCsiHandler({final: 'l'}, params => this.resetMode(params)); this._parser.setCsiHandler({prefix: '?', final: 'l'}, params => this.resetModePrivate(params)); - this._parser.setCsiHandler({final: 'm'}, params => this.charAttributes(params)); + this._parser.setCsiHandler({final: 'm'}, (params => this.charAttributes(params)) as (p: any) => void); this._parser.setCsiHandler({final: 'n'}, params => this.deviceStatus(params)); this._parser.setCsiHandler({prefix: '?', final: 'n'}, params => this.deviceStatusPrivate(params)); this._parser.setCsiHandler({intermediates: '!', final: 'p'}, params => this.softReset(params)); @@ -454,10 +454,43 @@ export class InputHandler extends Disposable implements IInputHandler { super.dispose(); } - public parse(data: string | Uint8Array): void { + // FIXME: cleanup async handling + private _parseStack = { + paused: false, + cursorStartX: 0, + cursorStartY: 0, + decodedLength: 0, + position: 0 + }; + + private _preserveStack(cursorStartX: number, cursorStartY: number, decodedLength: number, position: number): void { + this._parseStack.paused = true; + this._parseStack.cursorStartX = cursorStartX; + this._parseStack.cursorStartY = cursorStartY; + this._parseStack.decodedLength = decodedLength; + this._parseStack.position = position; + } + + public parse(data: string | Uint8Array, promiseResult?: boolean): void | Promise { + let result: void | Promise; let buffer = this._bufferService.buffer; - const cursorStartX = buffer.x; - const cursorStartY = buffer.y; + let cursorStartX = buffer.x; + let cursorStartY = buffer.y; + let start = 0; + const wasPaused = this._parseStack.paused; + + if (wasPaused) { + // assumption: _parseBuffer never mutates between async calls + if (result = this._parser.parse(this._parseBuffer, this._parseStack.decodedLength, promiseResult)) { + return result; + } + cursorStartX = this._parseStack.cursorStartX; + cursorStartY = this._parseStack.cursorStartY; + this._parseStack.paused = false; + if (data.length > MAX_PARSEBUFFER_LENGTH) { + start = this._parseStack.position + MAX_PARSEBUFFER_LENGTH; + } + } this._logService.debug('parsing data', data); @@ -469,22 +502,33 @@ export class InputHandler extends Disposable implements IInputHandler { } // Clear the dirty row service so we know which lines changed as a result of parsing - this._dirtyRowService.clearRange(); + // Important: do not clear between async calls, otherwise we lost pending update information. + if (!wasPaused) { + this._dirtyRowService.clearRange(); + } // process big data in smaller chunks if (data.length > MAX_PARSEBUFFER_LENGTH) { - for (let i = 0; i < data.length; i += MAX_PARSEBUFFER_LENGTH) { + for (let i = start; i < data.length; i += MAX_PARSEBUFFER_LENGTH) { const end = i + MAX_PARSEBUFFER_LENGTH < data.length ? i + MAX_PARSEBUFFER_LENGTH : data.length; const len = (typeof data === 'string') ? this._stringDecoder.decode(data.substring(i, end), this._parseBuffer) : this._utf8Decoder.decode(data.subarray(i, end), this._parseBuffer); - this._parser.parse(this._parseBuffer, len); + if (result = this._parser.parse(this._parseBuffer, len)) { + this._preserveStack(cursorStartX, cursorStartY, len, i); + return result; + } } } else { - const len = (typeof data === 'string') - ? this._stringDecoder.decode(data, this._parseBuffer) - : this._utf8Decoder.decode(data, this._parseBuffer); - this._parser.parse(this._parseBuffer, len); + if (!wasPaused) { + const len = (typeof data === 'string') + ? this._stringDecoder.decode(data, this._parseBuffer) + : this._utf8Decoder.decode(data, this._parseBuffer); + if (result = this._parser.parse(this._parseBuffer, len)) { + this._preserveStack(cursorStartX, cursorStartY, len, 0); + return result; + } + } } buffer = this._bufferService.buffer; diff --git a/src/common/input/WriteBuffer.ts b/src/common/input/WriteBuffer.ts index c7d82458..c6178024 100644 --- a/src/common/input/WriteBuffer.ts +++ b/src/common/input/WriteBuffer.ts @@ -37,8 +37,9 @@ export class WriteBuffer { private _pendingData = 0; private _bufferOffset = 0; - constructor(private _action: (data: string | Uint8Array) => void) { } + constructor(private _action: (data: string | Uint8Array, promiseResult?: boolean) => void | Promise) { } + // FIXME: does not work that way anymore with async handlers!!! public writeSync(data: string | Uint8Array): void { // force sync processing on pending data chunks to avoid in-band data scrambling // does the same as innerWrite but without event loop @@ -76,16 +77,64 @@ export class WriteBuffer { this._callbacks.push(callback); } - protected _innerWrite(): void { - const startTime = Date.now(); + protected _innerWrite(d: number = 0, promiseResult: boolean = true): void { + let result: void | Promise; + const startTime = d || Date.now(); while (this._writeBuffer.length > this._bufferOffset) { const data = this._writeBuffer[this._bufferOffset]; - const cb = this._callbacks[this._bufferOffset]; - this._bufferOffset++; - this._action(data); - this._pendingData -= data.length; + if (result = this._action(data, promiseResult)) { + /** + * If we get a promise as return value, we re-schedule the continuation + * as thenable on the promise and exit right away. + * + * The exit here means, that we block input processing at the current active chunk, + * the exact execution position within the chunk is preserved by the saved + * stack content in InputHandler and EscapeSequenceParser. + * + * Resuming happens automatically from that saved stack state. + * Also the resolved promise value is passed along the callstack to + * `EscapeSequenceParser.parse` to correctly resume the stopped handler loop. + * + * Exceptions on async handlers will be logged to console async, but do not interrupt + * the input processing (continues with next handler at the current input position). + * FIXME: No clear exception handling rules for sync handlers yet (will exit whole processing?). + */ + + /** + * If a promise takes long to resolve, we should schedule continuation behind setTimeout. + * This might already be too late, if our .then enters really late (executor + prev thens took very long). + * This cannot be solved here for the handler itself (it is the handlers responsibility to slice hard work), + * but we can at least schedule a screen update as we gain control. + */ + const continuation: (r: boolean) => void = (r: boolean) => Date.now() - startTime >= WRITE_TIMEOUT_MS + ? setTimeout(() => this._innerWrite(0, r)) + : this._innerWrite(startTime, r); + + /** + * Optimization considerations: + * The continuation above favors FPS over throughput by eval'ing `startTime` on resolve. + * This might schedule too many screen updates with bad throughput drops (in case a slow + * resolving handler sliced its work properly behind setTimeout calls). We cannot spot + * this condition here, also the renderer has no way to spot nonsense updates either. + * FIXME: A proper fix for this would track the FPS at the renderer entry level separately. + * + * If favoring of FPS shows bad throughtput impact, use the following instead. It favors + * throughput by eval'ing `startTime` upfront pulling at least one more chunk into the + * current microtask queue (executed before setTimeout). + */ + // const continuation: (r: boolean) => void = Date.now() - startTime >= WRITE_TIMEOUT_MS + // ? r => setTimeout(() => this._innerWrite(0, r)) + // : r => this._innerWrite(startTime, r); + + result.then(continuation, err => { setTimeout(() => { throw err; }); continuation(true); }); + return; + } + + const cb = this._callbacks[this._bufferOffset]; if (cb) cb(); + this._bufferOffset++; + this._pendingData -= data.length; if (Date.now() - startTime >= WRITE_TIMEOUT_MS) { break; @@ -99,7 +148,7 @@ export class WriteBuffer { this._callbacks = this._callbacks.slice(this._bufferOffset); this._bufferOffset = 0; } - setTimeout(() => this._innerWrite(), 0); + setTimeout(() => this._innerWrite()); } else { this._writeBuffer = []; this._callbacks = []; diff --git a/src/common/parser/EscapeSequenceParser.ts b/src/common/parser/EscapeSequenceParser.ts index 96d206b9..daf9e6eb 100644 --- a/src/common/parser/EscapeSequenceParser.ts +++ b/src/common/parser/EscapeSequenceParser.ts @@ -449,7 +449,45 @@ export class EscapeSequenceParser extends Disposable implements IEscapeSequenceP this.precedingCodepoint = 0; } + // FIXME: cleanup async handling + private _parseStack: { + paused: boolean; + type: 'ESC' | 'CSI'; // FIXME: support for DCS and OSC + handlers: CsiHandlerType[] | EscHandlerType[]; + handlerPos: number; + transition: number; + currentState: ParserState; + collect: number; + pos: number; + } = { + paused: false, + type: 'ESC', + handlers: [], + handlerPos: 0, + transition: 0, + currentState: 0, + collect: 0, + pos: 0 + }; + private _preserveStack( + type: 'ESC' | 'CSI', + handlers: CsiHandlerType[] | EscHandlerType[], + handlerPos: number, + transition: number, + currentState: ParserState, + collect: number, + pos: number + ): void { + this._parseStack.paused = true; + this._parseStack.type = type; + this._parseStack.handlers = handlers; + this._parseStack.handlerPos = handlerPos; + this._parseStack.transition = transition; + this._parseStack.currentState = currentState; + this._parseStack.collect = collect; + this._parseStack.pos = pos; + } /** * Parse UTF32 codepoints in `data` up to `length`. @@ -465,7 +503,7 @@ export class EscapeSequenceParser extends Disposable implements IEscapeSequenceP * - OSC_STRING:OSC_PUT * - DCS_PASSTHROUGH:DCS_PUT */ - public parse(data: Uint32Array, length: number): void { + public parse(data: Uint32Array, length: number, promiseResult?: boolean): void | Promise { let code = 0; let transition = 0; let currentState = this.currentState; @@ -475,8 +513,58 @@ export class EscapeSequenceParser extends Disposable implements IEscapeSequenceP const params = this._params; const table: Uint8Array = this._transitions.table; + let res: any; + let start = 0; + if (this._parseStack.paused) { + const handlers = this._parseStack.handlers; + let handlerPos = this._parseStack.handlerPos - 1; + transition = this._parseStack.transition; + currentState = this._parseStack.currentState; + collect = this._parseStack.collect; + start = this._parseStack.pos; + + // we have to resume the old handler loop if: + // - return value of the promise was `false` + // - handlers are not exhausted yet + // FIXME: removing handlers from within a handler of the same sequence + // is not supported atm (also true for sync handlers)!! + if (promiseResult === false && handlerPos > -1) { + switch (this._parseStack.type) { + case 'CSI': + for (; handlerPos >= 0; handlerPos--) { + if ((res = (handlers as CsiHandlerType[])[handlerPos](params)) !== false) { + if (res instanceof Promise) { + this._parseStack.handlerPos = handlerPos; + return res; + } + break; + } + } + break; + case 'ESC': + for (; handlerPos >= 0; handlerPos--) { + if ((res = (handlers as EscHandlerType[])[handlerPos]()) !== false) { + if (res instanceof Promise) { + this._parseStack.handlerPos = handlerPos; + return res; + } + break; + } + } + break; + } + } + // cleanup before continuing with the main loop + this.precedingCodepoint = 0; + this._parseStack.paused = false; + start++; + currentState = transition & TableAccess.TRANSITION_STATE_MASK; + } + + // console.log('startPos', start, length); + // process input string - for (let i = 0; i < length; ++i) { + for (let i = start; i < length; ++i) { code = data[i]; // normal transition & action lookup @@ -534,7 +622,13 @@ export class EscapeSequenceParser extends Disposable implements IEscapeSequenceP let j = handlers ? handlers.length - 1 : -1; for (; j >= 0; j--) { // undefined or true means success and to stop bubbling - if (handlers[j](params) !== false) { + // FIXME: remove setHandler interface, always use addHandler with proper return value in true|false + // background - result of undefined leads to nonsense instanceof Promise test below + if ((res = handlers[j](params)) !== false) { + if (res && res instanceof Promise) { + this._preserveStack('CSI', handlers, j, transition, currentState, collect, i); + return res; + } break; } } @@ -568,7 +662,11 @@ export class EscapeSequenceParser extends Disposable implements IEscapeSequenceP let jj = handlersEsc ? handlersEsc.length - 1 : -1; for (; jj >= 0; jj--) { // undefined or true means success and to stop bubbling - if (handlersEsc[jj]() !== false) { + if ((res = handlersEsc[jj]()) !== false) { + if (res && res instanceof Promise) { + this._preserveStack('ESC', handlersEsc, jj, transition, currentState, collect, i); + return res; + } break; } } diff --git a/src/common/parser/Types.d.ts b/src/common/parser/Types.d.ts index e2fac89f..8d55c51a 100644 --- a/src/common/parser/Types.d.ts +++ b/src/common/parser/Types.d.ts @@ -160,7 +160,7 @@ export interface IEscapeSequenceParser extends IDisposable { * Parse UTF32 codepoints in `data` up to `length`. * @param data The data to parse. */ - parse(data: Uint32Array, length: number): void; + parse(data: Uint32Array, length: number, promiseResult?: boolean): void | Promise; /** * Get string from numercial function identifier `ident`. diff --git a/test/benchmark/Terminal.benchmark.ts b/test/benchmark/Terminal.benchmark.ts index 76139cb6..7d0f5d84 100644 --- a/test/benchmark/Terminal.benchmark.ts +++ b/test/benchmark/Terminal.benchmark.ts @@ -29,7 +29,7 @@ perfContext('Terminal: ls -lR /usr/lib', () => { chunks.push(data as unknown as Buffer); length += data.length; }); - await new Promise(resolve => p.on('exit', () => resolve())); + await new Promise(resolve => p.on('exit', () => resolve())); contentUtf8 = Buffer.concat(chunks, length); // translate to content string const buffer = new Uint32Array(contentUtf8.length); @@ -44,24 +44,46 @@ perfContext('Terminal: ls -lR /usr/lib', () => { } }); - perfContext('write', () => { + // perfContext('write/string/sync', () => { + // let terminal: Terminal; + // before(() => { + // terminal = new Terminal({cols: 80, rows: 25, scrollback: 1000}); + // }); + // new ThroughputRuntimeCase('', () => { + // terminal.writeSync(content); + // return {payloadSize: contentUtf8.length}; + // }, {fork: false}).showAverageThroughput(); + // }); + // + // perfContext('write/Utf8/sync', () => { + // let terminal: Terminal; + // before(() => { + // terminal = new Terminal({cols: 80, rows: 25, scrollback: 1000}); + // }); + // new ThroughputRuntimeCase('', () => { + // terminal.writeSync(content); + // return {payloadSize: contentUtf8.length}; + // }, {fork: false}).showAverageThroughput(); + // }); + + perfContext('write/string/async', () => { let terminal: Terminal; before(() => { terminal = new Terminal({cols: 80, rows: 25, scrollback: 1000}); }); - new ThroughputRuntimeCase('', () => { - terminal.writeSync(content); + new ThroughputRuntimeCase('', async () => { + await new Promise(res => terminal.write(content, res)); return {payloadSize: contentUtf8.length}; }, {fork: false}).showAverageThroughput(); }); - perfContext('writeUtf8', () => { + perfContext('write/Utf8/async', () => { let terminal: Terminal; before(() => { terminal = new Terminal({cols: 80, rows: 25, scrollback: 1000}); }); - new ThroughputRuntimeCase('', () => { - terminal.writeSync(content); + new ThroughputRuntimeCase('', async () => { + await new Promise(res => terminal.write(content, res)); return {payloadSize: contentUtf8.length}; }, {fork: false}).showAverageThroughput(); });