This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 713ad01e16 perf: add Thread.onSpinWait() to CAS spin loops (#3149)
713ad01e16 is described below
commit 713ad01e16cb1e5b30902697a67b3d9e1863ffb7
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Tue Jun 23 19:57:00 2026 +0800
perf: add Thread.onSpinWait() to CAS spin loops (#3149)
* perf: add Thread.onSpinWait() to CAS spin loops
Motivation:
CAS retry loops in AbstractDispatcher and Mailbox spin without any
CPU hint when contention causes compareAndSet to fail. JDK 9's
Thread.onSpinWait() hints to the CPU that we're in a spin-wait
loop, improving power efficiency and reducing context-switch
latency.
Modification:
- AbstractDispatcher.scala: add onSpinWait() in shutdown schedule
CAS retry loop body
- Mailbox.scala: convert setAsScheduled() and setAsIdle() from
@tailrec recursive CAS loops to while loops with onSpinWait(),
re-reading currentStatus on each retry
Result:
Better CPU behavior under contention on the message dispatch hot
path (setAsScheduled is called on every registerForExecution).
The onSpinWait pattern is already established in the codebase
(AbstractNodeQueue, AbstractBoundedNodeQueue, AffinityPool).
Tests:
sbt "actor/compile" — passed
References:
Refs #3136
* perf: extend Thread.onSpinWait() to remaining CAS spin loops
Motivation:
The initial PR covered three CAS spin loops in AbstractDispatcher and
Mailbox. Internal review identified four additional @tailrec CAS loops
that spin on compareAndSet failure without any CPU hint, which can
waste power and slow down SMT siblings under contention:
- `AbstractDispatcher.ifSensibleToDoSoThenScheduleShutdown` (2 CAS
failure paths), on the detach and taskCleanup paths where shutdown
contention is realistic.
- `Mailbox.resume`, `suspend`, `becomeClosed`, called during actor
suspend/resume/stop.
Modification:
Rewrite each @tailrec method into an explicit while loop and insert
`Thread.onSpinWait()` between the failed CAS and the volatile re-read.
`Thread.onSpinWait()` is a JDK 9+ CPU hint (x86 PAUSE / ARM YIELD)
that neither yields to the OS scheduler nor introduces a delay — it
actually reduces starvation rather than increasing it.
Result:
Lower power draw and faster CAS recovery under contention on the
dispatcher shutdown and mailbox status-change paths. MiMa clean —
method signatures are unchanged and all affected methods are private.
Tests:
sbt "actor/compile" -- passed
sbt "actor-tests/Test/testOnly ActorMailboxSpec" -- 33/33 passed
sbt "actor-tests/Test/testOnly DispatcherShutdownSpec" -- 1/1 passed
sbt "actor-tests/Test/testOnly DispatcherActorsSpec" -- 1/1 passed
sbt "actor/mimaReportBinaryIssues" -- no issues
References:
Discovered during internal review of PR #3149.
* refactor: address code review feedback in CAS spin loops
Motivation:
Code review feedback on readability and dead code concerns
in the CAS retry loops.
Modification:
- Rewrite setAsScheduled from while({block}) to while(true) style
for consistency with other CAS loops
- Replace trailing 'false // unreachable' with explicit
'throw new IllegalStateException("unreachable")'
- Fix misleading exception message from 'actor class marker' to
'shutdown schedule marker' in AbstractDispatcher
Result:
More consistent and readable CAS loop implementations with proper
dead code handling.
References:
Refs #3149
---
.../apache/pekko/dispatch/AbstractDispatcher.scala | 30 ++++----
.../scala/org/apache/pekko/dispatch/Mailbox.scala | 80 ++++++++++++++--------
2 files changed, 67 insertions(+), 43 deletions(-)
diff --git
a/actor/src/main/scala/org/apache/pekko/dispatch/AbstractDispatcher.scala
b/actor/src/main/scala/org/apache/pekko/dispatch/AbstractDispatcher.scala
index f5e9e277dc..08a4b16767 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/AbstractDispatcher.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/AbstractDispatcher.scala
@@ -190,18 +190,22 @@ abstract class MessageDispatcher(val configurator:
MessageDispatcherConfigurator
case _ => eventStream.publish(Error(t,
getClass.getName, getClass, t.getMessage))
}
- @tailrec
private final def ifSensibleToDoSoThenScheduleShutdown(): Unit = {
- if (inhabitants <= 0) shutdownSchedule match {
- case UNSCHEDULED =>
- if (updateShutdownSchedule(UNSCHEDULED, SCHEDULED))
scheduleShutdownAction()
- else ifSensibleToDoSoThenScheduleShutdown()
- case SCHEDULED =>
- if (updateShutdownSchedule(SCHEDULED, RESCHEDULED)) ()
- else ifSensibleToDoSoThenScheduleShutdown()
- case RESCHEDULED =>
- case unexpected =>
- throw new IllegalArgumentException(s"Unexpected actor class marker:
$unexpected") // will not happen, for exhaustiveness check
+ while (inhabitants <= 0) {
+ shutdownSchedule match {
+ case UNSCHEDULED =>
+ if (updateShutdownSchedule(UNSCHEDULED, SCHEDULED)) {
+ scheduleShutdownAction()
+ return
+ }
+ case SCHEDULED =>
+ if (updateShutdownSchedule(SCHEDULED, RESCHEDULED)) return
+ case RESCHEDULED =>
+ return
+ case unexpected =>
+ throw new IllegalArgumentException(s"Unexpected shutdown schedule
marker: $unexpected") // will not happen, for exhaustiveness check
+ }
+ Thread.onSpinWait()
}
}
@@ -255,14 +259,14 @@ abstract class MessageDispatcher(val configurator:
MessageDispatcherConfigurator
try {
if (inhabitants == 0) shutdown() // Warning, racy
} finally {
- while (!updateShutdownSchedule(shutdownSchedule, UNSCHEDULED)) {}
+ while (!updateShutdownSchedule(shutdownSchedule, UNSCHEDULED)) {
Thread.onSpinWait() }
}
case RESCHEDULED =>
if (updateShutdownSchedule(RESCHEDULED, SCHEDULED))
scheduleShutdownAction()
else run()
case UNSCHEDULED =>
case unexpected =>
- throw new IllegalArgumentException(s"Unexpected actor class marker:
$unexpected") // will not happen, for exhaustiveness check
+ throw new IllegalArgumentException(s"Unexpected shutdown schedule
marker: $unexpected") // will not happen, for exhaustiveness check
}
}
}
diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala
b/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala
index af728fabf9..03ebf96a50 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala
@@ -150,14 +150,19 @@ private[pekko] abstract class Mailbox(val messageQueue:
MessageQueue)
*
* @return true if the suspend count reached zero
*/
- @tailrec
- final def resume(): Boolean = currentStatus match {
- case Closed =>
- setStatus(Closed); false
- case s =>
+ final def resume(): Boolean = {
+ var s = currentStatus
+ while (true) {
+ if (s == Closed) {
+ setStatus(Closed)
+ return false
+ }
val next = if (s < suspendUnit) s else s - suspendUnit
- if (updateStatus(s, next)) next < suspendUnit
- else resume()
+ if (updateStatus(s, next)) return next < suspendUnit
+ Thread.onSpinWait()
+ s = currentStatus
+ }
+ throw new IllegalStateException("unreachable")
}
/**
@@ -166,47 +171,62 @@ private[pekko] abstract class Mailbox(val messageQueue:
MessageQueue)
*
* @return true if the previous suspend count was zero
*/
- @tailrec
- final def suspend(): Boolean = currentStatus match {
- case Closed =>
- setStatus(Closed); false
- case s =>
- if (updateStatus(s, s + suspendUnit)) s < suspendUnit
- else suspend()
+ final def suspend(): Boolean = {
+ var s = currentStatus
+ while (true) {
+ if (s == Closed) {
+ setStatus(Closed)
+ return false
+ }
+ if (updateStatus(s, s + suspendUnit)) return s < suspendUnit
+ Thread.onSpinWait()
+ s = currentStatus
+ }
+ throw new IllegalStateException("unreachable")
}
/**
* set new primary status Closed. Caller does not need to worry about whether
* status was Scheduled or not.
*/
- @tailrec
- final def becomeClosed(): Boolean = currentStatus match {
- case Closed =>
- setStatus(Closed); false
- case s => updateStatus(s, Closed) || becomeClosed()
+ final def becomeClosed(): Boolean = {
+ var s = currentStatus
+ while (true) {
+ if (s == Closed) {
+ setStatus(Closed)
+ return false
+ }
+ if (updateStatus(s, Closed)) return true
+ Thread.onSpinWait()
+ s = currentStatus
+ }
+ throw new IllegalStateException("unreachable")
}
/**
* Set Scheduled status, keeping primary status as is.
*/
- @tailrec
final def setAsScheduled(): Boolean = {
- val s = currentStatus
- /*
- * Only try to add Scheduled bit if pure Open/Suspended, not Closed or with
- * Scheduled bit already set.
- */
- if ((s & shouldScheduleMask) != Open) false
- else updateStatus(s, s | Scheduled) || setAsScheduled()
+ var s = currentStatus
+ while (true) {
+ if ((s & shouldScheduleMask) != Open) return false
+ if (updateStatus(s, s | Scheduled)) return true
+ Thread.onSpinWait()
+ s = currentStatus
+ }
+ throw new IllegalStateException("unreachable")
}
/**
* Reset Scheduled status, keeping primary status as is.
*/
- @tailrec
final def setAsIdle(): Boolean = {
- val s = currentStatus
- updateStatus(s, s & ~Scheduled) || setAsIdle()
+ var s = currentStatus
+ while (!updateStatus(s, s & ~Scheduled)) {
+ Thread.onSpinWait()
+ s = currentStatus
+ }
+ true
}
/*
* AtomicReferenceFieldUpdater for system queue.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
