perf: avoid serialized copies of large worker buffers

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+- Copy large ArrayBuffers directly into receiver-owned contents for Worker and MessagePort messages, preserving source buffers and stored serialization bytes.
+
 - Honor elapsed `node:vm` timeouts when native SQLite writes outlast delayed timer delivery, releasing write cursors before propagating the timeout.
 
 - Sync oven-sh/bun through `a8fcd319be58ad4d284330b86602cfb3c8e49c83`, including native-binding schema migration, source-position updates, and runtime hardening while retaining fork compatibility and source-sharing fixes.

src/jsc/bindings/webcore/MessagePort.cpp

@@ -120,7 +120,7 @@ ExceptionOr<void> MessagePort::postMessage(JSC::JSGlobalObject& state, JSC::JSVa
     }
 
     Vector<RefPtr<MessagePort>> ports;
-    auto messageData = SerializedScriptValue::create(state, messageValue, WTF::move(options.transfer), ports, SerializationForStorage::No, SerializationContext::WorkerPostMessage);
+    auto messageData = SerializedScriptValue::create(state, messageValue, WTF::move(options.transfer), ports, SerializationForStorage::No, SerializationContext::WorkerPostMessage, SerializationForCrossProcessTransfer::No, SerializationForSingleRecipient::Yes);
     if (messageData.hasException()) {
         // Satisfy the exception-check verifier for create()'s simulated throw
         // WITHOUT clearing the pending exception (propagateException needs it).

src/jsc/bindings/webcore/SerializedScriptValue.cpp

@@ -731,14 +731,14 @@ class CloneSerializer : public CloneBase {
 #endif
         Vector<uint8_t>& out, SerializationContext context, ArrayBufferContentsArray& sharedBuffers,
         Vector<void*>& serializedBlockListRefs,
-        SerializationForStorage forStorage, SerializationForCrossProcessTransfer forTransfer)
+        SerializationForStorage forStorage, SerializationForCrossProcessTransfer forTransfer, std::unique_ptr<ArrayBufferContentsArray>* clonedBuffers = nullptr)
     {
         CloneSerializer serializer(lexicalGlobalObject, messagePorts, arrayBuffers,
 #if ENABLE(WEBASSEMBLY)
             wasmModules,
             wasmMemoryHandles,
 #endif
-            out, context, sharedBuffers, forStorage, forTransfer);
+            out, context, sharedBuffers, forStorage, forTransfer, clonedBuffers);
         auto code = serializer.serialize(value);
         serializedBlockListRefs = WTF::move(serializer.m_serializedBlockListRefs);
         return code;
@@ -752,7 +752,7 @@ class CloneSerializer : public CloneBase {
         WasmModuleArray& wasmModules,
         WasmMemoryHandleArray& wasmMemoryHandles,
 #endif
-        Vector<uint8_t>& out, SerializationContext context, ArrayBufferContentsArray& sharedBuffers, SerializationForStorage forStorage, SerializationForCrossProcessTransfer forTransfer)
+        Vector<uint8_t>& out, SerializationContext context, ArrayBufferContentsArray& sharedBuffers, SerializationForStorage forStorage, SerializationForCrossProcessTransfer forTransfer, std::unique_ptr<ArrayBufferContentsArray>* clonedBuffers = nullptr)
         : CloneBase(lexicalGlobalObject)
         , m_buffer(out)
         , m_emptyIdentifier(Identifier::fromString(lexicalGlobalObject->vm(), emptyString()))
