cwait
provides a queue handler (TaskQueue
) and a wrapper (Task
) for promises,
to limit how many are being resolved simultaneously. It can wrap any ES6-compatible promises.
This allows for example limiting simultaneous downloads with minor changes to existing code.
Just wrap your existing "download finished" promise and use it as before.
This is a tiny library with a single dependency, usable both in browsers and Node.js.
Create a new TaskQueue
passing it whatever Promise
constructor you're using (ES6, Bluebird, some other shim...)
and the maximum number of promise-returning functions to run concurrently.
Then just call queue.wrap(<function>)
instead of <function>
to limit simultaneous execution.
Simple Node.js example:
import * as Promise from 'bluebird';
import {TaskQueue} from 'cwait';
/** Queue allowing 3 concurrent function calls. */
var queue = new TaskQueue(Promise, 3);
Promise.map(list, download); // Download all listed files simultaneously.
Promise.map(list, queue.wrap(download)); // Download 3 files at a time.
See test/test.ts
for some runnable code or run it like this:
git clone https://github.com/charto/cwait.git
cd cwait
npm install
npm test
Recursive loops that run in parallel require special care.
Nested concurrency-limited calls (that are not tail-recursive) must be wrapped in queue.unblock()
.
Here's a simple example that fails:
var queue = new (require('cwait').TaskQueue)(Promise, 3);
var rec = queue.wrap(function(n) {
console.log(n);
return(n && Promise.resolve(rec(n - 1)));
});
rec(10);
It only prints numbers 10, 9 and 8. More calls don't get scheduled because there are already 3 promises pending. For example Node.js exits immediately afterwards because the program is not blocked waiting for any system calls.
Passing a promise to queue.unblock(promise)
tells queue
that
the current function will wait for promise
to resolve before continuing.
One additional concurrent function is then allowed until the promise resolves.
Be careful not to call queue.unblock()
more than once (concurrently) from inside a wrapped function!
Otherwise the queue may permit more simultaneous tasks than the intended limit.
Here is a corrected example:
var queue = new (require('cwait').TaskQueue)(Promise, 3);
var rec = queue.wrap(function(n) {
console.log(n);
return(n && queue.unblock(Promise.resolve(rec(n - 1))));
});
rec(10);
The following code recursively calculates the 10th Fibonacci number (55) running 3 recursive steps in parallel, each with an artificial 10-millisecond delay.
At the end, it prints the result (55) and the number of concurrent calls (3).
var queue = new (require('cwait').TaskQueue)(Promise, 3);
var maxRunning = 0;
var running = 0;
var delay = 10;
var fib = queue.wrap(function(n) {
// "Calculation" begins. Track maximum concurrent executions.
if(++running > maxRunning) maxRunning = running;
return(new Promise(function(resolve, reject) {
setTimeout(function() {
// "Calculation" ends.
--running;
// Each Fibonacci number is the sum of the previous two, except
// the first ones are 0, 1 (starting from the 0th number).
// Calculate them in parallel and unblock the queue until ready.
resolve(n < 2 ? n :
queue.unblock(Promise.all([
fib(n - 1),
fib(n - 2)
])).then(function(r) {
// Sum results from parallel recursion.
return(r[0] + r[1]);
})
);
}, delay);
}));
});
fib(10).then(function(x) {
console.log('Result: ' + x);
console.log('Concurrency: ' + maxRunning);
});
Docs generated using docts
Class
Task
Task wraps a promise, delaying it until some resource gets less busy.
Source code:<>
Methods:
new( ) ⇒
Task<PromiseType>
<>
▪ func() => PromiseType
▪ PromisePromisyClass<PromiseType>
.delay( ) ⇒PromiseType
<>
Wrap task result in a new promise so it can be resolved later.
.resume( ) ⇒PromiseType
<>
Start the task and call onFinish when done.
▪ onFinish() => void
Class
TaskQueue
Source code:
<>
Methods:
new( ) ⇒
TaskQueue<PromiseType>
<>
▪ PromisePromisyClass<PromiseType>
▪ concurrencynumber
.add( ) ⇒PromiseType
<>
Add a new task to the queue.
It will start when the number of other concurrent tasks is low enough.
▪ func() => PromiseType
.unblock( ) ⇒PromiseType
<>
Consider current function idle until promise resolves.
Useful for making recursive calls.
▪ promisePromiseType
.wrap( ) ⇒(...args: any[]) => PromiseType
<>
Wrap a function returning a promise, so that before running
it waits until concurrent invocations are below this queue's limit.
▪ func(...args: any[]) => PromiseType
▫ thisObject?any
Properties:
.concurrency
number
Number of promises allowed to resolve concurrently.
Copyright (c) 2015-2017 BusFaster Ltd