You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by as...@apache.org on 2026/09/27 08:33:44 UTC
(maka) branch main updated: perf(runtime-host): bound fragmented frame copy and scan work (#5356)
This is an automated email from the ASF dual-hosted git repository.
Astro-Han pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/main by this push:
new e3c25828ad perf(runtime-host): bound fragmented frame copy and scan work (#5356)
e3c25828ad is described below
commit e3c25828adcd76efee28ebd90f77631ff8bb9657
Author: seekskyworld <14...@users.noreply.github.com>
AuthorDate: Sun Sep 27 16:33:39 2026 +0800
perf(runtime-host): bound fragmented frame copy and scan work (#5356)
The Runtime Host framed transport re-concatenated the whole pending
buffer with Buffer.concat on every incoming fragment and then searched
it for a newline from the start. A large frame arriving in many small
fragments therefore cost copy and scan work quadratic in the number of
fragments.
Pending bytes now live in a geometrically grown backing buffer, so an
incomplete frame is copied a bounded number of times, and the newline
search resumes where the previous scan stopped, with the cursor reset
whenever bytes are consumed, shifted or reallocated. Framing, the frame
decoder, the 2 MiB pending-byte limit and queue back-pressure are
unchanged.
No behaviour change beyond lower CPU and copy cost for fragmented frames.
Fixes #5355
Generated-by: OpenAI Codex
---
.../__tests__/framed-byte-stream-transport.test.ts | 173 +++++++++++++++++++++
.../src/transport/framed-byte-stream-transport.ts | 46 +++++-
2 files changed, 214 insertions(+), 5 deletions(-)
diff --git a/packages/runtime-host/src/__tests__/framed-byte-stream-transport.test.ts b/packages/runtime-host/src/__tests__/framed-byte-stream-transport.test.ts
new file mode 100644
index 0000000000..c8e559b8b5
--- /dev/null
+++ b/packages/runtime-host/src/__tests__/framed-byte-stream-transport.test.ts
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import assert from 'node:assert/strict';
+import { test } from 'node:test';
+import { deferred } from '@maka/core/test-only/async-primitives';
+import {
+ FramedByteStreamTransport,
+ type RuntimeHostByteStream,
+} from '../transport/framed-byte-stream-transport.js';
+
+test('fragmented Host frames require linear byte copying and delimiter scanning', async (t) => {
+ const stream = new TestByteStream();
+ const transport = new FramedByteStreamTransport(stream);
+ const expected = { payload: 'x'.repeat(192 * 1024) };
+ const wire = Buffer.from(`${JSON.stringify(expected)}\n`);
+ let copiedBytes = 0;
+ let searchedBytes = 0;
+ const concat = Buffer.concat;
+ const copy = Buffer.prototype.copy;
+ const indexOf: (
+ this: Buffer,
+ value: string | number | Uint8Array,
+ offset?: number | BufferEncoding,
+ encoding?: BufferEncoding,
+ ) => number = Buffer.prototype.indexOf;
+ const concatMock = t.mock.method(
+ Buffer,
+ 'concat',
+ (...[list, totalLength]: Parameters<typeof Buffer.concat>) => {
+ copiedBytes += totalLength ?? list.reduce((sum, part) => sum + part.byteLength, 0);
+ return concat(list, totalLength);
+ },
+ );
+ const copyMock = t.mock.method(
+ Buffer.prototype,
+ 'copy',
+ function (this: Buffer, ...[target, targetStart, start, end]: Parameters<Buffer['copy']>) {
+ const written = copy.call(this, target, targetStart, start, end);
+ copiedBytes += written;
+ return written;
+ },
+ );
+ const indexMock = t.mock.method(
+ Buffer.prototype,
+ 'indexOf',
+ function (this: Buffer, ...[value, offset, encoding]: Parameters<typeof indexOf>) {
+ if (value === 0x0a)
+ searchedBytes += this.byteLength - (typeof offset === 'number' ? offset : 0);
+ return indexOf.call(this, value, offset, encoding);
+ },
+ );
+ try {
+ for (let offset = 0; offset < wire.byteLength; offset += 256) {
+ stream.data(wire.subarray(offset, offset + 256));
+ }
+ } finally {
+ concatMock.mock.restore();
+ copyMock.mock.restore();
+ indexMock.mock.restore();
+ }
+ try {
+ assert.deepEqual(await transport.read(1_000), expected);
+ assert.ok(
+ copiedBytes < wire.byteLength * 8,
+ `copied ${copiedBytes} bytes for ${wire.byteLength} input bytes`,
+ );
+ assert.ok(
+ searchedBytes < wire.byteLength * 8,
+ `searched ${searchedBytes} bytes for ${wire.byteLength} input bytes`,
+ );
+ } finally {
+ transport.abort();
+ await transport.closed;
+ }
+});
+
+test('retains split UTF-8 when the producer reuses its buffer and a frame shares its tail', async () => {
+ const stream = new TestByteStream();
+ const transport = new FramedByteStreamTransport(stream);
+ const expected = { text: '客户端🙂'.repeat(40) };
+ const first = Buffer.from(`${JSON.stringify(expected)}\r\n`);
+ const next = Buffer.from(`${JSON.stringify({ next: true })}\n`);
+ const wire = Buffer.concat([first, next]);
+ const reusable = Buffer.alloc(7);
+ try {
+ for (let offset = 0; offset < wire.byteLength; offset += reusable.byteLength) {
+ const size = wire.copy(reusable, 0, offset, offset + reusable.byteLength);
+ stream.data(reusable.subarray(0, size));
+ reusable.fill(0);
+ }
+ stream.end();
+ assert.deepEqual(await transport.read(1_000), expected);
+ assert.deepEqual(await transport.read(1_000), { next: true });
+ await assert.rejects(transport.read(1_000), { code: 'read_eof' });
+ } finally {
+ transport.abort();
+ await transport.closed;
+ }
+});
+
+test('accepts the exact inbound byte budget and drains every complete frame', async () => {
+ const stream = new TestByteStream();
+ const transport = new FramedByteStreamTransport(stream);
+ const limit = 2 * 1024 * 1024;
+ const frame = Buffer.from(`${JSON.stringify({ text: 'x'.repeat(128 * 1024 - 12) })}\n`);
+ const wire = Buffer.concat(Array.from({ length: 16 }, () => frame));
+ assert.equal(wire.byteLength, limit);
+ try {
+ stream.data(wire);
+ stream.end();
+ for (let index = 0; index < 16; index += 1) {
+ assert.deepEqual(await transport.read(1_000), JSON.parse(frame.toString()));
+ }
+ await assert.rejects(transport.read(1_000), { code: 'read_eof' });
+ } finally {
+ transport.abort();
+ await transport.closed;
+ }
+});
+
+test('rejects an inbound chunk one byte over the budget before delivering frames', async () => {
+ const stream = new TestByteStream();
+ const transport = new FramedByteStreamTransport(stream);
+ try {
+ const wire = Buffer.alloc(2 * 1024 * 1024 + 1, '0\n');
+ stream.data(wire);
+ await assert.rejects(transport.read(1_000), { code: 'inbound_queue_full' });
+ await transport.closed;
+ } finally {
+ transport.abort();
+ await transport.closed;
+ }
+});
+
+class TestByteStream implements RuntimeHostByteStream {
+ readonly completion = deferred<void>();
+ readonly closed = this.completion.promise;
+ data: (chunk: Buffer) => void = () => {};
+ end: () => void = () => {};
+ onData(listener: (chunk: Buffer) => void): void {
+ this.data = listener;
+ }
+ onEnd(listener: () => void): void {
+ this.end = listener;
+ }
+ onError(_listener: (error: Error) => void): void {}
+ async write(_chunk: Buffer): Promise<void> {}
+ closeAfterFlush(): void {
+ this.completion.resolve();
+ }
+ abort(): void {
+ this.completion.resolve();
+ }
+ pause(): void {}
+ resume(): void {}
+}
diff --git a/packages/runtime-host/src/transport/framed-byte-stream-transport.ts b/packages/runtime-host/src/transport/framed-byte-stream-transport.ts
index 376493b14d..ceedf3b109 100644
--- a/packages/runtime-host/src/transport/framed-byte-stream-transport.ts
+++ b/packages/runtime-host/src/transport/framed-byte-stream-transport.ts
@@ -72,6 +72,9 @@ export class FramedByteStreamTransport implements RuntimeHostMessageTransport {
readonly #queue: QueuedFrame[] = [];
#queuedBytes = 0;
#buffered = Buffer.alloc(0);
+ #bufferStorage = Buffer.alloc(0);
+ #searchedBytes = 0;
+ #nextNewline: number | undefined;
#waiter: ReadWaiter | undefined;
#failure: Error | undefined;
#readTerminal: RuntimeHostTransportError | undefined;
@@ -156,16 +159,43 @@ export class FramedByteStreamTransport implements RuntimeHostMessageTransport {
this.#failInboundOverflow();
return;
}
- this.#buffered =
- this.#buffered.byteLength === 0 ? Buffer.from(chunk) : Buffer.concat([this.#buffered, chunk]);
+ this.#appendIncoming(chunk);
this.#drainInbound();
}
+ // Geometric growth copies an incomplete frame a bounded number of times;
+ // scanning resumes at the previous tail instead of revisiting every fragment.
+ #appendIncoming(chunk: Buffer): void {
+ const length = this.#buffered.byteLength + chunk.byteLength;
+ let start = this.#buffered.byteOffset - this.#bufferStorage.byteOffset;
+ if (this.#buffered.byteLength === 0 || start + length > this.#bufferStorage.byteLength) {
+ const capacity = Math.min(
+ MAX_BUFFERED_BYTES,
+ Math.max(length, this.#bufferStorage.byteLength * 2),
+ );
+ const storage = Buffer.allocUnsafe(capacity);
+ this.#buffered.copy(storage);
+ this.#bufferStorage = storage;
+ this.#buffered = storage.subarray(0, this.#buffered.byteLength);
+ start = 0;
+ }
+ chunk.copy(this.#bufferStorage, start + this.#buffered.byteLength);
+ this.#buffered = this.#bufferStorage.subarray(start, start + length);
+ }
+
+ #findNewline(): number {
+ if (this.#nextNewline !== undefined) return this.#nextNewline;
+ const newline = this.#buffered.indexOf(0x0a, this.#searchedBytes);
+ this.#searchedBytes = newline === -1 ? this.#buffered.byteLength : newline + 1;
+ if (newline !== -1) this.#nextNewline = newline;
+ return newline;
+ }
+
#drainInbound(): void {
if (this.#failure) return;
try {
while (true) {
- const newline = this.#buffered.indexOf(0x0a);
+ const newline = this.#findNewline();
if (newline === -1) {
if (this.#buffered.byteLength > RUNTIME_HOST_MAX_MESSAGE_BYTES) {
throw new RuntimeHostProtocolError(
@@ -188,16 +218,22 @@ export class FramedByteStreamTransport implements RuntimeHostMessageTransport {
encodedBytes === this.#buffered.byteLength
? Buffer.alloc(0)
: this.#buffered.subarray(encodedBytes);
+ this.#searchedBytes = 0;
+ this.#nextNewline = undefined;
+ if (this.#buffered.byteLength === 0) this.#bufferStorage = Buffer.alloc(0);
const frames = this.#decoder.push(encoded);
if (frames.length !== 1) {
throw new Error('Runtime Host decoder did not produce one complete frame');
}
this.#deliver(frames[0], encodedBytes);
}
- if (this.#ended && !this.#decoderEnded && this.#buffered.indexOf(0x0a) === -1) {
+ if (this.#ended && !this.#decoderEnded && this.#findNewline() === -1) {
if (this.#buffered.byteLength !== 0) {
this.#decoder.push(this.#buffered);
this.#buffered = Buffer.alloc(0);
+ this.#bufferStorage = Buffer.alloc(0);
+ this.#searchedBytes = 0;
+ this.#nextNewline = undefined;
}
this.#decoder.end();
this.#decoderEnded = true;
@@ -224,7 +260,7 @@ export class FramedByteStreamTransport implements RuntimeHostMessageTransport {
}
#updateReadFlow(): void {
- const nextNewline = this.#buffered.indexOf(0x0a);
+ const nextNewline = this.#findNewline();
const nextFrameBytes = nextNewline === -1 ? undefined : nextNewline + 1;
const blocked =
this.#queue.length >= MAX_QUEUED_FRAMES ||
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org