Kotlin协程-一个协程的生命周期

fun main() {
    GlobalScope.launch {
        delay(1000L) // 非阻塞的等待 1 秒钟（默认时间单位是毫秒）
        println("World!") // 在延迟后打印输出
    }
    delay(100L)
    println("Hello,") // 协程已在等待时主线程还在继续

    Thread.sleep(2000L) // 阻塞主线程 2 秒钟来保证 JVM 存活
    println("out launch done")
}

flowchat
st=>start: 创建
p=>operation: 暂停
r=>operation: 继续
e=>end: 结束
con=>condition: 完成?

st->p->r->con
con(yes)->e
con(no)->p

public fun CoroutineScope.launch(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,//默认的立即启动方式
    block: suspend CoroutineScope.() -> Unit
): Job {
    val newContext = newCoroutineContext(context)//创建context
    val coroutine = if (start.isLazy)
        LazyStandaloneCoroutine(newContext, block) else
        StandaloneCoroutine(newContext, active = true)//创建立即启动的coroutine
    coroutine.start(start, coroutine, block)
    return coroutine
}

coroutine.start(start, coroutine, block)

public fun <R> start(start: CoroutineStart, receiver: R, block: suspend R.() -> T) {
    initParentJob()
    start(block, receiver, this)
}

@InternalCoroutinesApi
public operator fun <T> invoke(block: suspend () -> T, completion: Continuation<T>) =
    when (this) {
        CoroutineStart.DEFAULT -> block.startCoroutineCancellable(completion) //start实际执行的是这行代码
        CoroutineStart.ATOMIC -> block.startCoroutine(completion)
        CoroutineStart.UNDISPATCHED -> block.startCoroutineUndispatched(completion)
        CoroutineStart.LAZY -> Unit // will start lazily
    }

block.startCoroutineCancellable(completion)

@InternalCoroutinesApi
public fun <T> (suspend () -> T).startCoroutineCancellable(completion: Continuation<T>) = runSafely(completion) {
    createCoroutineUnintercepted(completion).intercepted().resumeCancellableWith(Result.success(Unit))
}

public interface Continuation<in T> {
    public val context: CoroutineContext
    public fun resumeWith(result: Result<T>)
}

@SinceKotlin("1.3")
// State machines for named suspend functions extend from this class
internal abstract class ContinuationImpl(
    completion: Continuation<Any?>?,
    private val _context: CoroutineContext?
) : BaseContinuationImpl(completion) {
    ...

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

context[ContinuationInterceptor]

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

inline fun resumeCancellableWith(result: Result<T>) {
    val state = result.toState()
    if (dispatcher.isDispatchNeeded(context)) {
        _state = state
        resumeMode = MODE_CANCELLABLE
        dispatcher.dispatch(context, this)//进入派发流程
    } else {
        executeUnconfined(state, MODE_CANCELLABLE) {
            if (!resumeCancelled()) {
                resumeUndispatchedWith(result)
            }
        }
    }
}

internal actual fun createDefaultDispatcher(): CoroutineDispatcher =
    if (useCoroutinesScheduler) DefaultScheduler else CommonPool

/**
 * Creates context for the new coroutine. It installs [Dispatchers.Default] when no other dispatcher nor
 * [ContinuationInterceptor] is specified, and adds optional support for debugging facilities (when turned on).
 *
 * See [DEBUG_PROPERTY_NAME] for description of debugging facilities on JVM.
 */
@ExperimentalCoroutinesApi
public actual fun CoroutineScope.newCoroutineContext(context: CoroutineContext):  CoroutineContext { //创建context
    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
}

public actual val Default: CoroutineDispatcher = createDefaultDispatcher()

override fun dispatch(context: CoroutineContext, block: Runnable): Unit =
    try {
        coroutineScheduler.dispatch(block)
    } catch (e: RejectedExecutionException) {
        DefaultExecutor.dispatch(context, block)
    }

fun dispatch(block: Runnable, taskContext: TaskContext = NonBlockingContext, tailDispatch: Boolean = false) {
    trackTask() // this is needed for virtual time support
    val task = createTask(block, taskContext) //封装任务
    // try to submit the task to the local queue and act depending on the result
    val currentWorker = currentWorker() //获取当前线程
    val notAdded = currentWorker.submitToLocalQueue(task, tailDispatch) //加入worker本地执行队列
    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")
        }
    }
    val skipUnpark = tailDispatch && currentWorker != null
    // Checking 'task' instead of 'notAdded' is completely okay
    if (task.mode == TASK_NON_BLOCKING) {
        if (skipUnpark) return
        signalCpuWork() //执行CPU密集型协程
    } else {
        // Increment blocking tasks anyway
        signalBlockingWork(skipUnpark = skipUnpark) //执行阻塞型协程
    }
}

