Implementing Concurrency with Kotlin [Tutorial]

Correct concurrent code is one that allows for a certain variability in the order of its execution while still being deterministic on the result. For this to be possible, different parts of the code need to have some level of independence, and some degree of orchestration may also be required.

This tutorial is an excerpt taken from the book, Learning Concurrency in Kotlin written by Miguel Angel Castiblanco Torres. In this book, you will learn how to handle errors and exceptions, as well as how to leverage multi-core processing.

In this article, we will be introduced to concurrency and how Kotlin approaches concurrency challenges.

The best way to understand concurrency is by comparing sequential code with concurrent code. Let's start by looking at some non-concurrent code:

fun getProfile(id: Int) : Profile {
    val basicUserInfo = getUserInfo(id)
    val contactInfo = getContactInfo(id)

return createProfile(basicUserInfo, contactInfo)
}

If I ask you what is going to be obtained first, the user information or the contact information – assuming no exceptions – you will probably agree with me that 100% of the time the user information will be retrieved first. And you will be correct. That is, first and foremost, because the contact information is not being requested until the contact information has already been retrieved:

Timeline of getProfile

And that's the beauty of sequential code: you can easily see the exact order of execution, and you will never have surprises on that front. But sequential code has two big issues:

• It may perform poorly compared to concurrent code
• It may not take advantage of the hardware that it's being executed on

Let's say, for example, that both getUserInfo and getContactInfo call a web service, and each service will constantly take around one second to return a response. That means that getProfile will take not less than two seconds to finish, always. And since it seems like getContactInfo doesn't depend on getUserInfo, the calls could be done concurrently, and by doing so it would be possible can halve the execution time of getProfile.

Let's imagine a concurrent implementation of getProfile:

suspend fun getProfile(id: Int) {
    val basicUserInfo = asyncGetUserInfo(id) 
    val contactInfo = asyncGetContactInfo(id)

createProfile(basicUserInfo.await(), contactInfo.await())
}

In this updated version of the code, getProfile() is a suspending function – notice the suspend modifier in its definition – and the implementation of asyncGetUserInfo() and asyncGetContactInfo() are asynchronous – which will not be shown in the example code to keep things simple.

Because asyncGetUserInfo() and asyncGetContactInfo() are written to run in different threads, they are said to be concurrent. For now, let's think of it as if they are being executed at the same time – we will see later that it's not necessarily the case, but will do for now. This means that the execution of asyncGetContactInfo() will not depend on the completion of asyncGetUserInfo(), so the requests to the web services could be done around at the same time. And since we know that each service takes around one second to return a response, createProfile() will be called around one second after getProfile() is started, sooner than it could ever be in the sequential version of the code, where it will always take at least two seconds to be called. Let's take a look at how this may look:

Concurrent timeline for getProfile

But in this updated version of the code, we don't really know if the user information will be obtained before the contact information. Remember, we said that each of the web services takes around one second, and we also said that both requests will be started at around the same time.

This means that if asyncGetContactInfo is faster than asyncGetUserInfo, the contact information will be obtained first; but the user information could be obtained first if asyncGetUserInfo returns first; and since we are at it, it could also happen that both of them return the information at the same time. This means that our concurrent implementation of getProfile, while possibly performing twice as fast as the sequential one, has some variability in its execution.

That's the reason there are two await() calls when calling createProfile(). What this is doing is suspending the execution of getProfile() until both asyncGetUserInfo() and asyncGetContactInfo() have completed. Only when both of them have completed createProfile() will be executed. This guarantees that regardless of which of the concurrent call ends first, the result of getProfile() will be deterministic.

And that's where the tricky part of concurrency is. You need to guarantee that no matter the order in which the semi-independent parts of the code are completed, the result needs to be deterministic. For this example, what we did was suspend part of the code until all the moving parts completed, but as we will see later in the book, we can also orchestrate our concurrent code by having it communicate between coroutines.

Concurrency in Kotlin

Now that we have covered the basics of concurrency, it's a good time to discuss the specifics of concurrency in Kotlin. This section will showcase the most differentiating characteristics of Kotlin when it comes to concurrent programming, covering both philosophical and technical topics.

Non-blocking

Threads are heavy, expensive to create, and limited—only so many threads can be created—So when a thread is blocked it is, in a way, being wasted. Because of this, Kotlin offers what is called Suspendable Computations; these are computations that can suspend their execution without blocking the thread of execution. So instead of, for example, blocking thread X to wait for an operation to be made in a thread Y, it's possible to suspend the code that has to wait and use thread X for other computations in the meantime.

Furthermore, Kotlin offers great primitives like channels, actors, and mutual exclusions, which provide mechanisms to communicate and synchronize concurrent code effectively without having to block a thread.

Being explicit

