I recently read the source code of Coroutines. It’s pretty fun, and I’ve learned a lot from the source code. I am going to share some I learned here. I would break the learning into several parts. I hope everyone can find something useful for your projects.

I download the latest master branch code at HERE, and if you are interested, you can just download the official release version to try. So, it’s up to you to choose either the official release version or the latest master branch. And the Kotlin version is 1.3.61 check README here.

Part1 - Introduction

Before starting the introduction, let’s think of a question: how do you handle the CPU-consuming tasks or networking accessing works? We can’t do these tasks in the main thread (in Android, which is UIThread). Then we need to think about using the multiple threading techniques to solve our questions. In Java, you can use Thread to do this.

1
2
3
4
5
6
7
8
9

Runnable task = () -> {
  String threadName = Thread.currentThread().getName();
  System.out.println("Hello " + threadName);
};

Thread thread = new Thread(task);
thread.start();

System.out.println("Done!");

You can see the result:

1
2

Hello Thread-0
Done!

or

1
2

Done!
Hello Thread-0

But creating threads frequently is a resource-consuming behavior, you would want to create a ThreadPool to reduce the cost.

In Android, we have several options for concurrency tasks. For example, you can use the AsyncTask, Background Service, and Handler in the Android projects.

The second question is: how do you communicate between two different threads? In Java, you need to manipulate the Thread interaction by using IPC mechanism, for example, shared memory, Pipe, named Pipe, mapped memory, semaphore, socket, message, and signals. There are many techniques we can use. What about concurrency in Kotlin’s world? Do we have any weapons for arming? Fortunately, we can still use the same techniques from the Java world. But Kotlin gives us a more powerful weapon, Coroutines, to use.

Let’s see an example first. We have three functions which need the results from the previous one finished the job.

1
2
3
4
5
6
7
8
9
10

fun requestToken(): Token { ... }
fun createPost(token: Token, item: Item): Post { ... }
fun processPost(post: Post) { ... }

// post a specific item
fun postItem(item: Item) {
  val token = requestToken()
  val post = createPost(token, item)
  processPost(post)
}

These three functions are probably time-consuming tasks. Hence, it’s not able to put them in the UI thread in the Android project. What can we do now?

By using a Thread

As I mentioned before, create a thread to run it.

1
2
3

Thread({
  postItem(item)
}).start()

Keeping this in mind: thread is not cheap. You would need to do context-switch, which is costly.

By using a Callback

We can also use a callback function plus threading to catch the result from each function.

1
2
3
4
5
6
7
8
9
10
11
12

fun postItem(item: Item) {
  preparePostAsync { token ->
    submitPostAsync(token, item) { post ->
      processPost(post)
    }
  }
}

fun preparePostAsync(callback: (Token) -> Unit) {
  // make request and return immediately
  // arrange callback to be invoked later
}

What’s the drawback of using callback? The most known pitfall is the callback hell. That makes you hardly debug and read the source code.

By using Java Future

In Java 8, it introduces the CompletableFuture. This feature lets you connect several tasks.

1
2
3
4
5
6
7
8
9

fun requestToken(): CompletableFuture { ... }
fun createPost(token: Token, item: Item): CompletableFuture { ... }
fun processPost(post: Post) { ... }

fun postItem(item: Item) {
  requestToken()
    .thenCompose { token -> createPost(token, item) }
    .thenAccept { post -> processPost(post) }
}

Rx Extension

I believe there must be many developers who use this library.

1
2
3
4
5
6

Single.fromCallable { requestToken() }
  .map { token -> createPost(token, item) }
  .subscribe(
    { post -> processPost(post) },
    { e -> e.printStackTrace() }
  )

I think the biggest problem for me when using Rx is that the exception log is pretty hard to read. And it also creates too many instances that Rx needs and consumes too much memory.

Here’s why Coroutines show and want to give developers another solution.

Coroutines

1
2
3
4
5
6
7
8
9
10
11

suspend fun requestToken(): Token { ... }
suspend fun createPost(token: Token, item: Item): Post { ... }
fun processPost(post: Post) { ... }

