Rename worker threads in blocking regions

benchmarks/src/main/scala/cats/effect/benchmarks/WorkStealingBenchmark.scala

@@ -174,7 +174,8 @@ class WorkStealingBenchmark {
       (Scheduler.fromScheduledExecutor(executor), () => executor.shutdown())
     }
 
-    val compute = new WorkStealingThreadPool(256, "io-compute", manyThreadsRuntime)
+    val compute =
+      new WorkStealingThreadPool(256, "io-compute", "io-blocker", manyThreadsRuntime)
 
     val cancelationCheckThreshold =
       System.getProperty("cats.effect.cancelation.check.threshold", "512").toInt

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

@@ -33,9 +33,11 @@ private[unsafe] abstract class IORuntimeCompanionPlatform { this: IORuntime.type
   def createDefaultComputeThreadPool(
       self: => IORuntime,
       threads: Int = Math.max(2, Runtime.getRuntime().availableProcessors()),
-      threadPrefix: String = "io-compute"): (WorkStealingThreadPool, () => Unit) = {
+      threadPrefix: String = "io-compute",
+      blockerThreadPrefix: String = "io-compute-blocker")
+      : (WorkStealingThreadPool, () => Unit) = {
     val threadPool =
-      new WorkStealingThreadPool(threads, threadPrefix, self)
+      new WorkStealingThreadPool(threads, threadPrefix, blockerThreadPrefix, self)
 
     val unregisterMBeans =
       if (isStackTracing) {

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

@@ -58,6 +58,7 @@ import java.util.concurrent.locks.LockSupport
 private[effect] final class WorkStealingThreadPool(
     threadCount: Int, // number of worker threads
     private[unsafe] val threadPrefix: String, // prefix for the name of worker threads
+    private[unsafe] val blockerThreadPrefix: String, // prefix for the name of worker threads currently in a blocking region
     self0: => IORuntime
 ) extends ExecutionContext {
 

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

@@ -596,6 +596,10 @@ private final class WorkerThread(
       // Logically enter the blocking region.
       blocking = true
 
+      val prefix = pool.blockerThreadPrefix
+      // Set the name of this thread to a blocker prefixed name.
+      setName(s"$prefix-$nameIndex")
+
       val cached = pool.cachedThreads.pollFirst()
       if (cached ne null) {
         // There is a cached worker thread that can be reused.
@@ -630,6 +634,10 @@ private final class WorkerThread(
     queue = pool.localQueues(newIdx)
     parked = pool.parkedSignals(newIdx)
     fiberBag = pool.fiberBags(newIdx)
+
+    // Reset the name of the thread to the regular prefix.
+    val prefix = pool.threadPrefix
+    setName(s"$prefix-$newIdx")
   }
 
   /**

tests/jvm/src/test/scala/cats/effect/RunnersPlatform.scala

@@ -35,7 +35,8 @@ trait RunnersPlatform extends BeforeAfterAll {
     val (compute, compDown) =
       IORuntime.createDefaultComputeThreadPool(
         runtime0,
-        threadPrefix = s"io-compute-${getClass.getName}")
+        threadPrefix = s"io-compute-${getClass.getName}",
+        blockerThreadPrefix = s"io-blocker-${getClass.getName}")
 
     runtime0 = IORuntime(
       compute,

tests/jvm/src/test/scala/cats/effect/unsafe/StripedHashtableSpec.scala

@@ -37,7 +37,8 @@ class StripedHashtableSpec extends BaseSpec {
       val (compute, compDown) =
         IORuntime.createDefaultComputeThreadPool(
           rt,
-          threadPrefix = s"io-compute-${getClass.getName}")
+          threadPrefix = s"io-compute-${getClass.getName}",
+          blockerThreadPrefix = s"io-blocker-${getClass.getName}")
 
       IORuntime(
         compute,

tests/jvm/src/test/scala/cats/effect/unsafe/WorkerThreadNameSpec.scala

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2020-2022 Typelevel
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cats.effect.unsafe
+
+import cats.effect.{BaseSpec, IO}
+import cats.effect.testkit.TestInstances
+import cats.syntax.all._
+
+class WorkerThreadNameSpec extends BaseSpec with TestInstances {
+
+  override def runtime(): IORuntime = {
+    lazy val rt: IORuntime = {
+      val (blocking, blockDown) =
+        IORuntime.createDefaultBlockingExecutionContext(threadPrefix =
+          s"io-blocking-${getClass.getName}")
+      val (scheduler, schedDown) =
+        IORuntime.createDefaultScheduler(threadPrefix = s"io-scheduler-${getClass.getName}")
+      val (compute, compDown) =
+        IORuntime.createDefaultComputeThreadPool(
+          rt,
+          threads = 1,
+          threadPrefix = s"io-compute-${getClass.getName}",
+          blockerThreadPrefix = s"io-blocker-${getClass.getName}")
+
+      IORuntime(
+        compute,
+        blocking,
+        scheduler,
+        { () =>
+          compDown()
+          blockDown()
+          schedDown()
+        },
+        IORuntimeConfig()
+      )
+    }
+
+    rt
+  }
+
+  "WorkerThread" should {
+    "rename itself when entering and exiting blocking region" in real {
+      for {
+        computeThread <- threadInfo
+        (computeThreadName, _) = computeThread
+        blockerThread <- IO.blocking(threadInfo).flatten
+        (blockerThreadName, blockerThreadId) = blockerThread
+        _ <- IO.cede
+        // Force the previously blocking thread to become a compute thread by converting
+        // the pool of compute threads (size=1) to blocker threads
+        resetComputeThreads <- List.fill(2)(threadInfo <* IO.blocking(())).parSequence
+      } yield {
+        // Start with the regular prefix
+        computeThreadName must startWith("io-compute")
+        // Check that entering a blocking region changes the name
+        blockerThreadName must startWith("io-blocker")
+        // Check that the same thread is renamed again when it is readded to the compute pool
+        resetComputeThreads.exists {
+          case (name, id) => id == blockerThreadId && name.startsWith("io-compute")
+        } must beTrue
+      }
+    }
+  }
+
+  private val threadInfo =
+    IO((Thread.currentThread().getName(), Thread.currentThread().getId()))
+
+}