Designing for Concurrency

Concurrent design patterns enable us to handle multiple tasks simultaneously while effectively organizing their life cycles. By leveraging these patterns, you can sidestep issues like resource leaks and deadlocks.

In this chapter, we’re going to explore concurrent design patterns in Kotlin. We won’t implement all of them, as some are quite complex and involve numerous edge cases that are beyond the scope of this book. Instead, we’ll discuss some common constructs that you’ll encounter frequently while writing concurrent code in Kotlin and examine the design patterns they represent. Throughout, we’ll leverage essential components we’ve already covered, such as coroutines, channels, and functional programming concepts.

The topics we’ll cover in this chapter include:

• Deferred Value
• Barrier
• Scheduler
• Pipeline
• Fan-Out
• Fan-In
• Racing
• Mutex
• Sidekick...

Technical requirements

There are no additional requirements compared to the previous chapter.

You can find the source code used in this chapter on GitHub at the following location:

https://github.com/PacktPublishing/Kotlin-Design-Patterns-and-Best-Practices_Third-Edition/tree/main/Chapter08.

Deferred Value

The Deferred Value design pattern aims to provide a reference to the outcome of an asynchronous operation. You’ll find similar implementations in Java and Scala through Futures, and in JavaScript through Promises.

We touched upon deferred values earlier in Chapter 6, Threads and Coroutines, where we noted that Kotlin’s async() function returns a Deferred type, which is Kotlin’s take on this pattern.

Interestingly, this Deferred type is not just an example of the Deferred Value design pattern; it’s also an embodiment of the Proxy design pattern discussed in Chapter 3, Understanding Structural Patterns, as well as the State design pattern featured in Chapter 4, Getting Familiar with Behavioral Patterns.

To instantiate a new container for an asynchronous outcome, you can use Kotlin’s CompletableDeferred constructor like so:

val deferred = CompletableDeferred<String>()

To populate the Deferred value with a result...

Barrier

The Barrier design pattern enables us to pause and wait for multiple concurrent tasks to finish before moving on. This is particularly useful when assembling objects from diverse data sources.

Consider this data class:

data class FavoriteCharacter(
    val name: String,
    val catchphrase: String,
    val picture: ByteArray = Random.nextBytes(42)
)

Imagine the catchphrase comes from one service, and the picture comes from another. You’d like to fetch both data points concurrently:

fun CoroutineScope.getCatchphraseAsync(characterName: String) = async { … }
fun CoroutineScope.getPictureAsync(characterName: String) = async { … }

The most straightforward way to fetch this data concurrently would look like this:

