Kotlin CoroutinesでCoroutineScope.launchの処理を追う

public interface CoroutineScope {
    public val coroutineContext: CoroutineContext
}

public fun CoroutineScope.launch(
    context: CoroutineContext,
    start: CoroutineStart,
    block: suspend CoroutineScope.() -> Unit
): Job {
    
    // ScopeのContext + 指定されたContext ...①
    val newContext = newCoroutineContext(context)
 
    // Coroutineそのものを作る ...②
    val coroutine =
        StandaloneCoroutine(newContext, active = true)
 
    // 指定された開始方法でblockを開始 ...③
    coroutine.start(start, coroutine, block)
 
    return coroutine
}

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

internal fun createViewModelScope(): CloseableCoroutineScope {
    val dispatcher =
        try {
            // In platforms where `Dispatchers.Main` is not available, Kotlin Multiplatform will
            // throw
            // an exception (the specific exception type may depend on the platform). Since there's
            // no
            // direct functional alternative, we use `EmptyCoroutineContext` to ensure that a
            // coroutine
            // launched within this scope will run in the same context as the caller.
            Dispatchers.Main.immediate
        } catch (_: NotImplementedError) {
            // In Native environments where `Dispatchers.Main` might not exist (e.g., Linux):
            EmptyCoroutineContext
        } catch (_: IllegalStateException) {
            // In JVM Desktop environments where `Dispatchers.Main` might not exist (e.g., Swing):
            EmptyCoroutineContext
        }
    return CloseableCoroutineScope(coroutineContext = dispatcher + SupervisorJob())
}

private open class StandaloneCoroutine(
    parentContext: CoroutineContext,
    active: Boolean
) : AbstractCoroutine<Unit>(parentContext, initParentJob = true, active = active) {
    override fun handleJobException(exception: Throwable): Boolean {
        handleCoroutineException(context, exception)
        return true
    }
}

@OptIn(InternalForInheritanceCoroutinesApi::class)
@InternalCoroutinesApi
public abstract class AbstractCoroutine<in T>(
    parentContext: CoroutineContext,
    initParentJob: Boolean,
    active: Boolean
) : JobSupport(active), Job, Continuation<T>, CoroutineScope

// start: CoroutineStart
// continue: StandaloneCoroutine
// block: suspend CoroutineScope.() -> Unit
coroutine.start(start, coroutine, block)

public fun <R> start(start: CoroutineStart, receiver: R, block: suspend R.() -> T) {
    // MEMO: R: CoroutineScope、実態はStandaloneCoroutine
    //       T: Unit
    //       receiver：R
    //       this: AbstractCoroutine<Unit>、実体は StandaloneCoroutine
    start(block, receiver, this)
}

@InternalCoroutinesApi
public operator fun <R, T> invoke(block: suspend R.() -> T, receiver: R, completion: Continuation<T>): Unit =
    // MEMO: R: CoroutineScope、実態はStandaloneCoroutine
    //       T: Unit
    //       completion: Continuation<Unit>、実態はStandaloneCoroutine
    when (this) {
        DEFAULT -> block.startCoroutineCancellable(receiver, completion)
        ATOMIC -> block.startCoroutine(receiver, completion)
        UNDISPATCHED -> block.startCoroutineUndispatched(receiver, completion)
        LAZY -> Unit // will start lazily
    }

// MEMO: R: CoroutineScope、実態はStandaloneCoroutine
//       T: Unit
//       receiver: CoroutineScope、実態はStandaloneCoroutine
//       completion: Continuation<Unit>、実態はStandaloneCoroutine
internal fun <R, T> (suspend (R) -> T).startCoroutineCancellable(
    receiver: R, completion: Continuation<T>,
) = runSafely(completion) {
    createCoroutineUnintercepted(receiver, completion).intercepted().resumeCancellableWithInternal(Result.success(Unit))
}

@SinceKotlin("1.3")
public actual fun <R, T> (suspend R.() -> T).createCoroutineUnintercepted(
    receiver: R,
    completion: Continuation<T>
): Continuation<Unit> {
    val probeCompletion = probeCoroutineCreated(completion)
    return if (this is BaseContinuationImpl)
        create(receiver, probeCompletion)
    else {
        createCoroutineFromSuspendFunction(probeCompletion) {
            (this as Function2<R, Continuation<T>, Any?>).invoke(receiver, it)
        }
    }
}

