Playing with promises and concurrency in TypeScript

I still remember how hard it was to understand promises the first time I learned them. Modern languages have built constructs like JavaScript's async/await to make it easier to synchronize processes and keep things simple. With async/await, you can write asynchronous code as if it was synchronous, but as with any abstraction, understanding the underlying mechanisms will help you to understand what's going on when things start behaving in unexpected ways.

We recently found an issue that drove our entire team nuts for a couple of hours. We deployed a lambda function to AWS that performed several tasks concurrently. To do that, we wrote code that looked like this:

async function task1(): Promise<void> {...}
async function task2(): Promise<void> {...}
async function task3(): Promise<void> {...}

async function doAllTheThings(): Promise<void[]> {
  await Promise.all([task1(), task2(), task3()])
}

This worked nicely... until one of the tasks failed. When that happens, Promise.all is automatically rejected, and the doAllTheThings method stops "awaiting", no matter what's the state of the other tasks. In our case, it looks like once the lambda returned the error response, it was killed, canceling the remaining ongoing tasks without notice. There's a shiny new Promise.allSettled method introduced in Node 12.9 that waits for all promises to either be fulfilled or rejected before emitting a response, but it's not available if you're not targeting ES2020 (which is our case). Discussing ideas to solve this without using the new Promise.allSettled method, we realized that the team had different levels of understanding of what was actually going on under the hoods, so... time for a back-to-basics article!

What is actually a promise?

Starting from the beginning, a Promise is not black magic sorcery, it's just an object that is smartly designed to hide the complexities of dealing with asynchronous code. You can even create your own custom promise objects if you want. This is a simple promise implementation to help you understand the basics:

class MyOwnPromise<TValue> {
    /* The promise holds the value, which is initially "unknown", meaning
     * that the promise is "pending". Once this value is set, 
     * we say that the promise is "resolved". The error will also be "cached" */
    private value?: TValue = null
    private error?: Error = null

    // Callbacks set before fulfillment are put in these queues
    private thenQueue: Array<(value: TValue) => void> = []
    private catchQueue: Array<(error: Error) => void> = []

    /* The constructor gets an `execFunction` with the user's code.
     * This function is asynchronous and calls the `resolveCallback` when the process
     * is fulfilled or the `rejectCallback` function when it isn't. */
    public constructor(
        readonly execFunction: (
            resolveCallback: (value: TValue) => void, 
            rejectCallback: (error: Error) => void
        ) => void
    ) {
        /* The user's `execFunction` is called in the constructor, this is important
         * to understand that any promise you create starts running right away, 
         * no matter if you await for it or not */
        execFunction(
            (value) => {
                /* When the user calls to the `resolveCallback`, we save the value
                 * and call the `then` callbacks to notify the user about the
                 * resolved value */
                this.value = value
                for (const thenCallback of this.thenQueue) {
                    thenCallback(value)
                }
            },
            (error) => {
                /* When the user calls to the `rejectCallback` with an error,
                 * we cache the error and call the `catch` callbacks in the queue. */
                this.error = error
                for (const catchCallback of this.catchQueue) {
                    catchCallback(error)
                }
            }
        )
    }

    /* The user calls to `then` to receive a callback when the promise
     * is resolved. */
    public then(callback: (value: TValue) => void) {
        if (this.value) {
            /* When the promise is resolved, the callback is called right away.
             * This is important, and means that the user don't need to know 
             * if a promise is fulfilled or not before registering their callback.*/
            callback(this.value)
        } else {
            // When is pending, we append the callback to the queue
            this.thenQueue.push(callback)
        }
    }

    public catch(callback: (error: Error) => void) {
        if (this.error) {
            /* In the same way, when there's an error, we just send it to 
             * the callback right away. */
            callback(this.error)
        } else {
            /* When there are no errors, we append the callback to the queue */
            this.catchQueue.push(callback)
        }
    }
}

Using this code as a reference, we can see that the most important properties of the promises that are important to remember:

• They encapsulate asynchronous code, but they are not processes.
• The asynchronous code is started right away no matter if the promise is awaited or not.
• Promises smartly cache their values and errors, so the user doesn't need to know the internal state of the promise to deal with it.

What do async/await actually do?

Okay, this might sound like it has nothing to do with async/await, so let's see first how our MyOwnPromise is used. Let's say I want to "promisify" a task that takes some time and can fail with an exception. I can do this:

// I create my promise, my code will be started right away
const myOwnPromise = new MyOwnPromise<string>((resolve, reject) => {
    try {
        resolve(longRunningTask())
    } catch (error) {
        reject(error)
    }
})

// If I want to receive the value once the process is completed:
myOwnPromise.then((value) => console.log('My value is ' + value))

// To log the errors:
myOwnPromise.catch((error) => console.error('Ooops: ' + error.message))

Async functions and the await keyword were introduced in ECMAScript 2017. The async keyword transforms a function into a "promise builder". It forces a function to return a promise that is fulfilled with any value it returns or rejected with any exception raised. This means that we can rewrite our previous code like this:

// I just need to wrap my `longRunningTask` in an async function to build my promise. I can do this only because my task runs synchronously. If it was asynchronous, I'd need to wrap it in a new promise object anyway.
const myOwnPromise = (async () => {
    return longRunningTask()
})()

// Everything else works in the same way because the returned object is a promise
myOwnPromise.then((value) => console.log('My value is ' + value))
myOwnPromise.catch((error) => console.error('Ooops: ' + error.message))

The await keyword waits for a promise to be fulfilled and unwraps its value. If the promise is rejected, it throws the error as an exception. So we can finally rewrite our code like this:

const myOwnPromise = (async () => {
    return longRunningTask()
})()

