early version of async support for CSI and ESC handlers

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 {

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<boolean> {
+    let result: void | Promise<boolean>;
     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;

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<boolean>) { }
 
+  // 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<boolean>;
+    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 = [];

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<boolean> {
     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;
             }
           }

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<boolean>;
 
   /**
    * Get string from numercial function identifier `ident`.

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<void>(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<void>(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<void>(res => terminal.write(content, res));
       return {payloadSize: contentUtf8.length};
     }, {fork: false}).showAverageThroughput();
   });