Merging upstream version 125.0.1.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

4 files changed, 203 insertions, 73 deletions

diff --git a/toolkit/components/promiseworker/PromiseWorker.sys.mjs b/toolkit/components/promiseworker/PromiseWorker.sys.mjs
index 59721e5663..bf5ae179cb 100644
--- a/toolkit/components/promiseworker/PromiseWorker.sys.mjs
+++ b/toolkit/components/promiseworker/PromiseWorker.sys.mjs
@@ -16,27 +16,6 @@
  */
 
 /**
- * An implementation of queues (FIFO).
- *
- * The current implementation uses one array, runs in O(n ^ 2), and is optimized
- * for the case in which queues are generally short.
- */
-function Queue() {
-  this._array = [];
-}
-Queue.prototype = {
-  pop: function pop() {
-    return this._array.shift();
-  },
-  push: function push(x) {
-    return this._array.push(x);
-  },
-  isEmpty: function isEmpty() {
-    return !this._array.length;
-  },
-};
-
-/**
  * Constructors for decoding standard exceptions received from the
  * worker.
  */
@@ -114,14 +93,31 @@ const EXCEPTION_CONSTRUCTORS = {
  *
  * @param {WorkerOptions} options The option parameter for ChromeWorker.
  *
+ * @param {Record<String, function>} functions Functions that the worker can call.
+ *
+ * Functions can be synchronous functions or promises and return a value
+ * that is sent back to the worker. The function can also send back a
+ * `BasePromiseWorker.Meta` object containing the data and an array of transferrable
+ * objects to transfer data to the worker with zero memory copy via `postMessage`.
+ *
+ * Example of sunch a function:
+ *
+ *   async function threadFunction(message) {
+ *     return new BasePromiseWorker.Meta(
+ *         ["data1", "data2", someBuffer],
+ *         {transfers: [someBuffer]}
+ *     );
+ *   }
+ *
  * @constructor
  */
-export var BasePromiseWorker = function (url, options = {}) {
+export var BasePromiseWorker = function (url, options = {}, functions = {}) {
   if (typeof url != "string") {
     throw new TypeError("Expecting a string");
   }
   this._url = url;
   this._options = options;
+  this._functions = functions;
 
   /**
    * A set of methods, with the following
@@ -137,24 +133,24 @@ export var BasePromiseWorker = function (url, options = {}) {
   this.ExceptionHandlers = Object.create(EXCEPTION_CONSTRUCTORS);
 
   /**
-   * The queue of deferred, waiting for the completion of their
+   * The map of deferred, waiting for the completion of their
    * respective job by the worker.
    *
-   * Each item in the list may contain an additional field |closure|,
+   * Each item in the map may contain an additional field |closure|,
    * used to store strong references to value that must not be
    * garbage-collected before the reply has been received (e.g.
    * arrays).
    *
-   * @type {Queue<{deferred:deferred, closure:*=}>}
+   * @type {Map<string, {deferred, closure, id}>}
    */
-  this._queue = new Queue();
+  this._deferredJobs = new Map();
 
   /**
    * The number of the current message.
    *
    * Used for debugging purposes.
    */
-  this._id = 0;
+  this._deferredJobId = 0;
 
   /**
    * The instant at which the worker was launched.
@@ -172,6 +168,11 @@ BasePromiseWorker.prototype = {
     // By Default, ignore all logs.
   },
 
+  _generateDeferredJobId() {
+    this._deferredJobId += 1;
+    return "ThreadToWorker-" + this._deferredJobId;
+  },
+
   /**
    * Instantiate the worker lazily.
    */
@@ -205,9 +206,15 @@ BasePromiseWorker.prototype = {
         error.filename,
         error.lineno
       );
+
       error.preventDefault();
-      let { deferred } = this._queue.pop();
-      deferred.reject(error);
+
+      if (this._deferredJobs.size > 0) {
+        this._deferredJobs.forEach(job => {
+          job.deferred.reject(error);
+        });
+        this._deferredJobs.clear();
+      }
     };
 
     /**
@@ -218,41 +225,74 @@ BasePromiseWorker.prototype = {
      * - {fail: some_error} in case of error, where
      *    some_error is an instance of |PromiseWorker.WorkerError|
      *
-     * Messages may also contain a field |id| to help
-     * with debugging.
-     *
-     * Messages may also optionally contain a field |durationMs|, holding
-     * the duration of the function call in milliseconds.
+     * Messages also contains the following fields:
+     * - |id| an integer matching the deferred function to resolve (mandatory)
+     * - |fun| a string matching a function to call (optional)
+     * - |durationMs| holding the duration of the function call in milliseconds. (optional)
      *
      * @param {*} msg The message received from the worker.
      */
     worker.onmessage = msg => {
-      this.log("Received message from worker", msg.data);
-      let handler = this._queue.pop();
-      let deferred = handler.deferred;
       let data = msg.data;
-      if (data.id != handler.id) {
-        throw new Error(
-          "Internal error: expecting msg " +
-            handler.id +
-            ", " +
-            " got " +
-            data.id +
-            ": " +
-            JSON.stringify(msg.data)
-        );
-      }
+      let messageId = data.id;
+
+      this.log(`Received message ${messageId} from worker`);
+
       if ("timeStamps" in data) {
         this.workerTimeStamps = data.timeStamps;
       }
-      if ("ok" in data) {
-        // Pass the data to the listeners.
-        deferred.resolve(data);
-      } else if ("fail" in data) {
-        // We have received an error that was serialized by the
-        // worker.
-        deferred.reject(new WorkerError(data.fail));
+
+      // If fun is provided by the worker, we look into the functions
+      if ("fun" in data) {
+        if (data.fun in this._functions) {
+          Promise.resolve(this._functions[data.fun](...data.args)).then(
+            ok => {
+              if (ok instanceof BasePromiseWorker.Meta) {
+                if ("transfers" in ok.meta) {
+                  worker.postMessage(
+                    { ok: ok.data, id: messageId },
+                    ok.meta.transfers
+                  );
+                } else {
+                  worker.postMessage({ ok: ok.data, id: messageId });
+                }
+              } else {
+                worker.postMessage({ id: messageId, ok });
+              }
+            },
+            fail => {
+              worker.postMessage({ id: messageId, fail });
+            }
+          );
+        } else {
+          worker.postMessage({
+            id: messageId,
+            fail: `function ${data.fun} not found`,
+          });
+        }
+        return;
+      }
+
+      // If the message id matches one of the promise that waits, we resolve/reject with the data
+      if (this._deferredJobs.has(messageId)) {
+        let handler = this._deferredJobs.get(messageId);
+        let deferred = handler.deferred;
+
+        if ("ok" in data) {
+          // Pass the data to the listeners.
+          deferred.resolve(data);
+        } else if ("fail" in data) {
+          // We have received an error that was serialized by the
+          // worker.
+          deferred.reject(new WorkerError(data.fail));
+        }
+        return;
       }
+
+      // in any other case, this is an unexpected message from the worker.
+      throw new Error(
+        `Unexpected message id ${messageId}, data: ${JSON.stringify(data)} `
+      );
     };
     return worker;
   },
@@ -303,9 +343,9 @@ BasePromiseWorker.prototype = {
         });
       }
 
