Asynchronous code in Java: CompletableFuture and races

Problem

The last item, sold twice

The sale is in full swing. One pair of headphones is left. checkout checks the stock, waits for the bank to confirm the payment and only then decreases the stock. Two buyers press "Buy" at the same moment:

// One thread plays the event loop: the continuations run in it one by one.
static final ExecutorService loop = Executors.newSingleThreadExecutor();

// after(ms): runs the task in loop after ms milliseconds.
static Executor after(long ms) {
    return CompletableFuture.delayedExecutor(ms, TimeUnit.MILLISECONDS, loop);
}

// The bank confirms the payment in 100 ms.
static CompletableFuture<Void> pay(String buyer) {
    return CompletableFuture.runAsync(() -> {}, after(100));
}

static int stock = 1;
static int sold = 0;

static CompletableFuture<Void> checkout(String buyer) {
    int left = stock;
    if (left == 0) {
        return CompletableFuture.completedFuture(null);
    }
    return pay(buyer).thenRunAsync(() -> {
        stock = left - 1;
        sold++;
    }, loop);
}

var first = checkout("Ann");
var second = checkout("Bob");
CompletableFuture.allOf(first, second).join();
System.out.println("Sold: " + sold + ", stock: " + stock);

Sold: 2, stock: 0

Two are sold, and there is one item. One buyer will get an e-mail "Your order has been shipped" for headphones that do not exist.

Everything here runs in one thread: loop is Executors.newSingleThreadExecutor(). In the lesson on race conditions two threads collided on count++. Here there is no second thread.

The output is the same on every run: this is not bad luck, it is the order in which the code runs. To see it, we need to know what await does.

The idea

One thread and many waits

Most of the work in an online shop is work that waits (see work that waits and work that computes): the payment waits for the bank, the e-mail waits for the mail server, the stock waits for the warehouse. A thread that sits and waits is wasted.

An event loop is one thread and a queue of pieces of code that are ready. The thread takes the next piece and runs it to the end. When the code reaches a wait, it does not hold the thread: it says "wake me up when the bank answers" and lets go. The thread takes the next ready piece. When the bank answers, the rest of the function (the continuation) goes into the queue and waits for its turn.

Java has no async and await. The closest thing is CompletableFuture: a promise of a result, to which you attach a continuation with thenRunAsync or thenApplyAsync. You say which executor the continuation runs in. Here they all run in loop, an executor with one thread that plays the event loop. That is how Netty and Vert.x work, and many Java servers are built on them.

Three requests to three services, started at once:

static int waiting = 0;
static int most = 0;

static CompletableFuture<Void> call(String name, int ms) {
    waiting++;
    most = Math.max(most, waiting);
    return CompletableFuture.runAsync(() -> {
        waiting--;
        System.out.println(name + ": done after " + ms + " ms");
    }, after(ms));
}

CompletableFuture.allOf(
    call("payment", 300),
    call("e-mail", 100),
    call("warehouse", 200)
).join();
System.out.println("Waiting at once: " + most);

e-mail: done after 100 ms
warehouse: done after 200 ms
payment: done after 300 ms
Waiting at once: 3

after(ms) gives an executor that runs the task in loop after ms milliseconds: this is the timer. call increases waiting and returns at once. allOf(...).join() waits for all three.

"Waiting at once: 3": all three requests wait at the same time, and there is only one thread. Everything takes about 300 ms (the longest), not 600 (the sum). The thread did not wait for any of them.

Try it

How many wait at once

The same call, but the first request is awaited on its own:

What will the program print?

call("payment", 300).join();
CompletableFuture.allOf(call("e-mail", 100), call("warehouse", 200)).join();
System.out.println("Waiting at once: " + most);

Answer

payment: done after 300 ms
e-mail: done after 100 ms
warehouse: done after 200 ms
Waiting at once: 2

e-mail: done after 100 ms
warehouse: done after 200 ms
payment: done after 300 ms
Waiting at once: 3

