(tinkerpop) branch master updated: Reduce async overhead in GraphBinary reads (#3630)

[co...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

Cole-Greer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git


The following commit(s) were added to refs/heads/master by this push:
     new d56f99f3f9 Reduce async overhead in GraphBinary reads (#3630)
d56f99f3f9 is described below

commit d56f99f3f901536f64885ffb3c516e05b5fd3dfe
Author: kirill-stepanishin <ki...@improving.com>
AuthorDate: Tue Sep 15 10:39:13 2026 -0700

    Reduce async overhead in GraphBinary reads (#3630)
    
    Adds a synchronous fast path to gremlin-javascript StreamReader
    
    Every primitive read in StreamReader awaits #ensure(n) first. For a buffered submit() response a chunk of
    the payload is already in memory, so that await often does nothing but still costs a Promise and a
    microtask per read, which dominates client CPU on large responses. Reads now return right away when the bytes are already buffered and only await when the buffer needs refilling.
    
    Assisted-by: Claude Code:claude-opus-4-8
---
 .../structure/io/binary/internals/StreamReader.js  | 94 +++++++++++++---------
 1 file changed, 57 insertions(+), 37 deletions(-)

diff --git a/gremlin-js/gremlin-javascript/lib/structure/io/binary/internals/StreamReader.js b/gremlin-js/gremlin-javascript/lib/structure/io/binary/internals/StreamReader.js
index b1f1274812..c134afb54d 100644
--- a/gremlin-js/gremlin-javascript/lib/structure/io/binary/internals/StreamReader.js
+++ b/gremlin-js/gremlin-javascript/lib/structure/io/binary/internals/StreamReader.js
@@ -23,7 +23,8 @@ import { Buffer } from 'buffer';
  * Async byte reader that provides a uniform interface over both a complete Buffer
  * (for non-streaming submit()) and a ReadableStream (for streaming HTTP responses).
  *
- * Handles chunk boundaries transparently and blocks (awaits) until the requested bytes are available.
+ * Chunk boundaries are handled transparently. A read returns immediately when the
+ * requested bytes are already buffered, and otherwise awaits more of the stream.
  */
 export default class StreamReader {
   /** @type {Buffer} */
@@ -106,6 +107,28 @@ export default class StreamReader {
     this.#offset = 0;
   }
 
+  /**
+   * @param {number} n
+   * @returns {boolean} true when the next `n` bytes are already buffered and can be read
+   *   without pulling more of the stream
+   */
+  #available(n) {
+    return this.#offset + n <= this.#buffer.length;
+  }
+
+  /**
+   * Advance past the next `n` bytes and return the offset where they start. The caller must
+   * have already made those bytes available, via #available or #ensure.
+   * @param {number} n
+   * @returns {number} offset of the first of the `n` bytes
+   */
+  #advance(n) {
+    const at = this.#offset;
+    this.#offset = at + n;
+    this.#position += n;
+    return at;
+  }
+
   /**
    * Total number of bytes consumed so far (monotonically increasing).
    * Useful for error diagnostics.
@@ -121,83 +144,80 @@ export default class StreamReader {
    * @returns {Promise<Buffer>}
    */
   async readBytes(n) {
-    await this.#ensure(n);
-    const result = this.#buffer.subarray(this.#offset, this.#offset + n);
-    this.#offset += n;
-    this.#position += n;
-    return result;
+    if (!this.#available(n)) {
+      await this.#ensure(n);
+    }
+    const at = this.#advance(n);
+    return this.#buffer.subarray(at, at + n);
   }
 
   /**
    * @returns {Promise<number>} unsigned 8-bit integer
    */
   async readUInt8() {
-    await this.#ensure(1);
-    this.#position++;
-    return this.#buffer[this.#offset++];
+    if (!this.#available(1)) {
+      await this.#ensure(1);
+    }
+    return this.#buffer[this.#advance(1)];
   }
 
   /**
    * @returns {Promise<number>} signed 8-bit integer
    */
   async readByte() {
-    await this.#ensure(1);
-    this.#position++;
-    return this.#buffer.readInt8(this.#offset++);
+    if (!this.#available(1)) {
+      await this.#ensure(1);
+    }
+    return this.#buffer.readInt8(this.#advance(1));
   }
 
   /**
    * @returns {Promise<number>} signed 16-bit big-endian integer
    */
   async readInt16BE() {
-    await this.#ensure(2);
-    const v = this.#buffer.readInt16BE(this.#offset);
-    this.#offset += 2;
-    this.#position += 2;
-    return v;
+    if (!this.#available(2)) {
+      await this.#ensure(2);
+    }
+    return this.#buffer.readInt16BE(this.#advance(2));
   }
 
   /**
    * @returns {Promise<number>} signed 32-bit big-endian integer
    */
   async readInt32BE() {
-    await this.#ensure(4);
-    const v = this.#buffer.readInt32BE(this.#offset);
-    this.#offset += 4;
-    this.#position += 4;
-    return v;
+    if (!this.#available(4)) {
+      await this.#ensure(4);
+    }
+    return this.#buffer.readInt32BE(this.#advance(4));
   }
 
   /**
    * @returns {Promise<bigint>} signed 64-bit big-endian integer
    */
   async readBigInt64BE() {
-    await this.#ensure(8);
-    const v = this.#buffer.readBigInt64BE(this.#offset);
-    this.#offset += 8;
-    this.#position += 8;
-    return v;
+    if (!this.#available(8)) {
+      await this.#ensure(8);
+    }
+    return this.#buffer.readBigInt64BE(this.#advance(8));
   }
 
   /**
    * @returns {Promise<number>} 32-bit big-endian float
    */
   async readFloatBE() {
-    await this.#ensure(4);
-    const v = this.#buffer.readFloatBE(this.#offset);
-    this.#offset += 4;
-    this.#position += 4;
-    return v;
+    if (!this.#available(4)) {
+      await this.#ensure(4);
+    }
+    return this.#buffer.readFloatBE(this.#advance(4));
   }
 
   /**
    * @returns {Promise<number>} 64-bit big-endian double
    */
   async readDoubleBE() {
-    await this.#ensure(8);
-    const v = this.#buffer.readDoubleBE(this.#offset);
-    this.#offset += 8;
-    this.#position += 8;
-    return v;
+    if (!this.#available(8)) {
+      await this.#ensure(8);
+    }
+    return this.#buffer.readDoubleBE(this.#advance(8));
   }
 }