Skip to content

Commit f25bfec

Browse files
kateinoigakukunkateinoigakukun
committed
MessageBroker
1 parent 09d5311 commit f25bfec

File tree

8 files changed

+752
-374
lines changed

8 files changed

+752
-374
lines changed

‎Runtime/src/index.ts‎

Lines changed: 70 additions & 196 deletions
Original file line numberDiff line numberDiff line change
@@ -10,120 +10,7 @@ import {
1010
} from "./types.js";
1111
import * as JSValue from "./js-value.js";
1212
import { Memory } from "./memory.js";
13-
14-
type TransferMessage = {
15-
type: "transfer";
16-
data: {
17-
object: any;
18-
transferring: pointer;
19-
destinationTid: number;
20-
};
21-
};
22-
23-
type RequestTransferMessage = {
24-
type: "requestTransfer";
25-
data: {
26-
objectRef: ref;
27-
objectSourceTid: number;
28-
transferring: pointer;
29-
destinationTid: number;
30-
};
31-
};
32-
33-
type TransferErrorMessage = {
34-
type: "transferError";
35-
data: {
36-
error: string;
37-
};
38-
};
39-
40-
type MainToWorkerMessage = {
41-
type: "wake";
42-
} | RequestTransferMessage | TransferMessage | TransferErrorMessage;
43-
44-
type WorkerToMainMessage = {
45-
type: "job";
46-
data: number;
47-
} | RequestTransferMessage | TransferMessage | TransferErrorMessage;
48-
49-
/**
50-
* A thread channel is a set of functions that are used to communicate between
51-
* the main thread and the worker thread. The main thread and the worker thread
52-
* can send messages to each other using these functions.
53-
*
54-
* @example
55-
* ```javascript
56-
* // worker.js
57-
* const runtime = new SwiftRuntime({
58-
* threadChannel: {
59-
* postMessageToMainThread: postMessage,
60-
* listenMessageFromMainThread: (listener) => {
61-
* self.onmessage = (event) => {
62-
* listener(event.data);
63-
* };
64-
* }
65-
* }
66-
* });
67-
*
68-
* // main.js
69-
* const worker = new Worker("worker.js");
70-
* const runtime = new SwiftRuntime({
71-
* threadChannel: {
72-
* postMessageToWorkerThread: (tid, data) => {
73-
* worker.postMessage(data);
74-
* },
75-
* listenMessageFromWorkerThread: (tid, listener) => {
76-
* worker.onmessage = (event) => {
77-
listener(event.data);
78-
* };
79-
* }
80-
* }
81-
* });
82-
* ```
83-
*/
84-
export type SwiftRuntimeThreadChannel =
85-
| {
86-
/**
87-
* This function is used to send messages from the worker thread to the main thread.
88-
* The message submitted by this function is expected to be listened by `listenMessageFromWorkerThread`.
89-
* @param message The message to be sent to the main thread.
90-
* @param transfer The array of objects to be transferred to the main thread.
91-
*/
92-
postMessageToMainThread: (message: WorkerToMainMessage, transfer: any[]) => void;
93-
/**
94-
* This function is expected to be set in the worker thread and should listen
95-
* to messages from the main thread sent by `postMessageToWorkerThread`.
96-
* @param listener The listener function to be called when a message is received from the main thread.
97-
*/
98-
listenMessageFromMainThread: (listener: (message: MainToWorkerMessage) => void) => void;
99-
}
100-
| {
101-
/**
102-
* This function is expected to be set in the main thread.
103-
* The message submitted by this function is expected to be listened by `listenMessageFromMainThread`.
104-
* @param tid The thread ID of the worker thread.
105-
* @param message The message to be sent to the worker thread.
106-
* @param transfer The array of objects to be transferred to the worker thread.
107-
*/
108-
postMessageToWorkerThread: (tid: number, message: MainToWorkerMessage, transfer: any[]) => void;
109-
/**
110-
* This function is expected to be set in the main thread and should listen
111-
* to messages sent by `postMessageToMainThread` from the worker thread.
112-
* @param tid The thread ID of the worker thread.
113-
* @param listener The listener function to be called when a message is received from the worker thread.
114-
*/
115-
listenMessageFromWorkerThread: (
116-
tid: number,
117-
listener: (message: WorkerToMainMessage) => void
118-
) => void;
119-
120-
/**
121-
* This function is expected to be set in the main thread and called
122-
* when the worker thread is terminated.
123-
* @param tid The thread ID of the worker thread.
124-
*/
125-
terminateWorkerThread?: (tid: number) => void;
126-
};
13+
import { deserializeError, MainToWorkerMessage, MessageBroker, ResponseMessage, ITCInterface, serializeError, SwiftRuntimeThreadChannel, WorkerToMainMessage } from "./itc.js";
12714

