Async Implementation

Question

I'm seeking code review comments for the following implementation.

jsFiddle

/**
* You need to run a number of async tasks, but you're being rate limited.
* Run as many concurrent tasks as you can, queueing the rest for when the
* tasks are complete.
*/

/**
* @param concurrency: number of concurrent tasks that can be run
*/
var ConcurrencyQueue = function (concurrency) {
    this.concurrency = concurrency;
    this.queue = [];
    this.count = 0;
};

/**
* add an asynchronous task to the queue. it will get executed immediately
* until the concurrent task count reaches this.concurrency. then, queue it until
* tasks are complete.
* @param task: function which takes a callback argument (done)
*/
ConcurrencyQueue.prototype.enqueue = function (task) {
    var self = this;
    self.count = self.count + 1;

    var done = function () {
        self.count = self.count - 1;
        if (self.queue.length) {
            self.queue.splice(0, 1)[0](done);
        }
    };

    if (self.count > self.concurrency) {
        self.queue.push(task);
    }

    if (self.count <= self.concurrency) {
        task(done);
    }
};

/**
 * tasks passed to concurrency queue are async and call the callback "done"
 * after they complete.
 */

function exampleTask (done) {
    console.log('started');
    setTimeout(function () {
        console.log('completed');
        done();
    }, Math.floor(Math.random() * 2000));
};

var queue = new ConcurrencyQueue(2);
queue.enqueue(exampleTask);
queue.enqueue(exampleTask);
queue.enqueue(exampleTask);
queue.enqueue(exampleTask);
queue.enqueue(exampleTask);
queue.enqueue(exampleTask);
queue.enqueue(exampleTask);

Answer 1

I have very little to say, maybe others will have more.

if (self.count > self.concurrency) {
    self.queue.push(task);
}

if (self.count <= self.concurrency) {
    task(done);
}

Why not if / else? At least it'll be clear it's one or the other.

this.queue.shift() is cleaner than this.queue.splice(0, 1)[0].

There's also a nitpick about self.count = self.count + 1; where += 1 could be used instead, but it really depends on the coding style.

About the callbacks, you could move to the newer Promise which is part of ECMAScript 2015.

ConcurrencyQueue.prototype.enqueue = function (task) {
    this.count += 1;

    if (this.count > this.concurrency) {
        this.queue.push(task);
    } else {
        this.execute(task);
    }
};

ConcurrencyQueue.prototype.execute = function(task) {
    var p = new Promise(task);

    /* `() => {` is a simpler notation for `function() {` in recent JS,
        but `this` refers to the encompassing `this` */
    var done = () => {
        this.count -= 1;
        if (this.queue.length) {
            this.execute(this.queue.shift());
        }
    };

    /* First argument is in case of success, second in case of errors */
    p.then(done, done);
}

jsFiddle

Answer 2

        self.queue.splice(0, 1)[0](done);

I would recommend to rewrite this code as follow

setTimeout(function(){
        self.queue.splice(0, 1)[0](done);
    },0);

There are few Cons of your approach:

1. If underlain code fails, it will show in stack trace function, that originally called done
2. It will hold all stack frames from task that was completed before(that was calling done), especially if tasks wasn't async actually. And it will hold both local stack variables and closures.

That's good approach to start new tasks in async manner.

I'd suggest there would be good to add try/catch, so if task failed to start, we can schedule next task.

Regards to @coyotte508 for promises, it's cool and first thing I thought about, but that's another step in learning curve, especially with ES6, @Jack can use Q, as it's ES5 compliant. And promise approach gives free async exception handling in p.then(done, done);