As an exercise, I wrote a promise queue in Typescript. The point of it is to parallelise a maximum number of promises, and not just Promise.all()
to batter the event queue.
Of course there are professional queuing solutions that exist like RabbitMQ, which would destroy this performance wise. I get that and again, this is purely just an exercise.
I've littered some questions throughout the comments in the code. Any suggestions?
type RunQueueEntry = {
queueIndex: number;
promise: Promise<unknown>;
};
async function runQueue(tasks: Array<() => Promise<unknown>>, parallelLimit: number) {
if (!tasks.length || !parallelLimit) throw new Error('runQueue::BAD_PARAMETERS');
// I'm worried that the break logic might be hard to follow...
return new Promise(resolve => {
let results: unknown[] = [];
// This is purely so I can use the Atomic function
// This limits the amount of tasks to 256, as it relies on a single byte.
//
// Could using atomics in this way have serious performance ramifications?
// Is there a better way to prevent race conditions between calls?
// I tried to use queueMicroTask originally, to try and break the loop that way. I ran into headaches.
let taskIndex = Buffer.from([tasks.length - 1]);
let runQueue: RunQueueEntry[] = [];
let queueStreamsLeft = parallelLimit;
// Recursive function that uses object pooling, in that it just recycles the array slot in runQueue
const queueNext = (result: unknown, queueIndex: number) => {
results.push(result);
if (!tasks[Atomics.load(taskIndex, 0)]) {
queueStreamsLeft--;
if (queueStreamsLeft === 0) {
resolve(results);
}
} else {
runQueue[queueIndex].promise = tasks[Atomics.sub(taskIndex, 0, 1)]();
// Would it be quicker to add the callback functions in a separate loop?
runQueue[queueIndex].promise.then(res => {
queueNext(res, queueIndex);
});
}
};
for (let queueIndex = 0; queueIndex < parallelLimit; queueIndex++) {
if (tasks[Atomics.load(taskIndex, 0)]) {
// different from queueNext, as we have to construct the initial objects before pooling
runQueue[queueIndex] = {
queueIndex,
promise: tasks[Atomics.sub(taskIndex, 0, 1)]()
} as RunQueueEntry;
runQueue[queueIndex].promise.then(result => {
queueNext(result, queueIndex);
});
} else {
// If there's not enough objects to for the limit, don't bother
return Promise.all(runQueue.map(q => q.promise)).then(r => r.concat(results));
}
}
});
}
async function tFun(n: number) {
return n;
}
function buildBigQueue() {
let x = [];
for (let i = 0; i < 100; i++) {
x.push(() => tFun(i));
}
return x;
}
let time = process.hrtime();
runQueue(buildBigQueue(), 4).then(results => {
console.log(results);
console.log(process.hrtime(time));
});