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 d75bf95017 fix: keep lazy dispatch state on VarHandle (#3116)
d75bf95017 is described below
commit d75bf95017f4188ad861a04b721c00ee600d7388
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Sun Jun 21 11:45:14 2026 +0800
fix: keep lazy dispatch state on VarHandle (#3116)
Motivation:
PR #3035 used a VarHandle-backed volatile Int for LazyDispatch election
state.
Review raised concern about the field and the VarHandle access semantics.
Modification:
Keep the VarHandle-backed Int state, but use explicit
getVolatile/setVolatile
accessors with compareAndSet. This keeps the state in the LazyDispatch
object
without per-instance AtomicInteger allocation while preserving conservative
volatile access semantics.
Result:
LazyDispatch keeps the same IDLE/SCHEDULED scheduling protocol with explicit
volatile state access.
Tests:
- scalafmt --mode diff-ref=origin/main --non-interactive
- scalafmt --list --mode diff-ref=origin/main --non-interactive
- sbt "+stream / compile"
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.scaladsl.StageActorRefSpec" (11/11)
- git diff --check
References:
Refs #3035
---
.../org/apache/pekko/stream/stage/GraphStage.scala | 18 +++++++-----------
1 file changed, 7 insertions(+), 11 deletions(-)
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
b/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
index 5ea8c444a9..6c04d0cfdd 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
@@ -13,8 +13,8 @@
package org.apache.pekko.stream.stage
-import java.util.Spliterator
import java.lang.invoke.{ MethodHandles, VarHandle }
+import java.util.Spliterator
import java.util.concurrent.{ CompletionStage, ConcurrentHashMap }
import java.util.concurrent.atomic.AtomicReference
@@ -359,9 +359,8 @@ object GraphStageLogic {
* object (one allocation per StageActor, one fewer field deref on the
producer hot path).
* - Implements `Any => Unit` directly — serves as both the producer
callback and the drain callback,
* eliminating the separate `drainCallback` lambda allocation.
- * - `state` is a plain `@volatile var Int` driven by a static
`VarHandle` in the companion object
- * (via `MethodHandles.privateLookupIn`), same pattern as
`AbstractNodeQueue` itself;
- * avoids per-instance `AtomicInteger`.
+ * - `state` is a plain `@volatile var Int` driven by a static
`VarHandle`; volatile state access
+ * keeps the election protocol conservative and avoids per-instance
`AtomicInteger`.
* - `drainBatchSize` is read once into a stack-local at the top of
`drain` so the JIT can treat the loop
* bound as a constant.
* - Per-tell allocation = 1 Node (`AbstractNodeQueue.Node`, ~24 bytes) +
1 Tuple2 (~24 bytes). The
@@ -378,18 +377,14 @@ object GraphStageLogic {
extends AbstractNodeQueue[(ActorRef, Any)]
with (Any => Unit) {
- // IDLE/SCHEDULED election state. VarHandle avoids per-instance
AtomicInteger;
- // the handle lives in the companion object (true JVM static), same
pattern as
- // AbstractNodeQueue._tailDoNotCallMeDirectly.
+ // IDLE/SCHEDULED election state. VarHandle avoids per-instance
AtomicInteger.
@volatile var state: Int = SchedStateIdle
- // Typed local witnesses the Int-returning signature-polymorphic
VarHandle.get overload
- // (required for Scala 3 cross-compile; on Scala 2 this is a no-op).
private def getState(): Int = {
- val v: Int = LazyDispatch.stateHandle.get(this)
+ val v: Int = LazyDispatch.stateHandle.getVolatile(this)
v
}
- private def setState(v: Int): Unit = LazyDispatch.stateHandle.set(this,
v)
+ private def setState(v: Int): Unit =
LazyDispatch.stateHandle.setVolatile(this, v)
private def casState(expect: Int, update: Int): Boolean =
LazyDispatch.stateHandle.compareAndSet(this, expect, update)
@@ -448,6 +443,7 @@ object GraphStageLogic {
lookup.findVarHandle(classOf[LazyDispatch], "state", Integer.TYPE)
}
}
+
}
/**
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
