1 files changed, 233 insertions, 0 deletions
diff --git a/services/settings/RemoteSettingsWorker.sys.mjs b/services/settings/RemoteSettingsWorker.sys.mjs
new file mode 100644
index 0000000000..57bfb3f3d5
--- /dev/null
+++ b/services/settings/RemoteSettingsWorker.sys.mjs
@@ -0,0 +1,233 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this file,
+ * You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+/**
+ * Interface to a dedicated thread handling for Remote Settings heavy operations.
+ */
+import { XPCOMUtils } from "resource://gre/modules/XPCOMUtils.sys.mjs";
+
+import { setTimeout, clearTimeout } from "resource://gre/modules/Timer.sys.mjs";
+
+const lazy = {};
+
+XPCOMUtils.defineLazyPreferenceGetter(
+  lazy,
+  "gMaxIdleMilliseconds",
+  "services.settings.worker_idle_max_milliseconds",
+  30 * 1000 // Default of 30 seconds.
+);
+
+ChromeUtils.defineESModuleGetters(lazy, {
+  AsyncShutdown: "resource://gre/modules/AsyncShutdown.sys.mjs",
+  SharedUtils: "resource://services-settings/SharedUtils.sys.mjs",
+});
+
+// Note: we currently only ever construct one instance of Worker.
+// If it stops being a singleton, the AsyncShutdown code at the bottom
+// of this file, as well as these globals, will need adjusting.
+let gShutdown = false;
+let gShutdownResolver = null;
+
+class RemoteSettingsWorkerError extends Error {
+  constructor(message) {
+    super(message);
+    this.name = "RemoteSettingsWorkerError";
+  }
+}
+
+class Worker {
+  constructor(source) {
+    if (gShutdown) {
+      console.error("Can't create worker once shutdown has started");
+    }
+    this.source = source;
+    this.worker = null;
+
+    this.callbacks = new Map();
+    this.lastCallbackId = 0;
+    this.idleTimeoutId = null;
+  }
+
+  async _execute(method, args = [], options = {}) {
+    // Check if we're shutting down.
+    if (gShutdown && method != "prepareShutdown") {
+      throw new RemoteSettingsWorkerError("Remote Settings has shut down.");
+    }
+    // Don't instantiate the worker to shut it down.
+    if (method == "prepareShutdown" && !this.worker) {
+      return null;
+    }
+
+    const { mustComplete = false } = options;
+    // (Re)instantiate the worker if it was terminated.
+    if (!this.worker) {
+      this.worker = new ChromeWorker(this.source, { type: "module" });
+      this.worker.onmessage = this._onWorkerMessage.bind(this);
+      this.worker.onerror = error => {
+        // Worker crashed. Reject each pending callback.
+        for (const { reject } of this.callbacks.values()) {
+          reject(error);
+        }
+        this.callbacks.clear();
+        // And terminate it.
+        this.stop();
+      };
+    }
+    // New activity: reset the idle timer.
+    if (this.idleTimeoutId) {
+      clearTimeout(this.idleTimeoutId);
+    }
+    let identifier = method + "-";
+    // Include the collection details in the importJSONDump case.
+    if (identifier == "importJSONDump-") {
+      identifier += `${args[0]}-${args[1]}-`;
+    }
+    return new Promise((resolve, reject) => {
+      const callbackId = `${identifier}${++this.lastCallbackId}`;
+      this.callbacks.set(callbackId, { resolve, reject, mustComplete });
+      this.worker.postMessage({ callbackId, method, args });
+    });
+  }
+
+  _onWorkerMessage(event) {
+    const { callbackId, result, error } = event.data;
+    // If we're shutting down, we may have already rejected this operation
+    // and removed its callback from our map:
+    if (!this.callbacks.has(callbackId)) {
+      return;
+    }
+    const { resolve, reject } = this.callbacks.get(callbackId);
+    if (error) {
+      reject(new RemoteSettingsWorkerError(error));
+    } else {
+      resolve(result);
+    }
+    this.callbacks.delete(callbackId);
+
+    // Terminate the worker when it's unused for some time.
+    // But don't terminate it if an operation is pending.
+    if (!this.callbacks.size) {
+      if (gShutdown) {
+        this.stop();
+        if (gShutdownResolver) {
+          gShutdownResolver();
+        }
+      } else {
+        this.idleTimeoutId = setTimeout(() => {
+          this.stop();
+        }, lazy.gMaxIdleMilliseconds);
+      }
+    }
+  }
+
+  /**
+   * Called at shutdown to abort anything the worker is doing that isn't
+   * critical.
+   */
+  _abortCancelableRequests() {
+    // End all tasks that we can.
+    const callbackCopy = Array.from(this.callbacks.entries());
+    const error = new Error("Shutdown, aborting read-only worker requests.");
+    for (const [id, { reject, mustComplete }] of callbackCopy) {
+      if (!mustComplete) {
+        this.callbacks.delete(id);
+        reject(error);
+      }
+    }
+    // There might be nothing left now:
+    if (!this.callbacks.size) {
+      this.stop();
+      if (gShutdownResolver) {
+        gShutdownResolver();
+      }
+    }
+    // If there was something left, we'll stop as soon as we get messages from
+    // those tasks, too.
+    // Let's hurry them along a bit:
+    this._execute("prepareShutdown");
+  }
+
+  stop() {
+    this.worker.terminate();
+    this.worker = null;
+    this.idleTimeoutId = null;
+  }
+
+  async canonicalStringify(localRecords, remoteRecords, timestamp) {
+    return this._execute("canonicalStringify", [
+      localRecords,
+      remoteRecords,
+      timestamp,
+    ]);
+  }
+
+  async importJSONDump(bucket, collection) {
+    return this._execute("importJSONDump", [bucket, collection], {
+      mustComplete: true,
+    });
+  }
+
+  async checkFileHash(filepath, size, hash) {
+    return this._execute("checkFileHash", [filepath, size, hash]);
+  }
+
+  async checkContentHash(buffer, size, hash) {
+    // The implementation does little work on the current thread, so run the
+    // task on the current thread instead of the worker thread.
+    return lazy.SharedUtils.checkContentHash(buffer, size, hash);
+  }
+}
+
+// Now, first add a shutdown blocker. If that fails, we must have
+// shut down already.
+// We're doing this here rather than in the Worker constructor because in
+// principle having just 1 shutdown blocker for the entire file should be
+// fine. If we ever start creating more than one Worker instance, this
+// code will need adjusting to deal with that.
+try {
+  lazy.AsyncShutdown.profileBeforeChange.addBlocker(
+    "Remote Settings profile-before-change",
+    async () => {
+      // First, indicate we've shut down.
+      gShutdown = true;
+      // Then, if we have no worker or no callbacks, we're done.
+      if (
+        !RemoteSettingsWorker.worker ||
+        !RemoteSettingsWorker.callbacks.size
+      ) {
+        return null;
+      }
+      // Otherwise, there's something left to do. Set up a promise:
+      let finishedPromise = new Promise(resolve => {
+        gShutdownResolver = resolve;
+      });
+
+      // Try to cancel most of the work:
+      RemoteSettingsWorker._abortCancelableRequests();
+
+      // Return a promise that the worker will resolve.
+      return finishedPromise;
+    },
+    {
+      fetchState() {
+        const remainingCallbacks = RemoteSettingsWorker.callbacks;
+        const details = Array.from(remainingCallbacks.keys()).join(", ");
+        return `Remaining: ${remainingCallbacks.size} callbacks (${details}).`;
+      },
+    }
+  );
+} catch (ex) {
+  console.error(
+    "Couldn't add shutdown blocker, assuming shutdown has started."
+  );
+  console.error(ex);
+  // If AsyncShutdown throws, `profileBeforeChange` has already fired. Ignore it
+  // and mark shutdown. Constructing the worker will report an error and do
+  // nothing.
+  gShutdown = true;
+}
+
+export var RemoteSettingsWorker = new Worker(
+  "resource://services-settings/RemoteSettings.worker.mjs"
+);