Limiting the maximum number of coroutines that can run in a scope

user_1771187 0
Viewed 5177

I am translating our current application from Java to Kotlin and I came upon this problem.

The java implementation used to use threads to transfer data from the server.

It would create about 100 different threads that would request data, but from what I have seen no more than 4 would run at a time, the others would wait for a thread to finish before starting.

When translating it to Kotlin I used Coroutines

This creates a problem because apparently the server can't handle 100 requests actually being sent.

All coroutines are launched in the same scope , so it's something like this:

//this is a custom scope that launches on Dispatchers.IO + a job that I can use to cancel everything
transferScope.launch {
     //loadData is a suspending function that returns true/false 
     val futures = mDownloadJobs.map{ async { it.loadData() } }
     val responses = futures.awaitAll()
     //check that everything in responses is true etc....
}

Is there a way to make the specific transferScope to allow only up to 5 coroutines at a time and then when one finishes let a different one run? (I do not care about the order)

If it can't be done through the scope, is there a different way this can be achieved?

kotlin coroutine android
4 Answers

Require each coroutine to acquire a Kotlin Semaphore permit from a total of 5 permits before making a request.

Something like this:

    import kotlinx.coroutines.sync.Semaphore

    val requestSemaphore = Semaphore(5)

    val futures = mDownloadJobs.map {
        async {
            // Will limit number of concurrent requests to 5
            requestSemaphore.withPermit {
                it.loadData()
            }
        }
    }

    val responses = futures.awaitAll()
user_2877610

I believe you should Channel and limit the creation of coroutine that you are creating.

val channel = Channel<Job>()

transferScope.launch {
  mDownloadJobs.forEach { channel.send(it) }
  channel.close() // don't forget to close the channel
}

coroutineScope {
    val responses = mutableListOf<Any>()
    repeat(5).map { 
      launch {
        for (job in mDownloadJobsChannel) {
          responses.add(jobs.loadData())
        }
      }
    }
}

Parallelization in this case is 5 coroutines.

I did not test this code :D and I am sure there are cleaner ways to do this.

user_8943170

You can do something like this, group the requests into chunks of 4, launch coroutines to process them and wait till that group is finished before launching a new one.

requests.chunked(4).forEachIndexed { index, chunk ->
    coroutineScope {
        LOG("processing chunk $index")
        chunk.forEach {
            launch {
                delay(100)
            }
        }
        LOG("done processing $index")
    }
}
user_18139710

Dispatchers.IO claims to create a thread pool and restrict span out to that pool. It's docstring tells you how to change the pool size (a system property tho).

user_3753650
Related