How delay function is working in Kotlin without blocking the current thread?

Viewed 1129

Past few days I am learning coroutines, most of thee concepts are clear but I don't understand the implementation of the delay function.

How delay function is resuming the coroutine after the delayed time? For a simple program, there is only one main thread, and to resume the coroutine after the delayed time I assume there should be another timer thread that handles all the delayed invocations and invokes them later. Is it true? Can someone explain the implementation detail of the delay function?

2 Answers

TL; DR;

When using runBlocking, delay is internally wrapped and runs on same thread and when using any other dispatcher it suspends and is resumed by resuming the continuation by event-loop thread. Check the long answer below to understand the internals.

Long answer:

@Francesc answer is pointing correctly but is somewhat abstract, and still does not explains how actually delay works internally.

So, as he pointed to the delay function:

public suspend fun delay(timeMillis: Long) {
    if (timeMillis <= 0) return // don't delay
    return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->
        cont.context.delay.scheduleResumeAfterDelay(timeMillis, cont)
    }
}

What it does is "Obtains the current continuation instance inside suspend functions and suspends the currently running coroutine after running the block inside the lambda"

So this line cont.context.delay.scheduleResumeAfterDelay(timeMillis, cont) is going to be executed and then the current coroutine gets suspended i.e. frees the current thread it was stick on.

cont.context.delay points to

internal val CoroutineContext.delay: Delay get() = get(ContinuationInterceptor) as? Delay ?: DefaultDelay

that says if ContinuationInterceptor is implementation of Delay then return that otherwise use DefaultDelay which is internal actual val DefaultDelay: Delay = DefaultExecutor a DefaultExecutor which is internal actual object DefaultExecutor : EventLoopImplBase(), Runnable {...} an implementation of EventLoop and has a thread of its own to run on.

Note: ContinuationInterceptor is an implementation of Delay when coroutine is in the runBlocking block in order to make sure the delay run on same thread otherwise it is not. Check this snippet to see the results.

Now I couldn't find implemenation of Delay created by runBlocking since internal expect fun createEventLoop(): EventLoop is an expect function which is implemented from outside, not by the source. But the DefaultDelay is implemented as follows

public override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
    val timeNanos = delayToNanos(timeMillis)
    if (timeNanos < MAX_DELAY_NS) {
        val now = nanoTime()
        DelayedResumeTask(now + timeNanos, continuation).also { task ->
            continuation.disposeOnCancellation(task)
            schedule(now, task)
        }
    }
}

This is how scheduleResumeAfterDelay is implemented it creates a DelayedResumeTask with the continuation passed by delay, and then calls schedule(now, task) which calls scheduleImpl(now, delayedTask) which finally calls delayedTask.scheduleTask(now, delayedQueue, this) passing the delayedQueue in the object

@Synchronized
fun scheduleTask(now: Long, delayed: DelayedTaskQueue, eventLoop: EventLoopImplBase): Int {
    if (_heap === kotlinx.coroutines.DISPOSED_TASK) return SCHEDULE_DISPOSED // don't add -- was already disposed
    delayed.addLastIf(this) { firstTask ->
        if (eventLoop.isCompleted) return SCHEDULE_COMPLETED // non-local return from scheduleTask
        /**
         * We are about to add new task and we have to make sure that [DelayedTaskQueue]
         * invariant is maintained. The code in this lambda is additionally executed under
         * the lock of [DelayedTaskQueue] and working with [DelayedTaskQueue.timeNow] here is thread-safe.
         */
        if (firstTask == null) {
            /**
             * When adding the first delayed task we simply update queue's [DelayedTaskQueue.timeNow] to
             * the current now time even if that means "going backwards in time". This makes the structure
             * self-correcting in spite of wild jumps in `nanoTime()` measurements once all delayed tasks
             * are removed from the delayed queue for execution.
             */
            delayed.timeNow = now
        } else {
            /**
             * Carefully update [DelayedTaskQueue.timeNow] so that it does not sweep past first's tasks time
             * and only goes forward in time. We cannot let it go backwards in time or invariant can be
             * violated for tasks that were already scheduled.
             */
            val firstTime = firstTask.nanoTime
            // compute min(now, firstTime) using a wrap-safe check
            val minTime = if (firstTime - now >= 0) now else firstTime
            // update timeNow only when going forward in time
            if (minTime - delayed.timeNow > 0) delayed.timeNow = minTime
        }
        /**
         * Here [DelayedTaskQueue.timeNow] was already modified and we have to double-check that newly added
         * task does not violate [DelayedTaskQueue] invariant because of that. Note also that this scheduleTask
         * function can be called to reschedule from one queue to another and this might be another reason
         * where new task's time might now violate invariant.
         * We correct invariant violation (if any) by simply changing this task's time to now.
         */
        if (nanoTime - delayed.timeNow < 0) nanoTime = delayed.timeNow
        true
    }
    return SCHEDULE_OK
}

It finally sets the task into the DelayedTaskQueue with the current time.

// Inside DefaultExecutor
override fun run() {
    ThreadLocalEventLoop.setEventLoop(this)
    registerTimeLoopThread()
    try {
        var shutdownNanos = Long.MAX_VALUE
        if (!DefaultExecutor.notifyStartup()) return
        while (true) {
            Thread.interrupted() // just reset interruption flag
            var parkNanos = DefaultExecutor.processNextEvent() /* Notice here, it calls the processNextEvent */
            if (parkNanos == Long.MAX_VALUE) {
                // nothing to do, initialize shutdown timeout
                if (shutdownNanos == Long.MAX_VALUE) {
                    val now = nanoTime()
                    if (shutdownNanos == Long.MAX_VALUE) shutdownNanos = now + DefaultExecutor.KEEP_ALIVE_NANOS
                    val tillShutdown = shutdownNanos - now
                    if (tillShutdown <= 0) return // shut thread down
                    parkNanos = parkNanos.coerceAtMost(tillShutdown)
                } else
                    parkNanos = parkNanos.coerceAtMost(DefaultExecutor.KEEP_ALIVE_NANOS) // limit wait time anyway
            }
            if (parkNanos > 0) {
                // check if shutdown was requested and bail out in this case
                if (DefaultExecutor.isShutdownRequested) return
                parkNanos(this, parkNanos)
            }
        }
    } finally {
        DefaultExecutor._thread = null // this thread is dead
        DefaultExecutor.acknowledgeShutdownIfNeeded()
        unregisterTimeLoopThread()
        // recheck if queues are empty after _thread reference was set to null (!!!)
        if (!DefaultExecutor.isEmpty) DefaultExecutor.thread // recreate thread if it is needed
    }
}
// Called by run inside the run of DefaultExecutor
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
}

And then the event loop (run function) of internal actual object DefaultExecutor : EventLoopImplBase(), Runnable {...} finally handles the tasks by dequeuing the tasks and resuming the actual Continuation which was suspended the function by calling delay if the delay time has reached.

All suspending functions work the same way, when compiled it gets converted into a state machine with callbacks.

When you call delay what happens is that a message is posted on a queue with a certain delay, similar to Handler().postDelayed(delay) and, when the delay has lapsed, it calls back to the suspension point and resumes execution.

You can check the source code for the delay function to see how it works:

public suspend fun delay(timeMillis: Long) {
    if (timeMillis <= 0) return // don't delay
    return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->
        cont.context.delay.scheduleResumeAfterDelay(timeMillis, cont)
    }
}

So if the delay is positive, it schedules the callback in the delay time.

Related