Source code
Revision control
Copy as Markdown
Other Tools
/* 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,
/**
* Interface to a dedicated thread handling for Remote Settings heavy operations.
*/
const lazy = {};
XPCOMUtils.defineLazyPreferenceGetter(
lazy,
"gMaxIdleMilliseconds",
"services.settings.worker_idle_max_milliseconds",
30 * 1000 // Default of 30 seconds.
);
ChromeUtils.defineESModuleGetters(lazy, {
});
// 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(
);