Concurrency needs to be thought about and designed for, and because of that, it's important to make it explicit in terms of when a computation should run concurrently. Suspendable computations will run sequentially by default. Since they don't block the thread when suspended, there's no direct drawback:

fun main(args: Array<String>) = runBlocking {
    val time = measureTimeMillis {
        val name = getName()
        val lastName = getLastName()
        println("Hello, $name $lastName")
    }
    println("Execution took $time ms")
}

suspend fun getName(): String {
delay(1000)
return "Susan"
}

suspend fun getLastName(): String {
delay(1000)
return "Calvin"
}

In this code, main() executes the suspendable computations getName() and getLastName() in the current thread, sequentially.

Executing main() will print the following:



This is convenient because it's possible to write non-concurrent code that doesn't block the thread of execution. But after some time and analysis, it becomes clear that it doesn't make sense to have getLastName() wait until after getName() has been executed since the computation of the latter has no dependency on the former. It's better to make it concurrent:

fun main(args: Array<String>) = runBlocking {
    val time = measureTimeMillis {
        val name = async { getName() }
        val lastName = async { getLastName() }

println("Hello, ${name.await()} ${lastName.await()}")
}
println("Execution took $time ms")
}

Now, by calling async {...} it's clear that both of them should run concurrently, and by calling await() it's requested that main() is suspended until both computations have a result:

Readable

Concurrent code in Kotlin is as readable as sequential code. One of the many problems with concurrency in other languages like Java is that often it's difficult to read, understand, and/or debug concurrent code. Kotlin's approach allows for idiomatic concurrent code:

suspend fun getProfile(id: Int) {
    val basicUserInfo = asyncGetUserInfo(id) 
    val contactInfo = asyncGetContactInfo(id)

createProfile(basicUserInfo.await(), contactInfo.await())
}

By convention, a function that is going to run concurrently by default should indicate this in its name, either by starting with async or ending in Async.

This suspend method calls two methods that will be executed in background threads and waits for their completion before processing the information. Reading and debugging this code is as simple as it would be for sequential code.

In many cases, it is better to write a suspend function and call it inside an async {} or launch {} block, rather than writing functions that are already async. This is because it gives more flexibility to the callers of the function to have a suspend function; that way the caller can decide when to run it concurrently, for example. In other cases, you may want to write both the concurrent and the suspend function.

Leveraged

Creating and managing threads is one of the difficult parts of writing concurrent code in many languages. As seen before, it's important to know when to create a thread, and almost as important to know how many threads are optimal. It's also important to have threads dedicated to I/O operations, while also having threads to tackle CPU-bound operations. And communicating/syncing threads is a challenge in itself.

Kotlin has high-level functions and primitives that make it easier to implement concurrent code:

• To create a thread it's enough to call newSingleThreadContext(), a function that only takes the name of the thread. Once created, that thread can be used to run as many coroutines as needed.
• Creating a pool of threads is as easy, by calling newFixedThreadPoolContext() with the size and the name of the pool.
• CommonPool is a pool of threads optimal for CPU-bound operations. Its maximum size is the amount of cores in the machine minus one.
• The runtime will take charge of moving a coroutine to a different thread when needed.
• There are many primitives and techniques to communicate and synchronize coroutines, such as channels, mutexes, and thread confinement.

Flexible

Kotlin offers many different primitives that allow for simple-yet-flexible concurrency. You will find that there are many ways to do concurrent programming in Kotlin. Here is a list of some of the topics we will look at throughout the book:

• Channels: Pipes that can be used to safely send and receive data between coroutines.
• Worker pools: A pool of coroutines that can be used to divide the processing of a set of operations in many threads.
• Actors: A wrapper around a state that uses channels and coroutines as a mechanism to offer the safe modification of a state from many different threads.
• Mutual exclusions (Mutexes): A synchronization mechanism that allows the definition of a critical zone so that only one thread can execute at a time. Any coroutine trying to access the critical zone will be suspended until the previous coroutine leaves.
• Thread confinement: The ability to limit the execution of a coroutine so that it always happens in a specified thread.
• Generators (Iterators and sequences): Data sources that can produce information on demand and be suspended when no new information is required.

All of these are tools that are at your fingertips when writing concurrent code in Kotlin, and their scope and use will help you to make the right choices when implementing concurrent code.

To summarize, we learned the basics of concurrency and how to implement concurrency in Kotlin programming language. If you've enjoyed reading this excerpt, head over to the book, Learning Concurrency in Kotlin to learn how to use the Machine Learning Toolkit and best practices and tips to help you implement Splunk services effectively and efficiently.

KotlinConf 2018: Kotlin 1.3 RC out and Kotlin/Native hits beta

Kotlin 1.3 RC1 is here with compiler and IDE improvements

Kotlin 1.3 M1 arrives with coroutines, and new experimental features like unsigned integer types