diff --git a/core/jvm-native/src/main/scala/cats/effect/unsafe/LocalQueue.scala b/core/jvm-native/src/main/scala/cats/effect/unsafe/LocalQueue.scala index f8bcf5af1f..57f8037971 100644 --- a/core/jvm-native/src/main/scala/cats/effect/unsafe/LocalQueue.scala +++ b/core/jvm-native/src/main/scala/cats/effect/unsafe/LocalQueue.scala @@ -331,13 +331,7 @@ private final class LocalQueue extends LocalQueuePadding { val tl = tail while (true) { - // A load of the head of the queue using `acquire` semantics. - val hd = Head.updater.get(this) - val steal = msb(hd) - - // Check the usable capacity of the queue. - val len = unsignedShortSubtraction(tl, steal) - if (len <= LocalQueueCapacityMinusBatch) { + if (hasCapacityForBatch()) { // It is safe to transfer the fibers from the batch to the queue. val startPos = tl - 1 var i = 1 @@ -368,7 +362,8 @@ private final class LocalQueue extends LocalQueuePadding { return fiber } - // Not enough usable capacity, which means there is an ongoing steal + // Not enough usable capacity. Callers guarantee that the queue has spare + // capacity to hold the batch, which means there is an ongoing steal // operation. Spin until it completes. } @@ -379,6 +374,20 @@ private final class LocalQueue extends LocalQueuePadding { null } + /** + * Checks whether this queue has enough usable capacity to hold a batch of fibers. + * + * @note + * Can '''only''' be correctly called by the owner [[WorkerThread]]. + * + * @return + * `true` if a batch of fibers can be enqueued on this local queue, `false` otherwise + */ + def hasCapacityForBatch(): Boolean = { + val hd = Head.updater.get(this) + unsignedShortSubtraction(tail, msb(hd)) <= LocalQueueCapacityMinusBatch + } + /** * Dequeues a fiber from the head of the local queue. * diff --git a/core/jvm-native/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala b/core/jvm-native/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala index 558d33bffe..a29d9ea2e1 100644 --- a/core/jvm-native/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala +++ b/core/jvm-native/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala @@ -203,6 +203,9 @@ private[effect] final class WorkStealingThreadPool[P <: AnyRef]( * worker threads starting at a random index. If the stealing attempt was unsuccessful, this * method falls back to checking the external queue. * + * Stealing from external queue is skipped if the calling worker thread's own local queue is + * too full to take a batch of fibers from the external queue. + * * @param dest * the index of the worker thread attempting to steal work from other worker threads (used * to avoid stealing from its own local queue) @@ -239,7 +242,16 @@ private[effect] final class WorkStealingThreadPool[P <: AnyRef]( } // The worker thread could not steal any work. Fall back to checking the - // external queue. + // external queue, but only if the local queue can hold a batch of fibers. + // Expired timers stolen just before this call may have resumed enough + // fibers onto the local queue to leave no room for a batch, in which case + // `enqueueBatch` would spin forever, as it expects to be called only when + // there is capacity for the batch. + // See https://github.com/typelevel/cats-effect/issues/4674. + if (!destQueue.hasCapacityForBatch()) { + return null + } + val element = externalQueue.poll(random) if (element.isInstanceOf[Array[Runnable]]) { val batch = element.asInstanceOf[Array[Runnable]]