ConcurrentCaller changes

- Different constructor parameters
- id property for logging
- fcall() -> start()
- add() to enqueue without starting
- runAll() to run down queue and return promises for all current tasks
- wait() to wait for all running tasks to finish

resource/concurrentCaller.js

@@ -34,43 +34,61 @@ Components.utils.import("resource://zotero/bluebird.js");
  *
  * Example:
  *
- *   var caller = new ConcurrentCaller(2);
+ *   var caller = new ConcurrentCaller({
- *   caller.stopOnError = true;
+ *       numConcurrent: 2,
- *   caller.fcall([foo, bar, baz, qux);
+ *       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) {
+ConcurrentCaller = function (options = {}) {
-	if (typeof numConcurrent == 'undefined') {
+	if (typeof options == 'number') {
-		throw new Error("numConcurrent not provided");
+		this._log("ConcurrentCaller now takes an object rather than a number");
+		options = {
+			numConcurrent: options
+		};
 	}
 	
-	this.stopOnError = false;
+	if (!options.numConcurrent) throw new Error("numConcurrent must be provided");
-	this.onError = null;
 	
-	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._logger = options.logger || null;
-	this._interval = 0;
+	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,
+	if (!this._deferred || !this._deferred.promise.isPending()) {
-	// queue this function for later
+		this._deferred = Promise.defer();
-	if (this._numRunning == this._numConcurrent) {
+	}
+	
+	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;
 	}
-	
+	return promise;
-	this._numRunning++;
+}
-	
+
-	// Otherwise run it now
+
-	this._log("Running function (" + this._numRunning + "/" + this._numConcurrent + ")");
+/**
-	
+ * Start processing if not already running and wait for all tasks to complete
-	return this._onFunctionDone(Promise.try(func));
+ *
+ * @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) {
+ConcurrentCaller.prototype._processNext = function () {
-	var self = this;
+	if (this._numRunning >= this._numConcurrent) {
+		return false;
+	}
 	
-	return promise.then(function (result) {
+	// If there's a function to call and we're under the concurrent limit, run it now
-		self._numRunning--;
+	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 ("
+		this._log("Done with function ("
-			+ self._numRunning + "/" + self._numConcurrent + " running, "
+			+ this._numRunning + "/" + this._numConcurrent + " running, "
-			+ self._queue.length + " queued)");
+			+ 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 + ", "
+		this._log("Error in function (" + this._numRunning + "/" + this._numConcurrent + ", "
-			+ self._queue.length + " in queue)");
+			+ this._queue.length + " in queue)");
 		
-		if (self.onError) {
+		if (this.onError) {
-			self.onError(e);
+			this.onError(e);
 		}
 		
-		if (self.stopOnError && self._queue.length) {
+		if (this.stopOnError && this._queue.length) {
-			self._log("Stopping on error: " + e);
+			this._log("Stopping on error: " + e);
-			self._queue = [];
+			this._queue = [];
 		}
 		
-		throw e;
+		this._waitForPause().bind(this).then(function () {
-	})
+			this._processNext();
-	.finally(function () {
+		});
-		// If there's a function to call and we're under the concurrent limit, run it now
+		
-		var f = self._queue.shift();
+		f.deferred.reject(e);
-		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()));
-			});
-		}
 	});
-};
+	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);
 	}
 };

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);
+		})
+	})
+})