runtime/js/11_workers.js - main - githubmirror/deno

runtime/js/11_workers.js
396 строк11 KB
perf(workers): optimize message passing (#35110)
03 июл 2026, 12:26
03 июл 2026, 12:265a12492
// Copyright 2018-2026 the Deno authors. MIT license.

(function () {
const { core, primordials } = __bootstrap;
const {
  op_create_worker,
  op_host_post_message,
  op_host_post_message_raw,
  op_host_recv_ctrl,
  op_host_recv_message,
  op_host_recv_message_sync,
  op_host_terminate_worker,
} = core.ops;
const {
  ArrayPrototypeFilter,
  ArrayPrototypeJoin,
  Error,
  JSONStringify,
  ObjectPrototypeIsPrototypeOf,
  Promise,
  queueMicrotask,
  String,
  StringPrototypeStartsWith,
  Symbol,
  SymbolFor,
  SymbolIterator,
  SymbolToStringTag,
} = primordials;

const webidl = core.loadExtScript("ext:deno_webidl/00_webidl.js");
const { createFilteredInspectProxy } = core.loadExtScript(
  "ext:deno_web/01_console.js",
);
const { URL } = core.loadExtScript("ext:deno_web/00_url.js");
const { getLocationHref } = core.loadExtScript("ext:deno_web/12_location.js");
const { serializePermissions } = core.loadExtScript(
  "ext:runtime/10_permissions.js",
);
const { log } = core.loadExtScript("ext:runtime/06_util.js");
const {
  defineEventHandler,
  ErrorEvent,
  EventTarget,
  MessageEvent,
  setIsTrusted,
} = core.loadExtScript("ext:deno_web/02_event.js");
const {
  deserializeJsMessageData,
  MessagePortPrototype,
  serializeJsMessageData,
  serializeMessageData,
} = core.loadExtScript("ext:deno_web/13_message_port.js");
const { DOMException } = core.loadExtScript(
  "ext:deno_web/01_dom_exception.js",
);

function createWorker(
  specifier,
  hasSourceCode,
  sourceCode,
  permissions,
  name,
  workerType,
  closeOnIdle,
) {
  return op_create_worker({
    hasSourceCode,
    name,
    permissions: serializePermissions(permissions),
    sourceCode,
    specifier,
    workerType,
    closeOnIdle,
  });
}

function hostTerminateWorker(id) {
  op_host_terminate_worker(id);
}

function hostPostMessage(id, data) {
  op_host_post_message(id, data);
}

function hostRecvCtrl(id) {
  return op_host_recv_ctrl(id);
}

function hostRecvMessage(id) {
  return op_host_recv_message(id);
}

const privateWorkerRef = Symbol();

class Worker extends EventTarget {
  #id = 0;
  #name = "";
  #refCount = 1;
  #messagePromise = undefined;
  #controlPromise = undefined;

  // "RUNNING" | "CLOSED" | "TERMINATED"
  // "TERMINATED" means that any controls or messages received will be
  // discarded. "CLOSED" means that we have received a control
  // indicating that the worker is no longer running, but there might
  // still be messages left to receive.
  #status = "RUNNING";

  constructor(specifier, options = { __proto__: null }) {
    super();
    specifier = String(specifier);
    const {
      deno,
      name,
      type = "classic",
    } = options;

    if (options.env !== undefined || options.workerData !== undefined) {
      const unsupported = [];
      if (options.env !== undefined) unsupported[unsupported.length] = "env";
      if (options.workerData !== undefined) {
        unsupported[unsupported.length] = "workerData";
      }
      globalThis.console.warn(
        `Warning: ${
          ArrayPrototypeJoin(unsupported, ", ")
        } option(s) are not supported ` +
          "for web Workers and will be ignored. Use the " +
          "node:worker_threads module instead.",
      );
    }

    const workerType = webidl.converters["WorkerType"](type);

    if (
      StringPrototypeStartsWith(specifier, "./") ||
      StringPrototypeStartsWith(specifier, "../") ||
      StringPrototypeStartsWith(specifier, "/") || workerType === "classic"
    ) {
      const baseUrl = getLocationHref();
      if (baseUrl != null) {
        specifier = new URL(specifier, baseUrl).href;
      }
    }

    this.#name = name;
    let hasSourceCode, sourceCode;
    if (workerType === "classic") {
      hasSourceCode = true;
      sourceCode = `importScripts(${JSONStringify(specifier)});`;
    } else {
      hasSourceCode = false;
      sourceCode = "";
    }

    const id = createWorker(
      specifier,
      hasSourceCode,
      sourceCode,
      deno?.permissions,
      this.#name,
      workerType,
      false,
    );
    this.#id = id;
    this.#pollControl();
    this.#pollMessages();
  }

  [privateWorkerRef](ref) {
    if (ref) {
      this.#refCount++;
    } else {
      this.#refCount--;
    }

    if (!ref && this.#refCount == 0) {
      if (this.#controlPromise) {
        core.unrefOpPromise(this.#controlPromise);
      }
      if (this.#messagePromise) {
        core.unrefOpPromise(this.#messagePromise);
      }
    } else if (ref && this.#refCount == 1) {
      if (this.#controlPromise) {
        core.refOpPromise(this.#controlPromise);
      }
      if (this.#messagePromise) {
        core.refOpPromise(this.#messagePromise);
      }
    }
  }

  #handleError(e) {
    const event = new ErrorEvent("error", {
      cancelable: true,
      message: e.message,
      lineno: e.lineNumber ? e.lineNumber : undefined,
      colno: e.columnNumber ? e.columnNumber : undefined,
      filename: e.fileName,
      error: null,
    });

    this.dispatchEvent(event);
    // Don't bubble error event to window for loader errors (`!e.fileName`).
    // TODO(nayeemrmn): It's not correct to use `e.fileName` to detect user
    // errors. It won't be there for non-awaited async ops for example.
    if (e.fileName && !event.defaultPrevented) {
      globalThis.dispatchEvent(event);
    }

    return event.defaultPrevented;
  }

  #pollControl = async () => {
    while (this.#status === "RUNNING") {
      this.#controlPromise = hostRecvCtrl(this.#id);
      if (this.#refCount < 1) {
        core.unrefOpPromise(this.#controlPromise);
      }
      const { 0: type, 1: data } = await this.#controlPromise;

      // If terminate was called then we ignore all messages
      if (this.#status === "TERMINATED") {
        return;
      }

      switch (type) {
        case 1: { // TerminalError
          this.#status = "CLOSED";
        } /* falls through */
        case 2: { // Error
          if (!this.#handleError(data)) {
            throw new Error("Unhandled error in child worker.");
          }
          break;
        }
        case 3: { // Close
          log(`Host got "close" message from worker: ${this.#name}`);
          this.#status = "CLOSED";
          return;
        }
        default: {
          throw new Error(`Unknown worker event: "${type}"`);
        }
      }
    }
  };

  #dispatchWorkerMessage(data) {
    let message, transferables;
    try {
      const v = deserializeJsMessageData(data);
      message = v[0];
      transferables = v[1];
    } catch (err) {
      const event = new MessageEvent("messageerror", {
        cancelable: false,
        data: err,
      });
      setIsTrusted(event, true);
      this.dispatchEvent(event);
      return false;
    }
    const event = new MessageEvent("message", {
      cancelable: false,
      data: message,
      // Skip the transferables filter for the common no-transferables case.
      // Passing `undefined` lets the MessageEvent constructor take its cheap
      // `ports == null` branch (a single frozen empty array, no iterator
      // validation) instead of allocating a filtered array per message.
      ports: transferables.length === 0 ? undefined : ArrayPrototypeFilter(
        transferables,
        (t) => ObjectPrototypeIsPrototypeOf(MessagePortPrototype, t),
      ),
    });
    setIsTrusted(event, true);
    this.dispatchEvent(event);
    return true;
  }

  #pollMessages = async () => {
    while (this.#status !== "TERMINATED") {
      this.#messagePromise = hostRecvMessage(this.#id);
      if (this.#refCount < 1) {
        core.unrefOpPromise(this.#messagePromise);
      }
      const data = await this.#messagePromise;
      if (this.#status === "TERMINATED" || data === null) {
        return;
      }
      if (!this.#dispatchWorkerMessage(data)) return;
      // Drain messages already queued on the host side instead of taking the
      // async op + Promise path for each, mirroring the worker global
      // (99_main.js) and node:worker_threads receive loops. The whole burst is
      // processed within this event-loop turn; the batch limit prevents
      // starving the event loop under a sustained flood.
      for (let i = 0; i < 1000 && this.#status !== "TERMINATED"; i++) {
        const syncData = op_host_recv_message_sync(this.#id);
        if (syncData === null) break;
        // Each message dispatch is its own task. Yield a microtask before
        // delivering this already-dequeued message so a handler that re-armed
        // itself in a microtask after the previous dispatch (e.g. reassigning
        // `onmessage` inside a `.then`, as WPT
        // workers/Worker-structure-message.html does) is installed first --
        // otherwise the message reaches the stale handler and is lost. A
        // synchronous checkpoint can't help here: V8 won't run microtasks
        // reentrantly while we are already inside one.
        await new Promise((resolve) => queueMicrotask(() => resolve()));
        if (this.#status === "TERMINATED") return;
        if (!this.#dispatchWorkerMessage(syncData)) return;
      }
    }
  };

  postMessage(message, transferOrOptions = { __proto__: null }) {
    const prefix = "Failed to execute 'postMessage' on 'MessagePort'";
    webidl.requiredArguments(arguments.length, 1, prefix);
    if (this.#status !== "RUNNING") return;
    // Fast path: no transferables
    if (
      transferOrOptions === undefined ||
      transferOrOptions === null ||
      (arguments.length <= 1)
    ) {
      op_host_post_message_raw(
        this.#id,
        serializeMessageData(message, (err) => {
          throw new DOMException(err, "DataCloneError");
        }),
      );
      return;
    }
    message = webidl.converters.any(message);
    let options;
    if (
      webidl.type(transferOrOptions) === "Object" &&
      transferOrOptions !== undefined &&
      transferOrOptions[SymbolIterator] !== undefined
    ) {
      const transfer = webidl.converters["sequence<object>"](
        transferOrOptions,
        prefix,
        "Argument 2",
      );
      options = { transfer };
    } else {
      options = webidl.converters.StructuredSerializeOptions(
        transferOrOptions,
        prefix,
        "Argument 2",
      );
    }
    const { transfer } = options;
    const data = serializeJsMessageData(message, transfer);
    hostPostMessage(this.#id, data);
  }

  terminate() {
    if (this.#status !== "TERMINATED") {
      this.#status = "TERMINATED";
      hostTerminateWorker(this.#id);
    }
  }

  [SymbolFor("Deno.privateCustomInspect")](inspect, inspectOptions) {
    return inspect(
      createFilteredInspectProxy({
        object: this,
        evaluate: ObjectPrototypeIsPrototypeOf(WorkerPrototype, this),
        keys: [
          "onerror",
          "onmessage",
          "onmessageerror",
        ],
      }),
      inspectOptions,
    );
  }

  [SymbolToStringTag] = "Worker";
}

const WorkerPrototype = Worker.prototype;

defineEventHandler(Worker.prototype, "error");
defineEventHandler(Worker.prototype, "message");
defineEventHandler(Worker.prototype, "messageerror");

webidl.converters["WorkerType"] = webidl.createEnumConverter("WorkerType", [
  "classic",
  "module",
]);

return { Worker };
})();