Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -332,3 +332,5 @@
- Close Worker-owned FileHandles and tracked fs descriptors on termination or natural exit, after pending filesystem work returns. Adapts [oven-sh/bun#34260](https://github.com/oven-sh/bun/pull/34260); thanks robobun.

- Settle child IPC send callbacks after peer disconnection, preserving immediate write errors and completing queued sends at channel close. Adapts [oven-sh/bun#43000](https://github.com/oven-sh/bun/pull/43000); thanks robobun.

- Reduce nested-object Worker `postMessage` and `structuredClone` overhead by avoiding transient value-string atomization and discarded flat-object string copies, while preserving stored and cross-process serialized bytes.
126 changes: 126 additions & 0 deletions bench/postMessage/postMessage-session-entry.mjs
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
import assert from "node:assert/strict";
import { once } from "node:events";
import { performance } from "node:perf_hooks";
import { Worker, isMainThread, parentPort } from "node:worker_threads";
const payloadToken = value => value.inputTokens + value.outputTokens + value.origin.label.length;

if (!isMainThread) {
let count = 0;
let checksum = 0;
parentPort.on("message", message => {
if (message?.__benchmark === "barrier") {
parentPort.postMessage({ count, checksum });
count = 0;
checksum = 0;
} else if (message?.__benchmark === "close") {
parentPort.close();
} else {
count++;
checksum += payloadToken(message);
}
});
parentPort.postMessage({ ready: true });
} else {
const kind = "postMessage";
const name = "session-entry";
const payload = {
sessionId: "synthetic-session",
updatedAt: 1760000000000,
sessionFile: "synthetic/session.jsonl",
model: "fixture-model",
provider: "fixture-provider",
inputTokens: 1200,
outputTokens: 640,
totalTokens: 1840,
contextTokens: 16384,
label: "Synthetic session",
channel: "fixture-channel",
lastChannel: "fixture-channel",
origin: {
label: "Synthetic sender",
provider: "fixture-provider",
surface: "fixture-surface",
chatType: "direct",
from: "fixture:sender",
to: "fixture:recipient",
accountId: "fixture-account",
},
deliveryContext: { channel: "fixture-channel", to: "fixture:recipient", accountId: "fixture-account" },
systemSent: true,
abortedLastRun: false,
};
const expectedToken = payloadToken(payload);
const shape = { operations: 250000, warmupOperations: 25000, batchSize: 1000 };
let worker;
let workerExit;
let workerError;
let sentNs = 0;
let observedChecksum = 0;
worker = new Worker(new URL(import.meta.url));
worker.on("error", error => {
workerError = error;
});
workerExit = once(worker, "exit");
const [ready] = await once(worker, "message");
assert.equal(ready.ready, true);
async function phase(operations) {
let ns = 0;
let checksum = 0;
const elapsedStart = performance.now();
for (let offset = 0; offset < operations; offset += shape.batchSize) {
const count = Math.min(shape.batchSize, operations - offset);
const ack = once(worker, "message");
const started = performance.now();
for (let index = 0; index < count; index++) worker.postMessage(payload);
ns += (performance.now() - started) * 1e6;
worker.postMessage({ __benchmark: "barrier" });
const [receipt] = await ack;
assert.equal(receipt.count, count, "Worker did not consume every message");
assert.equal(receipt.checksum, count * expectedToken, "Worker clone checksum differs");
checksum += receipt.checksum;
if (workerError) throw workerError;
}
assert.equal(checksum, operations * expectedToken);
return { synchronousNs: ns, elapsedMs: performance.now() - elapsedStart, checksum };
}
try {
await phase(shape.warmupOperations);
const cpuBefore = process.cpuUsage();
const memoryBefore = process.memoryUsage();
const sample = await phase(shape.operations);
sentNs = sample.synchronousNs;
observedChecksum = sample.checksum;
const cpu = process.cpuUsage(cpuBefore);
const memoryAfter = process.memoryUsage();
const maxRssKiB = process.resourceUsage().maxRSS;
if (worker) {
worker.postMessage({ __benchmark: "close" });
const [code] = await workerExit;
assert.equal(code, 0, "Worker failed clean exit");
worker = undefined;
}
console.log(
JSON.stringify({
schema: 1,
kind,
name,
runtime: process.versions,
operations: shape.operations,
batchSize: shape.batchSize,
warmupOperations: shape.warmupOperations,
synchronousNs: sentNs,
nsPerOperation: sentNs / shape.operations,
elapsedMs: sample.elapsedMs,
checksum: observedChecksum,
expectedToken,
cpuMicros: cpu,
memoryBefore,
memoryAfter,
maxRssKiB,
cleanExit: true,
}),
);
} finally {
if (worker) await worker.terminate();
}
}
31 changes: 27 additions & 4 deletions src/jsc/bindings/webcore/SerializedScriptValue.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@
#include <wtf/MainThread.h>
#include <wtf/RunLoop.h>
#include <wtf/Vector.h>
#include <wtf/text/StringHash.h>
#include <wtf/threads/BinarySemaphore.h>

#include "ZigGlobalObject.h"
Expand Down Expand Up @@ -1645,8 +1646,12 @@ class CloneSerializer : public CloneBase {

void write(const Identifier& ident)
{
const String& str = ident.string();
StringConstantPool::AddResult addResult = m_constantPool.add(ident.impl(), m_constantPool.size());
writeString(ident.string());
}

void writeString(const String& str)
{
StringConstantPool::AddResult addResult = m_constantPool.add(str, m_constantPool.size());
if (!addResult.isNewEntry) {
write(StringPoolTag);
writeStringIndex(addResult.iterator->value);
Expand Down Expand Up @@ -1682,8 +1687,11 @@ class CloneSerializer : public CloneBase {

void write(const String& str)
{
// Stored and cross-process bytes retain the canonical atom's string encoding.
if (str.isNull())
write(m_emptyIdentifier);
else if (m_context == SerializationContext::WorkerPostMessage && m_forStorage == SerializationForStorage::No && m_forTransfer == SerializationForCrossProcessTransfer::No)
writeString(str);
else
write(Identifier::fromString(m_lexicalGlobalObject->vm(), str));
}
Expand All @@ -1693,7 +1701,7 @@ class CloneSerializer : public CloneBase {
bool isNull = str.isNull();
write(isNull);
if (!isNull)
write(Identifier::fromString(m_lexicalGlobalObject->vm(), str));
write(str);
}

void write(const Vector<uint8_t>& vector)
Expand Down Expand Up @@ -1983,7 +1991,7 @@ class CloneSerializer : public CloneBase {
ObjectPool m_objectPool;
ObjectPool m_transferredMessagePorts;
ObjectPool m_transferredArrayBuffers;
typedef HashMap<RefPtr<UniquedStringImpl>, uint32_t, IdentifierRepHash> StringConstantPool;
typedef HashMap<String, uint32_t> StringConstantPool;
StringConstantPool m_constantPool;
Identifier m_emptyIdentifier;
SerializationContext m_context;
Expand Down Expand Up @@ -4671,10 +4679,25 @@ ExceptionOr<Ref<SerializedScriptValue>> SerializedScriptValue::create(JSGlobalOb
// ArrayStorage / Undecided / holes forwarding → fall through to normal serialization path
}

unsigned simplePropertyCount = 0;
if (canUseObjectFastPath) {
// Reject nested values before allocating cross-thread string copies.
structure->forEachProperty(vm, [&](const PropertyTableEntry& entry) -> bool {
JSValue propertyValue = object->getDirect(entry.offset());
if (propertyValue.isCell() && !propertyValue.isString()) {
canUseObjectFastPath = false;
return false;
}
++simplePropertyCount;
return true;
});
}

if (canUseObjectFastPath) {
ASSERT(object != nullptr);

WTF::Vector<SimpleInMemoryPropertyTableEntry> properties;
properties.reserveInitialCapacity(simplePropertyCount);

structure->forEachProperty(vm, [&](const PropertyTableEntry& entry) -> bool {
// Only enumerable, data properties
Expand Down
42 changes: 42 additions & 0 deletions test/js/web/structured-clone-fastpath.test.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,49 @@
import { describe, expect, test } from "bun:test";
import { isASAN, rss } from "harness";
import { MessageChannel as NodeMessageChannel, receiveMessageOnPort } from "node:worker_threads";

describe("Structured Clone Fast Path", () => {
test.each(["clone", "port"])("%s preserves repeated strings and references after a primitive prefix", kind => {
const latin1 = "caf\u00e9";
const sliced = ("\u0100" + latin1).slice(1);
const child = { latin1, sliced, wide: "\u4e2d\u6587", empty: "", absent: undefined, nil: null };
const input: any = { first: latin1, second: sliced, third: "prefix", child, again: child };
input.self = input;
const { port1, port2 } = new NodeMessageChannel();
try {
let copy;
if (kind === "clone") copy = structuredClone(input);
else {
port1.postMessage(input);
copy = receiveMessageOnPort(port2)!.message;
}
expect(copy).toEqual(input);
expect(copy.self).toBe(copy);
expect(copy.child).toBe(copy.again);
expect(copy.child).not.toBe(child);
expect(Object.keys(copy)).toEqual(Object.keys(input));
} finally {
port1.close();
port2.close();
}
});

test("a rejected primitive-prefix probe still visits a nested getter once", () => {
let calls = 0;
const input = {
first: "prefix",
second: "prefix",
child: {
get value() {
calls++;
return "caf\u00e9";
},
},
};
expect(structuredClone(input)).toEqual({ first: "prefix", second: "prefix", child: { value: "caf\u00e9" } });
expect(calls).toBe(1);
});

test("structuredClone should work with empty object", () => {
const object = {};
const cloned = structuredClone(object);
Expand Down
Loading