Execute IO.blocking on WSTP without BlockContext indirection

core/jvm/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala

@@ -485,14 +485,16 @@ private[effect] final class WorkStealingThreadPool(
   }
 
   /**
-   * Checks if the blocking code can be executed in the current context (only returns true for
-   * worker threads that belong to this execution context).
+   * Checks if the blocking code can be executed in the current context and, if so, prepares it
+   * for blocking. Only returns true for worker threads that belong to this execution context.
    */
   private[effect] def canExecuteBlockingCode(): Boolean = {
     val thread = Thread.currentThread()
     if (thread.isInstanceOf[WorkerThread]) {
       val worker = thread.asInstanceOf[WorkerThread]
-      worker.canExecuteBlockingCodeOn(this)
+      if (worker.canExecuteBlockingCodeOn(this))
+        worker.prepareBlocking()
+      true
     } else {
       false
     }

core/jvm/src/main/scala/cats/effect/unsafe/WorkerThread.scala

@@ -776,7 +776,7 @@ private final class WorkerThread(
   }
 
   /**
-   * A mechanism for executing support code before executing a blocking action.
+   * Support code that must be run before executing a blocking action on this thread.
    *
    * The current thread creates a replacement worker thread (or reuses a cached one) that will
    * take its place in the pool and does a complete transfer of ownership of the data structures
@@ -797,16 +797,14 @@ private final class WorkerThread(
    *   There is no reason to enclose any code in a `try/catch` block because the only way this
    *   code path can be exercised is through `IO.delay`, which already handles exceptions.
    */
-  override def blockOn[T](thunk: => T)(implicit permission: CanAwait): T = {
+  def prepareBlocking(): Unit = {
     val rnd = random
 
     pool.notifyParked(rnd)
 
     if (blocking) {
-      // This `WorkerThread` is already inside an enclosing blocking region.
-      // There is no need to spawn another `WorkerThread`. Instead, directly
-      // execute the blocking action.
-      thunk
+      // This `WorkerThread` has already been prepared for blocking.
+      // There is no need to spawn another `WorkerThread`.
     } else {
       // Spawn a new `WorkerThread` to take the place of this thread, as the
       // current thread prepares to execute a blocking action.
@@ -819,7 +817,7 @@ private final class WorkerThread(
         cedeBypass = null
       }
 
-      // Logically enter the blocking region.
+      // Logically become a blocking thread.
       blocking = true
 
       val prefix = pool.blockerThreadPrefix
@@ -853,11 +851,17 @@ private final class WorkerThread(
         pool.blockedWorkerThreadCounter.incrementAndGet()
         clone.start()
       }
-
-      thunk
     }
   }
 
+  /**
+   * A mechanism for executing support code before executing a blocking action.
+   */
+  override def blockOn[T](thunk: => T)(implicit permission: CanAwait): T = {
+    prepareBlocking()
+    thunk
+  }
+
   private[this] def init(newIdx: Int): Unit = {
     _index = newIdx
     queue = pool.localQueues(newIdx)

core/shared/src/main/scala/cats/effect/IOFiber.scala

@@ -994,7 +994,7 @@ private final class IOFiber[A](
                 var error: Throwable = null
                 val r =
                   try {
-                    scala.concurrent.blocking(cur.thunk())
+                    cur.thunk()
                   } catch {
                     case t if NonFatal(t) =>
                       error = t