public open fun create(value: Any?, completion: Continuation<*>): Continuation<Unit> {
    throw UnsupportedOperationException("create(Any?;Continuation) has not been overridden")
}

viewModelScope.launch {
    println("A")
    delay(1000)
    println("B")
}
 
final class GeneratedLambda(
    completion: Continuation<Unit>?
) : SuspendLambda(
    arity = 2,
    completion = completion
) {
 
    private var label = 0
    private var p$: CoroutineScope? = null
 
    override fun create(
        value: Any?,
        completion: Continuation<*>
    ): Continuation<Unit> {
 
        val result = GeneratedLambda(
            completion as Continuation<Unit>
        )
 
        // receiverを保存
        result.p$ = value as CoroutineScope
 
        return result
    }
 
    override fun invokeSuspend(result: Any?): Any? {
        when (label) {
            0 -> {
                throwOnFailure(result)
 
                val scope = p$!!
 
                println("A")
 
                label = 1
 
                if (delay(1000, this) === COROUTINE_SUSPENDED) {
                    return COROUTINE_SUSPENDED
                }
            }
 
            1 -> {
                throwOnFailure(result)
            }
 
            else -> {
                error("call to 'resume' before 'invoke' with coroutine")
            }
        }
 
        println("B")
        return Unit
    }
}

// MEMO: R: CoroutineScope、実態はStandaloneCoroutine
//       T: Unit
//       receiver: CoroutineScope、実態はStandaloneCoroutine
//       completion: Continuation<Unit>、実態はStandaloneCoroutine
internal fun <R, T> (suspend (R) -> T).startCoroutineCancellable(
    receiver: R, completion: Continuation<T>,
) = runSafely(completion) {
    createCoroutineUnintercepted(receiver, completion).intercepted().resumeCancellableWithInternal(Result.success(Unit))
}

@SinceKotlin("1.3")
public actual fun <T> Continuation<T>.intercepted(): Continuation<T> =
    (this as? ContinuationImpl)?.intercepted() ?: this

public fun intercepted(): Continuation<Any?> =
    intercepted
        ?: (context[ContinuationInterceptor]?.interceptContinuation(this) ?: this)
            .also { intercepted = it }

@SinceKotlin("1.3")
public interface ContinuationInterceptor : CoroutineContext.Element {
    /**
     * The key that defines *the* context interceptor.
     */
    public companion object Key : CoroutineContext.Key<ContinuationInterceptor>
    ...
}

public final override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =
        DispatchedContinuation(this, continuation)

internal class DispatchedContinuation<in T>(
    @JvmField internal val dispatcher: CoroutineDispatcher,
    @JvmField val continuation: Continuation<T>
) : DispatchedTask<T>(MODE_UNINITIALIZED), CoroutineStackFrame, Continuation<T> by continuation

// MEMO: R: CoroutineScope、実態はStandaloneCoroutine
//       T: Unit
//       receiver: CoroutineScope、実態はStandaloneCoroutine
//       completion: Continuation<Unit>、実態はStandaloneCoroutine
internal fun <R, T> (suspend (R) -> T).startCoroutineCancellable(
    receiver: R, completion: Continuation<T>,
) = runSafely(completion) {
    createCoroutineUnintercepted(receiver, completion).intercepted().resumeCancellableWithInternal(Result.success(Unit))
}

internal fun <T> Continuation<T>.resumeCancellableWithInternal(
    result: Result<T>,
): Unit = when (this) {
    is DispatchedContinuation -> resumeCancellableWith(result)
    else -> resumeWith(result)
}

// 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")
internal inline fun resumeCancellableWith(result: Result<T>) {
    val state = result.toState()
    if (dispatcher.safeIsDispatchNeeded(context)) {
        _state = state
        resumeMode = MODE_CANCELLABLE
        dispatcher.safeDispatch(context, this)
    } else {
        executeUnconfined(state, MODE_CANCELLABLE) {
            if (!resumeCancelled(state)) {
                resumeUndispatchedWith(result)
            }
        }
    }
}

