Android background tutorial. Part 5: Coroutines in Kotlin
- Transfer
Kotlin Island
Previous texts in this series: about AsyncTask , about Loaders , about Executors and EventBus , about RxJava .
So this hour has come. This is the article for which the entire series was written: an explanation of how the new approach works “under the hood”. If you don’t yet know how to use it, here are some useful links to get you started:
And having mastered with coroutines, you may wonder what allowed Kotlin to provide this opportunity and how it works. Please note that here we will focus only on the compilation stage: you can write a separate article about execution.
The first thing we need to understand is that in corpus, actually no coroutines exist. The compiler turns the function with the suspend modifier into a function with the Continuation parameter . This interface has two methods:
abstract fun resume(value: T)
abstract fun resumeWithException(exception: Throwable)
Type T is the return type of your original suspend function. And here’s what actually happens: this function is executed in a certain thread (patience, we also get to this), and the result is passed to the resume function of that continuation, in the context of which the suspend function was called. If the function does not receive the result and throws an exception, then resumeWithException is thrown, throwing an error to the calling code.
Ok, but where did continuation come from? Of course, from the Corutin builder! Let's look at the code that creates any coroutine, for example, launch:
public actual fun launch(
context: CoroutineContext = DefaultDispatcher,
start: CoroutineStart = CoroutineStart.DEFAULT,
parent: Job? = null,
block: suspend CoroutineScope.() -> Unit
): Job {
val newContext = newCoroutineContext(context, parent)
val coroutine = if (start.isLazy)
LazyStandaloneCoroutine(newContext, block) else
StandaloneCoroutine(newContext, active = true)
coroutine.start(start, coroutine, block)
return coroutine
}
Here, builder creates a coroutine - an instance of the AbstractCoroutine class, which, in turn, implements the Continuation interface. The start method belongs to the Job interface. But finding the definition of the start method is very difficult. But we can come here from the other side. An attentive reader has already noticed that the first argument to the launch function is the CoroutineContext, and it is set to DefaultDispatcher by default. Dispatchers are classes that control the execution of coroutines, so they are definitely important for understanding what is happening. Let's look at the DefaultDispatcher declaration:
public actual val DefaultDispatcher: CoroutineDispatcher = CommonPool
So, in fact, this is CommonPool, although java docks tell us that this can change. What is CommonPool?
This is a coroutine manager using ForkJoinPool as an implementation of ExecutorService. Yes, it is: in the end, all your lambda coroutines are just Runnable, which got into Executor with a set of tricky transformations. But the devil, as always, is in the details.
Fork? Or join?
Judging by the results of the survey on my twitter, here I need to briefly explain what FJP is :)
First of all, ForkJoinPool is a modern executor created for use with Java 8 parallel streams. The original task was efficient parallelism when working with the Stream API, which essentially means splitting the streams to process part of the data and then combining it when all the data has been processed. To simplify, imagine that you have the following code:
IntStream
.range(1, 1_000_000)
.parallel()
.sum()
The amount of such a stream will not be calculated in one stream, instead, ForkJoinPool will recursively split the range into parts (first into two parts of 500,000, then each of them into 250,000, and so on), calculate the sum of each part, and combine the results into a single amount. Here's a visualization of such a process:
Threads are split for different tasks and re-combined after completion.
The FJP is based on the “job theft” algorithm: when a particular thread runs out of tasks, it goes to the queues of other pool threads and steals their tasks. For a better understanding, you can see the report of Alexei Shipilev or watch a presentation .
Well, we realized what our coroutines are doing! But how do they end up there?
This happens inside the CommonPool # dispatch method:
_pool.execute(timeSource.trackTask(block))
The dispatch method is called from the resume (Value: T) method in DispatchedContinuation. Sounds familiar! We remember that Continuation is an interface implemented in AbstractCoroutine. But how are they related?
The trick is inside the CoroutineDispatcher class. It implements the ContinuationInterceptor interface as follows:
public actual override fun interceptContinuation(continuation: Continuation): Continuation =
DispatchedContinuation(this, continuation)
See? You provide a simple block to the builder corutin. You do not need to implement any interfaces that you want to know nothing about. The coroutine library takes care of all this. It
hooks execution, replaces continuation with DispatchedContinuation, and sends it to executor, which guarantees the most efficient execution of your code.
Now the only thing we need to deal with is how dispatch is called from the start method. Let's fill this gap. The resume method is called from startCoroutine in the extension function of the block:
public fun (suspend R.() -> T).startCoroutine(
receiver: R,
completion: Continuation
) {
createCoroutineUnchecked(receiver, completion).resume(Unit)
}
And startCoroutine, in turn, is called by the "()" operator in the CoroutineStart enumeration. Your builder accepts it as the second parameter, and the default is CoroutineStart.DEFAULT. That's all!
That's the reason why I admire the corutin approach: it is not only a spectacular syntax, but also a brilliant implementation.
And for those who have read to the end, they get exclusive: a video of my report “A violinist is not needed: we refuse RxJava in favor of coroutine in Kotlin” from the Mobius conference . Enjoy :)