try {
    // I just need to "await" for it:
    const value = await myOwnPromise
    console.log('My value is ' + value))
} catch (error) {
    console.error('Ooops: ' + error.message)
}

So async/await basically makes async code feel like it's synchronous code, but the most important thing to remember at this point is that all the versions of the code in this section are equivalent, async and await are just convenient syntactic sugar to help us reason about our asynchronous code, but the promise objects are still being created under the hoods in the same way. Promises are run when they're created, and we can play a bit with the awaits to achieve different execution schemes:

1. Running a sequence of tasks:

This is the easy scenario. Awaiting the promises as they are created we can block them from running until the previous one is completed. You often do this when one task require previous tasks results:

const result1 = await task1()
const result2 = await task2(result1)
const result3 = await task3(result2)

2. Running the tasks concurrently placing the awaits strategically

If you create all the tasks without immediately awaiting for them, they'll run concurrently, but managing errors could be tricky. In this case, all tasks will run concurrently, but if, let's say, task2 fails, it will wait for task1 to finish before raising the exception.

const promise1 = task1()
const promise2 = task2()
const promise3 = task3()

const results = [
    await promise1,
    await promise2, // If this one fails, we won't notice until `promise1` is fulfilled
    await promise3
]

3. Running tasks concurrently wrapping them in another promise!

Once you understand how promises, async and await work, you can combine them to achieve advanced behaviors. For instance, we could emulate what Promise.all do wrapping the whole process in a new promise that is only resolved when all the promises have been resolved and rejected immediately on the first rejection:

function customPromiseAll(...promises: Array<Promise<any>>): Promise<void> {
    // We initialize an `fulfilled` boolean array of the size of `promises`
    const pending = promises.map(() => true)
    return new Promise((resolve, reject) => {
        promises.forEach((promise, index) => {
            promise
                .then((value) => {
                    // When a promise is resolved, we store this fact in our array
                    pending[index] = false
                    // If there are no pending promises, we can resolve our main promise
                    if (!pending.find((isPending) => isPending === true)) {
                        resolve()
                    }
                })
                .catch(reject) // Any failure in one of the promises will make the main promise rejected
        })
    })
}

And back to our original problem: Making sure that we properly wait for all our promises to be settled (resolved or rejected). We can implement our own version of Promise.allSettled like this:

type PromiseResult = {
    status: 'pending'
} | {
    status: 'fulfilled',
    value: any
} | {
    status: 'rejected',
    error: Error
}

function customPromiseAllSettled(...promises: Array<Promise<any>>): Promise<PromiseResult[]> {
    // We initialize a `results` array of the size of `promises`
    const results: PromiseResult[] = promises.map(() => {
        return { status: 'pending' }
    })

    return new Promise((resolve, reject) => {
        function resolveIfNothingPending(): void {
            // If there are no pending promises, we can resolve our main promise
            if (!results.find((result) => result.status === 'pending')) {
                resolve(results)
            }
        } 

        promises.forEach((promise, index) => {
            promise
                .then((value) => {
                    // When a promise is resolved, we store the value and status
                    results[index] = { value, status: 'fulfilled' } 
                    resolveIfNothingPending()
                })
                .catch((error) => {
                    // When there's an error, we store it in the `results` array too
                    results[index] = { error, status: 'rejected' }
                    resolveIfNothingPending()
                }) 
        })
    })
}

Conclusions

In this article, we've gone through the internals of promises, this little, but powerful object that significantly simplifies how you deal with asynchronous code. We've seen how there's little magic in this truly magical object, and how you can use promises, async and await to build complex behaviors. In a production project, I wouldn't recommend implementing your own promise classes or combinator functions if you can use functions like Promise.allSettled, but understanding promises well is definitely an ace up your sleeve in your developer tool kit. I hope you liked it!

Cover photo by Gerrie van der Walt on Unsplash

*Disclaimer: I edited the code samples in VSCode to make sure the syntax is correct, but haven't run them in real scenarios. If a code snippet doesn't work, let me know in the comments, I'll be happy to fix it!

Comments

[Olivier Guimbal]

In the same spirit, I very often use this helper, which is very handy to limit your concurrency while keeping things parallel.

For instance, if you want to gather 4938935 users based on a list of IDs, its not advisable to launch 4938935 requests in parallel, but you could use parallel(100, userIds, fetchUser) to always run 100 concurrent requests until all are processed.

/**
 * Similar to Promise.all(), 
 * but limits parallelization to a certain number of parallel executions.
 */
export async function parallel<T>(concurrent: number
         , collection: Iterable<T>
         , processor: (item: T) => Promise<any>) {
    // queue up simultaneous calls
    const queue = [];
    const ret = [];
    for (const fn of collection) {
        // fire the async function, add its promise to the queue, and remove
        // it from queue when complete
        const p = processor(fn).then(res => {
            queue.splice(queue.indexOf(p), 1);
            return res;
        });
        queue.push(p);
        ret.push(p);
        // if max concurrent, wait for one to finish
        if (queue.length >= concurrent) {
            await Promise.race(queue);
        }
    }
    // wait for the rest of the calls to finish
    await Promise.all(queue);
}

`

[Javier Toledo]

Interesting, that's a very cool way to use Promise.race!

[stereobooster]

Running the tasks in parallel

We need to specify which tasks can actually run in parallel - io tasks and code out of the main thread e.g. WebWorkers (browser) or Worker threads (NodeJS). Other tasks would run concurrently, but not parallel because of event loop.

[Javier Toledo]

Right, I'm only talking about concurrency here, I'll change the wording in the article to make sure it doesn't lead to confusion. Thanks for pointing it out!