Is there way to get FixedTreadPool behavior using coroutines?

user_9017417 0
Viewed 101

Is there way to get the same behavior as code snippet below has but using coroutines?

Updated code snippet:

fun main(args: Array<String>) = runBlocking {
    val executor = Executors.newFixedThreadPool(50)
    log.info("Start")
    val jobs = List(300) {
        executor.submit {
            log.info("worker #$it started")
            sleep(1000L)
            log.info("worker #$it done")
        }
    }
    jobs.forEach { it.get() }
    executor.shutdown()
    log.info("All done!")
}

How can I run 300 jobs with parallelism factor == 50, but without creation 50 real threads?

Update 2: Solution

After reading Coroutines Guide one more time, I've found that Fan-out example is exactly what I was looking for. Thus, my example will look as follows:

fun produceTasks() = produce {
    for (taskId in 1..300) {
        send(
                async(start = CoroutineStart.LAZY) {
                    delay(1000) // simulate long work
                    taskId
                }
        )
    }
    close()
}

fun launchWorker(index: Int, channel: ProducerJob<Deferred<Int>>) = launch {
    channel.consumeEach {
        val result = it.await()
        log.info("Worker #$index done task #$result")
    }
}

fun main(args: Array<String>) = runBlocking {
    val tasks = produceTasks()
    val workers = List(50) { launchWorker(it + 1, tasks) }
    workers.forEach { it.join() }
    log.info("Done")
}
kotlinx.coroutines kotlin
1 Answers

First of all - please check issue on Gitbub, possible there are official solution.

You can use coroutine channels for this logic

/**
 * How it works:
 * 1. Executors queue has N items
 * 2. Before each execution:
 * 2.1. Get item from queue (if it is possible)
 * 2.2. Execute action
 * 2.3. Return item to queue
 *
 * Example below allows us to execute methods in the resticted count of threads.
 * However we can create object pool from this Semaphore:
 * 1. Firstly - push N pooled objects into executors variable (e.g. we will have Channel<PooledObject>
 * 2. Then change signature of schedule method: just change arg "()" to PooledObject
 */
class Semapshore(maxParallelWorkers: Int) {
    private val executors = Channel<Unit>(maxParallelWorkers).apply {
        runBlocking {
            repeat(maxParallelWorkers) {
                send(Unit)
            }
        }
    }

    suspend fun <TResult> schedule(func: () -> TResult): TResult {
        val executor = executors.receive()

        return try {
            func()
        } finally {
            executors.send(executor)
        }
    }
}
user_1854980
Related