internal inner class Worker private constructor() : Thread() {
    init {
        isDaemon = true
    }

    // guarded by scheduler lock, index in workers array, 0 when not in array (terminated)
    @Volatile // volatile for push/pop operation into parkedWorkersStack
    var indexInArray = 0
        set(index) {
            name = "$schedulerName-worker-${if (index == 0) "TERMINATED" else index.toString()}"
            field = index
        }

    constructor(index: Int) : this() {
        indexInArray = index
    }

    inline val scheduler get() = this@CoroutineScheduler

    @JvmField
    val localQueue: WorkQueue = WorkQueue() //本地队列

private fun createNewWorker(): Int {
    synchronized(workers) {
        ...
        val worker = Worker(newIndex)
        workers[newIndex] = worker
        require(newIndex == incrementCreatedWorkers())
        worker.start()
        return cpuWorkers + 1
    }
}

private fun runWorker() {
    var rescanned = false
    while (!isTerminated && state != WorkerState.TERMINATED) {
        val task = findTask(mayHaveLocalTasks)
        // Task found. Execute and repeat
        if (task != null) {
            rescanned = false
            minDelayUntilStealableTaskNs = 0L
            executeTask(task) //执行

private fun executeTask(task: Task) {
    val taskMode = task.mode
    idleReset(taskMode)
    beforeTask(taskMode)
    runSafely(task)
    afterTask(taskMode)
}

fun runSafely(task: Task) {
    try {
        task.run() //真正的执行
    } catch (e: Throwable) {
        val thread = Thread.currentThread()
        thread.uncaughtExceptionHandler.uncaughtException(thread, e)
    } finally {
        unTrackTask()
    }
}

internal abstract class DispatchedTask<in T>(
    @JvmField public var resumeMode: Int
) : SchedulerTask() {
    internal abstract val delegate: Continuation<T>

    internal abstract fun takeState(): Any?

    internal open fun cancelResult(state: Any?, cause: Throwable) {}

    @Suppress("UNCHECKED_CAST")
    internal open fun <T> getSuccessfulResult(state: Any?): T =
        state as T

    internal fun getExceptionalResult(state: Any?): Throwable? =
        (state as? CompletedExceptionally)?.cause

    public final override fun run() {
        val taskContext = this.taskContext
        var fatalException: Throwable? = null
        try {
            val delegate = delegate as DispatchedContinuation<T>
            val continuation = delegate.continuation
            val context = continuation.context
            val state = takeState() // NOTE: Must take state in any case, even if cancelled
            withCoroutineContext(context, delegate.countOrElement) {
                val exception = getExceptionalResult(state)
                val job = if (resumeMode.isCancellableMode) context[Job] else null
                /*
                 * Check whether continuation was originally resumed with an exception.
                 * If so, it dominates cancellation, otherwise the original exception
                 * will be silently lost.
                 */
                if (exception == null && job != null && !job.isActive) {
                    val cause = job.getCancellationException()
                    cancelResult(state, cause)
                    continuation.resumeWithStackTrace(cause)
                } else {
                    if (exception != null) continuation.resumeWithException(exception)
                    else continuation.resume(getSuccessfulResult(state)) //调用continuation的resume
                }
            }

internal abstract class BaseContinuationImpl(
    // 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).
    public val completion: Continuation<Any?>?
) : Continuation<Any?>, CoroutineStackFrame, Serializable {
    // This implementation is final. This fact is used to unroll resumeWith recursion.
    public final override fun resumeWith(result: Result<Any?>) {
        // This loop unrolls recursion in current.resumeWith(param) to make saner and shorter stack traces on resume
        var current = this
        var param = result
        while (true) {
            // Invoke "resume" debug probe on every resumed continuation, so that a debugging library infrastructure
            // can precisely track what part of suspended callstack was already resumed
            probeCoroutineResumed(current)
            with(current) {
                val completion = completion!! // fail fast when trying to resume continuation without completion
                val outcome: Result<Any?> =
                    try {
                        val outcome = invokeSuspend(param) //真正调用我们写的代码的地方
                        if (outcome === COROUTINE_SUSPENDED) return
                        Result.success(outcome)
                    } catch (exception: Throwable) {
                        Result.failure(exception)
                    }
                releaseIntercepted() // this state machine instance is terminating
                if (completion is BaseContinuationImpl) {
                    // unrolling recursion via loop
                    current = completion
                    param = outcome
                } else {
                    // top-level completion reached -- invoke and return
                    completion.resumeWith(outcome)
                    return
                }
            }
        }
    }

fun main() {
    GlobalScope.launch {
        println("Hello!")
        delay(100L) // 非阻塞的等待 1 秒钟（默认时间单位是毫秒）
        println("World!") // 在延迟后打印输出
    }
    Thread.sleep(400L) // 阻塞主线程 2 秒钟来保证 JVM 存活
    println("out launch done")
}

    public final override fun resumeWith(result: Result<Any?>) {
        // This loop unrolls recursion in current.resumeWith(param) to make saner and shorter stack traces on resume
        var current = this
        var param = result
        while (true) {
            // Invoke "resume" debug probe on every resumed continuation, so that a debugging library infrastructure
            // can precisely track what part of suspended callstack was already resumed
            probeCoroutineResumed(current)
            with(current) {
                val completion = completion!! // fail fast when trying to resume continuation without completion
                val outcome: Result<Any?> =
                    try {
                        val outcome = invokeSuspend(param)
                        if (outcome === COROUTINE_SUSPENDED) return //直接返回
                        Result.success(outcome)
                    } catch (exception: Throwable) {
                        Result.failure(exception)
                    }
                releaseIntercepted() // this state machine instance is terminating
                if (completion is BaseContinuationImpl) {
                    // unrolling recursion via loop
                    current = completion
                    param = outcome
                } else {
                    // top-level completion reached -- invoke and return
                    completion.resumeWith(outcome)
                    return
                }
            }
        }
    }