Wrote a class which delays running an action. ```c...

# coroutines

[rakeeb]

Wrote a class which delays running an action.

class SuspendedInvocationTimer(
    val onTimerExpired: suspend () -> Unit,
) {

    suspend fun start(duration: Duration) = coroutineScope {
        delay(duration)
        if (isActive) {
            onTimerExpired()
        }
    }
}

Even when the parent coroutine scope is cancelled the onTimerExpired lambda is still invoked. I've tried a few variations like:

suspend fun start(duration: Duration) = coroutineScope {
    val job = launch { delay(duration) }
    job.join()
    val isCancelling = !job.isActive && !job.isCompleted && job.isCancelled
    val isCancelled = !job.isActive && job.isCompleted && job.isCancelled
    if (!isCancelling && !isCancelled) {
        onTimerExpired()
    }
}

or checking to see the parent coroutine scope as well, none of them seem to work. The onTimerExpired lambda always gets invoked even if the timer got cancelled.
What is the fault in my approach here?

[Zach Klippenstein (he/him) [MOD]]

how are you cancelling the timer?

[Zach Klippenstein (he/him) [MOD]]

also are you sure you need this class – why not just call delay() from wherever you're using it?

[Zach Klippenstein (he/him) [MOD]]

you shouldn't need the coroutineScope {}, you're not launching any coroutines

[Zach Klippenstein (he/him) [MOD]]

Are you using a NonCancellable job? Cancelling the coroutine calling start during the delay should cause delay to throw an exception, so it sounds like that's not happening

[rakeeb]

my system would either have a backoff or a heartbeat/nudge which produces events. this class was my idea of generalizing this usecase, which is why i didn't call the delay in place.

i didn't need the coroutineScope in the first iteration, apologies the first snippet was a copy paste error, I've been actively modifying this implementation. the current iteration of this timer is as follows:

class SuspendedInvocationTimer(
    private val onTimerExpired: suspend () -> Unit,
) {

    private val timerJob: AtomicReference<Job?> = AtomicReference()

    suspend fun start(duration: Duration) {
        cancel()
        timerJob.getAndSet(createCoroutineJob(duration))
    }

    suspend fun cancel() {
        val job = timerJob.get() ?: return
        job.cancelAndJoin()
        timerJob.getAndSet(null)
    }

    private suspend fun createCoroutineJob(duration: Duration) = coroutineScope {
        launch {
            delay(duration)
            timerJob.getAndSet(null)
            if (isActive) onTimerExpired()
        }
    }
}

The cancellation is explicitly made on the timer class itself.

[rakeeb]

thank you for your help Zach! you nudged me in the right direction.

using the coroutineScope was the issue, I interpreted its documentation as a way to group related coroutines together. I found out that it doesn't necessarily allow you to cancel individual coroutines inside it.

fun CoroutineScope.delayedTaskExt(
    duration: Duration,
    task: suspend () -> Unit
): Job = launch {
    delay(duration)
    task()
}

the following function me to achieve my desired behavior.

[Zach Klippenstein (he/him) [MOD]]

Looking at your other code though, just for completeness, you're misusing AtomicReference. The whole point of it is to use getAndSet to… get and set, atomically, but you're doing them as separate steps in cancel.

[Zach Klippenstein (he/him) [MOD]]

and there's no reason to use getAndSet if you're only setting – just use set

[Zach Klippenstein (he/him) [MOD]]

There are a lot of issues with SuspendedInvocationTimer.

Firstly, since createCoroutineJob won't return until the launched job finishes, timerJob will only ever be set after the onTimerExpired callback returns. I.e. it never tracks the "active" job, only already-finished jobs.

Now, if you fixed that and made it launch the job and return it immediately so it could be stored in timerJob, you'd still have a race condition and unnecessary delays in your code:
1. Coroutine A calls start while another job is active. This calls cancel which gets the job and then cancels it.
2. While cancelAndJoin is running, Coroutine B calls cancel, which gets the same job and joins on it. So now both A and B are waiting on the same job unnecessarily. This is also kind of surprising timing, since the cancel call happens after the start call but is operating on a previous job.
3. Coroutine A resumes: cancel clears timerJob, returns, and start launches the new job and sets it in timerJob.
4. Coroutine B resumes: cancel clears timerJob, which it thinks is still the old job it started canceling, but has already been set to the new job from Coroutine A's start call. Now it looks like the job was cancelled, because timerJob is null, but it's actually still running, and a subsequent cancel call will no-op.

[Zach Klippenstein (he/him) [MOD]]

the correct way to use AtomicReference for this sort of thing would be to do something like:

suspend fun start(duration: Duration) {
  val newJob = createCoroutineJob(duration)
  replaceJob(newJob)
}

suspend fun cancel() {
  replaceJob(null)
}

private fun replaceJob(newJob: Job?) {
  timerJob.getAndSet(newJob)?.cancel()
}

[rakeeb]

that's what i get for trying to speed run to a solution, thanks for keeping me honest Zach, you've raised some really good flaws with my approach and implementation. Thanks man! back to the drawing board ...

[rakeeb]

oh btw i was able to use the APIs for AtomicReference the way they are meant to be. a bit embarrassed with my ignorant first implementation. thanks again, i was able to come up with the following solution:

class CancelPreviousDeferredInvocationTimer(
    private val coroutineScope: CoroutineScope,
    private val onTimerExpired: suspend () -> Unit,
) {

    private val timerJob: AtomicReference<Job?> = AtomicReference()

    suspend fun start(duration: Duration) {
        val newJob = createJob(duration)
        newJob.invokeOnCompletion { timerJob.compareAndSet(newJob, null) }
        while (true) {
            val currentJob = timerJob.get()
            if (timerJob.compareAndSet(currentJob, newJob)) {
                currentJob?.cancelAndJoin()
                break
            }
        }
    }

    suspend fun cancel() {
        while(true) {
            val currentJob = timerJob.get() ?: break
            if (timerJob.compareAndSet(currentJob, null)) {
                currentJob.cancelAndJoin()
                break
            }
        }
    }

    fun isActive(): Boolean = timerJob.get()?.isActive ?: false

    private fun createJob(duration: Duration): Job = coroutineScope.launch {
        delay(duration)
        onTimerExpired()
    }

}

thanks again man! appreciated your help