FTR another solution was considered: let runBlocking release the thread into CPU-pool before joinBlocking, so that the blocked thread would be considered as IO-bound, and Dispatchers.Default could use another thread to run submitted CPU tasks. But this will lead to starvation because IO pool is limited, or this will lead to OutOfMemoryError if set to effectively-unlimited via kotlinx.coroutines.io.parallelism. To be able to progress, the thread has to execute dispatched tasks while being blocked, like FJP does.

There is still a problem with thread-locals. We can manage our own thread-locals and reset them before running the getSingleDispatchersDefaultTaskOrNull task in the current thread, and it would be great to have a handle or a function to clear ThreadContextElement stuff, something like

suspend fun <X> resetThreadContext(/*non-suspending*/ action: () -> X): X {
  restoreThreadContext()
  try {
    return action()
  }
  finally {
    updateThreadContext() // install thread context back
  }
}

runBlocking {
  val pumpQueueJob = launch {
    while (true) {
      val task = getSingleDispatchersDefaultTaskOrNull()
      if (task != null) {
        resetThreadContext { // clean context
          task.run()
        }
      }
      yield()
    }
  }
  try {
    decompose() 
  }
  finally {
    pumpQueueJob.cancel()
  }
}

I would love to know how we can implement non-transparent restoreThreadContext.

AFAIR there is no API in Java to clean up and later recover all thread locals.
But, given that we know the current ThreadContextElement (e.g. if it works for you to pass it as an argument or because function itself is suspend), we can probably provide an API for isolating these specific elements.

We've recently had a starved thread situation after incorrect migration to coroutines in IJ (https://youtrack.jetbrains.com/issue/IDEA-302358), but it was using Dispatchers.IO.limitedParallelism(xx).
Thus I'd like to update the request to:

1. API to retrieve a task from a dispatcher
2. API to determine the current dispatcher by the current CoroutineScheduler.Worker thread

We are interested in Dispatchers.Default, Dispatchers.IO and limitedParallelism on any of them.

Another possible approach is to provide a single-function API with the semantics like this:

// NB return type is intentionally Nothing
public suspend fun processEventsInCurrentThread(): Nothing {
    assert {
        ThreadLocalEventLoop.currentOrNull() != null // only allowed inside runBlocking
    }
    val thread = Thread.currentThread()
    if (thread !is CoroutineScheduler.Worker) {
        awaitCancellation()
    }
    while (true) {
        // TASK_NON_BLOCKING for CPU-bound thread or TASK_PROBABLY_BLOCKING in IO-bound thread
        val delayTime = thread.executeSingleTaskInTheCurrentMode()
        if (delayTime >= 0L) {
            delay(delayTime)
        } else {
            yield()
        }
    }
}

Attaching unfinished attempt to implement this as per request in a private conversation
3439.patch.zip