suspend fun postItem(item: Item) {
  GlobalScope.launch {
    val token = requestToken()
    val post = createPost(token, item)
    processPost(post)
  }
}

It looks just like the previous code snap, doesn’t it? If we want to add a try-catch block inside the postItem function. Just add it. We don’t need any extra effort to handle it.

First of all, let’s check how a coroutine works. A Coroutine is like a Thread in concept. But it’s much more lightweight than a real Thread. Several different coroutines can run on a Thread, and the Coroutine runtime manages them. You might remember the thread itself is not that cheap as you think because the system controls the threads, and it needs to do context-switch for you.

Another reason that a thread is costly is because it would spend more resource on the creation and termination. So the good practice is to use a thread pool (to create some threads inside first) in advance. Let’s back to Coroutines. The Coroutines don’t have the same problem. It’s very lightweight, which means you can create a Coroutine anytime, anywhere. It’s controlled by Kotlin runtime, not by system.

What’s the Suspend keyword? You would call suspending functions in the coroutines, just like regular functions. To call a suspending function, you need to declare the caller function as a suspending function as well. A suspending function can only be used in a coroutine or another suspending function.

When you try to launch a coroutine, it would be like this:

1
2
3

GlobalScope.launch {
  processPost(post)
}

Check the launch function itself:

1
2
3
4
5

public fun CoroutineScope.launch(
  context: CoroutineContext = EmptyCoroutineContext,
  start: CoroutineStart = CoroutineStart.DEFAULT,
  block: suspend CoroutineScope.() -> Unit
): Job

It launches a new coroutine without blocking the current thread and returns a reference to the coroutine as a Job. You can use this Job object to cancel the coroutine execution by calling Job.cancel(). A CoroutineScope defines a scope for new coroutines. It contains a CoroutineContext variable, which describes what factors would be involved in the CoroutineScope by setting up the Element.

Which thread would host a specific coroutine? In launch function, if you don’t assign a dispatcher, it would use Dispatchers. Default.

1
2
3
4
5
6
7
8
9
10
11
12

public fun CoroutineScope.launch(
  context: CoroutineContext = EmptyCoroutineContext,
  start: CoroutineStart = CoroutineStart.DEFAULT,
  block: suspend CoroutineScope.() -> Unit
): Job {
  val newContext = newCoroutineContext(context)
  val coroutine = if (start.isLazy)
  LazyStandaloneCoroutine(newContext, block) else
  StandaloneCoroutine(newContext, active = true)
  coroutine.start(start, coroutine, block)
  return coroutine
}

in newCoroutineContext, it will use Dispatchers.Default to find a thread to host a coroutine. There are several Dispatchers types:

1
2
3
4

public actual val Default: CoroutineDispatcher
public actual val Main: MainCoroutineDispatcher
public actual val Unconfined: CoroutineDispatcher
public val IO: CoroutineDispatcher

In the launch function, if the context does not have any dispatcher nor any other ContinuationInterceptor, then Dispatchers. Default would be used. The Default is a CoroutineDispatcher, which has a shared thread pool to use.

What’s a Job? The most basic instances of Job are created with launch function of coroutine builder or with a Job() factory function. By default, a failure of any of the job’s children leads to an immediate failure of its parent and cancellation of the rest of its children. This behavior can be customized using SupervisorJob. Job has a simple life-cycle:

It has three different states: isActive, isCompleted, and isCancelled. Conceptually, execution of the job does not produce a result value. Jobs are launched solely for their side-effects. If you try to get a result from a Job, check the Deferred.

Here I introduce the basic concepts of Coroutines, which is very powerful and helps to reduce the complexity of multi-task in Kotlin world.

Happy coding, enjoy.

Part2 - Suspend, Resume and Dispatch

Part3 - Callback, Interaction and Cancellation

Part4 - Exception

Part5 - Concurrency

Reference

1. KotlinConf 2017 - Introduction to Coroutines by Roman Elizarov
2. Asynchronous Programming Techniques