Kotlin协程基本原理

viewModelScope.launch(Dispatchers.Main) {
    val result = withContext(Dispatchers.IO) {       
      searchRepository.getWebSites()    
    }
    if (result is Result.Success) {
       emitArticleUiState(showHot = true, showWebSites = result.data)    
    }
 }

@JvmStatic
public val IO: CoroutineDispatcher = DefaultScheduler.IO
val IO = blocking(systemProp(IO_PARALLELISM_PROPERTY_NAME, 64.coerceAtLeast(AVAILABLE_PROCESSORS)))


public fun blocking(parallelism: Int = BLOCKING_DEFAULT_PARALLELISM): CoroutineDispatcher {
    require(parallelism > 0) { "Expected positive parallelism level, but have $parallelism" }
    return LimitingDispatcher(this, parallelism, TaskMode.PROBABLY_BLOCKING)
}


private class LimitingDispatcher(
    val dispatcher: ExperimentalCoroutineDispatcher,
    val parallelism: Int,
    override val taskMode: TaskMode
) : ExecutorCoroutineDispatcher(), TaskContext, Executor {
    private val queue = ConcurrentLinkedQueue<Runnable>()
    private val inFlightTasks = atomic(0)
    override val executor: Executor
        get() = this
    override fun execute(command: Runnable) = dispatch(command, false)
    override fun close(): Unit = error("Close cannot be invoked on LimitingBlockingDispatcher")
    override fun dispatch(context: CoroutineContext, block: Runnable) = dispatch(block, false)
    private fun dispatch(block: Runnable, fair: Boolean) {
        var taskToSchedule = block
        while (true) {
            // Commit in-flight tasks slot
            val inFlight = inFlightTasks.incrementAndGet()
            // Fast path, if parallelism limit is not reached, dispatch task and return
            if (inFlight <= parallelism) {
                dispatcher.dispatchWithContext(taskToSchedule, this, fair)
                return
            }
            // Parallelism limit is reached, add task to the queue
            queue.add(taskToSchedule)
            if (inFlightTasks.decrementAndGet() >= parallelism) {
                return
            }
            taskToSchedule = queue.poll() ?: return
        }
    }

open class ExperimentalCoroutineDispatcher(
    private val corePoolSize: Int,
    private val maxPoolSize: Int,
    private val idleWorkerKeepAliveNs: Long,
    private val schedulerName: String = "CoroutineScheduler"
) : ExecutorCoroutineDispatcher() {
    constructor(
        corePoolSize: Int = CORE_POOL_SIZE,
        maxPoolSize: Int = MAX_POOL_SIZE,
        schedulerName: String = DEFAULT_SCHEDULER_NAME
    ) : this(corePoolSize, maxPoolSize, IDLE_WORKER_KEEP_ALIVE_NS, schedulerName)
    
    @Deprecated(message = "Binary compatibility for Ktor 1.0-beta", level = DeprecationLevel.HIDDEN)
    constructor(
        corePoolSize: Int = CORE_POOL_SIZE,
        maxPoolSize: Int = MAX_POOL_SIZE
    ) : this(corePoolSize, maxPoolSize, IDLE_WORKER_KEEP_ALIVE_NS)
    
    override val executor: Executor
        get() = coroutineScheduler
        
    // This is variable for test purposes, so that we can reinitialize from clean state
    private var coroutineScheduler = createScheduler()


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

private fun createScheduler() = CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs, schedulerName)internal class CoroutineScheduler(    private val corePoolSize: Int,    private val maxPoolSize: Int,    private val idleWorkerKeepAliveNs: Long = IDLE_WORKER_KEEP_ALIVE_NS,    private val schedulerName: String = DEFAULT_SCHEDULER_NAME) : Executor, Closeable {...}

internal fun dispatchWithContext(block: Runnable, context: TaskContext, fair: Boolean) {
    try {
        coroutineScheduler.dispatch(block, context, fair)
    } catch (e: RejectedExecutionException) {
        // Context shouldn't be lost here to properly invoke before/after task
        DefaultExecutor.enqueue(coroutineScheduler.createTask(block, context))
    }
}


fun dispatch(block: Runnable, taskContext: TaskContext = NonBlockingContext, fair: 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
    when (submitToLocalQueue(task, fair)) {
        ADDED -> return
        NOT_ADDED -> {
            // try to offload task to global queue
            if (!globalQueue.addLast(task)) {
                // Global queue is closed in the last step of close/shutdown -- no more tasks should be accepted
                throw RejectedExecutionException("$schedulerName was terminated")
            }
            requestCpuWorker()
        }
        else -> requestCpuWorker() // ask for help
    }
}

private fun submitToLocalQueue(task: Task, fair: Boolean): Int {
    val worker = currentWorker() ?: return NOT_ADDED
    /*
     * This worker could have been already terminated from this thread by close/shutdown and it should not
     * accept any more tasks into its local queue.
     */
    if (worker.state === WorkerState.TERMINATED) return NOT_ADDED
    var result = ADDED
    if (task.mode == TaskMode.NON_BLOCKING) {
        /*
         * If the worker is currently executing blocking task and tries to dispatch non-blocking task, it's one the following reasons:
         * 1) Blocking worker is finishing its block and resumes non-blocking continuation
         * 2) Blocking worker starts to create non-blocking jobs
         * First use-case is expected (as recommended way of using blocking contexts),
         * so we add non-blocking task to local queue, but also request CPU worker to mitigate second case
         */
        if (worker.isBlocking) {
            result = ADDED_REQUIRES_HELP
        } else {
            /*
             * If thread is not blocking, then it's just tries to finish its
             * local work in order to park (or grab another blocking task), do not add non-blocking tasks
             * to its local queue if it can't acquire CPU
             */
            val hasPermit = worker.tryAcquireCpuPermit()
            if (!hasPermit) {
                return NOT_ADDED
            }
        }
    }


    val noOffloadingHappened = if (fair) {
        worker.localQueue.addLast(task, globalQueue)
    } else {
        worker.localQueue.add(task, globalQueue)
    }
    if (noOffloadingHappened) {
        // When we're close to queue capacity, wake up anyone to steal work
        // Note: non-atomic bufferSize here is Ok (it is just a performance optimization)
        if (worker.localQueue.bufferSize > QUEUE_SIZE_OFFLOAD_THRESHOLD) {
            return ADDED_REQUIRES_HELP
        }
        return result
    }
    return ADDED_REQUIRES_HELP
}

private fun currentWorker(): Worker? = (Thread.currentThread() as? Worker)?.takeIf { it.scheduler == this }

fun main(args: Array<String>) = runBlocking<Unit> { 
    launch(Dispatchers.Unconfined) {
       println("${Thread.currentThread().name} : launch start") 
       async(Dispatchers.Default) {
           println("${Thread.currentThread().name} : async start") 
           delay(100) 
           println("${Thread.currentThread().name} : async end")        
       }.await()
       println("${Thread.currentThread().name} : launch end")   
    }
}

internal suspend fun awaitInternal(): Any? {
    // fast-path -- check state (avoid extra object creation)
    while (true) { // lock-free loop on state
        val state = this.state
        if (state !is Incomplete) {
            // already complete -- just return result
            if (state is CompletedExceptionally) { // Slow path to recover stacktrace
                recoverAndThrow(state.cause)
            }
            return state.unboxState()
        }
        if (startInternal(state) >= 0) break // break unless needs to retry
    }
    return awaitSuspend() // slow-path
}


private suspend fun awaitSuspend(): Any? = suspendCoroutineUninterceptedOrReturn { uCont ->
    /*
     * Custom code here, so that parent coroutine that is using await
     * on its child deferred (async) coroutine would throw the exception that this child had
     * thrown and not a JobCancellationException.
     */
    val cont = AwaitContinuation(uCont.intercepted(), this)
    cont.disposeOnCancellation(invokeOnCompletion(ResumeAwaitOnCompletion(this, cont).asHandler))
    cont.getResult()
}

private class AwaitContinuation<T>(
    delegate: Continuation<T>,
    private val job: JobSupport
) : CancellableContinuationImpl<T>(delegate, MODE_CANCELLABLE) {
    override fun getContinuationCancellationCause(parent: Job): Throwable {
        val state = job.state
        /*
         * When the job we are waiting for had already completely completed exceptionally or
         * is failing, we shall use its root/completion cause for await's result.
         */
        if (state is Finishing) state.rootCause?.let { return it }
        if (state is CompletedExceptionally) return state.cause
        return parent.getCancellationException()
    }
    
    protected override fun nameString(): String =
        "AwaitContinuation"
}

private fun trySuspend(): Boolean {
    _decision.loop { decision ->
        when (decision) {
            UNDECIDED -> if (this._decision.compareAndSet(UNDECIDED, SUSPENDED)) return true
            RESUMED -> return false
            else -> error("Already suspended")
        }
    }
}


private fun tryResume(): Boolean {
    _decision.loop { decision ->
        when (decision) {
            UNDECIDED -> if (this._decision.compareAndSet(UNDECIDED, RESUMED)) return true
            SUSPENDED -> return false
            else -> error("Already resumed")
        }
    }
}