fix: improve protocol handling performance in all cases (#2002)

Author: connor4312

CHANGELOG.md

@@ -5,6 +5,7 @@ This changelog records changes to stable releases since 1.50.2. "TBA" changes he
 ## Nightly (only)
 
 - fix: improve main-thread performance of source map rename ([vscode#210518](https://github.com/microsoft/vscode/issues/210518))
+- fix: improve protocol handling performance in all cases ([#2001](https://github.com/microsoft/vscode-js-debug/issues/2001))
 
 ## v1.89 (April 2024)
 

package-lock.json

@@ -78,7 +78,6 @@
     "reflect-metadata": "^0.2.1",
     "signale": "^1.4.0",
     "source-map-support": "^0.5.21",
-    "split2": "^4.2.0",
     "to-absolute-glob": "^3.0.0",
     "vscode-tas-client": "^0.1.84",
     "ws": "^8.16.0"
@@ -118,7 +117,6 @@
     "@types/prettier": "^2.7.3",
     "@types/signale": "^1.4.7",
     "@types/sinon": "^17.0.3",
-    "@types/split2": "^4.2.3",
     "@types/stream-buffers": "^3.0.7",
     "@types/tmp": "^0.2.6",
     "@types/to-absolute-glob": "^2.0.3",

package.json

@@ -2,12 +2,12 @@
  * Copyright (C) Microsoft Corporation. All rights reserved.
  *--------------------------------------------------------*/
 
-import split from 'split2';
 import { Duplex, Readable, Writable } from 'stream';
 import { EventEmitter } from '../common/events';
 import { HrTime } from '../common/hrnow';
 import { ILogger, LogTag } from '../common/logging';
 import { once } from '../common/objUtils';
+import { StreamSplitter } from '../common/streamSplitter';
 import { ITransport } from './transport';
 
 export class RawPipeTransport implements ITransport {
@@ -57,8 +57,8 @@ export class RawPipeTransport implements ITransport {
     this.streams = {
       read: read
         .on('error', error => this.logger.error(LogTag.Internal, 'pipeRead error', { error }))
-        .pipe(split('\0'))
-        .on('data', json => this.messageEmitter.fire([json, new HrTime()]))
+        .pipe(new StreamSplitter(0))
+        .on('data', (json: Buffer) => this.messageEmitter.fire([json.toString(), new HrTime()]))
         .on('end', this.onceEnded),
       write: pipeWrite.on('end', this.onceEnded).on('error', this.onWriteError),
     };

src/cdp/rawPipeTransport.ts

@@ -0,0 +1,29 @@
+/*---------------------------------------------------------
+ * Copyright (C) Microsoft Corporation. All rights reserved.
+ *--------------------------------------------------------*/
+
+import * as assert from 'assert';
+import { Writable } from 'stream';
+import { StreamSplitter } from './streamSplitter';
+
+describe('StreamSplitter', () => {
+  it('should split a stream', done => {
+    const chunks: string[] = [];
+    const splitter = new StreamSplitter('\n');
+    const writable = new Writable({
+      write(chunk, _encoding, callback) {
+        chunks.push(chunk.toString());
+        callback();
+      },
+    });
+
+    splitter.pipe(writable);
+    splitter.write('hello\nwor');
+    splitter.write('ld\n');
+    splitter.write('foo\nbar\nz');
+    splitter.end(() => {
+      assert.deepStrictEqual(chunks, ['hello', 'world', 'foo', 'bar', 'z']);
+      done();
+    });
+  });
+});

src/common/streamSplitter.test.ts

@@ -0,0 +1,70 @@
+/*---------------------------------------------------------
+ * Copyright (C) Microsoft Corporation. All rights reserved.
+ *--------------------------------------------------------*/
+
+// Based on VS Code's src/vs/base/node/nodeStreams.ts
+
+import { Transform } from 'stream';
+
+/**
+ * A Transform stream that splits the input on the "splitter" substring.
+ * The resulting chunks will contain (and trail with) the splitter match.
+ * The last chunk when the stream ends will be emitted even if a splitter
+ * is not encountered.
+ */
+export class StreamSplitter extends Transform {
+  private prefix: Buffer[] = [];
+  private readonly splitter: number;
+
+  /** Suffix added after each split chunk. */
+  protected splitSuffix = Buffer.alloc(0);
+
+  constructor(splitter: string | number | Buffer) {
+    super();
+    if (typeof splitter === 'string' && splitter.length === 1) {
+      this.splitter = splitter.charCodeAt(0);
+    } else if (typeof splitter === 'number') {
+      this.splitter = splitter;
+    } else {
+      throw new Error('not implemented here');
+    }
+  }
+
+  override _transform(
+    chunk: Buffer,
+    _encoding: string,
+    callback: (error?: Error | null, data?: unknown) => void,
+  ): void {
+    let offset = 0;
+    while (offset < chunk.length) {
+      const index = chunk.indexOf(this.splitter, offset);
+      if (index === -1) {
+        break;
+      }
+
+      const thisChunk = chunk.subarray(offset, index);
+      const toEmit =
+        this.prefix.length || this.splitSuffix.length
+          ? Buffer.concat([...this.prefix, thisChunk, this.splitSuffix])
+          : thisChunk;
+
+      this.push(toEmit);
+      this.prefix.length = 0;
+      offset = index + 1;
+    }
+
+    if (offset < chunk.length) {
+      this.prefix.push(chunk.subarray(offset));
+    }
+
+    callback();
+  }
+
+  override _flush(callback: (error?: Error | null, data?: unknown) => void): void {
+    for (const buf of this.prefix) {
+      this.push(buf);
+    }
+
+    callback();
+  }
+}

src/common/streamSplitter.ts

@@ -4,15 +4,14 @@
 
 import { ChildProcessWithoutNullStreams, spawn } from 'child_process';
 import { inject, injectable } from 'inversify';
-import split from 'split2';
-import { Transform } from 'stream';
 import { EnvironmentVars } from '../../common/environmentVars';
 import { ILogger } from '../../common/logging';
+import { StreamSplitter } from '../../common/streamSplitter';
 import * as urlUtils from '../../common/urlUtils';
 import { INodeLaunchConfiguration, OutputSource } from '../../configuration';
 import Dap from '../../dap/api';
 import { ILaunchContext } from '../targets';
-import { getNodeLaunchArgs, IProgramLauncher } from './processLauncher';
+import { IProgramLauncher, getNodeLaunchArgs } from './processLauncher';
 import { SubprocessProgram } from './program';
 
 /**
@@ -64,12 +63,12 @@ export class SubprocessProgramLauncher implements IProgramLauncher {
    */
   private captureStdio(dap: Dap.Api, child: ChildProcessWithoutNullStreams) {
     child.stdout
-      .pipe(EtxSplitter.stream())
-      .on('data', output => dap.output({ category: 'stdout', output }))
+      .pipe(new EtxSplitter())
+      .on('data', output => dap.output({ category: 'stdout', output: output.toString() }))
       .resume();
     child.stderr
-      .pipe(EtxSplitter.stream())
-      .on('data', output => dap.output({ category: 'stderr', output }))
+      .pipe(new EtxSplitter())
+      .on('data', output => dap.output({ category: 'stderr', output: output.toString() }))
       .resume();
   }
 
@@ -158,7 +157,7 @@ const formatArguments = (executable: string, args: ReadonlyArray<string>, cwd: s
 };
 
 const enum Char {
-  ETX = '\u0003',
+  ETX = 3,
 }
 
 /**
@@ -168,22 +167,28 @@ const enum Char {
  * finds any, it will switch from splitting the stream on newlines to
  * splitting on ETX characters.
  */
-export class EtxSplitter {
+export class EtxSplitter extends StreamSplitter {
   private etxSpotted = false;
 
-  public static stream(): Transform {
-    return split(new EtxSplitter());
+  constructor() {
+    super(Char.ETX);
+    this.splitSuffix = Buffer.from('\n');
   }
 
-  [Symbol.split](str: string) {
-    this.etxSpotted ||= str.includes(Char.ETX);
-    if (!this.etxSpotted) {
-      return [str, ''];
+  override _transform(
+    chunk: Buffer,
+    _encoding: string,
+    callback: (error?: Error | null | undefined, data?: unknown) => void,
+  ): void {
+    if (!this.etxSpotted && chunk.includes(Char.ETX)) {
+      this.etxSpotted = true;
     }
 
-    const split = str.split(Char.ETX);
+    if (!this.etxSpotted) {
+      this.push(chunk);
+      return callback();
+    }
 
-    // restore or add new lines between each record for proper debug console display
-    return split.length > 1 ? split.map((s, i) => (i < split.length - 1 ? `${s}\n` : s)) : split;
+    return super._transform(chunk, _encoding, callback);
   }
 }

src/targets/node/subprocessProgramLauncher.ts

@@ -8,12 +8,12 @@ import { promises as fsPromises } from 'fs';
 import { tmpdir } from 'os';
 import * as path from 'path';
 import { SinonSandbox, createSandbox } from 'sinon';
-import split from 'split2';
 import * as vscode from 'vscode';
 import { Commands, DebugType } from '../../common/contributionUtils';
 import { findOpenPort } from '../../common/findOpenPort';
 import { LocalFsUtils } from '../../common/fsUtils';
 import { delay } from '../../common/promiseUtil';
+import { StreamSplitter } from '../../common/streamSplitter';
 import { nodeAttachConfigDefaults } from '../../configuration';
 import { resolveProcessId } from '../../ui/processPicker';
 import { createFileTree } from '../createFileTree';
@@ -89,8 +89,11 @@ describe('pick and attach', () => {
       child = spawn('node', ['--inspect-brk', `--inspect-port=${port}`], { stdio: 'pipe' });
       child.on('error', console.error);
       child
-        .stderr!.pipe(split())
-        .on('data', (line: string) => (attached = attached || line.includes('Debugger attached')));
+        .stderr!.pipe(new StreamSplitter('\n'))
+        .on(
+          'data',
+          (line: string) => (attached = attached || line.toString().includes('Debugger attached')),
+        );
     });
 
     it('end to end', async function () {