2023-12-01

So what is concurrency? And why is it?

Concurrency is a complicated, but important, subject when it comes to programming, and as our industry and demands advance, so do our tools. You might be wondering, then, what is concurrency exactly and how does it differ from something like asynchronicity? The answer can be broken down to a simple difference: concurrency is the process of multiple things happening at once, whereas asynchronicity involves asking for something to happen and getting notified when it finishes, while doing something else in the meanwhile. In concurrency multiple processes at least have the appearance of happening at once, whereas with asynchronicity a process is ‘waiting’ for something to occur.

In Java, concurrency involves making use of multiple system threads to handle different processes at the same time, attempting to make full use of a computer’s power. While threads have been an element of Java since Java 1.5, the needs of developers and the evolving tech landscape required it to evolve over time. Creating threads is easy, but managing them? That is a whole different beast. For this purpose, the Executor Service was introduced in Java 5, with the process being further simplified by the addition of the ForkJoinPool and CompletableFutures in Java 8. These features were very useful at the time, and are still powerful, but as demands have increased, this model has started to lag behind. Java Concurrency’s problems can be summed up in two words: cost and ignorance.

Like renting a room at the Hilton

So what do I mean when I say that threads are costly? Well, you see, starting up a thread costs a lot in terms of system resources. It has a startup time of around 1 millisecond, takes up about 2 megabytes of memory (on the stack) and you need to switch contexts, which takes about 100 nanoseconds (dependant on your OS). Now one thread? Seems small enough, but when you’re talking about processes that take up thousands of threads at a time, you are starting to notice significant slowdown, which is of course a bit counterproductive. This also means that it is simply impossible to have something in the scale of a million threads, as it would require about 15 minutes of time and 2 terabytes of memory. This might seem like a lot, but consider how intense our computer use is nowadays and it starts making sense that maybe we would want to be able to do more things at the same time.

Furthermore, if you have such an expensive resource, you want to make use of it as much as possible, but trying to keep your system busy is not easy. An idle thread still takes up resources, but if you’re not careful, a thread might be idle for almost the entirety of its lifetime. When you use a single thread, for example, to handle things for which it has to wait, it simply waits, and all threads wait at the same speed. The Executor Service, ForkJoinPool and CompletableFuture were introduced to handle this problem, introducing tools that manage pools of threads for you and facilitate asynchronous code. Lets write some code for reference.

Lets take a common example of something that usually requires asynchronicity: an api call! For purposes of illustration and mutually assured destruction, I will be using the DadJokes API. Our calling function looks something like this:

private DadJoke callApi() {  
ResponseEntity<DadJoke> response =    
        template.getForEntity("https://icanhazdadjoke.com/", DadJoke.class);

    if (response.getStatusCode() != HttpStatus.OK){  
        throw new ConnectionException(response.getStatusCode());  
    }  
    return response.getBody();  
}

Say, I want to return an entity that contains an array of three dad jokes. If I were to write this in a synchronous, somewhat naive and inefficient way, I could do so as follows:

public DadJoke[] getDadJokes(){  
    DadJoke[] jokes = new DadJoke[3];  

    jokes[0] = callApi();  
    jokes[1] = callApi();  
    jokes[2] = callApi();  

    return jokes;  
}

I call the API three times, I receive three random dad jokes, I put them in an array and I send that array back. Relatively simple. Now, every call to the API is blocking. The next function waits for the previous to finish. It’s an issue, as it slows things down. Of course, it’s not a big issue for a single call, but what if we had more? What if we had hundreds? Well, lets use completable futures to make the three calls happen at the same time:

private CompletableFuture<DadJoke> getDadJokeFuture(){  
    return CompletableFuture  
            .supplyAsync(this::callApi)  
            .exceptionally(ex -> new DadJoke("error", ex.getMessage(),
                         HttpStatus.BAD_REQUEST.value()));  
}

public DadJoke[] getDadJokes() {   
    CompletableFuture<DadJoke> future1 = getDadJokeFuture();  
    CompletableFuture<DadJoke> future2 = getDadJokeFuture();  
    CompletableFuture<DadJoke> future3 = getDadJokeFuture();  

    return Stream.of(future1, future2, future3)  
            .map(CompletableFuture::join)  
            .toArray(DadJoke[]::new);  

}

