In the Introduction section, we already have a big picture of the Coroutines feature, including what it is, how we can use it, and other basic concepts. In this article, I will introduce another critical keyword suspend:
suspendfunpostItem(item: Item) { GlobalScope.launch { val token = requestToken() val post = createPost(token, item) processPost(post) } }
You can see the keyword suspend in front of a function. What does suspend keyword mean? If you compile the Kotlin code, you can see something like this:
1 2
// declaration: requestToken(kotlin.coroutines.Continuation<? super kotlin.Unit>) publicfinalrequestToken(Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
The code snap above is what we decompile into java code.
This is a bytecode/compiled-java-code sample for this suspend function - requestToken(). We can find there’s a parameter passed into requestToken(), which is a Continuation. What’s a Continuation?
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
/** * Interface representing a continuation after a suspension point that returns a value of type `T`. */ @SinceKotlin("1.3") publicinterfaceContinuation{ /** * The context of the coroutine that corresponds to this continuation. */ publicval context: CoroutineContext
/** * Resumes the execution of the corresponding coroutine passing a successful or failed [result] as the * return value of the last suspension point. */ publicfunresumeWith(result: Result) }
It’s an interface for resuming the suspended point. Here’s an example when you try to use CoroutineScope.Launch:
1 2 3 4 5
GlobalScope.launch { val token = requestToken() val post = createPost(token) processPost(post) }
Post post = (Post)var10000; MainActivity.this.processPost(post); return Unit.INSTANCE; }
In a Coroutine, it doesn’t use a callback function to return value. Instead, it uses the state machine to control what will respond to the next state. A suspend function will block the Coroutine when it’s in, but it wouldn’t block the thread. When it executes the suspend function, it won’t block the whole thread. The current thread can still do other jobs.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
Thread {
val context = newSingleThreadContext("Test Thread") GlobalScope.launch (context) { println("Start first coroutine.") delay(200) println("Resume first coroutine after delay 200ms") }
GlobalScope.launch (context){ println("Start second coroutine.") delay(100) println("Resume second coroutine after delay 100ms") }
Thread.sleep(500) }.start()
the result:
1 2 3 4
2019-08-17 21:25:22.437 I/System.out: Start first coroutine. 2019-08-17 21:25:22.473 I/System.out: Start second coroutine. 2019-08-17 21:25:22.574 I/System.out: Resume second coroutine after delay 100ms 2019-08-17 21:25:22.656 I/System.out: Resume first coroutine after delay 200ms
That’s why a Coroutine is very lightweight, and you can create it anytime without spending many resources on creating a Coroutine. After this, you might have a question: after resuming the suspended status, current Coroutine will still run on the same thread?
If you remember, I mentioned Dispatcher would decide which thread a Coroutine should run. There are two categories of dispatcher - confined & unconfined. The Default,Main, and IO belong to Confined category, and Unconfined is unconfined. Let’s create two different coroutines and use Confined & Unconfined dispatchers to see what will happen.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
Thread { GlobalScope.launch (Dispatchers.Main) { println("First coroutine runs in ${Thread.currentThread().name}") delay(100) println("First coroutine resumes and runs in ${Thread.currentThread().name}") }
GlobalScope.launch (Dispatchers.Unconfined){ println("Second coroutine runs in ${Thread.currentThread().name}") delay(100) println("Second coroutine resumes and runs in ${Thread.currentThread().name}") }
Thread.sleep(500) }.start()
Result:
1 2 3 4
2019-08-17 21:43:22.895 I/System.out: Second coroutine runs in Thread-4 2019-08-17 21:43:22.947 I/System.out: First coroutine runs in main 2019-08-17 21:43:23.005 I/System.out: Second coroutine resumes and runs in kotlinx.coroutines.DefaultExecutor 2019-08-17 21:43:23.050 I/System.out: First coroutine resumes and runs in main
When you use a confined dispatcher, no matter when you suspend the Coroutine, it would still run on the same thread after it resumes for the Dispatchers.Unconfined setting, it would first start on the caller thread but continue on a thread based on where the suspend function is.
So far, I introduce how the suspend function works, but it’s still worth going through how to create a Coroutine, start it and resume mechanism, especially it would be great to know how to switch different Coroutines. Let’s dig in a little bit deeper.
Go back to the CoroutineScope.Launch:
1 2 3 4 5 6 7 8 9 10 11 12 13
publicfun CoroutineScope.launch( context: CoroutineContext = EmptyCoroutineContext, start: CoroutineStart = CoroutineStart.DEFAULT, block: suspendCoroutineScope.() -> 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 general, the start parameter will be CoroutineStart.DEFAULT. It would go to here once the start function calls:
It creates a Coroutine, which is not intercepted without a receiver but with a result type T. This function creates a new and fresh instance of suspendable computation every time when it is invoked. To start executing this prepared Coroutine, it would call resume(Unit) on the returned Continuation instance. So, remember createCoroutineUnintercepted that will call resumeCancellable to execute a created Coroutine.
resumeCancellable function
1 2 3 4
internalfun Continuation.resumeCancellable(value: T) = when (this) { is DispatchedContinuation -> resumeCancellable(value) else -> resume(value) }
resume function
1 2
publicinlinefun Continuation.resume(value: T): Unit = resumeWith(Result.success(value))
In the resume function, it calls the resumeWith function in a Continuation. Let’s check how it works.
internalabstractclassBaseContinuationImpl( // This is `public val` so that it is private on JVM and cannot be modified by untrusted code, yet // it has a public getter (since even untrusted code is allowed to inspect its call stack). publicval completion: Continuation? ) : Continuation, CoroutineStackFrame, Serializable {
// This implementation is final. This fact is used to unroll resumeWith recursion. publicfinaloverridefunresumeWith(result: Result) { ...
BaseContinuationImpl is the implementation of the interface Continuation. You can see when a Coroutine tries to get the outcome, it will call invokeSuspend function which is generated by the compiler. If you still remember the example processPost we listed before, the compiled java code shows that the actual computation is in the invokeSuspend function. So the starting flow is:
In intercepted(), it tries to call context[ContinuationInterceptor]?.interceptContinuation(this) to get the Continuation, what’s the interceptContinuation()?
The CoroutineDispatcher implements this interface. Basically, the context[ContinuationInterceptor] is CoroutineDispatcher. When we use interceptContinuation(), it executes the DispatchedContinuation() in the CoroutineDispatcher class. What does the DispatchedContinuation() do?
overridefunresumeWith(result: Result) { // check if need to dispatch if (dispatcher.isDispatchNeeded(context)) { ...
// start dispatching dispatcher.dispatch(context, this) } else { // if not, execute the task as part of current event loop executeUnconfined(state, MODE_ATOMIC_DEFAULT) { withCoroutineContext(this.context, countOrElement) { continuation.resumeWith(result) } } } }
@Suppress("NOTHING_TO_INLINE")// we need it inline to save us an entry on the stack inlinefunresumeCancellable(value: T) { // check if need to dispatch if (dispatcher.isDispatchNeeded(context)) { ...
// start dispatching dispatcher.dispatch(context, this) } else { // if not, execute the task as part of current event loop executeUnconfined(value, MODE_CANCELLABLE) { if (!resumeCancelled()) { resumeUndispatched(value) } } } }
... }
This class intercepts the resumeCancellable and resumeWith functions. And it checks if this Coroutine will need to dispatch.
Alright, so far, we already know how to suspend, resume, and dispatch a Coroutine. It uses different layers to implement different functionalities. For example, BaseContinuationImpl implements the Coroutine resume, DispatchedContinuation implements the dispatching logic. Let’s move on to how to control the Job itself in the next post.
In the Introduction section, we already have a big picture of the Coroutines feature, including what it is, how we can use it, and other basic concepts. In this article, I will introduce another critical keyword suspend:
suspendfunpostItem(item: Item) { GlobalScope.launch { val token = requestToken() val post = createPost(token, item) processPost(post) } }
You can see the keyword suspend in front of a function. What does suspend keyword mean? If you compile the Kotlin code, you can see something like this:
1 2
// declaration: requestToken(kotlin.coroutines.Continuation<? super kotlin.Unit>) publicfinalrequestToken(Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
The code snap above is what we decompile into java code.
This is a bytecode/compiled-java-code sample for this suspend function - requestToken(). We can find there’s a parameter passed into requestToken(), which is a Continuation. What’s a Continuation?
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
/** * Interface representing a continuation after a suspension point that returns a value of type `T`. */ @SinceKotlin("1.3") publicinterfaceContinuation{ /** * The context of the coroutine that corresponds to this continuation. */ publicval context: CoroutineContext
/** * Resumes the execution of the corresponding coroutine passing a successful or failed [result] as the * return value of the last suspension point. */ publicfunresumeWith(result: Result) }
It’s an interface for resuming the suspended point. Here’s an example when you try to use CoroutineScope.Launch:
1 2 3 4 5
GlobalScope.launch { val token = requestToken() val post = createPost(token) processPost(post) }
Post post = (Post)var10000; MainActivity.this.processPost(post); return Unit.INSTANCE; }
In a Coroutine, it doesn’t use a callback function to return value. Instead, it uses the state machine to control what will respond to the next state. A suspend function will block the Coroutine when it’s in, but it wouldn’t block the thread. When it executes the suspend function, it won’t block the whole thread. The current thread can still do other jobs.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
Thread {
val context = newSingleThreadContext("Test Thread") GlobalScope.launch (context) { println("Start first coroutine.") delay(200) println("Resume first coroutine after delay 200ms") }
GlobalScope.launch (context){ println("Start second coroutine.") delay(100) println("Resume second coroutine after delay 100ms") }
Thread.sleep(500) }.start()
the result:
1 2 3 4
2019-08-17 21:25:22.437 I/System.out: Start first coroutine. 2019-08-17 21:25:22.473 I/System.out: Start second coroutine. 2019-08-17 21:25:22.574 I/System.out: Resume second coroutine after delay 100ms 2019-08-17 21:25:22.656 I/System.out: Resume first coroutine after delay 200ms
That’s why a Coroutine is very lightweight, and you can create it anytime without spending many resources on creating a Coroutine. After this, you might have a question: after resuming the suspended status, current Coroutine will still run on the same thread?
If you remember, I mentioned Dispatcher would decide which thread a Coroutine should run. There are two categories of dispatcher - confined & unconfined. The Default,Main, and IO belong to Confined category, and Unconfined is unconfined. Let’s create two different coroutines and use Confined & Unconfined dispatchers to see what will happen.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
Thread { GlobalScope.launch (Dispatchers.Main) { println("First coroutine runs in ${Thread.currentThread().name}") delay(100) println("First coroutine resumes and runs in ${Thread.currentThread().name}") }
GlobalScope.launch (Dispatchers.Unconfined){ println("Second coroutine runs in ${Thread.currentThread().name}") delay(100) println("Second coroutine resumes and runs in ${Thread.currentThread().name}") }
Thread.sleep(500) }.start()
Result:
1 2 3 4
2019-08-17 21:43:22.895 I/System.out: Second coroutine runs in Thread-4 2019-08-17 21:43:22.947 I/System.out: First coroutine runs in main 2019-08-17 21:43:23.005 I/System.out: Second coroutine resumes and runs in kotlinx.coroutines.DefaultExecutor 2019-08-17 21:43:23.050 I/System.out: First coroutine resumes and runs in main
When you use a confined dispatcher, no matter when you suspend the Coroutine, it would still run on the same thread after it resumes for the Dispatchers.Unconfined setting, it would first start on the caller thread but continue on a thread based on where the suspend function is.
So far, I introduce how the suspend function works, but it’s still worth going through how to create a Coroutine, start it and resume mechanism, especially it would be great to know how to switch different Coroutines. Let’s dig in a little bit deeper.
Go back to the CoroutineScope.Launch:
1 2 3 4 5 6 7 8 9 10 11 12 13
publicfun CoroutineScope.launch( context: CoroutineContext = EmptyCoroutineContext, start: CoroutineStart = CoroutineStart.DEFAULT, block: suspendCoroutineScope.() -> 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 general, the start parameter will be CoroutineStart.DEFAULT. It would go to here once the start function calls:
It creates a Coroutine, which is not intercepted without a receiver but with a result type T. This function creates a new and fresh instance of suspendable computation every time when it is invoked. To start executing this prepared Coroutine, it would call resume(Unit) on the returned Continuation instance. So, remember createCoroutineUnintercepted that will call resumeCancellable to execute a created Coroutine.
resumeCancellable function
1 2 3 4
internalfun Continuation.resumeCancellable(value: T) = when (this) { is DispatchedContinuation -> resumeCancellable(value) else -> resume(value) }
resume function
1 2
publicinlinefun Continuation.resume(value: T): Unit = resumeWith(Result.success(value))
In the resume function, it calls the resumeWith function in a Continuation. Let’s check how it works.
internalabstractclassBaseContinuationImpl( // This is `public val` so that it is private on JVM and cannot be modified by untrusted code, yet // it has a public getter (since even untrusted code is allowed to inspect its call stack). publicval completion: Continuation? ) : Continuation, CoroutineStackFrame, Serializable {
// This implementation is final. This fact is used to unroll resumeWith recursion. publicfinaloverridefunresumeWith(result: Result) { ...
BaseContinuationImpl is the implementation of the interface Continuation. You can see when a Coroutine tries to get the outcome, it will call invokeSuspend function which is generated by the compiler. If you still remember the example processPost we listed before, the compiled java code shows that the actual computation is in the invokeSuspend function. So the starting flow is:
In intercepted(), it tries to call context[ContinuationInterceptor]?.interceptContinuation(this) to get the Continuation, what’s the interceptContinuation()?
The CoroutineDispatcher implements this interface. Basically, the context[ContinuationInterceptor] is CoroutineDispatcher. When we use interceptContinuation(), it executes the DispatchedContinuation() in the CoroutineDispatcher class. What does the DispatchedContinuation() do?
overridefunresumeWith(result: Result) { // check if need to dispatch if (dispatcher.isDispatchNeeded(context)) { ...
// start dispatching dispatcher.dispatch(context, this) } else { // if not, execute the task as part of current event loop executeUnconfined(state, MODE_ATOMIC_DEFAULT) { withCoroutineContext(this.context, countOrElement) { continuation.resumeWith(result) } } } }
@Suppress("NOTHING_TO_INLINE")// we need it inline to save us an entry on the stack inlinefunresumeCancellable(value: T) { // check if need to dispatch if (dispatcher.isDispatchNeeded(context)) { ...
// start dispatching dispatcher.dispatch(context, this) } else { // if not, execute the task as part of current event loop executeUnconfined(value, MODE_CANCELLABLE) { if (!resumeCancelled()) { resumeUndispatched(value) } } } }
... }
This class intercepts the resumeCancellable and resumeWith functions. And it checks if this Coroutine will need to dispatch.
Alright, so far, we already know how to suspend, resume, and dispatch a Coroutine. It uses different layers to implement different functionalities. For example, BaseContinuationImpl implements the Coroutine resume, DispatchedContinuation implements the dispatching logic. Let’s move on to how to control the Job itself in the next post.