I’ve got a suspending function in an object/class ...

# coroutines

[Bradleycorn]

I’ve got a suspending function in an object/class that runs “indefinitely” (i.e. until it’s calling coroutine is canceled). A contrived example might be:

object MyObject {

    suspend fun repeatWork() {
        while (true) {
            // do something
            delay(5_000L)
        }
    }
}

That’s fine, but now multiple things can call it, and do “multiple copies” of the work:

class SomeClass {
    fun doWork() {
        someClassScope.launch { MyObject.repeatWork() } 
    }
}

class AnotherClass {
    fun doAnotherWork() {
        anotherClassScope.launch { MyObject.repeatWork() } 
    }
}

SomeClass().doWork()
AnotherClass().doAnotherWork()

I’m trying to ensure that only one “instance” of the repeatWork routine is running at a time, no matter how may things try to call it. I was thinking something like this:

object MyObject {

    private var repeatWorkJob: Job? = null

    suspend fun repeatWork() = coroutineScope {
        if (repeatWorkJob?.isActive == true) return
   
        repeatWorkJob = launch {
            while (true) {
                // do something
                delay(5_000L)
            }
        }
    }
}

Is this the best/right way to ensure there’s only one “instance” of the repeatWork routine running at one time?

[Francesc]

no, you can have multiple threads calling this method and still start multiple coroutines, you need to add a locking mechanism so that the coroutine can only be started once

[Francesc]

https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-mutex/

[ephemient]

or just create a lazy-started job once

[ephemient]

object MyObject {
    private val repeatWorkJob = GlobalScope.launch(start = CoroutineStart.LAZY) {
        while (true) {
            // do something
            delay(5_000L)
        }
    }

    fun repeatWork() {
        repeatWorkJob.start()
    }
}

[uli]

The question would be if the work should start running again after cancelation if repeatWork is called again.

[uli]

And what would be the scope to cancel the work if it has been requested from multiple scopes

[uli]

Could a SharedFlow be of any help?

[uli]

Could be a SharedFlow<Unit> but thinking about something to produce you might even be able to improve your architecture

[Bradleycorn]

@uli - Yeah, it would need to be “restarted” if called again after cancellation. And yeah, knowing when to cancel (when it’s called multiple times) is also a challenge.

[uli]

Well, for a sharedFlow you can set it up to run the producer as long as you have subscribers (i.e. it is being collected). And then after canceltation, the next subscriber would start the producer again.

[uli]

might be hacky abuse though if you don’t really produce anything??

[Bradleycorn]

yeah, that’s why I haven’t gone down that path (yet).

[Patrick Steiger]

I’ve created an utility for that

class JobSwapper {
  private var currentJob: Job? = null
  private val mutex = Mutex()
  private val jobMutex = Mutex()

  operator fun invoke(
    scope: CoroutineScope,
    block: suspend CoroutineScope.() -> Unit,
  ) {
    scope.launch {
      mutex.withLock {
        currentJob?.cancel()
        currentJob = launch {
          jobMutex.withLock {
            coroutineScope {
              block()
            }
          }
        }
      }
    }
  }
}

[Patrick Steiger]

Usage:

val jobSwapper = JobSwapper()
…
jobSwapper(scope) {
  // only runs when previous job is completed
}

[Bradleycorn]

thanks @Patrick Steiger. It’s not exactly what I need though.
Looks like JobSwapper will cancel the current job if it’s called again:
currentJob?.cancel()

I’m looking for something that would be more similar to `join`ing the job if it’s already running. But even that doesn’t solve all problems, because the job would be tied to the original caller’s scope. when that scope gets canceled the job would cancel, even though a second caller might still be interested in results.

I think I might have to do something like @uli suggested with a SharedFlow or something similar.

[ephemient]

well that's something that GlobalScope.launch(LAZY) will do (or async if you need the results), since it's outside of its launchers' scopes. but if you want to be able to restart it, or cancel it if all current waiters are cancelled, that'll be trickier

[Bradleycorn]

yep, and I need the cancel/restart functionality.

[ephemient]

then a Flow.shareIn(started = SharingStarted.WhileSubscribed()) might be the simplest solution

[Bradleycorn]

yeah, that’s what I’m currently exploring.