Add JSSending.receive(...) to receive multiple objects at once

Author: kateinoigakukun

Examples/OffscrenCanvas/Sources/MyApp/main.swift

@@ -11,7 +11,7 @@ protocol CanvasRenderer {
 struct BackgroundRenderer: CanvasRenderer {
     func render(canvas: JSObject, size: Int) async throws {
         let executor = try await WebWorkerTaskExecutor(numberOfThreads: 1)
-        let transfer = JSTransferring(canvas)
+        let transfer = JSSending.transfer(canvas)
         let renderingTask = Task(executorPreference: executor) {
             let canvas = try await transfer.receive()
             try await renderAnimation(canvas: canvas, size: size)

Runtime/src/index.ts

@@ -11,6 +11,7 @@ import {
 import * as JSValue from "./js-value.js";
 import { Memory } from "./memory.js";
 import { deserializeError, MainToWorkerMessage, MessageBroker, ResponseMessage, ITCInterface, serializeError, SwiftRuntimeThreadChannel, WorkerToMainMessage } from "./itc.js";
+import { decodeObjectRefs } from "./js-value.js";
 
 export type SwiftRuntimeOptions = {
     /**
@@ -208,7 +209,7 @@ export class SwiftRuntime {
                     } catch (error) {
                         responseMessage.data.response = {
                             ok: false,
-                            error: serializeError(new TypeError(`Failed to serialize response message: ${error}`))
+                            error: serializeError(new TypeError(`Failed to serialize message: ${error}`))
                         };
                         newBroker.reply(responseMessage);
                     }
@@ -648,24 +649,56 @@ export class SwiftRuntime {
                 // Main thread's tid is always -1
                 return this.tid || -1;
             },
-            swjs_request_transferring_object: (
-                object_ref: ref,
+            swjs_request_sending_object: (
+                sending_object: ref,
+                transferring_objects: pointer,
+                transferring_objects_count: number,
                 object_source_tid: number,
-                transferring: pointer,
+                sending_context: pointer,
             ) => {
                 if (!this.options.threadChannel) {
                     throw new Error("threadChannel is not set in options given to SwiftRuntime. Please set it to request transferring objects.");
                 }
                 const broker = getMessageBroker(this.options.threadChannel);
+                const memory = this.memory;
+                const transferringObjects = decodeObjectRefs(transferring_objects, transferring_objects_count, memory);
+                broker.request({
+                    type: "request",
+                    data: {
+                        sourceTid: this.tid ?? MAIN_THREAD_TID,
+                        targetTid: object_source_tid,
+                        context: sending_context,
+                        request: {
+                            method: "send",
+                            parameters: [sending_object, transferringObjects, sending_context],
+                        }
+                    }
+                })
+            },
+            swjs_request_sending_objects: (
+                sending_objects: pointer,
+                sending_objects_count: number,
+                transferring_objects: pointer,
+                transferring_objects_count: number,
+                object_source_tid: number,
+                sending_context: pointer,
+            ) => {
+                if (!this.options.threadChannel) {
+                    throw new Error("threadChannel is not set in options given to SwiftRuntime. Please set it to request transferring objects.");
+                }
+                const broker = getMessageBroker(this.options.threadChannel);
+                const memory = this.memory;
+                const sendingObjects = decodeObjectRefs(sending_objects, sending_objects_count, memory);
+                const transferringObjects = decodeObjectRefs(transferring_objects, transferring_objects_count, memory);
                 broker.request({
                     type: "request",
                     data: {
                         sourceTid: this.tid ?? MAIN_THREAD_TID,
                         targetTid: object_source_tid,
-                        context: transferring,
+                        context: sending_context,
                         request: {
-                            method: "transfer",
-                            parameters: [object_ref, transferring],
+                            method: "sendObjects",
+                            parameters: [sendingObjects, transferringObjects, sending_context],
                         }
                     }
                 })

Runtime/src/itc.ts

@@ -85,9 +85,16 @@ export type SwiftRuntimeThreadChannel =
 export class ITCInterface {
     constructor(private memory: Memory) {}
 
-    transfer(objectRef: ref, transferring: pointer): { object: any, transferring: pointer, transfer: Transferable[] } {
-        const object = this.memory.getObject(objectRef);
-        return { object, transferring, transfer: [object] };
+    send(sendingObject: ref, transferringObjects: ref[], sendingContext: pointer): { object: any, sendingContext: pointer, transfer: Transferable[] } {
+        const object = this.memory.getObject(sendingObject);
+        const transfer = transferringObjects.map(ref => this.memory.getObject(ref));
+        return { object, sendingContext, transfer };
+    }
+
+    sendObjects(sendingObjects: ref[], transferringObjects: ref[], sendingContext: pointer): { object: any[], sendingContext: pointer, transfer: Transferable[] } {
+        const objects = sendingObjects.map(ref => this.memory.getObject(ref));
+        const transfer = transferringObjects.map(ref => this.memory.getObject(ref));
+        return { object: objects, sendingContext, transfer };
     }
 
     release(objectRef: ref): { object: undefined, transfer: Transferable[] } {

Runtime/src/js-value.ts

@@ -1,5 +1,5 @@
 import { Memory } from "./memory.js";
-import { assertNever, JavaScriptValueKindAndFlags, pointer } from "./types.js";
+import { assertNever, JavaScriptValueKindAndFlags, pointer, ref } from "./types.js";
 
 export const enum Kind {
     Boolean = 0,
@@ -142,3 +142,11 @@ export const writeAndReturnKindBits = (
     }
     throw new Error("Unreachable");
 };
+
+export function decodeObjectRefs(ptr: pointer, length: number, memory: Memory): ref[] {
+    const result: ref[] = new Array(length);
+    for (let i = 0; i < length; i++) {
+        result[i] = memory.readUint32(ptr + 4 * i);
+    }
+    return result;
+}

Runtime/src/types.ts

@@ -115,10 +115,20 @@ export interface ImportedFunctions {
     swjs_listen_message_from_worker_thread: (tid: number) => void;
     swjs_terminate_worker_thread: (tid: number) => void;
     swjs_get_worker_thread_id: () => number;
-    swjs_request_transferring_object: (
-        object_ref: ref,
+    swjs_request_sending_object: (
+        sending_object: ref,
+        transferring_objects: pointer,
+        transferring_objects_count: number,
         object_source_tid: number,
-        transferring: pointer,
+        sending_context: pointer,
+    ) => void;
+    swjs_request_sending_objects: (
+        sending_objects: pointer,
+        sending_objects_count: number,
+        transferring_objects: pointer,
+        transferring_objects_count: number,
+        object_source_tid: number,
+        sending_context: pointer,
     ) => void;
 }
 

Sources/JavaScriptEventLoop/JSObject+Transferring.swift

@@ -16,6 +16,34 @@ import _CJavaScriptEventLoop
 
 /// A task executor that runs tasks on Web Worker threads.
 ///
+/// The `WebWorkerTaskExecutor` provides a way to execute Swift tasks in parallel across multiple
+/// Web Worker threads, enabling true multi-threaded execution in WebAssembly environments.
+/// This allows CPU-intensive tasks to be offloaded from the main thread, keeping the user
+/// interface responsive.
+///
+/// ## Multithreading Model
+///
+/// Each task submitted to the executor runs on one of the available worker threads. By default, 
+/// child tasks created within a worker thread continue to run on the same worker thread,
+/// maintaining thread locality and avoiding excessive context switching.
+///
+/// ## Object Sharing Between Threads
+///
+/// When working with JavaScript objects across threads, you must use the `JSSending` API to 
+/// explicitly transfer or clone objects:
+///
+/// ```swift
+/// // Create and transfer an object to a worker thread
+/// let buffer = JSObject.global.ArrayBuffer.function!.new(1024).object!
+/// let transferring = JSSending.transfer(buffer)
+///
+/// let task = Task(executorPreference: executor) {
+///     // Receive the transferred buffer in the worker
+///     let workerBuffer = try await transferring.receive()
+///     // Use the buffer in the worker thread
+/// }
+/// ```
+///
 /// ## Prerequisites
 ///
 /// This task executor is designed to work with [wasi-threads](https://github.com/WebAssembly/wasi-threads)
@@ -24,22 +52,40 @@ import _CJavaScriptEventLoop
 /// from spawned Web Workers, and forward the message to the main thread
 /// by calling `_swjs_enqueue_main_job_from_worker`.
 ///
-/// ## Usage
+/// ## Basic Usage
 ///
 /// ```swift
-/// let executor = WebWorkerTaskExecutor(numberOfThreads: 4)
+/// // Create an executor with 4 worker threads
+/// let executor = try await WebWorkerTaskExecutor(numberOfThreads: 4)
 /// defer { executor.terminate() }
 ///
+/// // Execute a task on a worker thread
+/// let task = Task(executorPreference: executor) {
+///     // This runs on a worker thread
+///     return performHeavyComputation()
+/// }
+/// let result = await task.value
+///
+/// // Run a block on a worker thread
 /// await withTaskExecutorPreference(executor) {
-///   // This block runs on the Web Worker thread.
-///   await withTaskGroup(of: Int.self) { group in
+///     // This entire block runs on a worker thread
+///     performHeavyComputation()
+/// }
+///
+/// // Execute multiple tasks in parallel
+/// await withTaskGroup(of: Int.self) { group in
 ///     for i in 0..<10 {
-///       // Structured child works are executed on the Web Worker thread.
-///       group.addTask { fibonacci(of: i) }
+///         group.addTask(executorPreference: executor) {
+///             // Each task runs on a worker thread
+///             return fibonacci(i)
+///         }
+///     }
+///     
+///     for await result in group {
+///         // Process results as they complete
 ///     }
-///   }
 /// }
-/// ````
+/// ```
 ///
 /// ## Known limitations
 ///
@@ -359,36 +405,89 @@ public final class WebWorkerTaskExecutor: TaskExecutor {
 
     private let executor: Executor
 
-    /// Create a new Web Worker task executor.
+    /// Creates a new Web Worker task executor with the specified number of worker threads.
+    ///
+    /// This initializer creates a pool of Web Worker threads that can execute Swift tasks
+    /// in parallel. The initialization is asynchronous because it waits for all worker
+    /// threads to be properly initialized before returning.
+    ///
+    /// The number of threads should typically match the number of available CPU cores
+    /// for CPU-bound workloads. For I/O-bound workloads, you might benefit from more
+    /// threads than CPU cores.
+    ///
+    /// ## Example
+    ///
+    /// ```swift
+    /// // Create an executor with 4 worker threads
+    /// let executor = try await WebWorkerTaskExecutor(numberOfThreads: 4)
+    ///
+    /// // Always terminate the executor when you're done with it
+    /// defer { executor.terminate() }
+    ///
+    /// // Use the executor...
+    /// ```
     ///
     /// - Parameters:
     ///   - numberOfThreads: The number of Web Worker threads to spawn.
-    ///   - timeout: The timeout to wait for all worker threads to be started.
-    ///   - checkInterval: The interval to check if all worker threads are started.
+    ///   - timeout: The maximum time to wait for all worker threads to be started. Default is 3 seconds.
+    ///   - checkInterval: The interval to check if all worker threads are started. Default is 5 microseconds.
+    /// - Throws: An error if any worker thread fails to initialize within the timeout period.
     public init(numberOfThreads: Int, timeout: Duration = .seconds(3), checkInterval: Duration = .microseconds(5)) async throws {
         self.executor = Executor(numberOfThreads: numberOfThreads)
         try await self.executor.start(timeout: timeout, checkInterval: checkInterval)
     }
 
-    /// Terminate child Web Worker threads.
-    /// Jobs enqueued to the executor after calling this method will be ignored.
+    /// Terminates all worker threads managed by this executor.
+    ///
+    /// This method should be called when the executor is no longer needed to free up 
+    /// resources. After calling this method, any tasks enqueued to this executor will 
+    /// be ignored and may never complete.
+    ///
+    /// It's recommended to use a `defer` statement immediately after creating the executor
+    /// to ensure it's properly terminated when it goes out of scope.
+    ///
+    /// ## Example
+    ///
+    /// ```swift
+    /// do {
+    ///     let executor = try await WebWorkerTaskExecutor(numberOfThreads: 4)
+    ///     defer { executor.terminate() }
+    ///
+    ///     // Use the executor...
+    /// }
+    /// // Executor is automatically terminated when exiting the scope
+    /// ```
     ///
-    /// NOTE: This method must be called after all tasks that prefer this executor are done.
-    /// Otherwise, the tasks may stuck forever.
+    /// - Important: This method must be called after all tasks that prefer this executor are done.
+    ///   Otherwise, the tasks may stuck forever.
     public func terminate() {
         executor.terminate()
     }
 
-    /// The number of Web Worker threads.
+    /// Returns the number of worker threads managed by this executor.
+    ///
+    /// This property reflects the value provided during initialization and doesn't change
+    /// during the lifetime of the executor.
+    ///
+    /// ## Example
+    ///
+    /// ```swift
+    /// let executor = try await WebWorkerTaskExecutor(numberOfThreads: 4)
+    /// print("Executor is running with \(executor.numberOfThreads) threads")
+    /// // Prints: "Executor is running with 4 threads"
+    /// ```
     public var numberOfThreads: Int {
         executor.numberOfThreads
     }
 
     // MARK: TaskExecutor conformance
 
-    /// Enqueue a job to the executor.
+    /// Enqueues a job to be executed by one of the worker threads.
+    ///
+    /// This method is part of the `TaskExecutor` protocol and is called by the Swift
+    /// Concurrency runtime. You typically don't need to call this method directly.
     ///
-    /// NOTE: Called from the Swift Concurrency runtime.
+    /// - Parameter job: The job to enqueue.
     public func enqueue(_ job: UnownedJob) {
         Self.traceStatsIncrement(\.enqueueExecutor)
         executor.enqueue(job)
@@ -431,9 +530,23 @@ public final class WebWorkerTaskExecutor: TaskExecutor {
     @MainActor private static var _swift_task_enqueueGlobalWithDelay_hook_original: UnsafeMutableRawPointer?
     @MainActor private static var _swift_task_enqueueGlobalWithDeadline_hook_original: UnsafeMutableRawPointer?
 
-    /// Install a global executor that forwards jobs from Web Worker threads to the main thread.
+    /// Installs a global executor that forwards jobs from Web Worker threads to the main thread.
+    ///
+    /// This method sets up the necessary hooks to ensure proper task scheduling between
+    /// the main thread and worker threads. It must be called once (typically at application 
+    /// startup) before using any `WebWorkerTaskExecutor` instances.
+    ///
+    /// ## Example
+    ///
+    /// ```swift
+    /// // At application startup
+    /// WebWorkerTaskExecutor.installGlobalExecutor()
+    ///
+    /// // Later, create and use executor instances
+    /// let executor = try await WebWorkerTaskExecutor(numberOfThreads: 4)
+    /// ```
     ///
-    /// This function must be called once before using the Web Worker task executor.
+    /// - Important: This method must be called from the main thread.
     public static func installGlobalExecutor() {
         MainActor.assumeIsolated {
             installGlobalExecutorIsolated()

Sources/JavaScriptEventLoop/WebWorkerTaskExecutor.swift

@@ -316,11 +316,20 @@ IMPORT_JS_FUNCTION(swjs_get_worker_thread_id, int, (void))
 
 int swjs_get_worker_thread_id_cached(void);
 
-/// Requests transferring a JavaScript object to another worker thread.
+/// Requests sending a JavaScript object to another worker thread.
 ///
 /// This must be called from the destination thread of the transfer.
-IMPORT_JS_FUNCTION(swjs_request_transferring_object, void, (JavaScriptObjectRef object,
-                                                     int object_source_tid,
-                                                     void * _Nonnull transferring))
+IMPORT_JS_FUNCTION(swjs_request_sending_object, void, (JavaScriptObjectRef sending_object,
+                                                       const JavaScriptObjectRef * _Nonnull transferring_objects,
+                                                       int transferring_objects_count,
+                                                       int object_source_tid,
+                                                       void * _Nonnull sending_context))
+
+IMPORT_JS_FUNCTION(swjs_request_sending_objects, void, (const JavaScriptObjectRef * _Nonnull sending_objects,
+                                                       int sending_objects_count,
+                                                       const JavaScriptObjectRef * _Nonnull transferring_objects,
+                                                       int transferring_objects_count,
+                                                       int object_source_tid,
+                                                       void * _Nonnull sending_context))
 
 #endif /* _CJavaScriptKit_h */

Sources/JavaScriptKit/Runtime/index.js

@@ -264,6 +264,12 @@ final class WebWorkerTaskExecutorTests: XCTestCase {
         executor.terminate()
     }
 
+    func testSendingWithoutReceiving() async throws {
+        let object = JSObject.global.Object.function!.new()
+        _ = JSSending.transfer(object)
+        _ = JSSending(object)
+    }
+
     func testTransferMainToWorker() async throws {
         let Uint8Array = JSObject.global.Uint8Array.function!
         let buffer = Uint8Array.new(100).buffer.object!
@@ -275,6 +281,9 @@ final class WebWorkerTaskExecutorTests: XCTestCase {
         }
         let byteLength = try await task.value
         XCTAssertEqual(byteLength, 100)
+
+        // Transferred Uint8Array should have 0 byteLength
+        XCTAssertEqual(buffer.byteLength.number!, 0)
     }
 
     func testTransferWorkerToMain() async throws {
@@ -306,7 +315,50 @@ final class WebWorkerTaskExecutorTests: XCTestCase {
             XCTFail("Should throw an error")
             return
         }
-        XCTAssertTrue(jsErrorMessage.contains("Failed to serialize response message"))
+        XCTAssertTrue(jsErrorMessage.contains("Failed to serialize message"), jsErrorMessage)
+    }
+
+    func testTransferMultipleTimes() async throws {
+        let executor = try await WebWorkerTaskExecutor(numberOfThreads: 1)
+        let Uint8Array = JSObject.global.Uint8Array.function!
+        let buffer = Uint8Array.new(100).buffer.object!
+        let transferring = JSSending.transfer(buffer)
+        let task1 = Task(executorPreference: executor) {
+            let buffer = try await transferring.receive()
+            return buffer.byteLength.number!
+        }
+        let byteLength1 = try await task1.value
+        XCTAssertEqual(byteLength1, 100)
+
+        let task2 = Task<String?, Never>(executorPreference: executor) {
+            do {
+                _ = try await transferring.receive()
+                return nil
+            } catch {
+                return String(describing: error)
+            }
+        }
+        guard let jsErrorMessage = await task2.value else {
+            XCTFail("Should throw an error")
+            return
+        }
+        XCTAssertTrue(jsErrorMessage.contains("Failed to serialize message"))
+    }
+
+    func testCloneMultipleTimes() async throws {
+        let executor = try await WebWorkerTaskExecutor(numberOfThreads: 1)
+        let object = JSObject.global.Object.function!.new()
+        object["test"] = "Hello, World!"
+
+        for _ in 0..<2 {
+            let cloning = JSSending(object)
+            let task = Task(executorPreference: executor) {
+                let object = try await cloning.receive()
+                return object["test"].string!
+            }
+            let result = try await task.value
+            XCTAssertEqual(result, "Hello, World!")
+        }
     }
 
     func testTransferBetweenWorkers() async throws {
@@ -327,6 +379,55 @@ final class WebWorkerTaskExecutorTests: XCTestCase {
         XCTAssertEqual(byteLength, 100)
     }
 
+    func testTransferMultipleItems() async throws {
+        let executor = try await WebWorkerTaskExecutor(numberOfThreads: 1)
+        let Uint8Array = JSObject.global.Uint8Array.function!
+        let buffer1 = Uint8Array.new(10).buffer.object!
+        let buffer2 = Uint8Array.new(11).buffer.object!
+        let transferring1 = JSSending.transfer(buffer1)
+        let transferring2 = JSSending.transfer(buffer2)
+        let task = Task(executorPreference: executor) {
+            let (buffer1, buffer2) = try await JSSending.receive(transferring1, transferring2)
+            return (buffer1.byteLength.number!, buffer2.byteLength.number!)
+        }
+        let (byteLength1, byteLength2) = try await task.value
+        XCTAssertEqual(byteLength1, 10)
+        XCTAssertEqual(byteLength2, 11)
+        XCTAssertEqual(buffer1.byteLength.number!, 0)
+        XCTAssertEqual(buffer2.byteLength.number!, 0)
+
+        // Mix transferring and cloning
+        let buffer3 = Uint8Array.new(12).buffer.object!
+        let buffer4 = Uint8Array.new(13).buffer.object!
+        let transferring3 = JSSending.transfer(buffer3)
+        let cloning4 = JSSending(buffer4)
+        let task2 = Task(executorPreference: executor) {
+            let (buffer3, buffer4) = try await JSSending.receive(transferring3, cloning4)
+            return (buffer3.byteLength.number!, buffer4.byteLength.number!)
+        }
+        let (byteLength3, byteLength4) = try await task2.value
+        XCTAssertEqual(byteLength3, 12)
+        XCTAssertEqual(byteLength4, 13)
+        XCTAssertEqual(buffer3.byteLength.number!, 0)
+        XCTAssertEqual(buffer4.byteLength.number!, 13)
+    }
+
+    func testCloneObjectToWorker() async throws {
+        let executor = try await WebWorkerTaskExecutor(numberOfThreads: 1)
+        let object = JSObject.global.Object.function!.new()
+        object["test"] = "Hello, World!"
+        let cloning = JSSending(object)
+        let task = Task(executorPreference: executor) {
+            let object = try await cloning.receive()
+            return object["test"].string!
+        }
+        let result = try await task.value
+        XCTAssertEqual(result, "Hello, World!")
+
+        // Further access to the original object is valid
+        XCTAssertEqual(object["test"].string!, "Hello, World!")
+    }
+
 /*
     func testDeinitJSObjectOnDifferentThread() async throws {
         let executor = try await WebWorkerTaskExecutor(numberOfThreads: 1)