GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

microsoft/qdk

Public

mirrored from https://github.com/microsoft/qdkAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
copilot/fix-wasm-logging-issue

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

qdk/source/npm/qsharp/src/workers

source/npm/qsharp/src/workers/main.ts

351lines · modecode

RawDownload
Latest commit unavailable.
1// Copyright (c) Microsoft Corporation.
2// Licensed under the MIT License.
3
4import type { IQSharpError } from "../../lib/web/qsc_wasm.js";
5import type { CancellationToken } from "../cancellation.js";
6import { QdkDiagnostics } from "../diagnostics.js";
7import { log } from "../log.js";
8import type {
9 CommonEventMessage,
10 EventMessage,
11 IServiceEventMessage,
12 IServiceEventTarget,
13 IServiceProxy,
14 MethodMap,
15 RequestMessage,
16 ResponseMessage,
17 ServiceMethods,
18 ServiceProtocol,
19 ServiceState,
20} from "./types.js";
21
22/**
23 * Creates and initializes a service in a web worker, and returns a proxy for the service
24 * to be used from the main thread.
25 *
26 * @param workerArg The service web worker or the URL of the web worker script.
27 * @param wasmModule The wasm module to initialize the service with
28 * @param serviceProtocol An object that describes the service: its constructor, methods and events
29 * @param workerType The type of worker to create: "classic" for browsers, "module" for Node.js
30 * @returns A proxy object that implements the service interface.
31 * This interface can now be used as if calling into the real service,
32 * and the calls will be proxied to the web worker.
33 */
34export function createProxy<
35 TService extends ServiceMethods<TService>,
36 TServiceEventMsg extends IServiceEventMessage,
37>(
38 workerArg: string | Worker,
39 wasmModule: WebAssembly.Module,
40 serviceProtocol: ServiceProtocol<TService, TServiceEventMsg>,
41 workerType: "classic" | "module",
42): TService & IServiceProxy {
43 // Create or use the WebWorker
44 const useModuleWorker: WorkerOptions = { type: workerType };
45
46 const worker =
47 typeof workerArg === "string"
48 ? new Worker(workerArg, useModuleWorker)
49 : workerArg;
50
51 // Log any errors from the worker
52 worker.addEventListener("error", (ev: Event) => {
53 log.error("Worker error:", ev);
54 });
55
56 // Send it the Wasm module to instantiate
57 worker.postMessage({
58 type: "init",
59 wasmModule,
60 qscLogLevel: log.getLogLevel(),
61 });
62
63 // When the log level changes, update the worker's log level as well.
64 const unsubscribeLevelChanged = log.addLevelChangedListener((level) => {
65 worker.postMessage({ type: "set-log-level", level });
66 });
67
68 // If you lose the 'this' binding, some environments have issues
69 const postMessage = worker.postMessage.bind(worker);
70 const onTerminate = () => {
71 unsubscribeLevelChanged();
72 worker.terminate();
73 };
74
75 // Create the proxy which will forward method calls to the worker
76 const proxy = createProxyInternal<TService, TServiceEventMsg>(
77 postMessage,
78 onTerminate,
79 serviceProtocol.methods,
80 );
81
82 // Let proxy handle response and event messages from the worker
83 worker.addEventListener("message", (ev: MessageEvent) => {
84 proxy.onMsgFromWorker(ev.data);
85 });
86 return proxy;
87}
88
89/** Holds state for a single request received by the proxy */
90type RequestState<
91 TService extends ServiceMethods<TService>,
92 TServiceEventMsg extends IServiceEventMessage,
93> = RequestMessage<TService> & {
94 resolve: (val: any) => void;
95 reject: (err: any) => void;
96 requestEventTarget?: IServiceEventTarget<TServiceEventMsg>;
97 cancellationToken?: CancellationToken;
98};
99
100/*
101The WorkerProxy works by queuing up requests to send over to the Worker, only
102ever having one in flight at a time. By queuing on the caller side, this allows
103for cancellation (it checks if a request is cancelled before sending to the worker).
104
105The queue contains an entry for each request with the data to send, the promise
106to resolve, the event handler, and the cancellation token. When a request completes
107the next one (if present) is fetched from the queue. If it is marked as cancelled,
108it is resolved immediately, else it is marked as the current request and the command
109sent to the worker. As events occurs on the current request the event handler is
110invoked. When the response is received this is used to resolve the promise and
111complete the request.
112*/
113
114/**
115 * Function to create the proxy for a type. To be used from the main thread.
116 *
117 * @param postMessage A function to post messages to the worker
118 * @param terminator A function to call to tear down the worker thread
119 * @param methods A map of method names to be proxied and some metadata @see MethodMap
120 * @returns The proxy object. The caller should then set the onMsgFromWorker
121 * property to a callback that will receive messages from the worker.
122 */
123function createProxyInternal<
124 TService extends ServiceMethods<TService>,
125 TServiceEventMsg extends IServiceEventMessage,
126>(
127 postMessage: (msg: RequestMessage<TService>) => void,
128 terminator: () => void,
129 methods: MethodMap<TService>,
130): TService &
131 IServiceProxy & {
132 onMsgFromWorker: (
133 msg: ResponseMessage<TService> | EventMessage<TServiceEventMsg>,
134 ) => void;
135 } {
136 const queue: RequestState<TService, TServiceEventMsg>[] = [];
137 const eventTarget = new EventTarget();
138 let curr: RequestState<TService, TServiceEventMsg> | undefined;
139 let state: ServiceState = "idle";
140
141 function setState(newState: ServiceState) {
142 if (state === newState) return;
143 state = newState;
144 if (proxy.onstatechange) proxy.onstatechange(state);
145 }
146
147 type ResultOf<TRespMsg> = TRespMsg extends { result: infer R } ? R : never;
148
149 function queueRequest(
150 msg: RequestMessage<TService>,
151 requestEventTarget?: IServiceEventTarget<TServiceEventMsg>,
152 cancellationToken?: CancellationToken,
153 ): Promise<ResultOf<ResponseMessage<TService>>> {
154 return new Promise((resolve, reject) => {
155 queue.push({
156 type: msg.type,
157 args: msg.args,
158 resolve,
159 reject,
160 requestEventTarget,
161 cancellationToken,
162 } as RequestState<TService, TServiceEventMsg>);
163
164 // If nothing was running when this got added, kick off processing
165 if (queue.length === 1) doNextRequest();
166 });
167 }
168
169 function doNextRequest() {
170 if (curr) return;
171
172 while ((curr = queue.shift())) {
173 if (curr.cancellationToken?.isCancellationRequested) {
174 curr.reject("cancelled");
175 continue;
176 } else {
177 break;
178 }
179 }
180 if (!curr) {
181 // Nothing else queued, signal that we're now idle and exit.
182 log.trace("Proxy: Worker queue is empty");
183 setState("idle");
184 return;
185 }
186
187 const msg = { type: curr.type, args: curr.args };
188 if (methods[curr.type] === "requestWithProgress") {
189 setState("busy");
190 }
191
192 log.trace("Proxy: Posting message to worker: %o", msg);
193 postMessage(msg);
194 }
195
196 function onMsgFromWorker(
197 msg:
198 | ResponseMessage<TService>
199 | EventMessage<TServiceEventMsg>
200 | CommonEventMessage,
201 ) {
202 if (log.getLogLevel() >= 4)
203 log.trace("Proxy: Received message from worker: %s", JSON.stringify(msg));
204
205 if (msg.messageType === "common-event") {
206 const commonEvent = msg; // assignment is necessary here for TypeScript to narrow the type
207 switch (commonEvent.type) {
208 case "telemetry-event":
209 {
210 const detail = commonEvent.detail;
211 log.logTelemetry(detail);
212 }
213 break;
214 case "log":
215 {
216 const detail = commonEvent.detail;
217 log.logWithLevel(detail.level, detail.target, ...detail.data);
218 }
219 break;
220 }
221 } else if (msg.messageType === "event") {
222 const event = new Event(msg.type) as Event & TServiceEventMsg;
223 event.detail = msg.detail;
224
225 log.trace("Proxy: Posting event: %o", msg);
226 // Post to a currently attached event target if there's a "requestWithProgress"
227 // in progress
228 curr?.requestEventTarget?.dispatchEvent(event);
229 // Also post to the general event target
230 eventTarget.dispatchEvent(event);
231 } else if (msg.messageType === "response") {
232 if (!curr) {
233 log.error("Proxy: No active request when message received: %o", msg);
234 return;
235 }
236 const result = {
237 success: msg.result.success,
238 data: msg.result.success ? msg.result.result : msg.result.error,
239 };
240 if (result.success) {
241 curr.resolve(result.data);
242 curr = undefined;
243 doNextRequest();
244 } else {
245 let err = result.data;
246
247 // The error may be a serialized error object.
248 err = deserializeIfError(err);
249
250 curr.reject(err);
251 curr = undefined;
252 doNextRequest();
253 }
254 }
255 }
256
257 // Create the proxy object to be returned
258 const proxy = {} as TService &
259 IServiceProxy & { onMsgFromWorker: typeof onMsgFromWorker };
260
261 // Assign each method with the desired proxying behavior
262 for (const methodName of Object.keys(methods) as (keyof TService &
263 string)[]) {
264 // @ts-expect-error - tricky to derive the type of the actual method here
265 proxy[methodName] = (...args: any[]) => {
266 let requestEventTarget:
267 | IServiceEventTarget<TServiceEventMsg>
268 | undefined = undefined;
269
270 switch (methods[methodName]) {
271 case "addEventListener":
272 {
273 // @ts-expect-error - can't get the typing of the rest parameters quite right
274 eventTarget.addEventListener(...args);
275 }
276 break;
277 case "removeEventListener":
278 {
279 // @ts-expect-error - can't get the typing of the rest parameters quite right
280 eventTarget.removeEventListener(...args);
281 }
282 break;
283 case "requestWithProgress": {
284 // For progress methods, the last argument is the event target
285 requestEventTarget = args[args.length - 1];
286 args = args.slice(0, args.length - 1);
287 }
288 // fallthrough
289 case "request": {
290 return queueRequest(
291 { type: methodName, args } as RequestMessage<TService>,
292 requestEventTarget,
293 );
294 }
295 }
296 };
297 }
298
299 proxy.onstatechange = null;
300 proxy.terminate = () => {
301 // Kill the worker without a chance to shutdown. May be needed if it is not responding.
302 log.info("Proxy: Terminating the worker");
303 if (curr) {
304 log.trace(
305 "Proxy: Terminating running worker item of type: %s",
306 curr.type,
307 );
308 curr.reject("terminated");
309 }
310 // Reject any outstanding items
311 while (queue.length) {
312 const item = queue.shift();
313 log.trace(
314 "Proxy: Terminating outstanding work item of type: %s",
315 item?.type,
316 );
317 item?.reject("terminated");
318 }
319 terminator();
320 };
321 proxy.onMsgFromWorker = onMsgFromWorker;
322
323 return proxy;
324}
325
326/**
327 * Deserializes an error if it is a known type.
328 *
329 * By default, browsers can only send certain types of errors between the main thread and a worker.
330 * See: https://developer.mozilla.org/en-US/docs/Web/API/Web_Workers_API/Structured_clone_algorithm#error_types
331 *
332 * Serializing our own custom errors allows us to send them between threads.
333 */
334function deserializeIfError(err: unknown) {
335 if (err !== null && typeof err === "object" && "name" in err) {
336 if (err.name === "QdkDiagnostics" && "data" in err) {
337 err = new QdkDiagnostics(err.data as IQSharpError[]);
338 } else if (
339 err.name === "WebAssembly.RuntimeError" &&
340 "message" in err &&
341 (typeof err.message === "string" || typeof err.message === "undefined") &&
342 "stack" in err &&
343 (typeof err.stack === "string" || typeof err.stack === "undefined")
344 ) {
345 const newErr = new WebAssembly.RuntimeError(err.message);
346 newErr.stack = err.stack;
347 err = newErr;
348 }
349 }
350 return err;
351}
352