diff --git a/resource/concurrentCaller.js b/resource/concurrentCaller.js index 89673d2bc..0bfe6e6aa 100644 --- a/resource/concurrentCaller.js +++ b/resource/concurrentCaller.js @@ -34,43 +34,61 @@ Components.utils.import("resource://zotero/bluebird.js"); * * Example: * - * var caller = new ConcurrentCaller(2); - * caller.stopOnError = true; - * caller.fcall([foo, bar, baz, qux); + * var caller = new ConcurrentCaller({ + * numConcurrent: 2, + * stopOnError: true + * }); + * yield caller.start([foo, bar, baz, qux); * * In this example, foo and bar would run immediately, and baz and qux would * be queued for later. When foo or bar finished, baz would be run, followed * by qux when another slot opened. * - * @param {Integer} numConcurrent The number of concurrent functions to run. + * Additional functions can be added at any time with another call to start(). The promises for + * all open start() calls will be resolved when all requests are finished. + * + * @param {Object} options + * @param {Integer} options.numConcurrent - The number of concurrent functions to run. + * @param {String} [options.id] - Identifier to use in debug output + * @param {Boolean} [options.stopOnError] + * @param {Function} [options.onError] + * @param {Integer} [options.interval] - Interval between the end of one function run and the + * beginning of another, in milliseconds + * @param {Function} [options.logger] */ -ConcurrentCaller = function (numConcurrent) { - if (typeof numConcurrent == 'undefined') { - throw new Error("numConcurrent not provided"); +ConcurrentCaller = function (options = {}) { + if (typeof options == 'number') { + this._log("ConcurrentCaller now takes an object rather than a number"); + options = { + numConcurrent: options + }; } - this.stopOnError = false; - this.onError = null; + if (!options.numConcurrent) throw new Error("numConcurrent must be provided"); - this._numConcurrent = numConcurrent; + this.stopOnError = options.stopOnError || false; + this.onError = options.onError || null; + + this._id = options.id; + this._numConcurrent = options.numConcurrent; this._numRunning = 0; this._queue = []; - this._logger = null; - this._interval = 0; + this._logger = options.logger || null; + this._interval = options.interval || 0; + this._pausing = false; this._pauseUntil = 0; + this._deferred = null; }; -/** - * Set the interval between the end of one function run and the beginning - * of another, in milliseconds - */ ConcurrentCaller.prototype.setInterval = function (ms) { + this._log("setInterval() is deprecated -- pass .interval to constructor"); this._interval = ms; }; ConcurrentCaller.prototype.setLogger = function (func) { + this._log("setLogger() is deprecated -- pass .logger to constructor"); this._logger = func; }; @@ -84,35 +102,81 @@ ConcurrentCaller.prototype.pause = function (ms) { /** - * @param {Function[]|Function} func One or more functions to run + * Add a task to the queue without starting it + * + * @param {Function|Function[]} - One or more functions to run + * @return {Promise[]} - An array of promises for passed functions, resolved once the queue is + * empty */ -ConcurrentCaller.prototype.fcall = function (func) { +ConcurrentCaller.prototype.add = function (func) { if (Array.isArray(func)) { - var promises = []; + let promises = []; for (let i = 0; i < func.length; i++) { - promises.push(this.fcall(func[i])); + promises.push(this.start(func[i])); } - return Promise.settle(promises); + return this._deferred.promise.return(promises); } - // If we're at the maximum number of concurrent functions, - // queue this function for later - if (this._numRunning == this._numConcurrent) { + if (!this._deferred || !this._deferred.promise.isPending()) { + this._deferred = Promise.defer(); + } + + var deferred = Promise.defer(); + this._queue.push({ + func: Promise.method(func), + deferred: deferred + }); + return this._deferred.promise.return(deferred.promise); +} + + +/** + * @param {Function|Function[]} - One or more functions to run + * @return {Promise[]} - An array of promises for passed functions, resolved once the queue is + * empty + */ +ConcurrentCaller.prototype.start = function (func) { + var promise = this.add(func); + var run = this._processNext(); + if (!run) { this._log("Already at " + this._numConcurrent + " -- queueing for later"); - var deferred = Promise.defer(); - this._queue.push({ - func: Promise.method(func), - deferred: deferred - }); - return deferred.promise; } - - this._numRunning++; - - // Otherwise run it now - this._log("Running function (" + this._numRunning + "/" + this._numConcurrent + ")"); - - return this._onFunctionDone(Promise.try(func)); + return promise; +} + + +/** + * Start processing if not already running and wait for all tasks to complete + * + * @return {Promise[]} - An array of promises for all currently queued tasks + */ +ConcurrentCaller.prototype.runAll = function () { + // If nothing queued, return immediately + if (!this._deferred) { + return Promise.resolve([]); + } + var promises = this._queue.map(x => x.deferred.promise); + do { + var run = this._processNext(); + } + while (run); + return this._deferred.promise.return(promises); +} + + +ConcurrentCaller.prototype.fcall = function (func) { + this._log("fcall() is deprecated -- use start()"); + return this.start(func); +} + + +/** + * Wait for all running tasks to complete + * + * @return {Promise} + */ +ConcurrentCaller.prototype.wait = function () { + return this._deferred ? this._deferred.promise : Promise.resolve(); } @@ -122,62 +186,85 @@ ConcurrentCaller.prototype.stop = function () { }; -ConcurrentCaller.prototype._onFunctionDone = function (promise) { - var self = this; +ConcurrentCaller.prototype._processNext = function () { + if (this._numRunning >= this._numConcurrent) { + return false; + } - return promise.then(function (result) { - self._numRunning--; + // If there's a function to call and we're under the concurrent limit, run it now + var f = this._queue.shift(); + if (!f) { + if (this._numRunning == 0 && !this._pausing) { + this._log("All tasks are done"); + this._deferred.resolve(); + } + else { + this._log("Nothing left to run -- waiting for running tasks to complete"); + } + return false; + } + + this._log("Running function (" + + this._numRunning + "/" + this._numConcurrent + " running, " + + this._queue.length + " queued)"); + + this._numRunning++; + f.func().bind(this).then(function (value) { + this._numRunning--; - self._log("Done with function (" - + self._numRunning + "/" + self._numConcurrent + " running, " - + self._queue.length + " queued)"); + this._log("Done with function (" + + this._numRunning + "/" + this._numConcurrent + " running, " + + this._queue.length + " queued)"); - return result; + this._waitForPause().bind(this).then(function () { + this._processNext(); + }); + + f.deferred.resolve(value); }) .catch(function (e) { - self._numRunning--; + this._numRunning--; - self._log("Error in function (" + self._numRunning + "/" + self._numConcurrent + ", " - + self._queue.length + " in queue)"); + this._log("Error in function (" + this._numRunning + "/" + this._numConcurrent + ", " + + this._queue.length + " in queue)"); - if (self.onError) { - self.onError(e); + if (this.onError) { + this.onError(e); } - if (self.stopOnError && self._queue.length) { - self._log("Stopping on error: " + e); - self._queue = []; + if (this.stopOnError && this._queue.length) { + this._log("Stopping on error: " + e); + this._queue = []; } - throw e; - }) - .finally(function () { - // If there's a function to call and we're under the concurrent limit, run it now - var f = self._queue.shift(); - if (f && self._numRunning < self._numConcurrent) { - // Wait until the specified interval has elapsed or the current - // pause (if there is one) is over, whichever is longer - let interval = self._interval; - let now = Date.now(); - if (self._pauseUntil > now && (self._pauseUntil - now > interval)) { - interval = self._pauseUntil - now; - } - Promise.delay(interval) - .then(function () { - self._log("Running new function (" - + self._numRunning + "/" + self._numConcurrent + " running, " - + self._queue.length + " queued)"); - - self._numRunning++; - f.deferred.resolve(self._onFunctionDone(f.func())); - }); - } + this._waitForPause().bind(this).then(function () { + this._processNext(); + }); + + f.deferred.reject(e); }); -}; + return true; +} + + +/** + * Wait until the specified interval has elapsed or the current pause (if there is one) is over, + * whichever is longer + */ +ConcurrentCaller.prototype._waitForPause = Promise.coroutine(function* () { + let interval = this._interval; + let now = Date.now(); + if (this._pauseUntil > now && (this._pauseUntil - now > interval)) { + interval = this._pauseUntil - now; + } + this._pausing = true; + yield Promise.delay(interval); + this._pausing = false; +}); ConcurrentCaller.prototype._log = function (msg) { if (this._logger) { - this._logger("[ConcurrentCaller] " + msg); + this._logger("[ConcurrentCaller] " + (this._id ? `[${this._id}] ` : "") + msg); } }; diff --git a/test/tests/concurrentCallerTest.js b/test/tests/concurrentCallerTest.js new file mode 100644 index 000000000..d237bdf00 --- /dev/null +++ b/test/tests/concurrentCallerTest.js @@ -0,0 +1,378 @@ +"use strict"; + +describe("ConcurrentCaller", function () { + Components.utils.import("resource://zotero/concurrentCaller.js"); + var logger = null; + // Uncomment to get debug output + //logger = Zotero.debug; + + describe("#start()", function () { + it("should run functions as slots open and wait for them to complete", function* () { + var numConcurrent = 2; + var running = 0; + var finished = 0; + var failed = false; + + var ids = ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']; + var funcs = ids.map(function (id) { + return Zotero.Promise.coroutine(function* () { + if (logger) { + Zotero.debug("Running " + id); + } + running++; + if (running > numConcurrent) { + failed = true; + throw new Error("Too many concurrent tasks"); + } + var min = 10; + var max = 25; + yield Zotero.Promise.delay( + Math.floor(Math.random() * (max - min + 1)) + min + ); + if (running > numConcurrent) { + failed = true; + throw new Error("Too many concurrent tasks"); + } + running--; + finished++; + if (logger) { + Zotero.debug("Finished " + id); + } + return id; + }); + }) + + var caller = new ConcurrentCaller({ + numConcurrent, + logger + }); + var results = yield caller.start(funcs); + + assert.equal(results.length, ids.length); + assert.equal(running, 0); + assert.equal(finished, ids.length); + }) + + it("should add functions to existing queue and resolve when all are complete (waiting for earlier)", function* () { + var numConcurrent = 2; + var running = 0; + var finished = 0; + var failed = false; + + var ids1 = ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm']; + var ids2 = ['n', 'o', 'p', 'q']; + var makeFunc = function (id) { + return Zotero.Promise.coroutine(function* () { + if (logger) { + Zotero.debug("Running " + id); + } + running++; + if (running > numConcurrent) { + failed = true; + throw new Error("Too many concurrent tasks"); + } + var min = 10; + var max = 25; + yield Zotero.Promise.delay( + Math.floor(Math.random() * (max - min + 1)) + min + ); + if (running > numConcurrent) { + failed = true; + throw new Error("Too many concurrent tasks"); + } + running--; + finished++; + if (logger) { + Zotero.debug("Finished " + id); + } + return id; + }); + }; + var funcs1 = ids1.map(makeFunc) + var funcs2 = ids2.map(makeFunc) + + var caller = new ConcurrentCaller({ + numConcurrent, + logger + }); + var promise1 = caller.start(funcs1); + yield Zotero.Promise.delay(10); + var promise2 = caller.start(funcs2); + + var results1 = yield promise1; + + assert.isTrue(promise2.isFulfilled()); + assert.equal(running, 0); + assert.equal(finished, ids1.length + ids2.length); + assert.equal(results1.length, ids1.length); + assert.sameMembers(results1.map(p => p.value()), ids1); + }) + + it("should add functions to existing queue and resolve when all are complete (waiting for later)", function* () { + var numConcurrent = 2; + var running = 0; + var finished = 0; + var failed = false; + + var ids1 = ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm']; + var ids2 = ['n', 'o', 'p', 'q']; + var makeFunc = function (id) { + return Zotero.Promise.coroutine(function* () { + if (logger) { + Zotero.debug("Running " + id); + } + running++; + if (running > numConcurrent) { + failed = true; + throw new Error("Too many concurrent tasks"); + } + var min = 10; + var max = 25; + yield Zotero.Promise.delay( + Math.floor(Math.random() * (max - min + 1)) + min + ); + if (running > numConcurrent) { + failed = true; + throw new Error("Too many concurrent tasks"); + } + running--; + finished++; + if (logger) { + Zotero.debug("Finished " + id); + } + return id; + }); + }; + var funcs1 = ids1.map(makeFunc) + var funcs2 = ids2.map(makeFunc) + + var caller = new ConcurrentCaller({ + numConcurrent, + logger + }); + var promise1 = caller.start(funcs1); + yield Zotero.Promise.delay(10); + var promise2 = caller.start(funcs2); + + var results2 = yield promise2; + + assert.isTrue(promise1.isFulfilled()); + assert.equal(running, 0); + assert.equal(finished, ids1.length + ids2.length); + assert.equal(results2.length, ids2.length); + }) + + it("should return a rejected promise if a single passed function fails", function* () { + var numConcurrent = 2; + + var caller = new ConcurrentCaller({ + numConcurrent, + logger + }); + var e = yield getPromiseError(caller.start(function () { + throw new Error("Fail"); + })); + assert.ok(e); + }) + + it("should stop on error if stopOnError is set", function* () { + var numConcurrent = 2; + var running = 0; + var finished = 0; + var failed = false; + + var ids1 = ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm']; + var ids2 = ['n', 'o', 'p', 'q']; + var makeFunc = function (id) { + return Zotero.Promise.coroutine(function* () { + if (logger) { + Zotero.debug("Running " + id); + } + running++ + if (running > numConcurrent) { + failed = true; + throw new Error("Too many concurrent tasks"); + } + var min = 10; + var max = 25; + yield Zotero.Promise.delay( + Math.floor(Math.random() * (max - min + 1)) + min + ); + if (id == 'g') { + running--; + finished++; + Zotero.debug("Throwing " + id); + // This causes an erroneous "possibly unhandled rejection" message in + // Bluebird 2.10.2 that I can't seem to get rid of (and the rejection + // is later handled), so pass " -- ignore" to tell Bluebird to ignore it + throw new Error("Fail -- ignore"); + } + if (running > numConcurrent) { + failed = true; + throw new Error("Too many concurrent tasks"); + } + running--; + finished++; + if (logger) { + Zotero.debug("Finished " + id); + } + return id; + }); + }; + var funcs1 = ids1.map(makeFunc) + var funcs2 = ids2.map(makeFunc) + + var caller = new ConcurrentCaller({ + numConcurrent, + stopOnError: true, + logger + }); + var promise1 = caller.start(funcs1); + var promise2 = caller.start(funcs2); + + var results1 = yield promise1; + + assert.isTrue(promise2.isFulfilled()); + assert.equal(running, 0); + assert.isBelow(finished, ids1.length); + assert.equal(results1.length, ids1.length); + assert.equal(promise2.value().length, ids2.length); + // 'a' should be fulfilled + assert.isTrue(results1[0].isFulfilled()); + // 'g' should be rejected + assert.isTrue(results1[6].isRejected()); + // 'm' should be pending + assert.isTrue(results1[12].isPending()); + // All promises in second batch should be pending + assert.isTrue(promise2.value().every(p => p.isPending())); + }) + + + it("should not stop on error if stopOnError isn't set", function* () { + var numConcurrent = 2; + var running = 0; + var finished = 0; + var failed = false; + + var ids1 = ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm']; + var ids2 = ['n', 'o', 'p', 'q']; + var makeFunc = function (id) { + return Zotero.Promise.coroutine(function* () { + if (logger) { + Zotero.debug("Running " + id); + } + running++ + if (running > numConcurrent) { + failed = true; + throw new Error("Too many concurrent tasks"); + } + var min = 10; + var max = 25; + yield Zotero.Promise.delay( + Math.floor(Math.random() * (max - min + 1)) + min + ); + if (id == 'g') { + running--; + finished++; + Zotero.debug("Throwing " + id); + // This causes an erroneous "possibly unhandled rejection" message in + // Bluebird 2.10.2 that I can't seem to get rid of (and the rejection + // is later handled), so pass " -- ignore" to tell Bluebird to ignore it + throw new Error("Fail -- ignore"); + } + if (running > numConcurrent) { + failed = true; + throw new Error("Too many concurrent tasks"); + } + running--; + finished++; + if (logger) { + Zotero.debug("Finished " + id); + } + return id; + }); + }; + var funcs1 = ids1.map(makeFunc) + var funcs2 = ids2.map(makeFunc) + + var caller = new ConcurrentCaller({ + numConcurrent, + logger + }); + var promise1 = caller.start(funcs1); + var promise2 = caller.start(funcs2); + + var results1 = yield promise1; + + assert.isTrue(promise2.isFulfilled()); + assert.equal(running, 0); + assert.equal(finished, ids1.length + ids2.length); + assert.equal(results1.length, ids1.length); + assert.equal(promise2.value().length, ids2.length); + // 'a' should be fulfilled + assert.isTrue(results1[0].isFulfilled()); + // 'g' should be rejected + assert.isTrue(results1[6].isRejected()); + // 'm' should be fulfilled + assert.isTrue(results1[12].isFulfilled()); + // All promises in second batch should be fulfilled + assert.isTrue(promise2.value().every(p => p.isFulfilled())); + }) + }) + + describe("#wait()", function () { + it("should return when all tasks are done", function* () { + var numConcurrent = 2; + var running = 0; + var finished = 0; + var failed = false; + + var ids1 = ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm']; + var ids2 = ['n', 'o', 'p', 'q']; + var makeFunc = function (id) { + return Zotero.Promise.coroutine(function* () { + if (logger) { + Zotero.debug("Running " + id); + } + running++; + if (running > numConcurrent) { + failed = true; + throw new Error("Too many concurrent tasks"); + } + var min = 10; + var max = 25; + yield Zotero.Promise.delay( + Math.floor(Math.random() * (max - min + 1)) + min + ); + if (running > numConcurrent) { + failed = true; + throw new Error("Too many concurrent tasks"); + } + running--; + finished++; + if (logger) { + Zotero.debug("Finished " + id); + } + return id; + }); + }; + var funcs1 = ids1.map(makeFunc) + var funcs2 = ids2.map(makeFunc) + + var caller = new ConcurrentCaller({ + numConcurrent, + logger + }); + var promise1 = caller.start(funcs1); + yield Zotero.Promise.delay(10); + var promise2 = caller.start(funcs2); + + yield caller.wait(); + + assert.isTrue(promise1.isFulfilled()); + assert.isTrue(promise2.isFulfilled()); + assert.equal(running, 0); + assert.equal(finished, ids1.length + ids2.length); + }) + }) +})