// kotlinx-coroutines-core/common/src/internal/DispatchedContinuation.kt
internal fun CoroutineDispatcher.safeIsDispatchNeeded(context: CoroutineContext): Boolean {
    try {
        return isDispatchNeeded(context)
    } catch (e: Throwable) {
        throw DispatchException(e, this, context)
    }
}
 
// kotlinx-coroutines-core/common/src/CoroutineDispatcher.kt
public abstract class CoroutineDispatcher {
    public open fun isDispatchNeeded(context: CoroutineContext): Boolean = true
}

public actual object Dispatchers {
    @JvmStatic
    public actual val Default: CoroutineDispatcher = DefaultScheduler
 
    @JvmStatic
    public actual val Main: MainCoroutineDispatcher get() = MainDispatcherLoader.dispatcher
 
    @JvmStatic
    public actual val Unconfined: CoroutineDispatcher = kotlinx.coroutines.Unconfined
 
    @JvmStatic
    public val IO: CoroutineDispatcher get() = DefaultIoScheduler
}

internal object MainDispatcherLoader {
 
    private val FAST_SERVICE_LOADER_ENABLED = systemProp(FAST_SERVICE_LOADER_PROPERTY_NAME, true)
 
    @JvmField
    val dispatcher: MainCoroutineDispatcher = loadMainDispatcher()
 
    private fun loadMainDispatcher(): MainCoroutineDispatcher {
        return try {
            val factories = if (FAST_SERVICE_LOADER_ENABLED) {
                FastServiceLoader.loadMainDispatcherFactory()
            } else {
                // We are explicitly using the
                // `ServiceLoader.load(MyClass::class.java, MyClass::class.java.classLoader).iterator()`
                // form of the ServiceLoader call to enable R8 optimization when compiled on Android.
                ServiceLoader.load(
                        MainDispatcherFactory::class.java,
                        MainDispatcherFactory::class.java.classLoader
                ).iterator().asSequence().toList()
            }
            @Suppress("ConstantConditionIf")
            factories.maxByOrNull { it.loadPriority }?.tryCreateDispatcher(factories)
                ?: createMissingDispatcher()
        } catch (e: Throwable) {
            // Service loader can throw an exception as well
            createMissingDispatcher(e)
        }
    }
}

internal class AndroidDispatcherFactory : MainDispatcherFactory {
 
    override fun createDispatcher(allFactories: List<MainDispatcherFactory>): MainCoroutineDispatcher {
        val mainLooper = Looper.getMainLooper() ?: throw IllegalStateException("The main looper is not available")
        return HandlerContext(mainLooper.asHandler(async = true))
    }
 
    override fun hintOnError(): String = "For tests Dispatchers.setMain from kotlinx-coroutines-test module can be used"
 
    override val loadPriority: Int
        get() = Int.MAX_VALUE / 2
}

// ui/kotlinx-coroutines-android/src/HandlerDispatcher.kt
override fun isDispatchNeeded(context: CoroutineContext): Boolean {
    return !invokeImmediately || Looper.myLooper() != handler.looper
}
 
// HandlerContext
internal class HandlerContext private constructor(
    private val handler: Handler,
    private val name: String?,
    private val invokeImmediately: Boolean
) : HandlerDispatcher(), Delay {
    /**
     * Creates [CoroutineDispatcher] for the given Android [handler].
     *
     * @param handler a handler.
     * @param name an optional name for debugging.
     */
    constructor(
        handler: Handler,
        name: String? = null
    ) : this(handler, name, false)
}

// 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")
internal inline fun resumeCancellableWith(result: Result<T>) {
    val state = result.toState()
    if (dispatcher.safeIsDispatchNeeded(context)) {
        _state = state
        resumeMode = MODE_CANCELLABLE
        dispatcher.safeDispatch(context, this)
    } else {
        executeUnconfined(state, MODE_CANCELLABLE) {
            if (!resumeCancelled(state)) {
                resumeUndispatchedWith(result)
            }
        }
    }
}

internal fun CoroutineDispatcher.safeDispatch(context: CoroutineContext, runnable: Runnable) {
    try {
        dispatch(context, runnable)
    } catch (e: Throwable) {
        throw DispatchException(e, this, context)
    }
}

