Merge pull request #4186 from iRevive/metrics-pool-id

Use global counter as an MBean ID for WSTP

Author: djspiewak

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

@@ -104,11 +104,11 @@ private[unsafe] abstract class IORuntimeCompanionPlatform { this: IORuntime.type
         if (mBeanServer ne null) {
           val registeredMBeans = mutable.Set.empty[ObjectName]
 
-          val hash = System.identityHashCode(threadPool).toHexString
+          val threadPoolId = threadPool.id
 
           try {
             val computePoolSamplerName = new ObjectName(
-              s"cats.effect.unsafe.metrics:type=ComputePoolSampler-$hash")
+              s"cats.effect.unsafe.metrics:type=ComputePoolSampler-$threadPoolId")
             val computePoolSampler = new ComputePoolSampler(threadPool)
             mBeanServer.registerMBean(computePoolSampler, computePoolSamplerName)
             registeredMBeans += computePoolSamplerName
@@ -125,7 +125,7 @@ private[unsafe] abstract class IORuntimeCompanionPlatform { this: IORuntime.type
 
             try {
               val localQueueSamplerName = new ObjectName(
-                s"cats.effect.unsafe.metrics:type=LocalQueueSampler-$hash-$i")
+                s"cats.effect.unsafe.metrics:type=LocalQueueSampler-$threadPoolId-$i")
               val localQueueSampler = new LocalQueueSampler(localQueue)
               mBeanServer.registerMBean(localQueueSampler, localQueueSamplerName)
               registeredMBeans += localQueueSamplerName
@@ -264,11 +264,10 @@ private[unsafe] abstract class IORuntimeCompanionPlatform { this: IORuntime.type
         }
 
       if (mBeanServer ne null) {
-        val hash = System.identityHashCode(fiberMonitor).toHexString
-
         try {
+          val mbeanId = LiveFiberSnapshotTrigger.IdCounter.getAndIncrement()
           val liveFiberSnapshotTriggerName = new ObjectName(
-            s"cats.effect.unsafe.metrics:type=LiveFiberSnapshotTrigger-$hash")
+            s"cats.effect.unsafe.metrics:type=LiveFiberSnapshotTrigger-$mbeanId")
           val liveFiberSnapshotTrigger = new LiveFiberSnapshotTrigger(fiberMonitor)
           mBeanServer.registerMBean(liveFiberSnapshotTrigger, liveFiberSnapshotTriggerName)
 

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

@@ -42,7 +42,7 @@ import java.time.Instant
 import java.time.temporal.ChronoField
 import java.util.Comparator
 import java.util.concurrent.{ConcurrentSkipListSet, ThreadLocalRandom}
-import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger, AtomicReference}
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger, AtomicLong, AtomicReference}
 
 import WorkStealingThreadPool._
 
@@ -77,6 +77,10 @@ private[effect] final class WorkStealingThreadPool[P <: AnyRef](
   import TracingConstants._
   import WorkStealingThreadPoolConstants._
 
+  // a unique identifier of the thread pool within a JVM
+  // used in the MBean name and as an identifier in metrics
+  private[unsafe] val id = WorkStealingThreadPool.IdCounter.getAndIncrement()
+
   /**
    * References to worker threads and their local queues.
    */
@@ -840,6 +844,8 @@ private[effect] final class WorkStealingThreadPool[P <: AnyRef](
 
 private object WorkStealingThreadPool {
 
+  private val IdCounter: AtomicLong = new AtomicLong(0)
+
   /**
    * A wrapper for a cancelation callback that is created asynchronously.
    */

core/jvm/src/main/scala/cats/effect/unsafe/metrics/LiveFiberSnapshotTrigger.scala

@@ -19,6 +19,8 @@ package metrics
 
 import scala.collection.mutable.ArrayBuffer
 
+import java.util.concurrent.atomic.AtomicLong
+
 /**
  * An implementation of the [[LiveFiberSnapshotTriggerMBean]] interface which simply delegates
  * to the corresponding method of the backing [[cats.effect.unsafe.FiberMonitor]].
@@ -34,3 +36,7 @@ private[unsafe] final class LiveFiberSnapshotTrigger(monitor: FiberMonitor)
     buffer.toArray
   }
 }
+
+private[unsafe] object LiveFiberSnapshotTrigger {
+  val IdCounter: AtomicLong = new AtomicLong(0)
+}

core/jvm/src/main/scala/cats/effect/unsafe/metrics/WorkStealingPoolMetrics.scala

@@ -25,10 +25,10 @@ import scala.concurrent.ExecutionContext
 sealed trait WorkStealingPoolMetrics {
 
   /**
-   * The hash code of the instrumented work-stealing thread pool. This hash uniquely identifies
-   * the specific thread pool.
+   * The identifier of the instrumented work-stealing thread pool. Uniquely identifies a
+   * specific thread pool within a JVM.
    */
-  def hash: String
+  def identifier: String
 
   /**
    * Compute-specific metrics of the work-stealing thread pool.
@@ -205,8 +205,8 @@ object WorkStealingPoolMetrics {
     ec match {
       case wstp: WorkStealingThreadPool[_] =>
         val metrics = new WorkStealingPoolMetrics {
-          val hash: String =
-            System.identityHashCode(wstp).toHexString
+          val identifier: String =
+            wstp.id.toString
 
           val compute: ComputeMetrics =
             computeMetrics(wstp)