Source code analysis of kotlin collaborative process startup

Reference article

Concept :

CoroutineScope

 * Defines a scope for new coroutines. Every **coroutine builder** (like [launch], [async], etc)
 * is an extension on [CoroutineScope] and inherits its [coroutineContext][CoroutineScope.coroutineContext]
 * to automatically propagate all its elements and cancellation.
 * 
 Define the scope of the new process . Each collaboration builder （ Such as launch 、 async etc. ） All are CoroutineScope An extension of , And inherit it coroutineContext To automatically propagate all its elements and cancel .

 * Every coroutine builder (like [launch], [async], etc)
 * and every scoping function (like [coroutineScope], [withContext], etc) provides _its own_ scope
 * with its own [Job] instance into the inner block of code it runs.
 * By convention, they all wait for all the coroutines inside their block to complete before completing themselves,
 * thus enforcing the structured concurrency. See [Job] documentation for more details.
 * 
 Each collaboration builder （ Such as launch 、 async etc. ） And each scope function （ Such as coroutineScope 、 withContext etc. ） Both provide their own scope and their own Job example . By convention , they   Wait for all the processes in its block to complete    Finish yourself again , thus   Enforce structured concurrency .

The routine code starts the process

        CoroutineScope(Dispatchers.IO).launch {
        
         }

Enclosed kotlin bytecode decomplie Code

      BuildersKt.launch$default(CoroutineScopeKt.CoroutineScope((CoroutineContext)Dispatchers.getIO()), (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 Unit.INSTANCE;
            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);
      

launch start-up kotlin The source implementation is as follows :

 Parameters ：
context -  It's co-operative CoroutineScope.coroutineContext Additional content of context .
start -  Co process startup options . The default value is CoroutineStart.DEFAULT .
block -  The coroutine code that will be called in the context of the scope provided .

public fun CoroutineScope.launch(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> Unit
): Job {
// 1. Create context for new collaboration . It does not specify another scheduler or ContinuationInterceptor Installed at Dispatchers.Default ,
// And add optional support for debugging tools （ When open ）.
    val newContext = newCoroutineContext(context)  
    //2  establish coroutine , It's usually StandaloneCoroutine
    val coroutine = if (start.isLazy)
        LazyStandaloneCoroutine(newContext, block) else
        StandaloneCoroutine(newContext, active = true)
        //3  Start the coroutines , CoroutineStart.DEFAULT  By default startCoroutineCancellable .
    coroutine.start(start, coroutine, block)
    return coroutine
}

CoroutineScope spread function launch

1.newCoroutineContext

@ExperimentalCoroutinesApi
public actual fun CoroutineScope.newCoroutineContext(context: CoroutineContext): CoroutineContext {
    val combined = coroutineContext + context
    val debug = if (DEBUG) combined + CoroutineId(COROUTINE_ID.incrementAndGet()) else combined
    return if (combined !== Dispatchers.Default && combined[ContinuationInterceptor] == null)
        debug + Dispatchers.Default else debug
}

1. Will launch Method passed in context And CoroutineScope Medium context combined
2. If combined There are no interceptors in the , A default interceptor will be passed in , namely Dispatchers.Default, This also explains why we There will be a default thread switching effect when there is no incoming interceptor

2. establish Continuation

  		 val coroutine = if (start.isLazy)
        LazyStandaloneCoroutine(newContext, block) else
        StandaloneCoroutine(newContext, active = true)

By default , We will create a StandloneCoroutine
It is worth noting that , This coroutine In fact, it's our collaborative process complete, Callback after success , Not the synergist itself
And then call coroutine.start, This indicates the start of the process

3. Start the process

 coroutine.start(start, coroutine, block)

It's actually calling theta

// Be similar to startCoroutineCancellable , But for the collaboration that has been created . fatalCompletion Only use when the interceptor throws an exception 
internal fun Continuation<Unit>.startCoroutineCancellable(fatalCompletion: Continuation<*>) =
    runSafely(fatalCompletion) {
        intercepted().resumeCancellableWith(Result.success(Unit))
    }

Next call

@InternalCoroutinesApi
public fun <T> Continuation<T>.resumeCancellableWith(
    result: Result<T>,
    onCancellation: ((cause: Throwable) -> Unit)? = null
): Unit = when (this) {
    is DispatchedContinuation -> resumeCancellableWith(result, onCancellation)
    else -> resumeWith(result)
}

3.1 resumeCancellableWith

 // We inline it to save an entry on the stack in cases where it shows (unconfined dispatcher)
    // It is used only in Continuation<T>.resumeCancellableWith
    @Suppress("NOTHING_TO_INLINE")
    inline fun resumeCancellableWith(
        result: Result<T>,
        noinline onCancellation: ((cause: Throwable) -> Unit)?
    ) {
        val state = result.toState(onCancellation)
        if (dispatcher.isDispatchNeeded(context)) {
            _state = state
            resumeMode = MODE_CANCELLABLE
            dispatcher.dispatch(context, this)
        } else {
            executeUnconfined(state, MODE_CANCELLABLE) {
                if (!resumeCancelled(state)) {
                    resumeUndispatchedWith(result)
                }
            }
        }
    }