class BasicWorkerPool<P, Q>(
        val concurrency: Int,
        val context: CoroutineContext = Dispatchers.Default,
        private val compute: suspend (P) -> Q
) : CoroutineScope {
    val job = Job(parent = context[Job])
    override val coroutineContext: CoroutineContext = context + job

    private val tasks = Channel<Pair<P, CompletableDeferred<Q>>>()

    init {
        require(concurrency > 0) { "concurrency cannot be <= 0" }
    }

    fun start() {
        repeat(concurrency) {
            launch {
                for (task in tasks) {
                    process(task)
                }
            }
        }
    }

    fun shutdown() {
        tasks.close()
    }

    fun shutdownNow() {
        tasks.close()
        job.cancel()
    }

    private suspend fun process(task: Pair<P, CompletableDeferred<Q>>) {
        val (p, deferredResult) = task
        try {
            println("processing $p in [$coroutineContext][${Thread.currentThread().name}]")
            val q = compute(p)
            deferredResult.complete(q)
        } catch (e: Throwable) {
            deferredResult.completeExceptionally(e)
        }
    }

    suspend fun execute(task: P): CompletableDeferred<Q> {
        val result = CompletableDeferred<Q>()
        tasks.send(task to result)
        return result
    }

    suspend fun executeAwait(task: P): Q {
        return execute(task).await()
    }
}

suspend fun foo1() = withContext(Dispatchers.Default) {
    val workerPool = BasicWorkerPool<Int, Int>(50, context = Dispatchers.Default) {
        delay(1000)
        it * 2
    }
    workerPool.start()
    val results = (1..100).mapIndexed { idx, _ ->
        workerPool.execute(idx)
    }
    println("submitted tasks, starting assertions")
    results.forEachIndexed { index, result ->
        check(index * 2 == result.await()) { "result not matching for $index" }
    }
    workerPool.shutdown()
    println("workerPool: ${workerPool.job.isActive};${workerPool.job.isCancelled};${workerPool.job.isCompleted}")
    println("done")
}

suspend fun foo2() = withContext(Dispatchers.Default) {
    val workerPool = BasicWorkerPool<Int, Int>(50, context = coroutineContext) {
        delay(1000)
        it * 2
    }
    workerPool.start()
    val results = (1..100).mapIndexed { idx, _ ->
        workerPool.execute(idx)
    }
    println("submitted tasks, starting assertions")
    results.forEachIndexed { index, result ->
        check(index * 2 == result.await()) { "result not matching for $index" }
    }
    workerPool.shutdown()
    println("workerPool: ${workerPool.job.isActive};${workerPool.job.isCancelled};${workerPool.job.isCompleted}")
    println("done")
}

fun main() = runBlocking {
    foo1() // works;
    foo2() // blocks indefinitely
}