What are processes, threads, coroutines? There are many vivid examples on the web, and I personally believe that the purpose of all this fuss is to maximize resources.
Note: processes and threads are not what I want to focus on, coroutines are, but without mentioning the first two, it feels like eating 🍔🍟 without 🥤.
Sad stories for workers
This article is more nonsense, do not like spray, patience to see, maybe there will be unexpected harvest
process
In the case of a process, a well-utilized resource in this article is the CPU’s time resource.
There are many steps to complete a task, and many strategies to execute the steps. Let’s say your strategy is one by one: you have to do one step before you move on to the next.
If the CPU follows your strategy, it will be lying down while performing IO operations on registers, memory, and hard drives. Because the CPU is too fast, very fast, and the registers are slower, the memory is slower, the hard drive is slower, so the CPU just waits for the data to finish, and that’s where the time is wasted.
To make good use of the time the CPU waits. You can have the CPU do the other steps first. (The worker writes here with tears in his eyes)
Then put the steps, in different processes to execute, CPU in executing a step, encountered lengthy wait IO, went to perform another step, is another process, but at this moment and found a problem, CPU is the waiting time is used up, but some time not executing processes inside the code, It’s switching between the process environment.
Here’s an inappropriate example:
You go to zhang SAN, let Zhang SAN help you write a web page, but Zhang SAN said you mentioned the function of the technical need to research (IO time-consuming operation), let you wait for him 20 minutes. You see, I first go to Find Li Si, let Li Si settle the backstage agreement first, from Zhang SAN to Li Si home took 5 minutes, and the discussion with Li Si took 10 minutes. Then you go back to John, and when you get to John’s house, John is just finished.
20 minutes did not wait in Zhang SAN’s home in vain, but you spent 10 minutes in Zhang SAN to Li Si’s home and back, 10 minutes you can let Wang Wu to change a few pages of the manuscript, Zhao Liu to mention a few scenes, Lao Tze minutes hundreds of thousands of yuan, this Zhang SAN and Li Si can not live together? In this case, Both John’s and Tom’s houses belong to processes, and switching between processes takes a little time.
thread
You think your time is wasted on worthless things, so you rent a building and put Sams and Lams together on a single floor so it’s easy to find them! By the way, please call wang Wu and Zhao Liu. In this case, we think of the building as a process, and Joe and Joe as threads. You can find Tom and Tom a lot faster by now. (The migrant worker writes here, tears are already flowing down)
As your business gets bigger and more social, and you get annoyed with the people you work for and everyone is scrambling to talk to you (threading on some operating systems is preemptive), time management becomes your primary concern. So you ask a secretary, meeting things let the secretary (operating system) to arrange, each person only two minutes at most, of course, the high-level can directly find you, timeout is also allowed (some operating systems are high priority can be preemptive type, and ordinary threads are non-forcible type).
Under the arrangement of the secretary (scheduling), zhang SAN Li Si will come to your office according to the arrangement of the secretary, and you have a meeting, a time is so long, talk about the next time. But then you realize that John is next, but it takes him a little while to get from his desk to your office. How can you stand the waste of time? This example may not be a good example, but it is a problem of resource contention and thread context switching. When executing multiple threads, they must compete for the same resource. It’s like having multiple employees competing for your boss, so you need to lock shared resources, and locking involves switching thread states, which can be a waste of time. Thread switching is also a waste of time.
coroutines
So you’ve introduced an online meeting system where you just have to meet with the right people, as arranged by your secretary, reducing the amount of time they spend in your office. The secretary notifies one employee that he has arrived and keeps all the other employees waiting. It also wasted a little time.
Therefore, if employees are allowed to communicate with each other in advance, because they know when they have time, they can take the initiative to transfer the meeting time to others and propose the scheduled meeting time, so that they can have the meeting directly when the time is up, which saves the notice of the secretary and saves a little time. This is the act of the coroutine actively ceding CPU ownership. Of course, thanks to the introduction of online meetings, you may be in a meeting with 10,000 people, but to you, you’re all looking at the same screen. The analogy here may not be appropriate either, but remember that sometimes no matter how many coroutines there are, there may only be one thread for the CPU. So in this case, We’re not talking about threads anymore, we’re talking about coroutines.
That’s pretty much the story, and here we focus on the waste of resources on processes and threads, which is essentially a waste of time, most of which is spent in space, and some of which can be saved by proper scheduling.
So the concepts of process, thread and so on, I feel can be regarded as the concepts of time. For example: Shuang Zi earns 2.08 million yuan a day, then you can compare yi Shuang to 2.08 million yuan, because in a certain level, yi Shuang and 2.08 million yuan are equivalent. Of course you can draw inferences.
With the coroutine, it’s gonna be good. How’s she doing?
From the above, it is more efficient for people to voluntarily allocate CPU time chips in the case of long waits. You may wonder, which employees all want the same time? Does it make sense to give it away? Yes, because the worst case scenario is that everyone is on their own busy time (IO) without coordinated time, but this can be reduced by coordinated time. These are IO intensive scenarios, and adopting a collaborative approach can greatly reduce the time wasted.
The act of people voluntarily giving up is collaboration, rather than grabbing the time slice as we did in the beginning. Suspending means resuming after a certain period of time. This is similar to a thread state switch, but not necessarily a true thread state switch, as discussed below.
As you can see from the story above, it’s up to your secretary to set the meeting schedule for everyone from John to John, while it’s up to your employees to decide when to give up and when to hold the meeting. There’s a big difference. In the second way, your secretary can save a lot of time talking to employees.
If John’s job is to have meetings with you without writing code, drawing scripts, or drawing prototypes, the act of sending John online and John offline will actually affect the work efficiency of John and John, which is compute-intensive, and serial is more efficient than parallel. Interestingly, however, some languages do a good job of scheduling coroutines, in some cases not switching contexts as often.
Implementation differences of coroutines
One day, Mr. Sun, our partner, came to visit you and found that you had a very good way of managing employees. Also invited a secretary, the market many such management professional talents, the cultivation of these talents has a system, looking for a very simple, so sun general company quickly realized through the secretary to arrange meetings. This is to say that the API design of different operating system threads is roughly the same.
But he has a problem: he finds out that his employees aren’t very collaborative (or he finds out that his employees are very collaborative and proudly shows off to you). Employees differ in their positions, skills, personality, quality and other aspects. Without training, it is difficult to require all of them to achieve good coordination at one time, and the training methods and effects are also different. So some companies might end up collaborating on plan A, and some might end up collaborating on Plan B. The point here is that coroutines are language level things, and each language implements them differently and has different effects.
summary
The above is written from the point of view of life, the example may not be so appropriate, but at least we know that the process has made efforts to use resources wisely, but there is still room for improvement. So threads are here, threads are trying, but there’s still room for improvement. At that time, coroutine appeared and made efforts. As for the space for improvement, I am sorry that I am still in love with coroutine.
There are also stack coroutines and stack – free coroutines. Often said call stack call stack, there is a stack actually because some coroutines may have call relationship, there is a big guy said very well: stack coroutine and stack free coroutine
I’m an Android app developer, and kotlin’s coroutines are stackless coroutines, so here’s my understanding of kotlin’s stackless coroutines. The following point of view may be different from the above, there is a gap between the examples in life and the code implementation, so most ordinary people suddenly difficult to understand the program development, because there is no programming thinking.
To understand coroutines, you must understand the concept of closures, starting with the most familiar callback.
Magic: callback functions
When first introduced to callback functions, I’m sure a lot of people think they’re something special. It’s really flexible, it’s amazing! Let’s see:
We often call an object passed in as a parameter in a function, and then use that object’s function to do something:
class LogPrinter {
fun print(a){... }}fun printLog(logPrinter:LogPrinter) {
logPrinter.print();
}
Copy the codeBut you need to describe this behavior in advance with classes. But we know that classes can describe more than just a behavior!
In event-driven design at the same time, there will be a thread is waiting for events and consumption, is a “producer consumer model of concurrent, and assuming that there is only one customer, in some applications may be referred to as the main thread, that the main thread is producers (in the case of existence oneself give oneself events), and consumers. Of course, most producers are other threads:
The behavior and rules of events are not fixed, so if you use classes to describe a behavior in advance, isn’t it a bit of an overuse? Do I have to declare a bunch of classes in advance? Is there something you can define only for behavior, when you really want to use it?
Yeah, isn’t that a function? This is where the flexibility of the callback function comes in.
Inner class I’ve seen, what the hell is declaring and implementing a function inside a function? Closures!
Does Java support passing functions as arguments? Always support, but java8 is more obvious!
Who is the this object used by the passed function? Here’s a story to help you understand:
Tang’s monk drove monkey Sun away, but the great sage was still afraid of his master’s accident, so he pulled out three hairs and gave them to Eight Quit. He told Eight Quit to blow one hair as soon as something happened, and my Lao Sun would naturally appear.
At the same time, in order to ensure the continuous consumption of events, there should be no time-consuming operations in the main thread, and time-consuming operations are commonly in the calculation of data, data acquisition, etc. How can the main thread do not wait and get the results at the appropriate time for processing?
It’s amazing how a callback can solve all of these problems. Why is it so amazing?
Yes, I’ve done all this to see if you understand closures. If you don’t, it might be confusing, but if you do, congratulations, this is pretty simple for you!
If you don’t know, check out higher-order functions, closures, and Java’s anonymous inner classes.
What’s special about callback magic
I’m going to talk about the power of the callback magic by starting with thread synchronization, and using the magic of closures in coroutines as a practical example.
Here’s a scenario: Load a web page resource and draw it. Since loading a web page is a time-consuming IO operation, we usually transfer the time-consuming operation to another thread, as shown in the following figure:
The main thread needs a resource to draw the display, so it has to wait for thread B to complete, otherwise there will be nothing left for the main thread to paint. Although the main thread does not perform a time-consuming operation, it blocks and waits for thread B to wake itself up, which is a simple thread synchronization operation.
Art comes from life, and this happens in our daily life, and our solution is: Let me know when you’re done. So the main thread should not paint until thread B notifies itself that it has finished loading the resource.
But we usually do other things during this period rather than wait, so we want the main thread to continue consuming other events rather than blocking.
At this point we can use callbacks to implement non-blocking notifications.
The function callback we pass in does not implement a thread switch. You can assume that thread B does a thread switch when the invoke callback is wrapped up and placed on a queue waiting for the main thread to consume it. More on thread switches later. The loadResource method looks something like this:
fun loadResource(callBack:(Resource) - >Unit) {
thread {
val result = load() / / time consuming
// Call callback.invoke ()
sendToMainThread(){
callBack.invoke(result)
}
}
}
Copy the codeThis code does this: “Let me know when you’re done” and the main thread is not blocked! No thread state changes from executing this code! This saves the overhead of thread state switching.
summary
A closer look at the order of execution reveals something special:
In code without callbacks, the main thread is executed in the following order:
- Enter the showHtmlPage method
- Let thread B execute the loadResource method
- sleep
- Wake up again and draw
- Exit the showHtmlPage method
- Consume other Events
In code with callbacks: the main thread is executed in order
- Enter the showHtmlPage method
- Let thread B execute the loadResource method
- Exit the showHtmlPage method
- Consume other Events
- Thread B tells it to draw
Attention! In this case, the internal order of the showHtmlPage function is constant. The order must be:
- Enter the showHtmlPage method
- Execute the loadResource method
- Perform rendering
But the callbacks are executed in a different order relative to the other events in the queue (which are actually blocks of code). So if you think of coroutines as blocks of code to be executed, you’re right! You got the hang of it! The code to be executed is “wrapped” with callbacks, which is an important prerequisite for stackless coroutines to execute state flow! Callbacks can be executed at the right time! This is a good example of how threads can make execution order send changes between wrapped blocks of code!
Magic advanced
Let’s look at another example like this:
You need to fetch resources from different addresses, using the simplest synchronization method first.
The main thread is blocked twice, and the state of the thread is switched between the two threads. If the loadResourceA takes 10 seconds and the loadResourceB takes 10 seconds, the total time from loading to drawing is estimated to be 20.10 seconds.
Note :(the thread state switch takes 0.1 seconds, just to make it easier to understand, if you really want to know the thread state switch time, you can Google it)
fun showHtmlPage(a) {
val addrA = "A"
val addrB = "B"
loadResource(addrA) { resourceA ->
loadResource(addrB) { resourceB ->
paint(resourceFromA, resource)
}
}
}
Copy the codeOnce the blocking problem is solved, the main thread has time to do other things. Even assuming that CPU and other scheduling doesn’t take time, the total time from loading to drawing of the code block is 20 seconds, which is shortened by thread switching, which is not a lot of seconds. Because we actually load resources in serial.
At this point we continue to use concurrency and callbacks to make good use of resources:
Note :(the thread state switch takes 0.1 seconds, just to make it easier to understand, if you really want to know the thread state switch time, you can Google it)
fun showHtmlPage(a) {
val addrA = "A"
val addrB = "B"
var resourceFromA:Resource
var resourceFromB:Resource
loadResource(addrA) {resource ->
if(resourceFromB! =null) {
paint(resource,resourceFromB)
}else {
resourceFromA = resource
}
}
loadResource(addrB) {resource ->
if(resourceFromA! =null) {
paint(resourceFromA,resource)
}else {
resourceFromB = resource
}
}
}
Copy the codeI made the two loading data parallel, and the total time from loading to drawing was also 10 seconds, which is a fuzzy estimate. This is a shortened serial time, because both resources are being loaded at the same time.
At this point you may be wondering: are your two callbacks competing for resources? No! As mentioned earlier, the loadResource method does thread switching inside ！！！！
To better understand why thread switching is necessary, let’s first assume that there is a resource contention: invoke is invoked by a different thread, and you have to lock it: (I will simply lock it.)
fun showHtmlPage(a) {
val addrA = "A"
val addrB = "B"
var resourceFromA:Resource
varResourceFromB :Resource loadResource(addrA) {Resource -> synchronized (obj) {if(resourceFromB! =null) {
paint(resource,resourceFromB)
}else{resourceFromA = resource}} loadResource(addrB) {resource -> synchronized (obj) {if(resourceFromA! =null) {
paint(resourceFromA,resource)
}else {
resourceFromB = resource
}
}
}
}
Copy the codeAfter adding the lock, the total time from loading to drawing took 11.2s, which is a few tenths of a second more because of the lock. Synchronized is a mutex that blocks other threads. So to solve a problem, we introduced a new problem.
The loadResource method has a thread switch inside it. The two callbacks must be serial on the main thread. There is no such thing as multiple threads running resourceFromA and resourceFromB simultaneously.
summary
That’s interesting too, right? This is also an idea of using coroutines to solve resource competition: by adjusting the execution environment of the callback code block, the code block that changes the common resource is transferred to a thread to execute, so as to solve the problem of multi-thread resource competition.
However, implementing thread switching in the loadResource method is obviously not a good design, so we need to design a scheduler that specifies which thread the callback is executed on. Will this solve the problem of multi-threaded concurrent resource competition to some extent? Loading resources initializes two threads. Does it take 100 threads to load 100 resources? The creation and release of thread resources is another resource consuming operation. What about this?
Resolve competition for resources
If developers make proper use of the scheduler: callbacks that specify operations on common resources are executed on the same thread, there will be no resource contention issues. This allows programmers to specify on which thread the code snippet (callback) will be executed at write time:
LoadResource (UIThread, addrB) {resourceB -> paint(resourceFromA, resource)}Copy the codeSo the scheduler is more flexible if implemented for a threaded environment. Then we can design a separate scheduler for the above main thread that conforms to the following model:
Address the performance cost of frequently creating and destroying threads
I’m sure many programmers know how to solve the problem of frequently creating and releasing threads, and yes, thread pools.
As I mentioned above:
If Joe’s job is to hold meetings with you, without writing code, drawing scripts, or drawing prototypes, putting Joe on and Off the line will actually hurt Joe’s and Joe’s productivity
Can I choose a different scheduler based on the IO – or computation-intensive? Because only the developer knows which type.
So we can:
The thread pool for an operational scheduler can create threads based on the number of CPU cores.
A thread pool for an IO – intensive scheduler can create many threads.
There are three basic schedulers! If there are many methods like loadResource, do they have to do the thread switch themselves? It doesn’t fit into a single responsibility, does it?
Create new magic to make callbacks thread-switch
Attention! The following is the code I wrote in Java to help me understand the idea. Kotlin’s code for implementing coroutines is not exactly like this!
Why Java? Because it feels weird for me to use Kotlin for this example.)
Define a generic callback interface
Callbacks in Java are anonymous inner classes, so we declare an interface to instantiate to the consumer. As a generic callback, this callback allows the user to wrap their own time-synchronization code, so we need to return the value.
public interface FakerCallBack {
public Object invokeSuspend(a);
}
Copy the codeContinuation delivery style
A callback is used to send the result to the caller asynchronously. For example:
General writing:
public String getWelcomeSpeech(a) {
return "welcome";
}
public void sayWelcome(a) {
System.out.println(getWelcomeSpeech());
}
Copy the codeContinuation transfer style:
interface FakerCompletionResult {
public void resumeWith(Object value);
}
public void getWelcomeSpeech(FakerCompletionResult completion) {
completion.resumeWith("welcome");
}
public void sayWelcome(a) {
getWelcomeSpeech(value -> {
System.out.println(value)
});
}
Copy the codeSo specify a callback that can pass the result
public interface FakerContinuation {
public void resumeWith(Object obj);
}
Copy the codeA Runnable – “Job
We want to switch coroutines, which callback does the thread recognize? Yes, Runnable is also an interface. To distinguish it from threads, I define a Job:
public interface FakerJob {
public void start(a);
}
Copy the codeThe callback to intercept
The callback interface we defined above does not have the ability to specify threads, so we need to define an outer callback to wrap the block of code passed in by the user, giving it the ability to switch threads, just like AOP.
public interface FakerInterceptor {
public void dispatch(FakerJob callBack);
}
Copy the codeThe scheduler
The scheduler is the implementation of the interceptor. Why? The callback must be intercepted before it can be executed in the specified thread.
I wrote two schedulers,
-
Fixed 20 thread pool implementation of IO intensive scheduler.
-
Use Handler to implement a scheduler that switches to main thread execution.
Unless you’re an Android developer, you probably don’t know what Handler is. Then you can write a single thread pool scheduler, and then implement their own message queue, as for how to implement delayed event execution, to simulate the thread sleep function. Just keep one idea in mind: delay is not blocking, it’s just suspending this block of code and letting the thread consume another block of code and come back when it’s time to execute. Of course, if the thread is already idle, the message queue consumers will also wait, it is not written to refer to the Android handler implementation.
public class IODispatchers implements FakerInterceptor {
private ExecutorService executor = Executors.newFixedThreadPool(20);
@Override
public void dispatch(FakerJob callBack) { executor.execute(callBack::start); }}public class MainDispatchers implements FakerInterceptor {
private Handler executor = new Handler(Looper.getMainLooper());
@Override
public void dispatch(FakerJob callBack) { executor.post(callBack::start); }}Copy the codeWrite a scheduler declaration class:
public class FakerDispatchers {
public static FakerInterceptor IO = new IODispatchers();
public static FakerInterceptor Main = new MainDispatchers();
}
Copy the codeThe callback structure
We also need to define a callBack constructor for the external API, so that users can easily use this powerful callBack, some do not need to perform a result or time-consuming operation, callBack can be directly passed null:
public class FakerCompletionBuilder {
public static void launch(FakerInterceptor dispatchers,FakerCallBack callBack, FakerContinuation fakerContinuation) {
dispatchers.dispatch(new FakerJob() {
@Override
public void start(a) {
if(callBack! =null) {
fakerContinuation.resumeWith(
callBack.invokeSuspend()
);
}else {
fakerContinuation.resumeWith(null); }}}); }}Copy the codeSpecifying thread execution
It’s a mule or a horse! Pull it out! I run the Activity code directly on the android real machine, kotlin is too lazy to change the code, similar, you should understand:
Modified in 2021.6.21
fun launchTest(a) {
FakerCompletionBuilder.launch(FakerDispatchers.IO, {
Log.d(">>> MainActivity"."launchTest ${Thread.currentThread().name}")
val r = loadResult("addrA")
}) {
FakerCompletionBuilder.launch(
FakerDispatchers.Main, null
) {
Log.d(">>> MainActivity"."launchTest r = $r ${Thread.currentThread().name}")}}}private fun loadResult(addr: String): String? {
try {
if (addr == addrA) {
Thread.sleep(1000)}else {
Thread.sleep(3000)}}catch (e: InterruptedException) {
e.printStackTrace()
}
return "load end $addr"
}
Copy the codeNote that thread. sleep is used to simulate elapsed time, so this is not a Thread state change. Running results:
>>> MainActivity: launchTest pool-1-thread-1
>>> MainActivity: launchTest result = load end addrA main
Copy the codeThe thread switches smoothly!
Competition for resources
Since this is an Android application, if the main thread blocks and waits, the result will be error, so I add a single thread scheduler:
public class SingleDispatchers implements FakerInterceptor {
private ExecutorService executor = Executors.newSingleThreadExecutor();
@Override
public void dispatch(FakerJob callBack) { executor.execute(callBack::start); }}public class FakerDispatchers {
public static FakerInterceptor IO = new IODispatchers();
public static FakerInterceptor Main = new MainDispatchers();
public static FakerInterceptor Single = new SingleDispatchers();
}
Copy the codeUse:
var i = 0
var j = 0
fun testResourceCompetition(a){(1.10000.).forEach {
load()
}
Thread.sleep(10000)
println("i $i")
println("j $j")}fun load(a) {
FakerCompletionBuilder.launch(FakerDispatchers.IO, null) {
j++
FakerCompletionBuilder.launch(FakerDispatchers.Single,null) {
i++
}
}
}
Copy the codeResults:
I/System.out: i 10000
I/System.out: j 9869
Copy the codeThe value of j is different every time, and the value of I is always 10000. Don’t need to lock to solve the resource competition, interesting!
Load multiple asynchronous resources
If we use the FakerCompletionBuilder implemented above to load resources in parallel:
fun loadDouble(a) {
var resultA: String? = null
var resultB: String? = null
FakerCompletionBuilder.launch(FakerDispatchers.IO, { loadResource(addrA) }) {result ->
FakerCompletionBuilder.launch(FakerDispatchers.Main, null) {
resultA = result.toString()
if(resultB ! =null) {
Log.d(">>> MainActivity"."result = $resultA $resultB")
}
}
}
FakerCompletionBuilder.launch(FakerDispatchers.IO, { loadResource(addrB) }) {result->
FakerCompletionBuilder.launch(FakerDispatchers.Main, null) {
resultB = result.toString()
if(resultA ! =null) {
Log.d(">>> MainActivity"."result = $resultA $resultB")}}}}Copy the codesummary
Implementing a simple thread-switch callback is not difficult, and the last example shows how to synchronize the results of two asynchronous threads. The real core code is executed in this order:
// We want to run these two in parallel
loadResource(addrA)
loadResource(addrB)
// But in the end it must be:
Log.d(">>> MainActivity"."result = $resultA $resultB")
Copy the codeTo achieve this effect, I use a callback to separate the three statements and wrap them separately, saying: I put the statements in different coroutines.
The coroutine of loadResource is executed by different threads, and the resulting coroutine is executed at the end using control flow.
Recall that the order of these callback blocks can be influenced by threads. However, from the last example, you can also use control flow to get the desired sequence of results from an out-of-order coroutine. No blocking, no locking! Interesting!
As you can see from the above code, the conditional statement has some duplicate code:
resultB = result.toString()
if(resultA ! =null) {
Log.d(">>> MainActivity"."result = $resultA $resultB")
}
resultA = result.toString()
if(resultB ! =null) {
Log.d(">>> MainActivity"."result = $resultA $resultB")}Copy the codeDo you think I should pull out a function? No! Don’t! Because I said in the callback function summary: I want to declare the desired function in a function, not declare another function in a class. Simply put: you don’t want to contaminate the class to which this function belongs in order to change its internal behavior! So let me solve this duplication of code.
When a callback meets a Switch, what sparks?
When there are many statements controlling conditions, we use the switch statement, and the control condition can give a status value. Yes, that’s the state machine:
public void test(state : Int) {
switch (state) {
case 0: {
loadResource(addrA);
break;
}
case 1: {
loadResource(addrB)
break;
}
case 2: {
Log.d(">>> MainActivity"."result = $resultA $resultB")
break; }}}Copy the codeYou can control the order in which these lines of code are executed by passing in the value of state.
But who calls this method? Who stores conversion values?
- Scenario 1: Write an external loop that changes the value of the state and calls this method
But don’t declare new functions? How to do this without defining a new member variable?
- Yes, closures! The callback. Anonymous inner class! Nice! We can write this:
public void loadDouble(a) {
// Instantiate an anonymous inner class, called a callback, that wraps the code inside the entire loadDouble method
FakerContinuation continuation= new FakerContinuation() {
// Initialization state
int state = 0;
// Receive two results
String resultA = null;
String resultB = null;
@Override
public void resumeWith(Object obj) {
switch (state) {
case 0: {
// Wrap the code to load A and throw it to the child thread. The launch method was not seen earlier, but will be described below
FakerCompletionBuilder.launch(FakerDispatchers.Main, FakerDispatchers.IO, this.new FakerCallBack() {
@Override
public Object invokeSuspend(a) {
// Load resource A
return loadResource("A"); }},new FakerContinuation() {
@Override
public void resumeWith(Object obj) {
// After loading resource A, store the valueresultA = obj.toString(); }}); FakerCompletionBuilder.launch(FakerDispatchers.Main, FakerDispatchers.IO,this.new FakerCallBack() {
@Override
public Object invokeSuspend(a) {
// Load resource B
return loadResource("B"); }},new FakerContinuation() {
@Override
public void resumeWith(Object obj) {
// After loading resource B, store the valueresultB = obj.toString(); }});// Change the state to 1
state = 1;
return; Exit the outermost resumeWith method
}
case 1: {
// The next time the resumeWith method is called, it enters the state and decides who will call it next, as described below
if (resultA == null || resultB == null) {
return; }}}// Output the result
Log.d("> > >", resultA + ""+ resultB); }};Execute the parent closure manually
continuation.resumeWith(null);
}
Copy the codeWe declared the member variables inside the closure by using the features of closure, and then started two coroutines to load resources. However, when the state was 0, we had already returned. Why did we re-enter switch?
That’s because I’ve added a new way to start coroutines:
// Pass the parent closure scheduler, and the parent closure
public static void launch(FakerInterceptor dispatchersParen, FakerInterceptor dispatchers, FakerContinuation continuationParen, FakerCallBack callBack, FakerContinuation continuation) {
dispatchers.dispatch(new FakerJob() {
@Override
public void start(a) {
Object obj = callBack.invokeSuspend();
continuation.resumeWith(obj);
// Notice that the parent closure's resumeWith is reexecuted using the parent closure's scheduler and passed the result, so the switch can execute
dispatchersParen.dispatch(new FakerJob() {
@Override
public void start(a) { continuationParen.resumeWith(obj); }}); }}); }Copy the codeWhat happened to no new methods? Ha ha, I have added a general coroutine start method, to achieve the above effect, the best way is to add this method! And this method is defined in the FakerCompletionBuilder, without contaminating the original classes and functions.
Why a parent closure scheduler switch? Of course! Because we expect the following to happen:
suspend fun showHtmlPage (a) {
val resultA = async{loadResource(addrA)} // Asynchronous thread execution
val resultB = async{loadResource(addrB)} // Asynchronous thread execution
log(resultA,resultB) // Main thread execution
}
Copy the codeNo lock! Judgment statement is no longer duplicate code! See! Switch and callback together, it is quite convoluted, but very interesting, right? This section is actually very close to the coroutine, the idea has been reflected, I suggest to savor it again!!
summary
The switch divides the statements that need to be executed asynchronously and synchronously into different closures, changes the value of the lable (changes the execution state) after each closure, and then re-calls the parent closure’s callback method through the child closure. When the parent closure’s resumeWith method reenters, the state changes. So a different code block is executed than the last one. In the parent closure, the implementation of the various code blocks execution state flow, and then achieve the synchronization of performance results.
When you think about it, you take an entire code block, break it up into different blocks based on the developer’s declaration (for example, launch{}), wrap it in closures, and then adjust the state for each block, changing the order in which all the blocks are executed. Attention! The developer’s claim is highlighted here, and it’s clear that the chunking is actually up to the developer! That’s what the program decides!
This confirms the above: the code to be executed is “wrapped” with a callback and executed at the right time! This is a good example of how thread callbacks cause code to execute in sequence! It is also one of the important prerequisites for stackless coroutine implementation of state flow!
The parent closure starts the child closure, and the child closure calls the parent closure again, which is essentially a special loop! But is it recursion? I’ll leave that to you!
depressed
But there are two areas of frustration:
1. Switch to the parent closure’s scheduler to call back the resumeWith method, and my main thread scheduler is throwing the closure into the queue. If all the closures are in the main thread, it will take a while to execute.
2. Why does the switch assign a value to the parent closure’s resumeWith method when we already return the result from the parent closure?
// Pass the parent closure scheduler, and the parent closure
public static void launch(FakerInterceptor dispatchersParen, FakerInterceptor dispatchers, FakerContinuation continuationParen, FakerCallBack callBack, FakerContinuation continuation) {...// Notice that the parent closure's resumeWith is reexecuted using the parent closure's scheduler and passed the result, so the switch can execute
dispatchersParen.dispatch(new FakerJob() {
@Override
public void start(a) {
// here!!continuationParen.resumeWith(obj); }}); }}); }Copy the code. },new FakerContinuation() {
@Override
public void resumeWith(Object obj) {
// After loading resource A, store the valueresultA = obj.toString(); }}); .Copy the codeThat’s because I don’t know who sent in the results, I can’t confirm whether it’s result A or result B! Simply put: I don’t know the order of callbacks to A and B!
This is not an actual implementation of the Kotlin coroutine, this is an example I wrote myself, why is there such a weird implementation? Don’t worry because you don’t want to lose await so early. This problem will be solved after we get to await!
Is it worth suspending
Switch to the parent closure’s scheduler to call back the resumeWith method, because my main thread scheduler is throwing the closure into the queue. If it’s all in the main thread, and it takes a while to execute, is it necessary to wait?
Yeah, I don’t think that’s necessary! So kotlin has flag declarations that are possible to hang! Instead of definitely hanging! Such as the suspend keyword at the beginning of the function and the return value of the closure execution.
Yi! Suspend is simply not executing the closure for the time being!
So we can use the simplest way to simulate, the premise is the same thread, please remember this!
Since the return value of launch in Kotlin is a job, I think async is more appropriate. Kotlin’s implementation doesn’t work like this, but to make it easier for you to see why suspension is not necessary and to illustrate the idea of suspension in a simpler way, I wrote this example by hand. Please be aware:
// Define a flag bit
final int SUSPEDN = 1;// Add a return value
public static Object async(FakerInterceptor dispatchersParen, FakerInterceptor dispatchers, FakerContinuation continuationParen, FakerCallBack callBack, FakerContinuation continuation) {
// Notice here that if the scheduler is not equal to the parent coroutine's scheduler, it is suspended
if(dispatchers ! = dispatchersParen) { dispatchers.dispatch(new FakerJob() {
@Override
public void start(a) {
Object obj = callBack.invokeSuspend();
continuation.resumeWith(obj);
dispatchersParen.dispatch(new FakerJob() {
@Override
public void start(a) { continuationParen.resumeWith(obj); }}); }});// Return the flag bit
return SUSPEDN;
} else { // Otherwise, there is no need to suspend
Object obj = callBack.invokeSuspend();
continuation.resumeWith(obj); // Child closures also pass results
return obj; // Execute directly, return the result directly}}Copy the codeReturn a flag bit that tells the switch whether to switch the state.
public void loadDouble(a) {
FakerContinuation continuation= new FakerContinuation() {
int state = 0;
String resultA = null;
String resultB = null;
Object resultLaunchB ;
@Override
public void resumeWith(Object obj) {
switch (state) {
case 0: {
Object resultLaunchA = FakerCompletionBuilder.async(FakerDispatchers.Main, FakerDispatchers.Main, this.new FakerCallBack() {
@Override
public Object invokeSuspend(a) {
return loadResource("A"); }},new FakerContinuation() {
@Override
public void resumeWith(Object obj) { resultA = obj.toString(); }}); resultLaunchB = FakerCompletionBuilder.async(FakerDispatchers.Main, FakerDispatchers.IO,this.new FakerCallBack() {
@Override
public Object invokeSuspend(a) {
return loadResource("B"); }},new FakerContinuation() {
@Override
public void resumeWith(Object obj) { resultB = obj.toString(); }}); state =1;
// Notice that this determines whether to skip the closure and wait for the next state switch, or continue executing the code
if (resultLaunchA == SUSPEDN) {
return;
}else {
resultA = resultLaunchA.toString();
}
if (resultLaunchB == SUSPEDN) {
return;
}else{ resultB = resultLaunchB.toString(); }}case 1: {
if (resultA == null || resultB == null) {
return;
}
break;
}
}
Log.d("> > >", resultA + ""+ resultB); }}; continuation.resumeWith(null);
}
Copy the codeThe return value of async determines whether to continue jumping out of the switch.
- Should two pieces of code be divided into two code blocks and executed sequentially;
- Or just execute it without separating the two pieces of code.
Now we know that the suspension is not certain, depending on the conditions, in Kotlin is declared by the return value.
Realize the delay
But there’s one thing that’s bound to hang! That’s delay! Delay is a perfect example of a suspended action.
We know that Thread. Sleep blocks. Why does Delay not block? It’s hanging? If you already understand the above, you already know the answer! Don’t understand it doesn’t matter, look down maybe it will be clear!
We want to achieve this effect:
log(1)
delay(1000) // The thread is not blocked
log(2)
Copy the codeAttention! I’m going to use the main thread scheduler implementation here, because it’s simple, and why is it simple? Android developers will see this and call it a pro!
Change the scheduler for the main thread first:
public class MainDispatchers implements FakerInterceptor {
private Handler executor = new Handler(Looper.getMainLooper());
@Override
public void dispatch(FakerJob callBack) {
executor.post(callBack::start);
}
// Yes, it is postDelayed. So the producer-consumer model is really awesome!
@Override
public void dispatch(Long time,FakerJob callBack) { executor.postDelayed(callBack::start,time); }}Copy the codeCreate a delay constructor:
public static Object delay(Long time, FakerInterceptor dispatchersParen, FakerContinuation continuationParen) {
dispatchersParen.dispatch(time, new FakerJob() {
@Override
public void start(a) {
// The delay endscontinuationParen.resumeWith(SUSPEDN_FINISH); }});/ / delay
return SUSPEDN;
}
Copy the codeUse:
public void test(a) {
FakerContinuation continuation = new FakerContinuation() {
int state = 0;
Object resultLaunchB;
@Override
public void resumeWith(Object obj) {
switch (state) {
case 0: {
Log.d("> > >"."1");
state = 1;
// Notice that this is a reference to the parent closure, so the delayed task is to re-call the parent closure's resumeWith method
Object result = FakerCompletionBuilder.delay(1000L, FakerDispatchers.Main, this);
if (result == SUSPEND) {
return; }}case 1: {
Log.d("> > >"."2"); }}}}; continuation.resumeWith(null);
}
Copy the codeSimple, is to join the queue, and then queue to go! When the time is up, it returns to the parent closure’s resumeWith and enters state 2. Here it is as it was in the story at the beginning of this article:
Log2 voluntarily gives up 1 second! Then the main thread executes other events in the consumption queue!
Implement await
Let’s take a look at another question mentioned above:
Why would a switch assign a value to the parent closure’s resumeWith method when we already return the result via the parent closure’s resumeWith?
Because it is not possible to control the return of loaded resources A and B without adding A flag bit to the return! Instead, I’ll give an object to store asynchronous results and notify the parent closure when the results arrive. What does that mean?
Most Java programmers know about futures because runnable returns no results, just as we did with Job:
public interface FakerJob {
public void start(a);
}
Copy the codeSo Java provides a Future to retrieve the result of runnable execution, but the Future is blocked, and we want to implement a non-blocking Future:
public interface LightFuture {
public Object await(FakerContinuation continuation);
}
Copy the codeDoes it smell like deferred? Take a look at the implementation, it’s all in the comment:
public class LightFutureImpl implements LightFuture.FakerContinuation {
// status, whether the asynchronous coroutine has completed execution
public boolean isCompleted = false;
// Stores the results of asynchronous coroutine execution
public Object result;
// Reference to the parent closure
public FakerContinuation continuation;
// This method is familiar, pass in the parent closure, if the result is already there, return directly, if not, tell the parent closure that it should be suspended!
public Object await(FakerContinuation continuation) {
this.continuation = continuation;
if (isCompleted) {
return result;
}
return FakerContinuation.SUSPEDN;
}
// The asynchronous coroutine informs the Future that the result has been fetched
@Override
public void resumeWith(Object obj) {
isCompleted = true;
result = obj;
// The parent coroutine does not save the result, but does not call back
if(continuation ! =null) {
// Notify the parent that the coroutine result has been obtained and that it is ready to proceed to the next statecontinuation.resumeWith(obj); }}}Copy the codeDefine our async constructor:
public static LightFuture async(FakerInterceptor dispatchersParen, FakerInterceptor dispatchers, FakerCallBack callBack) {
Instantiate a LightFutureImpl
LightFutureImpl lightFuture = new LightFutureImpl();
dispatchers.dispatch(new FakerJob() {
@Override
public void start(a) {
Object result = callBack.invokeSuspend();
// Note that this switches back to the parent coroutine scheduler
dispatchersParen.dispatch(new FakerJob() {
@Override
public void start(a) {
// Notice here that after the asynchronous coroutine completes, the result is stored in LightFutureImpllightFuture.resumeWith(result); }}); }});return lightFuture;
}
Copy the codeUse:
public void test(a) {
FakerContinuation continuation = new FakerContinuation() {
int state = 0;
// Get the Future of result B
LightFuture futureB;
Object resultB;
@Override
public void resumeWith(Object obj) {
// Note that a scope lable17 is defined here
lable17:
{
// Store the results passed by other coroutines
resultB = obj;
switch (state) {
case 0: {
// Start the coroutine and load resource A
LightFuture futureA = FakerCompletionBuilder.async(FakerDispatchers.Single, FakerDispatchers.IO, new FakerCallBack() {
@Override
public Object invokeSuspend(a) {
return loadResource("A"); }});// Start the coroutine and load resource B
futureB = FakerCompletionBuilder.async(FakerDispatchers.Single, FakerDispatchers.IO, new FakerCallBack() {
@Override
public Object invokeSuspend(a) {
return loadResource("B"); }});// Change the state to 1
state = 1;
// Try to fetch the result, notice that a reference to the parent closure is passed in
Object result = futureA.await(this);
// Suspend the parent closure and wait for the next state to arrive
if (result == FakerContinuation.SUSPEDN) {
return; }}// Coroutine A is executed and enters this state, but we do nothing to jump out of the switch. Why is it coroutine A
case 1: {
break;
}
case 2: {
// Jump to lable17, note lable17!
breaklable17; }}// From state 1 to here, output the result of coroutine A's execution
System.out.println("> > >"+obj.toString());
// State is cut to 2
state = 2;
// Try to fetch the result, notice that a reference to the parent closure is passed in
resultB = futureB.await(this);
// The parent closure is suspended until the coroutine B calls back to state 2
if (resultB == FakerContinuation.SUSPEDN) {
return; }}// lable17 end
// Select * from state 2; // Select * from state 2
System.out.println("> > >"+resultB.toString()); }}; continuation.resumeWith(null);
}
private String loadResource(String addr) {
try {
if (addr.equals("A")) {
Thread.sleep(1000);
} else {
Thread.sleep(2000); }}catch (InterruptedException e) {
e.printStackTrace();
}
return "load end " + addr;
}
Copy the codeAmazing, isn’t it? Why does coroutine A have to go into state 1?
Because we didn’t call it before state 1:
// Try to fetch the result, notice that a reference to the parent closure is passed in
resultB = futureB.await(this);
Copy the codeSo coroutine B doesn’t call the resumeWith method on the parent closure even if it’s done, because we have a nullation remember? :
public class LightFutureImpl implements LightFuture.FakerContinuation {
public boolean isCompleted = false;
public Object result;
public FakerContinuation continuation;
public Object await(FakerContinuation continuation) {
this.continuation = continuation;
if (isCompleted) {
return result;
}
return FakerContinuation.SUSPEDN;
}
@Override
public void resumeWith(Object obj) {
isCompleted = true;
result = obj;
/ / here
if(continuation ! =null) { continuation.resumeWith(obj); }}}Copy the codeSo state 1 must have been entered by coroutine A!
You’re just going to temporarily store the result somewhere else, either directly when you fetch it, or you’re going to hang it until the parent closure comes in. Results are synchronized but threads are not blocked. A simple version of async is implemented!
Attention! Kotlin’s await implementation is complex, and there are many debates on the web about blocking or not blocking
- Some people say “await” is dead while blocking :(I disagree)
@Override
public void resumeWith(@NotNull Object result) {
synchronized (this) {this.result = result;
notifyAll(); // The coroutine has ended, notifies wait() below to stop blocking}}public void await(a) throws Throwable {
synchronized (this) {while (true){
Object result = this.result;
if(result == null) wait(); Object.wait() is called, which blocks the current thread and returns when notify or notifyAll is called
else if(result instanceof Throwable){
throw (Throwable) result;
} else return; }}}Copy the code- Some people say that compilation turns synchronized code into a callback (amazing operation, but I don’t agree either)
summary
Kotlin’s LightFuture is a Deferred implementation that inherits the Job CompletableDeferredImpl because the Job has a state and the LightFuture has an isCompleted state. But I didn’t want it to be too complicated, so I didn’t want it to be too complicated to inherit FakerJob.
Kotlin’s source code for await is extremely complex and confusing to look at, and because of the dark magic I feel silly looking at the source code for Kotlin layer. Here you may have two questions:
- How do I prove that the implementation of await in this article is correct?
- If you wanted to implement coroutines in Java every time, wouldn’t you have to write switch? Isn’t the code long and stinky?
Black magic
Answer question 1:
How do I prove that the implementation of await in this article is correct? I do not guarantee that I am correct. I read the Java code after The Decompile of Kotlin code and assumed that this might be the way to implement it. The decompile Java code and The Kotlin code are compared below.
Answer question 2:
No, Java doesn’t support coroutines, but since coroutines are language level stuff, I was able to implement coroutines entirely from Java code. You can see why it was so easy for Kotlin to write a coroutine after decompile Java code is shown below.
Black magic is when the compiler does something to the Kotlin code we wrote, for example:
suspend fun showHtmlPage(a) = runBlocking {
val resultA = async { loadResource("addrA") }
val resultB = async { loadResource("addrB") }
Log.d("> > >", resultA.await().toString()+resultB.await().toString())
}
Copy the codeThe Java code after Decompile looks like this:
If you understand all of the ideas above, this part of the code will not be difficult to understand, give me some patience:
public final Object showHtmlPage(@NotNull Continuation $completion) {
return BuildersKt.runBlocking$default((CoroutineContext)null, (Function2)(new Function2((Continuation)null) {
// $FF: synthetic field
private Object L$0;
Object L$1;
Object L$2;
int label;
@Nullable
public final Object invokeSuspend(@NotNull Object $result) {
Object var10000;
String var5;
StringBuilder var6;
Object var7;
label17: {
Object var8 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
Deferred resultB;
switch(this.label) {
case 0:
ResultKt.throwOnFailure($result);
CoroutineScope $this$runBlocking = (CoroutineScope)this.L$0;
/ / attention to this
Deferred resultA = BuildersKt.async$default($this$runBlocking, (CoroutineContext)null, (CoroutineStart)null, (Function2)(new Function2((Continuation)null) {
int label;
@Nullable
public final Object invokeSuspend(@NotNull Object var1) {
Object var2 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
switch(this.label) {
case 0:
ResultKt.throwOnFailure(var1);
return MainActivity.this.loadResource("addrA");
default:
throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine"); }}@NotNull
public final Continuation create(@Nullable Object value, @NotNull Continuation completion) {
Intrinsics.checkNotNullParameter(completion, "completion");
Function2 var3 = new <anonymous constructor>(completion);
return var3;
}
public final Object invoke(Object var1, Object var2) {
return ((<undefinedtype>)this.create(var1, (Continuation)var2)).invokeSuspend(Unit.INSTANCE); }}),3, (Object)null);
// Notice here
resultB = BuildersKt.async$default($this$runBlocking, (CoroutineContext)null, (CoroutineStart)null, (Function2)(new Function2((Continuation)null) {
int label;
@Nullable
public final Object invokeSuspend(@NotNull Object var1) {
Object var2 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
switch(this.label) {
case 0:
ResultKt.throwOnFailure(var1);
return MainActivity.this.loadResource("addrB");
default:
throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine"); }}@NotNull
public final Continuation create(@Nullable Object value, @NotNull Continuation completion) {
Intrinsics.checkNotNullParameter(completion, "completion");
Function2 var3 = new <anonymous constructor>(completion);
return var3;
}
public final Object invoke(Object var1, Object var2) {
return ((<undefinedtype>)this.create(var1, (Continuation)var2)).invokeSuspend(Unit.INSTANCE); }}),3, (Object)null);
var6 = new StringBuilder();
var5 = "> > >";
this.L$0 = resultB;
this.L$1 = var5;
this.L$2 = var6;
this.label = 1;
// Notice here!
var10000 = resultA.await(this);
if (var10000 == var8) {
return var8;
}
break;
case 1:
var6 = (StringBuilder)this.L$2;
var5 = (String)this.L$1;
resultB = (Deferred)this.L$0;
ResultKt.throwOnFailure($result);
var10000 = $result;
break;
case 2:
var6 = (StringBuilder)this.L$1;
var5 = (String)this.L$0;
ResultKt.throwOnFailure($result);
var10000 = $result;
// Notice here
break label17;
default:
throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
}
var7 = var10000;
var6 = var6.append(String.valueOf(var7));
var5 = var5;
this.L$0 = var5;
this.L$1 = var6;
this.L$2 = null;
// Notice here
this.label = 2;
var10000 = resultB.await(this);
if (var10000 == var8) {
return var8;
}
}
var7 = var10000;
// Notice here
return Boxing.boxInt(Log.d(var5, var6.append(String.valueOf(var7)).toString()));
}
@NotNull
public final Continuation create(@Nullable Object value, @NotNull Continuation completion) {
Intrinsics.checkNotNullParameter(completion, "completion");
Function2 var3 = new <anonymous constructor>(completion);
var3.L$0 = value;
return var3;
}
public final Object invoke(Object var1, Object var2) {
return ((<undefinedtype>)this.create(var1, (Continuation)var2)).invokeSuspend(Unit.INSTANCE); }}),1, (Object)null);
}
private final String loadResource(String addr) {
try {
if (Intrinsics.areEqual(addr, "A")) {
Thread.sleep(2000L);
} else {
Thread.sleep(1000L); }}catch (InterruptedException var3) {
var3.printStackTrace();
}
return "load end " + addr;
}
Copy the codeYou can see that no matter what you do in the coroutine, it generates a switch statement, I guess for the sake of unity.
Is it similar to what I achieved above? So how to realize async in the end, I drew the conclusion here, and wrote the test code, and the result is more consistent with Kotlin’s async. I’ve been working here for less than a year and my skills are limited, so the real way to achieve this is through analysis and reasoning.
If you don’t worry about the implementation, but the idea, I’m glad you got the idea I wanted the reader to get: store the results, call back to the parent coroutine when appropriate, and re-execute the code block corresponding to the state through the parent coroutine’s state machine.
Through this idea, I realized the flow of several coroutines, as well as the synchronization of execution results, and no blocking!
“
Kotlin’s coroutines have many more powerful features that I haven’t mentioned, because the idea for this article was simple: To see how Kotlin’s coroutines are implemented using closures. I think the above is very simple, even if the expression is not good, you can run the code can suddenly understand! I really tried my best to iron children 😭, I hope you read this more than 10,000 words can have a harvest!