12815
export type SwiftRuntimeOptions = {
12916
/**
@@ -294,6 +181,51 @@ export class SwiftRuntime {
294181
importObjects = () => this.wasmImports;
295182

296183
get wasmImports(): ImportedFunctions {
184+
let broker: MessageBroker | null = null;
185+
const getMessageBroker = (threadChannel: SwiftRuntimeThreadChannel) => {
186+
if (broker) return broker;
187+
const itcInterface = new ITCInterface(this.memory);
188+
const newBroker = new MessageBroker(this.tid ?? -1, threadChannel, {
189+
onRequest: (message) => {
190+
let returnValue: ResponseMessage["data"]["response"];
191+
try {
192+
const result = itcInterface[message.data.request.method](...message.data.request.parameters);
193+
returnValue = { ok: true, value: result };
194+
} catch (error) {
195+
returnValue = { ok: false, error: serializeError(error) };
196+
}
197+
const responseMessage: ResponseMessage = {
198+
type: "response",
199+
data: {
200+
sourceTid: message.data.sourceTid,
201+
context: message.data.context,
202+
response: returnValue,
203+
},
204+
}
205+
try {
206+
newBroker.reply(responseMessage);
207+
} catch (error) {
208+
responseMessage.data.response = {
209+
ok: false,
210+
error: serializeError(new TypeError(`Failed to serialize response message: ${error}`))
211+
};
212+
newBroker.reply(responseMessage);
213+
}
214+
},
215+
onResponse: (message) => {
216+
if (message.data.response.ok) {
217+
const object = this.memory.retain(message.data.response.value.object);
218+
this.exports.swjs_receive_response(object, message.data.context);
219+
} else {
220+
const error = deserializeError(message.data.response.error);
221+
const errorObject = this.memory.retain(error);
222+
this.exports.swjs_receive_error(errorObject, message.data.context);
223+
}
224+
}
225+
})
226+
broker = newBroker;
227+
return newBroker;
228+
}
297229
return {
298230
swjs_set_prop: (
299231
ref: ref,
@@ -634,38 +566,18 @@ export class SwiftRuntime {
634566
"listenMessageFromMainThread is not set in options given to SwiftRuntime. Please set it to listen to wake events from the main thread."
635567
);
636568
}
569+
const broker = getMessageBroker(threadChannel);
637570
threadChannel.listenMessageFromMainThread((message) => {
638571
switch (message.type) {
639572
case "wake":
640573
this.exports.swjs_wake_worker_thread();
641574
break;
642-
case "requestTransfer": {
643-
const object = this.memory.getObject(message.data.objectRef);
644-
const messageToMainThread: TransferMessage = {
645-
type: "transfer",
646-
data: {
647-
object,
648-
destinationTid: message.data.destinationTid,
649-
transferring: message.data.transferring,
650-
},
651-
};
652-
try {
653-
this.postMessageToMainThread(messageToMainThread, [object]);
654-
} catch (error) {
655-
this.postMessageToMainThread({
656-
type: "transferError",
657-
data: { error: String(error) },
658-
});
659-
}
660-
break;
661-
}
662-
case "transfer": {
663-
const objectRef = this.memory.retain(message.data.object);
664-
this.exports.swjs_receive_object(objectRef, message.data.transferring);
575+
case "request": {
576+
broker.onReceivingRequest(message);
665577
break;
666578
}
667-
case "transferError": {
668-
console.error(message.data.error); // TODO: Handle the error
579+
case "response": {
580+
broker.onReceivingResponse(message);
669581
break;
670582
}
671583
default:
@@ -684,59 +596,19 @@ export class SwiftRuntime {
684596
"listenMessageFromWorkerThread is not set in options given to SwiftRuntime. Please set it to listen to jobs from worker threads."
685597
);
686598
}
599+
const broker = getMessageBroker(threadChannel);
687600
threadChannel.listenMessageFromWorkerThread(
688601
tid, (message) => {
689602
switch (message.type) {
690603
case "job":
691604
this.exports.swjs_enqueue_main_job_from_worker(message.data);
692605
break;
693-
case "requestTransfer": {
694-
if (message.data.objectSourceTid == MAIN_THREAD_TID) {
695-
const object = this.memory.getObject(message.data.objectRef);
696-
if (message.data.destinationTid != tid) {
697-
throw new Error("Invariant violation: The destination tid of the transfer request must be the same as the tid of the worker thread that received the request.");
698-
}
699-
this.postMessageToWorkerThread(message.data.destinationTid, {
700-
type: "transfer",
701-
data: {
702-
object,
703-
transferring: message.data.transferring,
704-
destinationTid: message.data.destinationTid,
705-
},
706-
}, [object]);
707-
} else {
708-
// Proxy the transfer request to the worker thread that owns the object
709-
this.postMessageToWorkerThread(message.data.objectSourceTid, {
710-
type: "requestTransfer",
711-
data: {
712-
objectRef: message.data.objectRef,
713-
objectSourceTid: tid,
714-
transferring: message.data.transferring,
715-
destinationTid: message.data.destinationTid,
716-
},
717-
});
718-
}
606+
case "request": {
607+
broker.onReceivingRequest(message);
719608
break;
720609
}
721-
case "transfer": {
722-
if (message.data.destinationTid == MAIN_THREAD_TID) {
723-
const objectRef = this.memory.retain(message.data.object);
724-
this.exports.swjs_receive_object(objectRef, message.data.transferring);
725-
} else {
726-
// Proxy the transfer response to the destination worker thread
727-
this.postMessageToWorkerThread(message.data.destinationTid, {
728-
type: "transfer",
729-
data: {
730-
object: message.data.object,
731-
transferring: message.data.transferring,
732-
destinationTid: message.data.destinationTid,
733-
},
734-
}, [message.data.object]);
735-
}
736-
break;
737-
}
738-
case "transferError": {
739-
console.error(message.data.error); // TODO: Handle the error
610+
case "response": {
611+
broker.onReceivingResponse(message);
740612
break;
741613
}
742614
default:
@@ -761,20 +633,22 @@ export class SwiftRuntime {
761633
object_source_tid: number,
762634
transferring: pointer,
763635
) => {
764-
if (this.tid == object_source_tid) {
765-
// Fast path: The object is already in the same thread
766-
this.exports.swjs_receive_object(object_ref, transferring);
767-
return;
636+
if (!this.options.threadChannel) {
637+
throw new Error("threadChannel is not set in options given to SwiftRuntime. Please set it to request transferring objects.");
768638
}
769-
this.postMessageToMainThread({
770-
type: "requestTransfer",
639+
const broker = getMessageBroker(this.options.threadChannel);
640+
broker.request({
641+
type: "request",
771642
data: {
772-
objectRef: object_ref,
773-
objectSourceTid: object_source_tid,
774-
transferring,
775-
destinationTid: this.tid ?? MAIN_THREAD_TID,
776-
},
777-
});
643+
sourceTid: this.tid ?? MAIN_THREAD_TID,
644+
targetTid: object_source_tid,
645+
context: transferring,
646+
request: {
647+
method: "transfer",
648+
parameters: [object_ref, transferring],
649+
}
650+
}
651+
})
778652
},
779653
};
780654
}

0 commit comments

Comments
 (0)