Do you know how could I restrain the number of threads? I would like to execute a task (fetch & sync data) with only 5 threads. Here what I have done so far, but it’s not working for me.

val entryCoroutineContext = newFixedThreadPoolContext(5, "sync_entry_fixed")
        CoroutineScope(entryCoroutineContext).launch {
            val syncEntries = syncEntryDao.findAll()

            syncEntries.forEach { syncEntry ->
                Timber.d("Got this data to sync: $syncEntry") // All data are shown here I exepct just 5 to 5
                syncEntry.locked = true
                syncEntryDao.update(syncEntry)

                val response = webServices.syncEntriesCoroutine(clientName, syncEntry.date, syncEntry.blueAppId, syncEntry.offset, syncEntry.limit)
                if (response.isSuccessful) {
                    ...
                }
            }

To me it looks like you're actually not doing any work in parallel - you're doing one

launch

, and everything inside of that runs blocking? The

.forEach

call does not parallelize anything, at least if

syncEntries

is a standard list, and not an already parallelized stream or flow.

Nope and it’s my problem. The log shows that: all the output are donne at the same time.

Timber.d("Got this data to sync: $syncEntry")

I'm having trouble understanding your desired outcome. Right now, your code is iterating through all entries in your

syncEntries

list, on a single thread, one by one in sequence (not at the same time). What you could do instead is this:

syncEntries.forEach { syncEntry ->
  launch(entryCoroutineContext) {
    Timber.d("Got this data to sync: $syncEntry")
    // etc.
  }
}

This will launch a new coroutine for every syncEntry, and will run all these coroutines on your pool of 5 threads concurrently. I'm not sure that's what you actually want, though.

Yes thanks a lot. It's what I wanted.

So, well, no, it will work only as long as you won't have any suspend call inside your launch. This I learned the hard way 🙂

@Jakub Gwóźdź you should be perfectly fine calling suspend functions inside both

launch

and

async

. It might not work if those suspend functions do unexpected stuff, but in general, it should work fine! If it does not, you probably have another bug there.

@robin I'm using playground on https://kotlinlang.org to run this snippet:

data class Counter(var current:Int = 0, var top:Int = 0) {
    fun inc() { 
        synchronized(this) { 
            current++
            top = top.coerceAtLeast(current) 
        }
    }
    fun dec() {
        synchronized(this) {
            current--
        }
    }
}

val counter = Counter()

val entryCoroutineContext = newFixedThreadPoolContext(5, "sync_entry_fixed")

suspend fun main() = coroutineScope {
    (1..10).forEach { syncEntry ->
  		launch(entryCoroutineContext) {
          	counter.inc()
	        println("I'm starting $syncEntry in thread ${Thread.currentThread().name}, counter is $counter")
	        
            Thread.sleep(syncEntry * 100L)

            counter.dec()
	        println("I'm ending $syncEntry in thread ${Thread.currentThread().name}, counter is $counter")
        }
  	}
}

And it limits concurrency perfectly. But as soon as I change Thread.sleep(...) to delay(...) - well, see for yourself. I was surprised too :)

Ah, I get what you're talking about now. Yeah, using coroutines on a fixed thread pool will not limit the number of Jobs that actually run concurrently, it will only limit the number of Threads the system will use to run the Jobs, which is an important distinction. Every time a coroutine hits a suspension point (e.g. calling a function like

delay

), it might suspend and give back the thread it runs on to another coroutine that's waiting to be executed. Which is one of the major benefits of programming with coroutines - why completely block the existing thread while you're waiting, when other work could be done on it? So using a limited thread pool is only a means to restrict how many actual OS threads can be occupied with the Jobs, but not how many are run on them. Keeping the code inside the coroutine completely blocking of course circumvents that, but that's more an accident than desired behavior, really. If you want to actually restrict the number of Jobs that run concurrently, then yes, a Semaphore is exactly what you would use for that.

And again - in ideal world that would be great, but sometimes you may need to use that one library that does blocking IO and you'd like to have larger amount of threads than these given by Dispatchers.Default 🙂

Absolutely, if you're dealing with a blocking library a larger, separate thread pool could make sense! But that's a different situation than what this thread was about initially.

No problem, make sense to me what you are talking about. It's really helpful for me, too many misunderstanding in my part. Thanks a lot.

@Jakub Gwóźdź Your mix of coroutines, semaphores and `Thread.sleep()`'s brings the cold sweat 😅

thread.sleep was only to emulate time-consuming task 🙂