@@ -764,6 +764,7 @@ class CloneSerializer : public CloneBase {
 #endif
         , m_forStorage(forStorage)
         , m_forTransfer(forTransfer)
+        , m_clonedBuffers(clonedBuffers)
     {
         write(CurrentVersion);
         fillTransferMap(messagePorts, m_transferredMessagePorts);
@@ -1295,6 +1296,26 @@ class CloneSerializer : public CloneBase {
                     write(index->value);
                     return true;
                 }
+                constexpr size_t largeBufferThreshold = 16 * 1024;
+                if (m_clonedBuffers && !arrayBuffer->isShared() && !arrayBuffer->isResizableOrGrowableShared() && arrayBuffer->byteLength() >= largeBufferThreshold) {
+                    auto copy = arrayBuffer->sliceWithClampedIndex(0, arrayBuffer->byteLength());
+                    ArrayBufferContents contents;
+                    if (!copy || !copy->transferTo(vm, contents)) {
+                        code = SerializationReturnCode::DataCloneError;
+                        return true;
+                    }
+                    if (!*m_clonedBuffers)
+                        *m_clonedBuffers = makeUnique<ArrayBufferContentsArray>();
+                    uint32_t clonedIndex = (*m_clonedBuffers)->size();
+                    (*m_clonedBuffers)->append(WTF::move(contents));
+                    // This mode has no explicit transfers, so reuse their backing index map.
+                    m_transferredArrayBuffers.add(obj, clonedIndex);
+                    m_gcBuffer.appendWithCrashOnOverflow(obj);
+                    // Move the copy to the sole receiver; the source buffer is never detached.
+                    write(ArrayBufferTransferTag);
+                    write(clonedIndex);
+                    return true;
+                }
                 if (!startObjectInternal(obj)) // handle duplicates
                     return true;
 
@@ -2021,6 +2042,7 @@ class CloneSerializer : public CloneBase {
 #endif
     SerializationForStorage m_forStorage;
     SerializationForCrossProcessTransfer m_forTransfer;
+    std::unique_ptr<ArrayBufferContentsArray>* m_clonedBuffers;
 };
 
 extern "C" void WebCore__CloneSerializer__write(CloneSerializer* serializer, const uint8_t* data, uint32_t length)
@@ -4536,12 +4558,12 @@ RefPtr<SerializedScriptValue> SerializedScriptValue::create(JSC::JSGlobalObject&
     return result.releaseReturnValue();
 }
 
-ExceptionOr<Ref<SerializedScriptValue>> SerializedScriptValue::create(JSGlobalObject& globalObject, JSValue value, Vector<JSC::Strong<JSC::JSObject>>&& transferList, Vector<RefPtr<MessagePort>>& messagePorts, SerializationForStorage forStorage, SerializationContext serializationContext, SerializationForCrossProcessTransfer forTransfer)
+ExceptionOr<Ref<SerializedScriptValue>> SerializedScriptValue::create(JSGlobalObject& globalObject, JSValue value, Vector<JSC::Strong<JSC::JSObject>>&& transferList, Vector<RefPtr<MessagePort>>& messagePorts, SerializationForStorage forStorage, SerializationContext serializationContext, SerializationForCrossProcessTransfer forTransfer, SerializationForSingleRecipient singleRecipient)
 {
-    return create(globalObject, value, WTF::move(transferList), messagePorts, forStorage, SerializationErrorMode::Throwing, serializationContext, forTransfer);
+    return create(globalObject, value, WTF::move(transferList), messagePorts, forStorage, SerializationErrorMode::Throwing, serializationContext, forTransfer, singleRecipient);
 }
 
-ExceptionOr<Ref<SerializedScriptValue>> SerializedScriptValue::create(JSGlobalObject& lexicalGlobalObject, JSValue value, Vector<JSC::Strong<JSC::JSObject>>&& transferList, Vector<RefPtr<MessagePort>>& messagePorts, SerializationForStorage forStorage, SerializationErrorMode throwExceptions, SerializationContext context, SerializationForCrossProcessTransfer forTransfer)
+ExceptionOr<Ref<SerializedScriptValue>> SerializedScriptValue::create(JSGlobalObject& lexicalGlobalObject, JSValue value, Vector<JSC::Strong<JSC::JSObject>>&& transferList, Vector<RefPtr<MessagePort>>& messagePorts, SerializationForStorage forStorage, SerializationErrorMode throwExceptions, SerializationContext context, SerializationForCrossProcessTransfer forTransfer, SerializationForSingleRecipient singleRecipient)
 {
     VM& vm = lexicalGlobalObject.vm();
     auto scope = DECLARE_THROW_SCOPE(vm);
@@ -4838,13 +4860,19 @@ ExceptionOr<Ref<SerializedScriptValue>> SerializedScriptValue::create(JSGlobalOb
     WasmMemoryHandleArray wasmMemoryHandles;
 #endif
     std::unique_ptr<ArrayBufferContentsArray> sharedBuffers = makeUnique<ArrayBufferContentsArray>();
+    std::unique_ptr<ArrayBufferContentsArray> arrayBufferContentsArray;
+    bool canMoveClonedBuffers = singleRecipient == SerializationForSingleRecipient::Yes
+        && context == SerializationContext::WorkerPostMessage
+        && forStorage == SerializationForStorage::No
+        && forTransfer == SerializationForCrossProcessTransfer::No
+        && transferList.isEmpty();
     Vector<void*> serializedBlockListRefs;
     auto code = CloneSerializer::serialize(&lexicalGlobalObject, value, messagePorts, arrayBuffers,
 #if ENABLE(WEBASSEMBLY)
         wasmModules,
         wasmMemoryHandles,
 #endif
-        buffer, context, *sharedBuffers, serializedBlockListRefs, forStorage, forTransfer);
+        buffer, context, *sharedBuffers, serializedBlockListRefs, forStorage, forTransfer, canMoveClonedBuffers ? &arrayBufferContentsArray : nullptr);
 
     auto releaseSerializedBlockListRefs = [&] {
         for (auto* ptr : serializedBlockListRefs)
@@ -4867,7 +4895,6 @@ ExceptionOr<Ref<SerializedScriptValue>> SerializedScriptValue::create(JSGlobalOb
     }
 
     // Most calls have no transfer list; this keeps them at one branch.
-    std::unique_ptr<ArrayBufferContentsArray> arrayBufferContentsArray;
     if (!transferList.isEmpty()) {
         if (auto exception = transferListChangedDuringSerialization(vm, transferList, arrayBuffers, messagePorts)) [[unlikely]] {
             releaseSerializedBlockListRefs();

src/jsc/bindings/webcore/SerializedScriptValue.h

@@ -92,6 +92,9 @@ enum class SerializationForStorage : bool { No,
     Yes };
 enum class SerializationForCrossProcessTransfer : bool { No,
     Yes };
+// Side buffers can be consumed only when there is one delivery, never for broadcasts.
+enum class SerializationForSingleRecipient : bool { No,
+    Yes };
 
 using ArrayBufferContentsArray = Vector<JSC::ArrayBufferContents>;
 #if ENABLE(WEBASSEMBLY)
@@ -108,7 +111,7 @@ class SerializedScriptValue : public ThreadSafeRefCounted<SerializedScriptValue>
     WTF_DEPRECATED_MAKE_FAST_ALLOCATED_WITH_HEAP_IDENTIFIER(SerializedScriptValue, SerializedScriptValue);
 
 public:
-    WEBCORE_EXPORT static ExceptionOr<Ref<SerializedScriptValue>> create(JSC::JSGlobalObject&, JSC::JSValue, Vector<JSC::Strong<JSC::JSObject>>&& transfer, Vector<RefPtr<MessagePort>>&, SerializationForStorage = SerializationForStorage::No, SerializationContext = SerializationContext::Default, SerializationForCrossProcessTransfer = SerializationForCrossProcessTransfer::No);
+    WEBCORE_EXPORT static ExceptionOr<Ref<SerializedScriptValue>> create(JSC::JSGlobalObject&, JSC::JSValue, Vector<JSC::Strong<JSC::JSObject>>&& transfer, Vector<RefPtr<MessagePort>>&, SerializationForStorage = SerializationForStorage::No, SerializationContext = SerializationContext::Default, SerializationForCrossProcessTransfer = SerializationForCrossProcessTransfer::No, SerializationForSingleRecipient = SerializationForSingleRecipient::No);
 
     WEBCORE_EXPORT static RefPtr<SerializedScriptValue> create(JSC::JSGlobalObject&, JSC::JSValue, SerializationForStorage = SerializationForStorage::No, SerializationErrorMode = SerializationErrorMode::Throwing, SerializationContext = SerializationContext::Default, SerializationForCrossProcessTransfer = SerializationForCrossProcessTransfer::No);
 
@@ -149,7 +152,7 @@ class SerializedScriptValue : public ThreadSafeRefCounted<SerializedScriptValue>
     WEBCORE_EXPORT ~SerializedScriptValue();
 
 private:
-    static ExceptionOr<Ref<SerializedScriptValue>> create(JSC::JSGlobalObject&, JSC::JSValue, Vector<JSC::Strong<JSC::JSObject>>&& transfer, Vector<RefPtr<MessagePort>>&, SerializationForStorage, SerializationErrorMode, SerializationContext, SerializationForCrossProcessTransfer);
+    static ExceptionOr<Ref<SerializedScriptValue>> create(JSC::JSGlobalObject&, JSC::JSValue, Vector<JSC::Strong<JSC::JSObject>>&& transfer, Vector<RefPtr<MessagePort>>&, SerializationForStorage, SerializationErrorMode, SerializationContext, SerializationForCrossProcessTransfer, SerializationForSingleRecipient = SerializationForSingleRecipient::No);
     WEBCORE_EXPORT SerializedScriptValue(Vector<unsigned char>&&, std::unique_ptr<ArrayBufferContentsArray>&& = nullptr);
 
     SerializedScriptValue(Vector<unsigned char>&&, std::unique_ptr<ArrayBufferContentsArray>, std::unique_ptr<ArrayBufferContentsArray> sharedBuffers

src/jsc/bindings/webcore/Worker.cpp

@@ -97,7 +97,7 @@ Worker::~Worker()
 ExceptionOr<void> Worker::postMessage(JSC::JSGlobalObject& state, JSC::JSValue messageValue, StructuredSerializeOptions&& options)
 {
     Vector<RefPtr<MessagePort>> ports;
-    auto serialized = SerializedScriptValue::create(state, messageValue, WTF::move(options.transfer), ports, SerializationForStorage::No, SerializationContext::WorkerPostMessage);
+    auto serialized = SerializedScriptValue::create(state, messageValue, WTF::move(options.transfer), ports, SerializationForStorage::No, SerializationContext::WorkerPostMessage, SerializationForCrossProcessTransfer::No, SerializationForSingleRecipient::Yes);
     if (serialized.hasException())
         return serialized.releaseException();
 
@@ -505,7 +505,7 @@ JSC_DEFINE_HOST_FUNCTION(jsFunctionPostMessage,
     }
 
     Vector<RefPtr<MessagePort>> ports;
-    ExceptionOr<Ref<SerializedScriptValue>> serialized = SerializedScriptValue::create(*globalObject, value, WTF::move(transferList), ports, SerializationForStorage::No, SerializationContext::WorkerPostMessage);
+    ExceptionOr<Ref<SerializedScriptValue>> serialized = SerializedScriptValue::create(*globalObject, value, WTF::move(transferList), ports, SerializationForStorage::No, SerializationContext::WorkerPostMessage, SerializationForCrossProcessTransfer::No, SerializationForSingleRecipient::Yes);
     RETURN_IF_EXCEPTION(scope, {});
     if (serialized.hasException()) {
         WebCore::propagateException(*globalObject, scope, serialized.releaseException());

test/js/web/broadcastchannel/broadcast-channel.test.ts

@@ -2,6 +2,31 @@ import { bunEnv, bunExe, tempDir } from "harness";
 import type { BroadcastChannelEventMap } from "node:worker_threads";
 import util from "util";
 
+test("large buffer broadcasts make independent copies for every receiver", async () => {
+  const name = crypto.randomUUID();
+  const sender = new BroadcastChannel(name);
+  const receivers = [new BroadcastChannel(name), new BroadcastChannel(name)];
+  try {
+    const pending = receivers.map(channel => {
+      const { promise, resolve, reject } = Promise.withResolvers<Uint8Array>();
+      channel.onmessage = event => resolve(event.data.bytes);
+      channel.onmessageerror = reject;
+      return promise;
+    });
+    const bytes = Buffer.alloc(4 * 1024 * 1024, 0x5a);
+    sender.postMessage({ bytes });
+    const [first, second] = await Promise.all(pending);
+    expect(Buffer.from(first)).toEqual(bytes);
+    expect(Buffer.from(second)).toEqual(bytes);
+    first.fill(0);
+    expect(Buffer.from(second)).toEqual(bytes);
+    expect(bytes).toEqual(Buffer.alloc(bytes.length, 0x5a));
+  } finally {
+    sender.close();
+    receivers.forEach(channel => channel.close());
+  }
+});
+
 test("postMessage results in correct event", done => {
   let c1 = new BroadcastChannel("eventType");
   let c2 = new BroadcastChannel("eventType");

test/js/web/structured-clone-fastpath.test.ts

@@ -1,8 +1,55 @@
 import { describe, expect, test } from "bun:test";
 import { isASAN, rss } from "harness";
-import { MessageChannel as NodeMessageChannel, receiveMessageOnPort } from "node:worker_threads";
+import { markAsUntransferable, MessageChannel as NodeMessageChannel, receiveMessageOnPort } from "node:worker_threads";
 
 describe("Structured Clone Fast Path", () => {
+  test.each([16 * 1024, 4 * 1024 * 1024])(
+    "MessagePort clones %i-byte shared backing references without detaching",
+    size => {
+      const bytes = Buffer.alloc(size, 0x5a);
+      markAsUntransferable(bytes.buffer);
+      const input: any = {
+        bytes,
+        again: bytes,
+        backing: bytes.buffer,
+        view: new DataView(bytes.buffer, bytes.byteOffset + 1, size - 2),
+      };
+      input.self = input;
+      const { port1, port2 } = new NodeMessageChannel();
+      try {
+        port1.postMessage(input);
+        const copy = receiveMessageOnPort(port2)!.message;
+        expect(copy.self).toBe(copy);
+        expect(copy.again).toBe(copy.bytes);
+        expect(copy.backing).toBe(copy.bytes.buffer);
+        expect(copy.view.buffer).toBe(copy.backing);
+        expect(copy.view.byteOffset).toBe(bytes.byteOffset + 1);
+        expect(copy.view.byteLength).toBe(size - 2);
+        expect(Buffer.from(copy.bytes)).toEqual(bytes);
+        copy.bytes.fill(0);
+        expect(bytes).toEqual(Buffer.alloc(size, 0x5a));
+      } finally {
+        port1.close();
+        port2.close();
+      }
+    },
+  );
+
+  test("rejected messages leave a previously cloned large buffer attached", () => {
+    const bytes = Buffer.alloc(16 * 1024, 0x5a);
+    const { port1, port2 } = new NodeMessageChannel();
+    try {
+      expect(() => port1.postMessage({ bytes, invalid() {} })).toThrow(
+        expect.objectContaining({ name: "DataCloneError" }),
+      );
+      expect(bytes).toEqual(Buffer.alloc(16 * 1024, 0x5a));
+      expect(receiveMessageOnPort(port2)).toBeUndefined();
+    } finally {
+      port1.close();
+      port2.close();
+    }
+  });
+
   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);