Could I set fixed pool when using coroutines?

Support
1

I was playing with coroutines with kotlin, and met some questions when I need to execute several jobs with a fixed pool.

(1..10).forEach { n ->
        GlobalScope.launch { 
            println(n)
            delay(1000)
        }
    }

Who can teach me how to print 3 numbers for every second?

2

Documentation: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/new-fixed-thread-pool-context.html

3

I used a newFixedThreadPollContext and delay function in my coroutines, the program still printed all numbers at the first time, this was not what I expected.

I knew the new context limit the threads for the executor, so when using Thread.sleep (instead of delay), I got an expected result

val context = newFixedThreadPoolContext(3, "co")
    (1..10).forEach { n ->
        CoroutineScope(context).launch {
            println(n)
            Thread.sleep(1000)
        }
    }

However, this issued an ‘inappropriate blocking method call’ warning, and I also think use sleep in coroutines is not a good idea. So is there an alternative way?

4

You should really read the documentation on coroutines. It is all there. The idea of coroutines is that delay method does not block the thread. When you call delay, the execution is suspended and yields the processor to other tasks like running next iteration in your forEach cycle. When you call Thread.sleep, you emulate heavy duty blocking call which indeed blocks the processor. You will get the result as from regular fixed thread pool execution, but it is not the workflow for which coroutines are made. Hence the warning. Using Thread.sleep is OK for testing purposes, but not for actual production code. Also please remember not to launch blocking operation on default dispatcher.
Also it is better to use a context in launch method instead of using it as a scope. In this particular example it does not matter, but in general it is better to separate context and scope, they serve different purposes.

5

You are calling delay (which is a suspending function) as the last statement within your coroutine. You should understand that suspending functions do not block the thread, they suspend the coroutine only. Since you don’t have any other statement left in your coroutine, there is nothing to suspend or delay.

The solution to your problem will be unintuitive if you don’t understand how coroutines work. If you do, it is quite straight forward. I second @darksnake in that you should read the documentation.

6

Your is an open issue.

7

This is not what question was about. Changing implementation of newFixedThreadPool won’t change the fact that delay yield the thread for other coroutines. In the presented example, all tasks will still be completed simultaneously, without waiting for first batch to finish.

8

@darksnake I agree with you, unfortunately something like fun Mutex(permits: Int = 1) : Mutex was discarded, so it is an open issue…

9

You should foreach inside coroutine.

        GlobalScope.launch(dispatcher) {
            repeat(10) { n ->
                println(n)
                delay(1000L)
            }
        }

for more code to understand thread pool and coroutine:

    val computationExecutor: Executor = Executors.newFixedThreadPool(
            Runtime.getRuntime().availableProcessors(),
            ThreadFactoryImpl("computation", Thread.MAX_PRIORITY)
    )

    val dispatcher = object : CoroutineDispatcher() {
        override fun dispatch(context: CoroutineContext, block: Runnable) {
            computationExecutor.execute(block)
        }
    }

    val t = System.currentTimeMillis()
    repeat(2) { a ->
        GlobalScope.launch(dispatcher) {
            repeat(3) { b ->
                Log.d("tag", "a: $a, b: $b, ${Thread.currentThread()}, millis: ${System.currentTimeMillis() - t}")
                delay(1000L)
            }
        }
    }

output:

a: 1, b: 0, Thread[computation-2,10,main], millis: 11
a: 0, b: 0, Thread[computation-1,10,main], millis: 13
a: 1, b: 1, Thread[computation-3,10,main], millis: 1018
a: 0, b: 1, Thread[computation-4,10,main], millis: 1021
a: 1, b: 2, Thread[computation-2,10,main], millis: 2021
a: 0, b: 2, Thread[computation-1,10,main], millis: 2023
10

You could do something link this…

package com.example.workers

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.channels.produce
import kotlin.system.measureTimeMillis

class ChannellibgradleApplication

fun main(args: Array) {
var myList = mutableListOf(3000,1200,1400,3000,1200,1400,3000)
runBlocking {
var myChannel = produce(CoroutineName(“MyInts”)) {
myList.forEach { send(it) }
}

    println("Starting coroutineScope  ")
    var time = measureTimeMillis {
        coroutineScope {
            var workers = 2
            repeat(workers)
            {
                launch(CoroutineName("Sleep 1")) { theHardWork(myChannel) }
            }
        }
    }
    println("Ending coroutineScope  $time ms")
}

}

suspend fun theHardWork(channel : ReceiveChannel)
{
for(m in channel) {
println(“Starting Sleep $m”)
delay(m.toLong())
println(“Ending Sleep $m”)
}
}