Implement metrics for external queue

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

@@ -281,9 +281,10 @@ 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)
-        // Loop again for a chance to insert the original fiber to be enqueued
-        // on the local queue.
+        //  (loop through each batch)
+        for (batch <- batches) {
+          external.offer(batch, random)
+        }
       }
 
       // None of the three final outcomes have been reached, loop again for a
@@ -702,7 +703,6 @@ private final class LocalQueue extends LocalQueuePadding {
           totalSpilloverCount += SpilloverBatchSize
           Tail.updater.lazySet(this, tl)
         }
-
         external.offer(batch, random)
         return
       }

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

@@ -17,6 +17,7 @@
 package cats.effect.unsafe
 
 import java.util.concurrent.{ConcurrentLinkedQueue, ThreadLocalRandom}
+import java.util.concurrent.atomic.AtomicLong
 
 /**
  * A striped queue implementation inspired by the [[https://scal.cs.uni-salzburg.at/dq/ Scal]]
@@ -33,6 +34,12 @@ import java.util.concurrent.{ConcurrentLinkedQueue, ThreadLocalRandom}
  */
 private[effect] final class ScalQueue[A <: AnyRef](threadCount: Int) {
 
+  // Metrics counters for tracking external queue submissions
+  private val singletonsSubmittedCount = new AtomicLong(0)
+  private val singletonsPresentCount = new AtomicLong(0)
+  private val batchesSubmittedCount = new AtomicLong(0)
+  private val batchesPresentCount = new AtomicLong(0)
+
   /**
    * Calculates the next power of 2 using bitwise operations. This value actually represents the
    * bitmask for the next power of 2 and can be used for indexing into the array of concurrent
@@ -80,7 +87,15 @@ private[effect] final class ScalQueue[A <: AnyRef](threadCount: Int) {
   def offer(a: A, random: ThreadLocalRandom): Unit = {
     val idx = random.nextInt(numQueues)
     queues(idx).offer(a)
-    ()
+
+    // Track metrics - if it's a runnable but not an array, it's a singleton task
+    if (!a.isInstanceOf[Array[?]]) {
+      singletonsSubmittedCount.incrementAndGet()
+      val _ = singletonsPresentCount.incrementAndGet()
+    } else {
+      batchesSubmittedCount.incrementAndGet()
+      val _ = batchesPresentCount.incrementAndGet()
+    }
   }
 
   /**
@@ -104,7 +119,7 @@ private[effect] final class ScalQueue[A <: AnyRef](threadCount: Int) {
    * @param random
    *   an uncontended source of randomness, used for randomly choosing a destination queue
    */
-  def offerAll(as: Array[? <: A], random: ThreadLocalRandom): Unit = {
+  def offerAll(as: Array[A], random: ThreadLocalRandom): Unit = {
     val nq = numQueues
     val len = as.length
     var i = 0
@@ -114,6 +129,10 @@ private[effect] final class ScalQueue[A <: AnyRef](threadCount: Int) {
       queues(idx).offer(fiber)
       i += 1
     }
+
+    // Track as batch submissions
+    batchesSubmittedCount.incrementAndGet()
+    val _ = batchesPresentCount.incrementAndGet()
   }
 
   /**
@@ -129,15 +148,23 @@ private[effect] final class ScalQueue[A <: AnyRef](threadCount: Int) {
     val nq = numQueues
     val from = random.nextInt(nq)
     var i = 0
-    var a = null.asInstanceOf[A]
+    var element = null.asInstanceOf[A]
 
-    while ((a eq null) && i < nq) {
+    while ((element eq null) && i < nq) {
       val idx = (from + i) & mask
-      a = queues(idx).poll()
+      element = queues(idx).poll()
       i += 1
     }
 
-    a
+    if (element != null) {
+      if (element.isInstanceOf[Array[?]]) {
+        val _ = batchesPresentCount.decrementAndGet()
+      } else {
+        val _ = singletonsPresentCount.decrementAndGet()
+      }
+    }
+
+    element
   }
 
   /**
@@ -156,7 +183,7 @@ private[effect] final class ScalQueue[A <: AnyRef](threadCount: Int) {
    * @param a
    *   the element to be removed
    */
-  def remove(a: A): Unit = {
+  def remove(a: A): Boolean = {
     val nq = numQueues
     var i = 0
     var done = false
@@ -165,6 +192,16 @@ private[effect] final class ScalQueue[A <: AnyRef](threadCount: Int) {
       done = queues(i).remove(a)
       i += 1
     }
+
+    if (done) {
+      if (a.isInstanceOf[Array[?]]) {
+        val _ = batchesPresentCount.decrementAndGet()
+      } else {
+        val _ = singletonsPresentCount.decrementAndGet()
+      }
+    }
+
+    done
   }
 
   /**
@@ -220,5 +257,42 @@ private[effect] final class ScalQueue[A <: AnyRef](threadCount: Int) {
       queues(i).clear()
       i += 1
     }
+
+    // Reset present counters when clearing the queue
+    singletonsPresentCount.set(0)
+    batchesPresentCount.set(0)
   }
+
+  /**
+   * Returns the total number of singleton tasks submitted to this queue.
+   */
+  def getSingletonsSubmittedCount(): Long = singletonsSubmittedCount.get()
+
+  /**
+   * Returns the number of singleton tasks currently in this queue.
+   */
+  def getSingletonsPresentCount(): Long = singletonsPresentCount.get()
+
+  /**
+   * Returns the total number of batch tasks submitted to this queue.
+   */
+  def getBatchesSubmittedCount(): Long = batchesSubmittedCount.get()
+
+  /**
+   * Returns the number of batch tasks currently in this queue.
+   */
+  def getBatchesPresentCount(): Long = batchesPresentCount.get()
+}
+
+object ScalQueue {
+
+  /**
+   * Creates a new Scal queue.
+   *
+   * @param threadCount
+   *   the number of threads to load balance between
+   * @return
+   *   a new Scal queue instance
+   */
+  def apply[A <: AnyRef](threadCount: Int): ScalQueue[A] = new ScalQueue(threadCount)
 }

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

@@ -121,9 +121,8 @@ private[effect] final class WorkStealingThreadPool[P <: AnyRef](
       worker.ownsPoller(poller)
     } else false
   }
-
   private[this] val externalQueue: ScalQueue[AnyRef] =
-    new ScalQueue(threadCount << 2)
+    ScalQueue[AnyRef](threadCount << 2)
 
   /**
    * Represents two unsigned 16 bit integers. The 16 most significant bits track the number of
@@ -239,7 +238,6 @@ private[effect] final class WorkStealingThreadPool[P <: AnyRef](
       destQueue.enqueueBatch(batch, destWorker)
     } else if (element.isInstanceOf[Runnable]) {
       val fiber = element.asInstanceOf[Runnable]
-
       if (isStackTracing) {
         destWorker.active = fiber
         parkedSignals(dest).lazySet(false)
@@ -529,6 +527,43 @@ private[effect] final class WorkStealingThreadPool[P <: AnyRef](
     externalQueue.offer(fiber, random)
     notifyParked(random)
     ()
+
+  }
+
+  /**
+   * Offers a batch of runnables to the external queue and updates batch metrics.
+   *
+   * @param batch
+   *   the batch of runnables to be offered to the external queue
+   * @param random
+   *   a reference to an uncontended source of randomness
+   * @return
+   *   true if the batch was successfully offered
+   */
+  private[unsafe] def offerBatchToExternalQueue(
+      batch: Array[Runnable],
+      random: ThreadLocalRandom): Boolean = {
+    externalQueue.offer(batch, random)
+    true // Assume success
+  }
+
+  /**
+   * Offers multiple batches of runnables to the external queue and updates batch metrics.
+   *
+   * @param batches
+   *   the batches of runnables to be offered to the external queue
+   * @param random
+   *   a reference to an uncontended source of randomness
+   * @return
+   *   true if the batches were successfully offered
+   */
+  private[unsafe] def offerAllBatchesToExternalQueue(
+      batches: Array[AnyRef],
+      random: ThreadLocalRandom): Boolean = {
+    for (batch <- batches) {
+      externalQueue.offer(batch.asInstanceOf[Array[Runnable]], random)
+    }
+    true // Assume success
   }
 
   /**
@@ -844,6 +879,37 @@ private[effect] final class WorkStealingThreadPool[P <: AnyRef](
     }
     sum
   }
+
+  /**
+   * Returns the total number of singleton tasks submitted to the external queue.
+   */
+  private[unsafe] def getSingletonsSubmittedCount(): Long =
+    externalQueue.getSingletonsSubmittedCount()
+
+  /**
+   * Returns the total number of batch tasks submitted to the external queue.
+   */
+  private[unsafe] def getBatchesSubmittedCount(): Long =
+    externalQueue.getBatchesSubmittedCount()
+
+  /**
+   * Returns the number of singleton tasks currently in the external queue.
+   */
+  private[unsafe] def getSingletonsPresentCount(): Long =
+    externalQueue.getSingletonsPresentCount()
+
+  /**
+   * Returns the number of batch tasks currently in the external queue.
+   */
+  private[unsafe] def getBatchesPresentCount(): Long = externalQueue.getBatchesPresentCount()
+
+  private[unsafe] def logQueueMetrics(): Unit = {
+    println(s"[Thread Pool ${id}] Queue Metrics:")
+    println(s"  Singletons submitted: ${getSingletonsSubmittedCount()}")
+    println(s"  Singletons present: ${getSingletonsPresentCount()}")
+    println(s"  Batches submitted: ${getBatchesSubmittedCount()}")
+    println(s"  Batches present: ${getBatchesPresentCount()}")
+  }
 }
 
 private object WorkStealingThreadPool {

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

@@ -303,6 +303,13 @@ private[effect] final class WorkerThread[P <: AnyRef](
   private[unsafe] def ownsPoller(poller: P): Boolean =
     poller eq _poller
 
+  /**
+   * Returns the thread pool that owns this worker thread.
+   *
+   * @return
+   *   reference to the owning WorkStealingThreadPool
+   */
+  private[unsafe] def getPool(): WorkStealingThreadPool[P] = pool
   private[unsafe] def ownsTimers(timers: TimerHeap): Boolean =
     sleepers eq timers
 

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

@@ -85,6 +85,38 @@ sealed trait WorkStealingThreadPoolMetrics {
    */
   def suspendedFiberCount(): Long
 
+  /**
+   * Returns the total number of singleton tasks submitted to the external queue.
+   *
+   * @note
+   *   the value may differ between invocations
+   */
+  def singletonsSubmittedCount(): Long
+
+  /**
+   * Returns the total number of batch tasks submitted to the external queue.
+   *
+   * @note
+   *   the value may differ between invocations
+   */
+  def batchesSubmittedCount(): Long
+
+  /**
+   * Returns the number of singleton tasks currently in the external queue.
+   *
+   * @note
+   *   the value may differ between invocations
+   */
+  def singletonsPresentCount(): Long
+
+  /**
+   * Returns the number of batch tasks currently in the external queue.
+   *
+   * @note
+   *   the value may differ between invocations
+   */
+  def batchesPresentCount(): Long
+
   /**
    * The list of worker-specific metrics of this work-stealing thread pool.
    */
@@ -263,6 +295,10 @@ object WorkStealingThreadPoolMetrics {
     def blockedWorkerThreadCount(): Int = wstp.getBlockedWorkerThreadCount()
     def localQueueFiberCount(): Long = wstp.getLocalQueueFiberCount()
     def suspendedFiberCount(): Long = wstp.getSuspendedFiberCount()
+    def batchesSubmittedCount(): Long = wstp.getBatchesSubmittedCount()
+    def singletonsPresentCount(): Long = wstp.getSingletonsPresentCount()
+    def batchesPresentCount(): Long = wstp.getBatchesPresentCount()
+    def singletonsSubmittedCount(): Long = wstp.getSingletonsSubmittedCount()
 
     val workerThreads: List[WorkerThreadMetrics] =
       List.range(0, workerThreadCount()).map(workerThreadMetrics(wstp, _))

tests/js/src/test/scala/cats/effect/IOPlatformSuite.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2024 Typelevel
+ * Copyright 2020-2025 Typelevel
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.