Executing asynchronous processes with a specified concurrency level

Introduction

I have added a new feature to my own library, Rotery.

https://www.npmjs.com/package/rotery
https://github.com/somnicattus/rotery

Rotery is a utility library for performing iterative processing efficiently. It features support for iterator and asyncIterator, functional programming with the pipe function, and powerful type support from TypeScript.

Feature Overview

The new feature added this time is "Concurrency Control." For example, the following code controls the execution so that 5 HTTP requests are executed in parallel.

const responses = await Rt.pipe(
    urls,
    Rt.map.sync(async url => {
        const response = await fetch(url);
        return await response.json();
    }),
    Rt.throttle(5), // This maintains up to 5 concurrent HTTP fetch requests.
    Rt.toArray.async, // The results are ordered by the completion time.
);

In the code above, Rt.map.sync(... generates HTTP requests through lazy evaluation using an iterator, and throttle(5) limits the concurrency of the generated HTTP requests to 5.

Implementation Characteristics of Rotery

While there are other libraries with concurrency control features, Rotery's concurrency control is implemented using asynchronous generators, allowing it to be integrated naturally into the pipe() function.

Additionally, each process during parallel execution outputs its result as soon as it completes and starts the next pending process. This minimizes the waiting time for asynchronous processing.

Here is the implementation source code. It is very compact, totaling about 50 lines including line breaks and comments, which I am personally satisfied with.

Source Code

const isNotEmptyElement = () => true;

async function* _buffer<T>(input: Series<T>, size: number): AsyncGenerator<Awaited<T>> {
    if (size <= 0 || !Number.isInteger(size))
        throw new RangeError(`"size" must be a positive integer (got ${size.toString()}).`);

    const awaited = await input;
    // eslint-disable-next-line @typescript-eslint/no-explicit-any
    const iterator = (Array.isArray as (v: unknown) => v is readonly any[])(awaited)
        ? awaited.values()
        : awaited;

    const pull = async () => {
        const next = iterator.next();
        return next instanceof Promise
            ? await next
            : // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
              ({ value: await next.value, done: next.done } as IteratorResult<MaybePromise<T>>);
    };

    const pullers = Array.from({ length: size }, (_, k) => pull().then(result => ({ k, result })));

    // eslint-disable-next-line unicorn/no-array-callback-reference
    while (pullers.some(isNotEmptyElement)) {
        // eslint-disable-next-line unicorn/no-array-callback-reference
        const item = await Promise.race(pullers.filter(isNotEmptyElement));
        if (item.result.done) {
            // eslint-disable-next-line @typescript-eslint/no-array-delete, @typescript-eslint/no-dynamic-delete
            delete pullers[item.k];
            continue;
        }
        yield await item.result.value;
        // eslint-disable-next-line @typescript-eslint/no-floating-promises
        pullers.splice(
            item.k,
            1,
            pull().then(result => ({ k: item.k, result })),
        );
    }
}

Code explanation by Microsoft Copilot

This TypeScript source code defines an asynchronous generator function _buffer. This function buffers an input asynchronous sequence and generates elements while maintaining a buffer of a specified size. Below is an explanation of each part of the code.

1. isNotEmptyElement function

const isNotEmptyElement = () => true;

This function always returns true. This is used later to check if elements in the pullers array exist.

2. _buffer function

async function* _buffer<T>(input: Series<T>, size: number): AsyncGenerator<Awaited<T>> {

This function takes a generic type T and returns an asynchronous generator. input is an asynchronous sequence, and size is the size of the buffer.

3. Size validation

if (size <= 0 || !Number.isInteger(size))
    throw new RangeError(`"size" must be a positive integer (got ${size.toString()}).`);

It checks that the buffer size is a positive integer, and if not, throws a RangeError.

4. Waiting for input

const awaited = await input;

Since the input sequence is asynchronous, it waits for the result using await.

5. Obtaining the iterator

const iterator = (Array.isArray as (v: unknown) => v is readonly any[])(awaited)
    ? awaited.values()
    : awaited;

If the input is an array, it obtains an iterator for its values; otherwise, it is used as an iterator as is.

6. pull function

const pull = async () => {
    const next = iterator.next();
    return next instanceof Promise
        ? await next
        : ({ value: await next.value, done: next.done } as IteratorResult<MaybePromise<T>>);
};

This function retrieves the next element, waits if it is a Promise, and returns it as is otherwise.

7. Creating the pullers array

const pullers = Array.from({ length: size }, (_, k) => pull().then(result => ({ k, result })));

It creates a pullers array based on the buffer size, where each element holds the result of the pull function.

8. Main loop

while (pullers.some(isNotEmptyElement)) {
    const item = await Promise.race(pullers.filter(isNotEmptyElement));
    if (item.result.done) {
        delete pullers[item.k];
        continue;
    }
    yield await item.result.value;
    pullers.splice(
        item.k,
        1,
        pull().then(result => ({ k: item.k, result })),
    );
}

This loop executes as long as elements exist in the pullers array. It uses Promise.race to obtain the first completed pull. If the result indicates completion, it is removed from pullers; otherwise, it yields the value and adds a new pull to pullers.

This code is designed to efficiently buffer asynchronous sequences and generate elements while maintaining a buffer of the specified size.

Conclusion

Questions, feature requests, and bug reports regarding this library are welcome. Please feel free to reach out in the comments section of this article, or via GitHub Issues and Pull Requests.