payment: done after 300 ms
e-mail: done after 100 ms
warehouse: done after 200 ms
Waiting at once: 3

The first call(...).join() waits for the payment to the end. Only after that does allOf start the other two at once, so the most is 2, and the order of the output is the order in which they finish. The whole thing takes about 500 ms.

await right after the call means "wait for this before you go on". For them to wait together, start them all first, then wait.

See it

A race between two awaits

Back to the last item. This is how the two orders interleave:

static int stock = 1;
static int sold = 0;

static CompletableFuture<Void> checkout(String buyer) {
    int left = stock;
    if (left == 0) {
        return CompletableFuture.completedFuture(null);
    }
    return pay(buyer).thenRunAsync(() -> {
        stock = left - 1;
        sold++;
    }, loop);
}

Step by step, in words

One item is left. Two orders arrive at the same moment: Ann and Bob.
Ann's checkout reads the stock: left is 1. The rest is in thenRunAsync: it runs when the bank answers, and the thread is free until then.
The thread is free, so Bob's checkout runs. It reads 1 as well: Ann has not written anything yet.
The bank answers Ann. Her code goes on after the thenRunAsync: stock = left - 1 = 0, sold = 1. Correct so far.
Bob goes on with his old left = 1: stock = 0 again, sold = 2. One item, two sales, with only one thread.

Between two awaits (here: between the start of the method and thenRunAsync) the code runs without interruption. But at an await the function lets go of the thread, and meanwhile anything else can run. When it goes on, the world may be different: left is a value read before the wait.

This is the same "read, change, write" as with threads, only the interruption is not just anywhere, it is exactly at the await. That is also why the output is the same on every run: the interruption is not random.

The idea

The fix: reserve the item before the await

The check and the decrease must be one piece of code with no await between them. The order reserves the item at once and only then waits for the bank:

static int stock = 1;
static int sold = 0;

static CompletableFuture<Void> checkout(String buyer) {
    if (stock == 0) {
        return CompletableFuture.completedFuture(null);
    }
    stock--;   // reserve the item before the await
    return pay(buyer).thenRunAsync(() -> sold++, loop);
}

Sold: 1, stock: 0

The check and stock-- come before pay(...) and run in one go. Here checkout is called from one thread, so that is enough. If many threads called it (the requests of a web server), you would also need synchronized around the check and the decrease, as in the lesson on locks.

If the payment fails, the item must be returned. That is the exercise at the end.

More precisely, if you are curious

Sometimes the whole piece must be locked together with the wait: for example, only one order at a time may write to an old warehouse system that cannot take two. Then you need a lock that understands await:

In a loop with one thread every piece between two continuations runs alone anyway. With many threads you use a Semaphore or a queue with one thread for the warehouse.

How it is in the other languages: in JavaScript and in asyncio there is one thread, and the piece between two awaits runs alone. In C# the continuations are in the thread pool, so a lock is needed there too. Java has no await: there the same thing is done with CompletableFuture, and since Java 21 also with virtual threads, where you write ordinary blocking code and the JVM unmounts the thread by itself while it waits.

Try it

Find the bug

The cart adds up the prices of three orders. The price comes from the warehouse with await. The three orders cost €25 each, and the total comes out as 25:

static int total = 0;

static CompletableFuture<Void> addOrder(int id) {
    int before = total;
    return priceOf(id).thenAcceptAsync(price -> {
        total = before + price;
    }, loop);
}

CompletableFuture.allOf(addOrder(1), addOrder(2), addOrder(3)).join();
System.out.println("Total: " + "€" + total);

Total: €25

Pick the line with the bug.

int before = total reads the total before the price arrives. All three orders read 0, and the continuations write 0 + 25. The addition must be in the continuation: priceOf(id).thenAcceptAsync(price -> total += price, loop).

The bug is not in the threads but in the order: a value read before an await is stale after it.

The idea

Do not block the event loop