I make a little helper function which simply gives me a CompletableFuture that calls the api. I then run the completable futures in parallel and join the results together, forming the array. The completable future uses system threads, by default as many threads as your system has cores. This works! But it’s taxing… and looks a bit confusing. Furthermore, you can only deal with exceptions when tasks are finished. Threads are ignorant of each other and if one task fails, the others will unnecessarily continue their work, even though our code will fail anyway. Furthermore, if a thread gets interrupted or throws an exception somehow, neither will spread to subtasks. All of this is, of course, a waste of very precious resources! It’s like renting a room in the Hilton for five days but only staying there for one!

Weaving a tapestry

Technology keeps moving and our use of computers is through the roof. The above model of concurrency was a solution to problems that appeared almost ten years ago as of the date of writing, but it is becoming increasingly problematic. How do we deal with this issue? Enter Virtual Threads and Structured Concurrency. To explain how virtual threads and structured concurrency tackle the problems of cost and ignorance, let me first explain what virtual threads are.

Let me just give you a small code snippet of how to instantiate a virtual thread. Virtual threads have various factory methods that one can use to instantiate them. Let us compare the output of virtual threads with system threads.

public static void main(String[] args) throws InterruptedException {  
    Thread platformThread = Thread.ofPlatform()  
            .name("system-", 0)  
            .start(() -> {  
                System.out.println(String.format("system %s", Thread.currentThread()));  
            });  

    platformThread.join();  

    Thread virtualThread = Thread.ofVirtual()  
            .name("virtual-", 0)  
            .start(() -> {  
                System.out.println(String.format("virtual %s", Thread.currentThread()));  
            });  
    virtualThread.join();  
}

Here you can see how to instantiate the two different kinds of thread. They output the following:

system Thread[#21,system-0,5,main]
virtual VirtualThread[#22,virtual-0]/runnable@ForkJoinPool-1-worker-1

Interestingly, the virtual thread seems to have a bit more going on. What is all this text after the slash supposed to indicate? Well you see, virtual threads are special, lightweight threads that run on top of system threads. They are managed by the JVM and are detachable. This means that one does not need to make special system calls to create them, the JVM is aware of their functioning and they can be detached and reattached on demand, meaning that the system threads are free to work on other things while the virtual threads are idle. They are cheap and that means you can make a million of them. Observe:

public static void main(String[] args) throws InterruptedException {  
    var threads =  
            IntStream.range(0, 1_000_000)  
                    .mapToObj(index ->  
                            Thread.ofVirtual()  
                                    .name("virtual-", index)  
                                    .unstarted(() -> {  
                                        try {  
                                            System.out.printf("virtual %s%n", Thread.currentThread());  
                                            Thread.sleep(2_000);  
                                        } catch (InterruptedException e) {  
                                            throw new RuntimeException(e);  
                                        }  
                                    }))  
                    .toList();  

    Instant begin = Instant.now();  
    threads.forEach(Thread::start);  
    for (Thread thread : threads) {  
        thread.join();  
    }  
    Instant end = Instant.now();  
    System.out.println("Duration = " + Duration.between(begin, end).toSeconds());  
}

One. Million. Threads.

So…. what? How does it work then? How fast is it? Lets see a part of the output.

virtual VirtualThread[#1000017,virtual-999989]/runnable@ForkJoinPool-1-worker-3
virtual VirtualThread[#999940,virtual-999912]/runnable@ForkJoinPool-1-worker-6
virtual VirtualThread[#1000018,virtual-999990]/runnable@ForkJoinPool-1-worker-6
virtual VirtualThread[#1000019,virtual-999991]/runnable@ForkJoinPool-1-worker-6
virtual VirtualThread[#1000010,virtual-999982]/runnable@ForkJoinPool-1-worker-6
virtual VirtualThread[#1000024,virtual-999996]/runnable@ForkJoinPool-1-worker-1
virtual VirtualThread[#1000025,virtual-999997]/runnable@ForkJoinPool-1-worker-1
virtual VirtualThread[#1000026,virtual-999998]/runnable@ForkJoinPool-1-worker-1
virtual VirtualThread[#1000027,virtual-999999]/runnable@ForkJoinPool-1-worker-2
virtual VirtualThread[#1000022,virtual-999994]/runnable@ForkJoinPool-1-worker-2
virtual VirtualThread[#1000000,virtual-999972]/runnable@ForkJoinPool-1-worker-2
virtual VirtualThread[#1000021,virtual-999993]/runnable@ForkJoinPool-1-worker-2
virtual VirtualThread[#1000020,virtual-999992]/runnable@ForkJoinPool-1-worker-2
virtual VirtualThread[#999999,virtual-999971]/runnable@ForkJoinPool-1-worker-2
Duration = 19

What this block of code shows is that there are actually… a lot of threads. More than that, these threads only really use a limited amount of system threads! I can inform you that the max system threads used was actually 8. It wasn’t instantaneous, naturally. The duration at the end indicates that 19 seconds passed between start and finish, but that’s a whole lot better than impossible, right? Not to mention that it doesn’t eat memory like skittles. It can also carry multiple threads on a single system thread. Now that’s what I call multithreading.

In essence, starting up virtual threads up is cheap as dirt and means that starting up and managing virtual threads is a lot less of an issue. That said, managing a bunch of threads probably still seems like a big headache. Well, you could still make use of the Executor Services like before to build pools of virtual threads, or, well, there is structured concurrency. It is comparable to Kotlin coroutines or Go’s goroutines.

Note that as of Java 21, Structured Concurrency is still a preview feature so things are bound to change and functionality is still limited. Now without further ado, lets rewrite our api call using Structured Concurrency.

public DadJoke[] getDadJokes(){  

    try (StructuredTaskScope<DadJoke> scope = new StructuredTaskScope<>()) {  

        Subtask<DadJoke> fork1 = scope.fork(this::callApi);  
        Subtask<DadJoke> fork2 = scope.fork(this::callApi);  
        Subtask<DadJoke> fork3 = scope.fork(this::callApi);  

        scope.join();  

        if (forkIsUnsuccessful(fork1, fork2, fork3)){  

            throw new StructuredConcurrencyException("A thread failed");  
        }  

        return new DadJoke[]{fork1.get(), fork2.get(), fork3.get()};  

    } catch (InterruptedException e) {  
        e.printStackTrace();   
    }  
}

@SafeVarargs  
private <T> boolean forkIsUnsuccessful(Subtask<T>... tasks){  

    return Arrays  
            .stream(tasks)  
            .anyMatch(task -> task.state() == State.FAILED);  
};

First we instantiate our scope in a try-with-resources statement. What is that, a scope? Well, it is an object that allows us to perform structured concurrency. It supports cases where a task is split up into several subtasks that are all performed… concurrently. The life of the concurrent operation is confined to its syntax block, like a sequential operation. It works a lot like an Executor Service in that you supply it with tasks. The fork method defines the subtasks to perform. A subtask holds a value, as well as a State. This state can be used to check whether a task was successful or not. The join method is blocking and waits until all subtasks are done, after which the value and state of the subtasks can be extracted. Each subtask runs on a separate virtual thread, and it’s fine to block these as they are just so much cheaper than system threads. Importantly, for this to work, the values within the scope are immutable, but can be shared between threads.

In this above code, all the checking and so forth is done manually, but there are special factory methods of the StructuredStaskScope that allow it to shutdown on success and failure. The ShutdownOnSuccess scope simply shuts down the scope after the first subtask is successful. The ShutdownOnFailure scope, of course, does the opposite, and you can follow this up by throwing an exception.

Something like this

public DadJoke[] getDadJokes(){  

    try (StructuredTaskScope.ShutdownOnFailure scope = new
                    StructuredTaskScope.ShutdownOnFailure()) {  

        Subtask<DadJoke> fork1 = scope.fork(this::callApi);  
        Subtask<DadJoke> fork2 = scope.fork(this::callApi);  
        Subtask<DadJoke> fork3 = scope.fork(this::callApi);  

        scope.join();  
        scope.throwIfFailed(ex -> new
                StructuredConcurrencyException(ex.getMessage()));  
                // it throws an Execution Exception by default,
                // which is a checked exception.

        return new DadJoke[]{fork1.get(), fork2.get(), fork3.get()};  

    } catch (InterruptedException e) {  
        e.printStackTrace();  
    }  
}

That was virtual threads and structured concurrency in java in a shellnut! I hope it informed you a bit about how you can weave your own multithreaded code tapestry. It is there to allow you to boost performance of certain tasks within a specific scope. This allows you to apply readily evident control flow to your multithreaded processes as all threads complete their tasks before exit. Let me know if there is anything I missed or anything I didn’t focus on which you would like more clarity on. I’d also love to know whether there are any specific topics you would like to see me expand on! Write something beautiful.