investigating deadlock in kotlin

FluxDeadLockInvestigation.kt

package net.FluxDeadLockInvestigation

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.reactor.asFlux
import org.junit.jupiter.api.Test
import reactor.core.scheduler.Schedulers


class FluxDeadLockInvestigation(){

    @Test
    fun deadlockInvstigation() = runBlocking<Unit>{
        (0..3000).map { async {
            //noDeadLock()
            yesDeadLock()
        }}.forEach { it.await()}
    }
    val thPool = newSingleThreadContext("single")
    suspend fun yesDeadLock(){
        withContext(Dispatchers.IO){
            var channel = Channel<Int>(Channel.RENDEZVOUS)
            val producer =
                launch(newSingleThreadContext("nconte")){
                    while (isActive  && !channel.isClosedForSend && !channel.isClosedForReceive ){
                        try{
                            channel.send(1)
                        }catch (t:Throwable){
                        }
                    }
                }

            withContext(Dispatchers.Default){
                // withContext(thPool){  // <<-- with any other context, even a single thread, this long cause deadlocks
                channel.asFlux()
                    .publishOn(Schedulers.elastic(),1)
                    .doFinally {  producer.cancel() }
                    .limitRate(1)
                    .take(30)
                    .blockLast()
            }
            1
        }
    }
}

https://stackoverflow.com/questions/56826500/channel-asflux-seems-to-cause-deadlock-in-high-concurrency-scenario

even a more minimal demonstration of the issue

class FluxDeadLockInvestigation(){

    @Test
    fun deadlockInvstigation() = runBlocking<Unit>(Dispatchers.IO){
        (0..3000).map { async {
            yesDeadLock()
        }}.forEach { it.await()}
    }
    val thPool = newSingleThreadContext("single")
    val thPool2 = newFixedThreadPoolContext(8,"single")
    suspend fun yesDeadLock(){
        withContext(Dispatchers.IO){
            var channel = Channel<Int>(Channel.RENDEZVOUS)
            val producer =
                launch(newSingleThreadContext("nconte")){
                    while (isActive  && !channel.isClosedForSend && !channel.isClosedForReceive ){
                        try{
                            channel.send(1)
                            cancel()
                        }catch (t:Throwable){
                        }
                    }
                }

            withContext(Dispatchers.Default){ // <<--- DEADLOCK
                //withContext(Dispatchers.IO){  // <<-- NO DEADLOCK
                //withContext(thPool){  // <<-- NO DEADLOCK
                //withContext(thPool2){  // <<-- NO DEADLOCK
                channel.asFlux()
                    //.subscribeOn(Schedulers.elastic())
                    .publishOn(Schedulers.elastic(),1)
                    //.doFinally {  producer.cancel() }
                    .limitRate(1)
                    //.take(30)
                    .take(1)
                    .blockLast()
            }
            1
        }
    }
}