Kotlin协程-特殊的阻塞协程

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

/**
 * Runs a new coroutine and **blocks** the current thread _interruptibly_ until its completion.
 * This function should not be used from a coroutine. It is designed to bridge regular blocking code
 * to libraries that are written in suspending style, to be used in `main` functions and in tests.
 ...
 */
@Throws(InterruptedException::class)
public fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T {

public fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T {
    val currentThread = Thread.currentThread()
    val contextInterceptor = context[ContinuationInterceptor]
    val eventLoop: EventLoop?
    val newContext: CoroutineContext
    if (contextInterceptor == null) {
        // create or use private event loop if no dispatcher is specified
        eventLoop = ThreadLocalEventLoop.eventLoop //默认派发器
        newContext = GlobalScope.newCoroutineContext(context + eventLoop)
    } else {
        // See if context's interceptor is an event loop that we shall use (to support TestContext)
        // or take an existing thread-local event loop if present to avoid blocking it (but don't create one)
        eventLoop = (contextInterceptor as? EventLoop)?.takeIf { it.shouldBeProcessedFromContext() } //继承自上下文的派发器
            ?: ThreadLocalEventLoop.currentOrNull()
        newContext = GlobalScope.newCoroutineContext(context)
    }
    val coroutine = BlockingCoroutine<T>(newContext, currentThread, eventLoop)
    coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
    return coroutine.joinBlocking() 
}

@ThreadLocal
internal object ThreadLocalEventLoop {
    private val ref = CommonThreadLocal<EventLoop?>()

    internal val eventLoop: EventLoop //eventloop对象
        get() = ref.get() ?: createEventLoop().also { ref.set(it) }

    internal fun currentOrNull(): EventLoop? =
        ref.get()

    internal fun resetEventLoop() {
        ref.set(null)
    }

    internal fun setEventLoop(eventLoop: EventLoop) {
        ref.set(eventLoop)
    }
}

internal expect fun createEventLoop(): EventLoop

internal class BlockingEventLoop(
    override val thread: Thread
) : EventLoopImplBase()

internal actual fun createEventLoop(): EventLoop = BlockingEventLoop(Thread.currentThread())

public final override fun dispatch(context: CoroutineContext, block: Runnable) = enqueue(block) //重载 dispatch函数，调用入队函数

public fun enqueue(task: Runnable) {
    if (enqueueImpl(task)) { //入队
        // todo: we should unpark only when this delayed task became first in the queue
        unpark()
    } else {
        DefaultExecutor.enqueue(task)
    }
}

@Suppress("UNCHECKED_CAST")
private fun enqueueImpl(task: Runnable): Boolean { //真正入队
    _queue.loop { queue ->
        if (isCompleted) return false // fail fast if already completed, may still add, but queues will close
        when (queue) {
            null -> if (_queue.compareAndSet(null, task)) return true //在这里入队
            is Queue<*> -> {
                when ((queue as Queue<Runnable>).addLast(task)) {
                    Queue.ADD_SUCCESS -> return true
                    Queue.ADD_CLOSED -> return false
                    Queue.ADD_FROZEN -> _queue.compareAndSet(queue, queue.next())
                }
            }
            else -> when {
                queue === CLOSED_EMPTY -> return false
                else -> {
                    // update to full-blown queue to add one more
                    val newQueue = Queue<Runnable>(Queue.INITIAL_CAPACITY, singleConsumer = true)
                    newQueue.addLast(queue as Runnable)
                    newQueue.addLast(task)
                    if (_queue.compareAndSet(queue, newQueue)) return true
                }
            }
        }
    }
}

null -> if (_queue.compareAndSet(null, task)) return true

   fun joinBlocking(): T {
        registerTimeLoopThread()
        try {
            eventLoop?.incrementUseCount()
            try {
                while (true) {
                    @Suppress("DEPRECATION")
                    if (Thread.interrupted()) throw InterruptedException().also { cancelCoroutine(it) }
                    val parkNanos = eventLoop?.processNextEvent() ?: Long.MAX_VALUE //执行队列里的下一个任务
                    // note: process next even may loose unpark flag, so check if completed before parking
                    if (isCompleted) break
                    parkNanos(this, parkNanos)
                }
            } finally { // paranoia
                eventLoop?.decrementUseCount()
            }
        } finally { // paranoia
            unregisterTimeLoopThread()
        }
        // now return result
        val state = this.state.unboxState()
        (state as? CompletedExceptionally)?.let { throw it.cause }
        return state as T
    }

override fun processNextEvent(): Long {
    // unconfined events take priority
    if (processUnconfinedEvent()) return nextTime
    // queue all delayed tasks that are due to be executed
    val delayed = _delayed.value
    if (delayed != null && !delayed.isEmpty) { //判断是否到延时时间，否则重新入队
        val now = nanoTime()
        while (true) {
            // make sure that moving from delayed to queue removes from delayed only after it is added to queue
            // to make sure that 'isEmpty' and `nextTime` that check both of them
            // do not transiently report that both delayed and queue are empty during move
            delayed.removeFirstIf {
                if (it.timeToExecute(now)) {//重新入队
                    enqueueImpl(it)
                } else
                    false
            } ?: break // quit loop when nothing more to remove or enqueueImpl returns false on "isComplete"
        }
    }
    // then process one event from queue
    dequeue()?.run() //出队并执行
    return nextTime
}

@Suppress("UNCHECKED_CAST")
private fun dequeue(): Runnable? {
    _queue.loop { queue ->
        when (queue) {
            null -> return null
            is Queue<*> -> {
                val result = (queue as Queue<Runnable>).removeFirstOrNull()
                if (result !== Queue.REMOVE_FROZEN) return result as Runnable?
                _queue.compareAndSet(queue, queue.next())
            }
            else -> when {
                queue === CLOSED_EMPTY -> return null
                else -> if (_queue.compareAndSet(queue, null)) return queue as Runnable //出队并把当前queue设为null
            }
        }
    }
}