node package manager
Stop writing boring code. Discover, share, and reuse within your team. Create a free org »

qrate

qrate

Build Status npm version

Introduction

The queue library based on the async.queue utility but modified to allow a queue's throughput to be controlled in terms of:

  • concurrency - the maximum number of workers running at any point in time
  • rateLimit - the maximum number of workers started per second

The default behaviour is a concurrency of 1 (one worker at a time) and a rateLimit of null (no rate limiting).

The qrate library can be used as a drop-in replacement for the async.queue function.

Installation

Install with

npm install qrate

or to import it into your Node.js project:

npm install --save qrate

Usage

A queue is created by calling qrate passing in the the worker function you want to operate on each item in the queue. The returned q can then be used to push data into the queue.

 
// require qrate library 
const qrate = require('qrate');
 
// mark the start time of this script 
const start = new Date().getTime();
 
// worker function that calls back after 100ms 
const worker = function(data, done) {
 
  // your worker code goes here 
  // 'data' contains the queue to work on 
  // call 'done' when finished. 
 
 
  // output a message including a timestamp 
  console.log('Processing', data, '@', new Date().getTime() - start, 'ms');
 
  // call the 'done' function after 100ms 
  setTimeout(done, 100);
};
 
// create a queue with default properties (concurrency = 1, rateLimit = null) 
// using our 'worker' function to process each item in the queue 
const q = qrate(worker);
 
// add ten things to the queue 
for (var i = 0; i < 10; i++) {
  q.push({ i: i });
}

The queue has the default concurrency of 1, so worker starts after its predecessor finishes:

Processing { i: 0 } @ 21 ms
Processing { i: 1 } @ 129 ms
Processing { i: 2 } @ 233 ms
Processing { i: 3 } @ 338 ms
Processing { i: 4 } @ 441 ms
Processing { i: 5 } @ 545 ms
Processing { i: 6 } @ 650 ms
Processing { i: 7 } @ 751 ms
Processing { i: 8 } @ 852 ms
Processing { i: 9 } @ 958 ms

We can increase the number of workers running in parallel by passing a concurrency value as a second parameter:

 
// create a queue where up to three workers run at any time 
var q = qrate(worker, 3);

which speeds things up significantly:

Processing { i: 0 } @ 27 ms
Processing { i: 1 } @ 33 ms
Processing { i: 2 } @ 35 ms
Processing { i: 3 } @ 134 ms
Processing { i: 4 } @ 135 ms
Processing { i: 5 } @ 135 ms
Processing { i: 6 } @ 235 ms
Processing { i: 7 } @ 235 ms
Processing { i: 8 } @ 236 ms
Processing { i: 9 } @ 340 ms

So far we have not done anything that a normal async.queue could do. This is where the third parameter comes in.

Rate limiting the queue

If you want to limit the rate of throughput of the queue (e.g. 5 jobs per second), then you can pass a third rateLimit parameter to qrate. The rateLimit indicates the maximum number workers per second you want the queue to start:

  • rateLimit = 1 - one per second
  • rateLimit = 5 - five per second
  • rateLimit = 0.5 - one every two seconds
  • rateLimit = null - as fast as possible (default)
// concurrency 1, rateLimit 2 workers per second 
var q = qrate(worker, 1, 2);

which produces the output:

Processing { i: 0 } @ 16 ms
Processing { i: 1 } @ 126 ms
Processing { i: 2 } @ 1007 ms
Processing { i: 3 } @ 1111 ms
Processing { i: 4 } @ 2013 ms
Processing { i: 5 } @ 2118 ms
Processing { i: 6 } @ 3018 ms
Processing { i: 7 } @ 3124 ms
Processing { i: 8 } @ 4025 ms
Processing { i: 9 } @ 4127 ms

Notice how in the early part of each second, two workers are executed in turn, then the queue waits until the next second boundary before resuming work again.

Rate-limiting is useful if you want to ensure that the number of API calls your code generates stays below the API provider's quota, e.g. five API calls per second.

Killing the queue

A rate-limited qrate queue sets up a timer to handle the throttling of a rate-limited queue. The queue can be cleaned up by calling the q.kill() function.

If your application is working through a single list of work, the you can provide a q.drain function that is called when a queue is emptied and call q.kill in that function:

q.drain = funcion() {
  console.log('the queue is empty');
  q.kill();
};

or, simply tie the drain and kill functions together:

q.drain = q.kill;

In other applications, you may wish to keep the queue alive and periodically feed it with fresh work.