0
\$\begingroup\$

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));
});
\$\endgroup\$

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Browse other questions tagged or ask your own question.