suspend fun fetchFavoriteCharacter(name: String) = coroutineScope {
    val catchphrase = getCatchphraseAsync(name).await()
    val picture = getPictureAsync(name).await()
    FavoriteCharacter(name, catchphrase, picture)...

Scheduler

The goal of the Scheduler design pattern is to decouple what is being run from how it’s being run and optimize the use of resources when doing so.

In Kotlin, Dispatchers are an implementation of the Scheduler design pattern that decouples the coroutine (that is, the what) from underlying thread pools (that is, the how).

We’ve already seen dispatchers briefly in Chapter 6, Threads and Coroutines.

To remind you, the coroutine builders such as launch() and async() can specify which dispatcher to use. Here’s an example of how you specify it explicitly:

runBlocking { 
    // This will use the Dispatcher from the parent coroutine 
    launch { 
        // Prints: main 
        println(Thread.currentThread().name) 
    } 
    launch(Dispatchers.Default) { 
        // Prints DefaultDispatcher-worker-1 
        println(Thread.currentThread().name) 
    } 
}

The default dispatcher creates as many threads in the underlying thread pool as you...

Pipeline

The Pipeline design pattern is like having a team of experts working together to handle complex tasks. Each expert specializes in one part of the job, and they work simultaneously to get things done faster. Let’s explore this idea with an example.

Remember back in Chapter 4, Getting Familiar with Behavioral Patterns, when we talked about creating an HTML page parser? Back then, we assumed we already had the HTML pages to work with. Now, let’s design a process to create a never-ending stream of pages.

First, we need someone to fetch news pages from the internet every now and then. Think of this as our producer. In code, it looks like this:

fun CoroutineScope.producePages() = produce { 
    fun getPages(): List<String> { 
        // In reality, this would fetch pages from the web
        return listOf(
            "<html><body><h1>Cool stuff</h1></body></html>",
            "<html><body...

Fan-Out

The purpose of the Fan-Out design pattern is to divide the workload among multiple concurrent processors, or workers, efficiently. To grasp this concept better, let’s revisit the previous section but consider a specific problem: what if there’s a significant disparity in the amount of work at different stages in our pipeline?

For instance, fetching HTML content might take much longer than parsing it. In such cases, it makes sense to distribute the heavy lifting across multiple coroutines. In the previous example, each channel had only one coroutine reading from it. However, it’s possible for multiple coroutines to consume from a single channel, effectively sharing the workload.

To simplify the problem we’re about to discuss, let’s assume we have only one coroutine producing some results:

fun CoroutineScope.generateWork() = produce {
    for (i in 1..10_000) {
        send("page$i")
    }
    close()
}

And we’...

Fan-In

The objective of the Fan-In design pattern is to consolidate results generated by multiple workers. This pattern becomes invaluable when workers produce results that need to be gathered and managed.

Unlike the Fan-Out design pattern we discussed earlier, which involves multiple coroutines reading from the same channel, Fan-In reverses the roles. In this pattern, multiple coroutines can contribute their results by writing them to the same shared channel.

Combining the Fan-Out and Fan-In design patterns lays a solid foundation for building MapReduce algorithms. To illustrate this concept, we’ll make a slight modification to the workers used in the previous example:

private fun CoroutineScope.doWorkAsync(
    channel: ReceiveChannel<String>,
    resultChannel: Channel<String>
) = async(Dispatchers.Default) {
    for (p in channel) {
        resultChannel.send(p.repeat(2))
    }
}

Now, each worker sends the results of its computations to a common...

Racing

The Racing design pattern is a concurrency pattern that involves running multiple tasks that produce the same type of data concurrently and selecting the result from the task that completes first, discarding the results from the other tasks.

This pattern is useful in scenarios where you want to maximize responsiveness by accepting the result from the fastest task, even if multiple tasks are competing.

In Kotlin, you can implement the Racing pattern using the select function on channels. Here’s an example using two weather sources, preciseWeather and weatherToday, where you fetch weather information from both sources and accept the result from the source that responds first:

runBlocking {
    val winner = select<Pair<String, String>> {
        preciseWeather().onReceive { preciseWeatherResult ->
            preciseWeatherResult
        }
        weatherToday().onReceive { weatherTodayResult ->
            weatherTodayResult
        }
  ...

Mutex

Mutex, also known as mutual exclusion, serves as a way to safeguard a shared state that might be accessed by multiple coroutines simultaneously.

Let’s kick off with the familiar scenario we all dread—the counter example. Imagine multiple concurrent tasks attempting to update the same counter:

var counter = 0
val jobs = List(10) {
    async(Dispatchers.Default) {
        repeat(1000) {
            counter++
        }
    }
}
jobs.awaitAll()
println(counter)

As you might have guessed, the result displayed is less than 10,000, which is quite embarrassing!

To address this issue, we can introduce a locking mechanism that ensures only one coroutine interacts with the variable at any given time, making the operation atomic. Each coroutine tries to obtain ownership of the counter. If another coroutine is already updating it, our coroutine waits patiently and then attempts to acquire the lock again. After updating, it must release the lock to allow other...

Sidekick

The Sidekick design pattern enables us to delegate some tasks from our primary worker to a secondary worker.

So far, we’ve talked about using select solely as a receiver. However, it’s also possible to use select to send items to another channel. To illustrate, let’s consider an example.

First, we initialize batman as an actor coroutine that can process 10 messages every second:

val batman = actor<String> {
    for (c in channel) {
        println("Batman is dealing with $c")
        delay(100)
    }
}

Next, we introduce robin, another actor coroutine, albeit a slower one, capable of processing just four messages per second:

val robin = actor<String> {
    for (c in channel) {
        println("Robin is dealing with $c")
        delay(250)
    }
}

Here, we have a superhero and his sidekick represented as two actor coroutines. The superhero, being more adept, usually takes less time to handle villains...

Summary

In this chapter, we examined a variety of concurrency design patterns in Kotlin, with a focus on core components like coroutines, channels, and deferred values. Deferred values serve as placeholders for values that will be computed asynchronously. The Barrier design pattern synchronizes multiple asynchronous tasks, allowing them to move forward together. With the Scheduler pattern, we can separate the task logic from its runtime execution.

We also discussed the Pipeline, Fan-In, and Fan-Out patterns, which facilitate the distribution of tasks and the collection of results. The Mutex pattern is used to manage concurrent execution, ensuring tasks don’t conflict with one another. The Racing pattern is geared toward improving application responsiveness. Lastly, the Sidekick Channel pattern acts as a backup, taking on work when the primary task struggles to keep up.

These patterns equip you with the tools to manage your application’s concurrency in an efficient...

Questions

1. What does it mean when we say that the select expression in Kotlin is biased?
2. When should you use a Mutex instead of a channel?
3. Which of the concurrent design patterns could help you implement a MapReduce or divide-and-conquer algorithm efficiently?

Learn more on Discord

Join our community’s Discord space for discussions with the author and other readers:

https://discord.com/invite/xQ7vVN4XSc