Implement metrics for external queue

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

@@ -48,11 +48,11 @@ class ScalQueueBenchmark {
   @Param(Array("4")) // keep this a power of 2
   var threads: Int = _
 
-  val thing = new AnyRef
+  val thing = new Runnable { def run(): Unit = {} }
 
   @Benchmark
   def scalConcurrentEnqueueDequeue(): Unit = {
-    val q = new ScalQueue[AnyRef](threads)
+    val q = new ScalQueue(threads)
     val latch = new CountDownLatch(threads)
 
     // every thread will send and receive this number of events

build.sbt

@@ -725,11 +725,16 @@ lazy val core = crossProject(JSPlatform, JVMPlatform, NativePlatform)
       // internal API change, makes CpuStarvationMetrics available on all platforms
       ProblemFilters.exclude[MissingClassProblem](
         "cats.effect.metrics.JvmCpuStarvationMetrics$NoOpCpuStarvationMetrics"),
+      // introduced by #4292, external queue metrics
+      // WorkStealingThreadPoolMetrics is a sealed trait, so we control all of its implementations.
+      ProblemFilters.exclude[ReversedMissingMethodProblem](
+        "cats.effect.unsafe.metrics.WorkStealingThreadPoolMetrics.externalQueue"),
       ProblemFilters.exclude[MissingClassProblem]("cats.effect.metrics.CpuStarvationMetrics"),
       // package-private classes moved to the `cats.effect.unsafe.metrics` package
       ProblemFilters.exclude[MissingClassProblem]("cats.effect.metrics.CpuStarvation"),
       ProblemFilters.exclude[MissingClassProblem]("cats.effect.metrics.CpuStarvation$"),
-      ProblemFilters.exclude[MissingClassProblem]("cats.effect.metrics.CpuStarvationMBean")
+      ProblemFilters.exclude[MissingClassProblem]("cats.effect.metrics.CpuStarvationMBean"),
+      ProblemFilters.exclude[Problem]("cats.effect.unsafe.ScalQueue*")
     ) ++ {
       if (tlIsScala3.value) {
         // Scala 3 specific exclusions

core/jvm-native/src/main/scala/cats/effect/unsafe/LocalQueue.scala

@@ -193,7 +193,7 @@ private final class LocalQueue extends LocalQueuePadding {
    *   a reference to an uncontended source of randomness, to be passed along to the striped
    *   concurrent queues when executing their enqueue operations
    */
-  def enqueue(fiber: Runnable, external: ScalQueue[AnyRef], random: ThreadLocalRandom): Unit = {
+  def enqueue(fiber: Runnable, external: ScalQueue, random: ThreadLocalRandom): Unit = {
     // A plain, unsynchronized load of the tail of the local queue.
     val tl = tail
 
@@ -281,7 +281,7 @@ private final class LocalQueue extends LocalQueuePadding {
 
         // Enqueue all of the batches of fibers on the batched queue with a bulk
         // add operation.
-        external.offerAll(batches, random)
+        external.offerAllBatches(batches, random)
         // Loop again for a chance to insert the original fiber to be enqueued
         // on the local queue.
       }
@@ -657,7 +657,7 @@ private final class LocalQueue extends LocalQueuePadding {
    *   a reference to an uncontended source of randomness, to be passed along to the striped
    *   concurrent queues when executing their enqueue operations
    */
-  def drainBatch(external: ScalQueue[AnyRef], random: ThreadLocalRandom): Unit = {
+  def drainBatch(external: ScalQueue, random: ThreadLocalRandom): Unit = {
     // A plain, unsynchronized load of the tail of the local queue.
     val tl = tail
 
@@ -702,8 +702,7 @@ private final class LocalQueue extends LocalQueuePadding {
           totalSpilloverCount += SpilloverBatchSize
           Tail.updater.lazySet(this, tl)
         }
-
-        external.offer(batch, random)
+        external.offerBatch(batch, random)
         return
       }
     }