promise-pool
Concurrent control of async function invocation with a pool-like abstraction. Implicitly applies
backpressure to the imperative task producer.
Requires a version of Node that supports async
/await
.
import PromisePool from '@mixmaxhq/promise-pool';
async function sample() {
const pool = new PromisePool({numConcurrent: 10});
for (let i = 0; i < 1000000; ++i) {
await pool.start(async (i) => {
await sendResult(await getResult(i));
}, i);
}
const errors = await pool.flush();
if (errors.length) {
console.log('done with errors', errors);
} else {
console.log('done');
}
}
If this strikes you as similar to batch, that's because it is. It differs, however, in that it
has built-in backpressure, to simplify writing concurrent code that avoids loading everything into
memory. This module is a rewrite of the synchronize-pool module, but instead of using
synchronize, it uses async/await.
Install
We're hoping to use the promise-pool
package name, but it's currently occupied.
$ npm install @mixmaxhq/promise-pool
or
$ npm i @mixmaxhq/promise-pool
Changelog
-
2.0.0 Add maxPending
option to avoid problematic usage (see new Troubleshooting section)
-
1.1.1 Move ava
and ava-spec
to devDependencies
.
-
1.1.0 Adds transpilation so it can be used in Node 6 (and prior) environments.
-
1.0.0 Initial release.
Troubleshooting
cannot queue function in pool
If you're getting this error, then you're calling start
too many times
concurrently. Please read on - one of two things is happening:
- you're not waiting for the previous
start
call to resolve before calling start
again - you're adding items to the pool in multiple places
In the former case, you probably have code that looks like this:
import _ from 'lodash';
import mongoist from 'mongoist';
const db = mongoist(...);
async function startJobs() {
const pool = new PromisePool({numConcurrent: 4});
const users = await db.users.findAsCursor().toArray();
_.each(users, async (user) => {
await pool.start(async () => {
await queue.publish(user);
});
});
await pool.flush();
}
Instead, you need to use some iteration method that preserves backpressure, like the for
-of
loop:
async function startJobs() {
const pool = new PromisePool({numConcurrent: 4});
const users = await db.users.findAsCursor().toArray();
for (const user of users) {
await pool.start(async () => {
await queue.publish(user);
});
}
await pool.flush();
}
Or even better, couple this with a call to promise-iterate
to only load users as you need them:
import {asyncIterate} from 'promise-iterate';
async function startJobs() {
const pool = new PromisePool({numConcurrent: 4});
const users = await db.users.findAsCursor();
await asyncIterate(users, async (user) => {
await pool.start(async () => {
await queue.publish(user);
});
});
await pool.flush();
}
The other case is where you're using promise-pool
in multiple places, and
thus you'll need to use maxPending
to tell promise-pool
that you know what
you're doing:
async function initializeAllUsers() {
const pool = new PromisePool({
numConcurrent: 4,
maxPending: 2,
});
const users = await db.users.findAsCursor();
const [usersA, usersB] = tee(users);
await Promise.all([
startJobs(pool, usersA),
sendEmails(pool, usersB),
});
await pool.flush();
}
async function startJobs(pool, users) {
await asyncIterate(users, async (user) => {
await pool.start(async () => {
await queue.publish(user);
});
});
}
async function sendEmails(pool, users) {
await asyncIterate(users, async (user) => {
await pool.start(async () => {
await sendEmail(user);
});
});
}
License
The MIT License (MIT)
Copyright © 2017 Mixmax, Inc (mixmax.com)
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in allcopies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.