internal object DefaultScheduler : SchedulerCoroutineDispatcher(
    CORE_POOL_SIZE, MAX_POOL_SIZE,
    IDLE_WORKER_KEEP_ALIVE_NS, DEFAULT_SCHEDULER_NAME
) {
 
    override fun limitedParallelism(parallelism: Int, name: String?): CoroutineDispatcher {
        parallelism.checkParallelism()
        if (parallelism >= CORE_POOL_SIZE) {
            return namedOrThis(name)
        }
        return super.limitedParallelism(parallelism, name)
    }
 
    // Shuts down the dispatcher, used only by Dispatchers.shutdown()
    internal fun shutdown() {
        super.close()
    }
 
    // Overridden in case anyone writes (Dispatchers.Default as ExecutorCoroutineDispatcher).close()
    override fun close() {
        throw UnsupportedOperationException("Dispatchers.Default cannot be closed")
    }
 
    override fun toString(): String = "Dispatchers.Default"
}

internal open class SchedulerCoroutineDispatcher(
    private val corePoolSize: Int = CORE_POOL_SIZE,
    private val maxPoolSize: Int = MAX_POOL_SIZE,
    private val idleWorkerKeepAliveNs: Long = IDLE_WORKER_KEEP_ALIVE_NS,
    private val schedulerName: String = "CoroutineScheduler",
) : ExecutorCoroutineDispatcher() {
 
    override val executor: Executor
        get() = coroutineScheduler
 
    // This is variable for test purposes, so that we can reinitialize from clean state
    private var coroutineScheduler = createScheduler()
 
    private fun createScheduler() =
        CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs, schedulerName)
 
    override fun dispatch(context: CoroutineContext, block: Runnable): Unit = coroutineScheduler.dispatch(block)
    ....
}

@Suppress("NOTHING_TO_INLINE")
internal class CoroutineScheduler(
    @JvmField val corePoolSize: Int,
    @JvmField val maxPoolSize: Int,
    @JvmField val idleWorkerKeepAliveNs: Long = IDLE_WORKER_KEEP_ALIVE_NS,
    @JvmField val schedulerName: String = DEFAULT_SCHEDULER_NAME
) : Executor, Closeable {
 
   fun dispatch(block: Runnable, taskContext: TaskContext = NonBlockingContext, fair: Boolean = false) {
        trackTask() // this is needed for virtual time support
        val task = createTask(block, taskContext)
        val isBlockingTask = task.isBlocking
        // Invariant: we increment counter **before** publishing the task
        // so executing thread can safely decrement the number of blocking tasks
        val stateSnapshot = if (isBlockingTask) incrementBlockingTasks() else 0
        // try to submit the task to the local queue and act depending on the result
        val currentWorker = currentWorker()
        val notAdded = currentWorker.submitToLocalQueue(task, fair)
        if (notAdded != null) {
            if (!addToGlobalQueue(notAdded)) {
                // Global queue is closed in the last step of close/shutdown -- no more tasks should be accepted
                throw RejectedExecutionException("$schedulerName was terminated")
            }
        }
        // Checking 'task' instead of 'notAdded' is completely okay
        if (isBlockingTask) {
            // Use state snapshot to better estimate the number of running threads
            signalBlockingWork(stateSnapshot)
        } else {
            signalCpuWork()
        }
    }
}

fun signalCpuWork() {
    if (tryUnpark()) return
    if (tryCreateWorker()) return
    tryUnpark()
}

private fun tryUnpark(): Boolean {
    while (true) {
        val worker = parkedWorkersStackPop() ?: return false
        if (worker.workerCtl.compareAndSet(PARKED, CLAIMED)) {
            LockSupport.unpark(worker)
            return true
        }
    }
}

private fun tryCreateWorker(state: Long = controlState.value): Boolean {
    val created = createdWorkers(state)
    val blocking = blockingTasks(state)
    val cpuWorkers = (created - blocking).coerceAtLeast(0)
    /*
        * We check how many threads are there to handle non-blocking work,
        * and create one more if we have not enough of them.
        */
    if (cpuWorkers < corePoolSize) {
        val newCpuWorkers = createNewWorker()
        // If we've created the first cpu worker and corePoolSize > 1 then create
        // one more (second) cpu worker, so that stealing between them is operational
        if (newCpuWorkers == 1 && corePoolSize > 1) createNewWorker()
        if (newCpuWorkers > 0) return true
    }
    return false
}