Properly cancelling kotlin coroutine job

Question

I'm scratching my head around properly cancelling coroutine job. Test case is simple I have a class with two methods:

class CancellationTest {
   private var job: Job? = null
   private var scope = MainScope()

   fun run() {
      job?.cancel()
      job = scope.launch { doWork() }
   }

   fun doWork() {
      // gets data from some source and send it to BE
   }
}

Method doWork has an api call that is suspending and respects cancellation.

In the above example after counting objects that were successfully sent to backend I can see many duplicates, meaning that cancel did not really cancel previous invocation.

However if I use snippet found on the internet

internal class WorkingCancellation<T> {
    private val activeTask = AtomicReference<Deferred<T>?>(null)
    suspend fun cancelPreviousThenRun(block: suspend () -> T): T {
        activeTask.get()?.cancelAndJoin()

        return coroutineScope {
            val newTask = async(start = CoroutineStart.LAZY) {
                block()
            }

            newTask.invokeOnCompletion {
                activeTask.compareAndSet(newTask, null)
            }

            val result: T

            while (true) {
                if (!activeTask.compareAndSet(null, newTask)) {
                    activeTask.get()?.cancelAndJoin()
                    yield()
                } else {
                    result = newTask.await()
                    break
                }
            }

            result
        }
    }
}

It works properly, objects are not duplicated and sent properly to BE. One last thing is that I'm calling run method in a for loop - but anyways I'm not quire sure I understand why job?.cancel does not do its job properly and WorkingCancellation is actually working

Answer 1

class JobManager(private val coroutineScope: CoroutineScope) {

  private var currentJob: Job? = null

  fun cancelPreviousAndLaunch(block: suspend () -> Unit) {
    currentJob?.cancel()
    currentJob = coroutineScope.launch {
      block()
    }  
 } 
}


    private fun example() {
        jobManager.cancelPreviousAndLaunch {
   }
}

JobManager Class: private val coroutineScope: CoroutineScope : A CoroutineScope object is provided as a private variable. coroutineScope represents the scope of the scope of Coordinated jobs. While JobManager provides its own scope, an externally provided scope can also be used.

Current Job: Is it a job? =empty: Editing it as a custom variable that holds a Business object. This variable is used to track the currently running Coordinated job. cancelPreciousAndLaunch(block: suspend () -> Unit) Function: This function cancels the previous Coordinated job and then a new Coordinated job starts.

block:suspend() -> Unit: This is a suspendable function that contains the code of the newly initialized Coordinated part. Inside the function: currentJob?.cancel(): If there is a previous Coordinated job, it is canceled using the cancel() function.

currentJob = coroutineScope.launch {block() }: A new Coordinated job is started by calling the launch function via CoroutineScope. It contains the block function, the number of parts in Coordinate and is assigned to the currentJob variable.

Answer 2

In addition to Sam's answer, consider this example that mocks a continuous transaction, lets say location updates to a server.

var pingInterval = System.currentTimeMillis()
job = launch {
     while (true) {
         if (System.currentTimeMillis() > pingInterval) {
             Log.e("LocationJob", "Executing location updates... ")
             pingInterval += 1000L
         }
     }
}

Continuously it will "ping" the server with location udpates, or like any other common use-cases, say this will continuously fetch something from it.

Then I have a function here that's being called by a button that cancels this job operation.

fun cancel() {
    job.cancel()
    Log.e("LocationJob", "Location updates done.")
}

When this function is called, the job is cancelled, however the operation keeps on going because nothing ensures the coroutine scope to stop working, all actions above will print

 Ping server my location...
 Ping server my location...
 Ping server my location...
 Ping server my location...
 Location updates done.
 Ping server my location...
 Ping server my location...

Now if we insert ensureActive() inside the infinite loop

while (true) {
     ensureActive()
     if (System.currentTimeMillis() > pingInterval) {
          Log.e("LocationJob", "Ping server my location... ")
           pingInterval += 1000L
     }
}

Cancelling the job will guarantee that the operation will stop. I tested using delay though and it guaranteed total cancellation when the job it is being called in is cancelled. Emplacing ensureActive(), and cancelling after 2 seconds, prints

 Ping server my location...
 Ping server my location...
 Location updates done.

Answer 3

Short answer: cancellation only works out-of-the box if you call suspending library functions. Non-suspending code needs manual checks to make it cancellable.

Cancellation in Kotlin coroutines is cooperative, and requires the job being cancelled to check for cancellation and terminate whatever work it's doing. If the job doesn't check for cancellation, it can quite happily carry on running forever and never find out it has been cancelled.

Coroutines automatically check for cancellation when you call built-in suspending functions. If you look at the docs for commonly-used suspending functions like await() and yield(), you'll see that they always say "This suspending function is cancellable".

Your doWork isn't a suspend function, so it can't call any other suspending functions and consequently will never hit one of those automatic checks for cancellation. If you do want to cancel it, you will need to have it periodically check whether the job is still active, or change its implementation to use suspending functions. You can manually check for cancellation by calling ensureActive on the Job.