Throttle amount of promises open at a given time
EDIT
Jeff Bowman has vastly improved his answer to resolve meaningful values. Feel free to view the history of this answer to understand why the resolved values are so important/useful.
throttlep
This solution closely mimics the native Promise.all
How it's the same …
- Resolves promises as quickly as possible
- Resolves an array of values in the same order as the inputs
- Rejects as soon as it encounters one reject
How it's different …
- Number parameter limits the number of simultaneously-running Promises
- Array input accepts promise creators (thunks); not actual Promises
// throttlep :: Number -> [(* -> Promise)]
const throttlep = n=> Ps=>
new Promise ((pass, fail)=> {
// r is the number of promises, xs is final resolved value
let r = Ps.length, xs = []
// decrement r, save the resolved value in position i, run the next promise
let next = i=> x=> (r--, xs[i] = x, run(Ps[n], n++))
// if r is 0, we can resolve the final value xs, otherwise chain next
let run = (P,i)=> r === 0 ? pass(xs) : P().then(next(i), fail)
// initialize by running the first n promises
Ps.slice(0,n).forEach(run)
})
// -----------------------------------------------------
// make sure it works
// delay :: (String, Number) -> (* -> Promise)
const delay = (id, ms)=>
new Promise (pass=> {
console.log (`running: ${id}`)
setTimeout(pass, ms, id)
})
// ps :: [(* -> Promise)]
let ps = new Array(10)
for (let i = 0; i < 10; i++) {
ps[i] = () => delay(i, Math.random() * 3000)
}
// run a limit of 3 promises in parallel
// the first error will reject the entire pool
throttlep (3) (ps) .then (
xs => console.log ('result:', xs),
err=> console.log ('error:', err.message)
)Console output
Inputs are run in order; Resolved results are in the same order as the inputs
running: 0
running: 1
running: 2
=> Promise {}
running: 3
running: 4
running: 5
running: 6
running: 7
running: 8
running: 9
result: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
Practical use
Let's look at a more practical code example. This code is tasked with fetching a set of images from a server. This is how we might use throttlep to throttle the amount of simultaneous requests to 3 at a time
// getImage :: String -> Promise<base64>
let getImage = url=> makeRequest(url).then(data => data.base64, reqErrorHandler)
// actions :: [(* -> Promise<base64>)]
let actions = [
()=> getImage('one.jpg'),
()=> getImage('two.jpg'),
()=> getImage('three.jpg'),
()=> getImage('four.jpg'),
()=> getImage('five.jpg')
]
// throttle the actions then do something...
throttlep (3) (actions) .then(results => {
// results are guaranteed to be ordered the same as the input array
console.log(results)
// [<base64>, <base64>, <base64>, <base64>, <base64>]
})
There isn't anything built-in for this, so you'll have to build your own. AFAIK, there's no library for this yet, either.
First, start with a "deferral" - a promise that allows external code to resolve it:
class Deferral<T> {
constructor() {
this.promise = new Promise<T>((resolve, reject) => {
this.resolve = resolve;
this.reject = reject;
});
}
promise: Promise<T>;
resolve: (thenableOrResult?: T | PromiseLike<T>) => void;
reject: (error: any) => void;
}
Then you can define a "wait queue", which represents all the code blocks that are waiting to enter the critical section:
class WaitQueue<T> {
private deferrals: Deferral<T>[];
constructor() {
this.deferrals = [];
}
get isEmpty(): boolean {
return this.deferrals.length === 0;
}
enqueue(): Promise<T> {
const deferral = new Deferral<T>();
this.deferrals.push(deferral);
return deferral.promise;
}
dequeue(result?: T) {
const deferral = this.deferrals.shift();
deferral.resolve(result);
}
}
Finally you can define an async semaphore, as such:
export class AsyncSemaphore {
private queue: WaitQueue<void>;
private _count: number;
constructor(count: number = 0) {
this.queue = new WaitQueue<void>();
this._count = count;
}
get count(): number { return this._count; }
waitAsync(): Promise<void> {
if (this._count !== 0) {
--this._count;
return Promise.resolve();
}
return this.queue.enqueue();
}
release(value: number = 1) {
while (value !== 0 && !this.queue.isEmpty) {
this.queue.dequeue();
--value;
}
this._count += value;
}
}
Example usage:
async function performActionsInParallel() {
const semaphore = new AsyncSemaphore(10);
const listOfActions = [...];
const promises = listOfActions.map(async (action) => {
await semaphore.waitAsync();
try {
await doSomething(action);
}
finally {
semaphore.release();
}
});
const results = await Promise.all(promises);
}
This method first creates a throttler and then immediately starts all the asynchronous operations. Each asynchronous operation will first (asynchronously) wait for the semaphore to be free, then perform the action, and finally release the semaphore (allowing another one in). When all asynchronous operations have completed, all the results are retrieved.
Warning: this code is 100% completely untested. I haven't even tried it once.
You can do this in one short function. (Returns values in order per naomik's suggestion. Thanks!)
/**
* Performs a list of callable actions (promise factories) so
* that only a limited number of promises are pending at any
* given time.
*
* @param listOfCallableActions An array of callable functions,
* which should return promises.
* @param limit The maximum number of promises to have pending
* at once.
* @returns A Promise that resolves to the full list of values
* when everything is done.
*/
function throttleActions(listOfCallableActions, limit) {
// We'll need to store which is the next promise in the list.
let i = 0;
let resultArray = new Array(listOfCallableActions.length);
// Now define what happens when any of the actions completes.
// Javascript is (mostly) single-threaded, so only one
// completion handler will call at a given time. Because we
// return doNextAction, the Promise chain continues as long as
// there's an action left in the list.
function doNextAction() {
if (i < listOfCallableActions.length) {
// Save the current value of i, so we can put the result
// in the right place
let actionIndex = i++;
let nextAction = listOfCallableActions[actionIndex];
return Promise.resolve(nextAction()).then(result => {
// Save results to the correct array index.
resultArray[actionIndex] = result;
}).then(doNextAction);
}
}
// Now start up the original <limit> number of promises.
// i advances in calls to doNextAction.
let listOfPromises = [];
while (i < limit && i < listOfCallableActions.length) {
listOfPromises.push(doNextAction());
}
return Promise.all(listOfPromises).then(() => resultArray);
}
// Test harness:
function delay(name, ms) {
return new Promise((resolve, reject) => setTimeout(() => {
console.log(name);
resolve(name);
}, ms));
}
var ps = [];
for (let i = 0; i < 10; i++) {
ps.push(() => delay("promise " + i, Math.random() * 3000));
}
throttleActions(ps, 3).then(result => console.log(result));