-      let id = ++this._id;
+      let id = this._generateDeferredJobId();
       let message = { fun, args, id };
-      this.log("Posting message", message);
+      this.log("Posting message", JSON.stringify(message));
       try {
         this._worker.postMessage(message, ...[transfers]);
       } catch (ex) {
@@ -320,15 +360,14 @@ BasePromiseWorker.prototype = {
       }
 
       let deferred = Promise.withResolvers();
-      this._queue.push({ deferred, closure, id });
-      this.log("Message posted");
+      this._deferredJobs.set(id, { deferred, closure, id });
 
       let reply;
       try {
         this.log("Expecting reply");
         reply = await deferred.promise;
       } catch (error) {
-        this.log("Got error", error);
+        this.log("Got error", JSON.stringify(error));
         reply = error;
 
         if (error instanceof WorkerError) {
@@ -389,7 +428,7 @@ BasePromiseWorker.prototype = {
   /**
    * Terminate the worker, if it has been created at all, and set things up to
    * be instantiated lazily again on the next `post()`.
-   * If there are pending Promises in the queue, we'll reject them and clear it.
+   * If there are pending Promises in the jobs, we'll reject them and clear it.
    */
   terminate() {
     if (!this.__worker) {
@@ -404,14 +443,12 @@ BasePromiseWorker.prototype = {
       this.log("Error whilst terminating ChromeWorker: " + ex.message);
     }
 
-    let error;
-    while (!this._queue.isEmpty()) {
-      if (!error) {
-        // We create this lazily, because error objects are not cheap.
-        error = new Error("Internal error: worker terminated");
-      }
-      let { deferred } = this._queue.pop();
-      deferred.reject(error);
+    if (this._deferredJobs.size) {
+      let error = new Error("Internal error: worker terminated");
+      this._deferredJobs.forEach(job => {
+        job.deferred.reject(error);
+      });
+      this._deferredJobs.clear();
     }
   },
 };
diff --git a/toolkit/components/promiseworker/tests/xpcshell/data/worker.js b/toolkit/components/promiseworker/tests/xpcshell/data/worker.js
index 30087bdc4a..94b6dd17ad 100644
--- a/toolkit/components/promiseworker/tests/xpcshell/data/worker.js
+++ b/toolkit/components/promiseworker/tests/xpcshell/data/worker.js
@@ -12,8 +12,9 @@ importScripts("resource://gre/modules/workers/require.js");
 var PromiseWorker = require("resource://gre/modules/workers/PromiseWorker.js");
 
 var worker = new PromiseWorker.AbstractWorker();
-worker.dispatch = function (method, args = []) {
-  return Agent[method](...args);
+
+worker.dispatch = async function (method, args = []) {
+  return await Agent[method](...args);
 };
 worker.postMessage = function (...args) {
   self.postMessage(...args);
@@ -34,6 +35,14 @@ var Agent = {
     return args;
   },
 
+  async bounceWithExtraCalls(...args) {
+    let result = await worker.callMainThread("echo", [
+      "Posting something unrelated",
+    ]);
+    args.push(result.ok);
+    return args;
+  },
+
   throwError(msg, ...args) {
     throw new Error(msg);
   },
diff --git a/toolkit/components/promiseworker/tests/xpcshell/test_Promise.js b/toolkit/components/promiseworker/tests/xpcshell/test_Promise.js
index f7581b664f..f9091a2b85 100644
--- a/toolkit/components/promiseworker/tests/xpcshell/test_Promise.js
+++ b/toolkit/components/promiseworker/tests/xpcshell/test_Promise.js
@@ -16,7 +16,22 @@ const { setTimeout } = ChromeUtils.importESModule(
 
 var WORKER_SOURCE_URI = "chrome://promiseworker/content/worker.js";
 do_load_manifest("data/chrome.manifest");
-var worker = new BasePromiseWorker(WORKER_SOURCE_URI);
+
+const UUID = crypto.randomUUID();
+
+const SOME_ARRAY = new Uint8Array(4);
+for (let i = 0; i < 4; ++i) {
+  SOME_ARRAY[i] = i;
+}
+
+async function echo(message) {
+  return new BasePromiseWorker.Meta([message, UUID, SOME_ARRAY.buffer], {
+    transfers: [SOME_ARRAY.buffer],
+  });
+}
+
+var worker = new BasePromiseWorker(WORKER_SOURCE_URI, {}, { echo });
+
 worker.log = function (...args) {
   info("Controller: " + args.join(" "));
 };
@@ -166,3 +181,34 @@ add_task(async function test_terminate() {
     "ChromeWorker instances should differ"
   );
 });
+
+function cloneArrayBuffer(original) {
+  const clone = new ArrayBuffer(original.byteLength);
+  const originalView = new Uint8Array(original);
+  const cloneView = new Uint8Array(clone);
+  cloneView.set(originalView);
+  return clone;
+}
+
+add_task(async function test_bidirectional() {
+  // Before we transfer the array, we clone it
+  const arrayCopy = cloneArrayBuffer(SOME_ARRAY.buffer);
+
+  let message = ["test_simple_args", Math.random()];
+
+  // Checking the array buffer size
+  Assert.equal(
+    SOME_ARRAY.buffer.byteLength,
+    4,
+    "The buffer is not detached yet"
+  );
+  let result = await worker.post("bounceWithExtraCalls", message);
+
+  // After the post call, the array was transferred and SOME_ARRAY should be empty
+  Assert.equal(SOME_ARRAY.buffer.byteLength, 0, "The buffer has been detached");
+
+  // The echo() function in the worker adds to the message a string, an uuid and has the transferred array
+  message.push(["Posting something unrelated", UUID, arrayCopy]);
+
+  Assert.deepEqual(result, message);
+});
diff --git a/toolkit/components/promiseworker/worker/PromiseWorker.template.worker.js b/toolkit/components/promiseworker/worker/PromiseWorker.template.worker.js
index c8203db48f..9ea26df9f5 100644
--- a/toolkit/components/promiseworker/worker/PromiseWorker.template.worker.js
+++ b/toolkit/components/promiseworker/worker/PromiseWorker.template.worker.js
@@ -102,19 +102,57 @@ function Meta(data, meta) {
  */
 function AbstractWorker(agent) {
   this._agent = agent;
+  this._deferredJobs = new Map();
+  this._deferredJobId = 0;
 }
+
 AbstractWorker.prototype = {
   // Default logger: discard all messages
   log() {},
 
+  _generateDeferredJobId() {
+    this._deferredJobId += 1;
+    return "WorkerToThread-" + this._deferredJobId;
+  },
+
+  /**
+   * Post and wait for an answer from the thread.
+   */
+  callMainThread(funcName, args) {
+    const messageId = this._generateDeferredJobId();
+
+    const message = {
+      id: messageId,
+      fun: funcName,
+      args,
+    };
+
+    return new Promise((resolve, reject) => {
+      this._deferredJobs.set(messageId, { resolve, reject });
+      this.postMessage(message);
+    });
+  },
+
   /**
    * Handle a message.
    */
   async handleMessage(msg) {
     let data = msg.data;
-    this.log("Received message", data);
     let id = data.id;
 
+    // if the id is found in _deferredJobs, we proceed with the message
+    if (this._deferredJobs.has(id)) {
+      const { resolve, reject } = this._deferredJobs.get(id);
+
+      if ("ok" in data) {
+        resolve(data);
+      } else if ("fail" in data) {
+        reject(data);
+      }
+      this._deferredJobs.delete(id);
+      return;
+    }
+
     let start;
     let options;
     if (data.args) {