How to properly initialize such kind of patterns? ...

# coroutines

[Tolriq]

How to properly initialize such kind of patterns?

Currently depending on the state of the device and load, the repeat call can trigger creation of tons of threads even if there's no tasks yet.
Generating a lot of unecessary load at init time.

class XXX  : CoroutineScope {

	override val coroutineContext = <http://Dispatchers.IO|Dispatchers.IO> + SupervisorJob()

	internal val tasks = Channel<Task>(UNLIMITED)

    fun startWorkers() {
        repeat(150) {
            launch {
                for (task in tasks) executeTask(task)
            }
        }
    }
}

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

What if you use the Default dispatcher (smaller, fixed-size thread pool, so you’re not gonna create a bunch of threads), then inside each of the workers you wait to switch to the IO dispatcher until you have the first task?

launch(Unconfined) {
  var initialTask = tasks.receive()
  withContext(IO) {
    executeTask(initialTask)
    for (task in tasks) executeTask(task)
  }
}

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

I think you could even use Unconfined to launch them, since the only work they’d be doing initially would be registering to receive from the channel.

[Tolriq]

The problem is that your solution loose the parallelism configuration. At some point I need to start at max X for loops. 150 in the posted sample for example.

[Tolriq]

Maybe @elizarov have an official pattern for that. I suppose I could wrap the repeat in a coroutine and add a small delay() after each launch as a workaround.

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

You only lose parallelism while waiting for the first emission from the channel, but that doesn’t matter – the coroutines aren’t doing any work then anyway, they’re just waiting. As soon as you switch to the IO dispatcher you get the full parallelism of the IO dispatcher back.

[Tolriq]

I’m talking about the parallelism in active running tasks. the repeat ensure the number of simultaneous tasks. Without 1 launch + for loop, you only have 1 running task. And if I do the launch in the for loop then I loose control over max running tasks too.

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

oh, sorry i meant you’d still have the repeat. i was just referring to the code inside it. So,

fun startWorkers() {
  repeat(150) {
    startWorker()
  }
}

private fun startWorker() {
  launch(Unconfined) {
    var initialTask = tasks.receive()
    withContext(IO) {
      executeTask(initialTask)
      for (task in tasks) executeTask(task)
    }
  }
}

[Tolriq]

Ho yes I'm stupid 🙂 Thanks this should work nicely.

[Tolriq]

For the record tasks.receive() throw on channel closing unlike the for loop. So call require try catch ClosedReceiveChannelException. And it seems there's maybe a side effect of using Unconfined leading to withContext not always switching. Will investigate more on that last one.

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

Oh yikes, that seems bad. Is there an issue?

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

Anyway, you can use Default too for that initial dispatcher.

[Tolriq]

That issue was due to me being too tired and inheriting the wrong coroutineContext,.With Unconfined there's strange difference in the number of launch but I did not investigate that much and use my global background dispatcher, that fit the needs and solve my issue.

[Tolriq]

repeat(maxConcurrentTasks) {
            launch(Dispatchers.Unconfined) {
                Log.e("AAAA", "Launch ${Thread.currentThread().name}" )
                val initialTask = try {
                    tasks.receive()
                } catch (e: Exception) {
                    return@launch
                }
                withContext([e-mail]) {
                    Log.e("AAAA", "Run ${Thread.currentThread().name}" )
                    executeTask(initialTask)
                    for (task in tasks) executeTask(task)
                }
            }
        }

Will only log 4 Launch main thread

[Tolriq]

Using Dispatchers.Default or IO will properly log 1 Launch per repeat.