The event loop works only if no piece of code holds the thread for long. The sales report is made with Thread.sleep(300) in loop. At the same time a buyer opens a product page, which waits 50 ms for the warehouse:

static void buildReport() throws InterruptedException {
    Thread.sleep(300);   // blocks: waits 300 ms without await
}

long start = System.nanoTime();
CompletableFuture<Void> report = CompletableFuture.runAsync(() -> {
    try {
        buildReport();
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    System.out.println("The report is ready");
}, loop);
CompletableFuture<Void> page = CompletableFuture.runAsync(() -> {
    long ms = (System.nanoTime() - start) / 1_000_000;
    System.out.println("Ann waits over 0.25 s: " + (ms > 250));
}, after(50));
CompletableFuture.allOf(report, page).join();

The report is ready
Ann waits over 0.25 s: true

The page waited over 0.25 s instead of 50 ms. While the report holds the thread, no other continuation can run: not this page, not payments, not timers. On a server with thousands of buyers, they all freeze at once.

The fix is to keep the heavy work out of the event loop's thread. It goes to a thread pool, and the event loop only waits for the result:

static final ExecutorService workers = Executors.newFixedThreadPool(2);

static CompletableFuture<Void> report() {
    return CompletableFuture.runAsync(() -> {
        try {
            buildReport();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }, workers).thenRunAsync(() -> System.out.println("The report is ready"), loop);
}

CompletableFuture<Void> page = CompletableFuture.runAsync(
    () -> System.out.println("Ann: the page is ready"), after(50));
CompletableFuture.allOf(report(), page).join();
workers.shutdown();

Ann: the page is ready
The report is ready

runAsync(..., workers) makes the report in the pool workers, and thenRunAsync(..., loop) brings the continuation back to loop. loop stays free, and the page is ready after 50 ms.

Try it

Fill in

Start the three requests at once and wait for them together:

(
    call("payment", 300),
    call("e-mail", 100),
    call("warehouse", 200)
).join();
System.out.println("Waiting at once: " + most);

Exercise

Reserve, pay, return

Solve it in your own editor.

Two items are left. Four buyers order at once: Ann, Bob, Vera and Gina. The payment takes 100 ms. Bob's card is declined: the payment throws an exception with the text "card declined".

checkout reserves the item before the await, then pays. If the payment fails, the item is returned. Print each buyer's result in the order of the orders, and at the end the stock and the number sold:

Ann: paid
Bob: card declined, the item is returned
Vera: out of stock
Gina: out of stock
Stock: 1, sold: 1

Vera and Gina see "out of stock", because Bob's item is still reserved until the bank answers. That is fair: the returned item goes to the next buyer.

More tasks

Try again. If a buyer finds no stock but another one may return an item, let them wait up to 200 ms and try again.
Timeout. The bank sometimes does not answer. If the payment takes over 1 s, decline the order and return the item. (orTimeout)
Report. Work out the total of the sales (€25 each) and print it at the end.

Test

Check yourself

1. In how many threads does your code run in JavaScript (Node) and in Python asyncio?In one: the event loop runs the pieces of code one by oneIn one thread per awaitIn as many as there are cores

2. Three requests wait 100, 200 and 300 ms. How long in total if you await them one by one, and how long with Promise.all, gather, Task.WhenAll or allOf?About 300 ms both waysAbout 600 ms and about 300 msAbout 600 ms both ways

3. In JavaScript and Python an async function reads shared data, then awaits, then writes. When can other code change the data?Only while it waits at the awaitNever: there is one threadBetween any two lines

4. Handling a request includes a computation that takes 2 seconds, without a single await. What happens to the other requests in Node or asyncio?Nothing: the async function runs separatelyThey move to another threadThey wait 2 seconds: the event loop is busy

5. How do you avoid selling the last item twice?Run the orders in two threadsCheck and decrease the stock with no await between them, then pay; if the payment fails, return the itemAdd one more await before the write

async is not a thread

Cookies