cwait

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.

Usage

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

Recursion

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

Advanced example with recursion

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

API

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}
 ▪ Promise ^{PromisyClass<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>} <>
 ▪ Promise ^{PromisyClass<PromiseType>}
 ▪ concurrency ^number
.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.
 ▪ promise ^PromiseType
.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.

License

The MIT License

Copyright (c) 2015-2017 BusFaster Ltd