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 89b590f65d Avoid receive timeout type pollution (#3332)
89b590f65d is described below
commit 89b590f65d0a31dbc5c2855a92136bef86b16f6f
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Tue Jul 21 19:58:36 2026 +0800
Avoid receive timeout type pollution (#3332)
* fix: avoid receive timeout type pollution
Motivation:
The ActorCell receive timeout path allocates on every message
and misclassifies timer messages, causing unnecessary timeout
rescheduling and state churn.
Modification:
- Avoid type pollution in the receive timeout hot path
- Guard timer receive timeout classification
Result:
Reduced allocations and correct timeout behavior for actors
using receive timeouts.
Tests:
- Existing ActorCellReceiveTimeoutSpec
References:
Refs #1668
* Remove @inline annotation for Scala 3 compatibility
Motivation:
The @inline annotation does not work for Scala 3. The JVM JIT will inline
this small method regardless of the annotation.
Modification:
Remove the @inline annotation from isNotInfluenceReceiveTimeout method.
Result:
Code compiles cleanly on both Scala 2.13 and Scala 3. Performance is
unaffected as the JIT compiler will still inline the method.
Tests:
Not run - annotation removal only, no behavior change
References:
Refs #3332
* Split isNotInfluenceReceiveTimeout into Scala 2/3 version-specific files
Motivation:
The @inline annotation does not work for Scala 3. Reviewer requested
splitting the method into version-specific files to preserve the inline
optimization for Scala 2 while supporting Scala 3.
Modification:
- Create ReceiveTimeoutCompat object in scala-2.13 with @inline annotation
- Create ReceiveTimeoutCompat object in scala-3 with inline keyword
- Move isNotInfluenceReceiveTimeout method from ReceiveTimeout to
ReceiveTimeoutCompat
- Update all call sites to use ReceiveTimeoutCompat
Result:
Code compiles cleanly on both Scala 2.13 and Scala 3 with appropriate
inline optimizations for each version.
Tests:
sbt "actor-tests/Test/testOnly org.apache.pekko.actor.ReceiveTimeoutSpec"
All 14 tests passed.
References:
Refs #3332
* fix: update references to ReceiveTimeoutCompat after method relocation
Motivation:
The isNotInfluenceReceiveTimeout method was moved from ReceiveTimeout to
ReceiveTimeoutCompat in the version-specific files, but two call sites
still referenced the old location, causing compilation failures across
all CI jobs.
Modification:
Update TimerSchedulerImpl (actor-typed) and
ReceiveTimeoutTypePollutionBenchmark
(bench-jmh) to reference ReceiveTimeoutCompat.isNotInfluenceReceiveTimeout.
Result:
All modules compile successfully on both Scala 2.13 and Scala 3.
Tests:
- sbt "actor-typed / Compile / compileIncremental" (Scala 2.13) - success
- sbt "++3.3.8; actor-typed / Compile / compileIncremental" - success
References:
Refs #3332
---
.../apache/pekko/actor/ReceiveTimeoutSpec.scala | 7 ++
.../actor/typed/internal/TimerSchedulerImpl.scala | 2 +-
.../pekko/actor/dungeon/ReceiveTimeoutCompat.scala | 25 +++++++
.../pekko/actor/dungeon/ReceiveTimeoutCompat.scala | 25 +++++++
.../scala/org/apache/pekko/actor/ActorCell.scala | 2 +-
.../pekko/actor/dungeon/ReceiveTimeout.scala | 17 +++--
.../pekko/actor/dungeon/TimerSchedulerImpl.scala | 2 +-
.../ReceiveTimeoutTypePollutionBenchmark.scala | 81 ++++++++++++++++++++++
8 files changed, 152 insertions(+), 9 deletions(-)
diff --git
a/actor-tests/src/test/scala/org/apache/pekko/actor/ReceiveTimeoutSpec.scala
b/actor-tests/src/test/scala/org/apache/pekko/actor/ReceiveTimeoutSpec.scala
index 1aa2e3d43f..ef4ecb344f 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/actor/ReceiveTimeoutSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/actor/ReceiveTimeoutSpec.scala
@@ -21,6 +21,7 @@ import scala.concurrent.Await
import scala.concurrent.duration._
import org.apache.pekko
+import pekko.actor.dungeon.{ ReceiveTimeoutCompat => ReceiveTimeoutSupport }
import pekko.testkit._
object ReceiveTimeoutSpec {
@@ -80,6 +81,12 @@ class ReceiveTimeoutSpec extends PekkoSpec() {
"An actor with receive timeout" must {
+ "classify messages that do not influence the timeout" in {
+ ReceiveTimeoutSupport.isNotInfluenceReceiveTimeout(Identify(None))
should ===(true)
+ ReceiveTimeoutSupport.isNotInfluenceReceiveTimeout(TransparentTick)
should ===(true)
+ ReceiveTimeoutSupport.isNotInfluenceReceiveTimeout(Tick) should
===(false)
+ }
+
"get timeout" taggedAs TimingTest in {
val timeoutLatch = TestLatch()
diff --git
a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/TimerSchedulerImpl.scala
b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/TimerSchedulerImpl.scala
index 8672770e2e..a8e57f17cb 100644
---
a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/TimerSchedulerImpl.scala
+++
b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/TimerSchedulerImpl.scala
@@ -117,7 +117,7 @@ import org.slf4j.Logger
val nextGen = timerGen.next()
val timerMsg =
- if (msg.isInstanceOf[NotInfluenceReceiveTimeout])
+ if
(pekko.actor.dungeon.ReceiveTimeoutCompat.isNotInfluenceReceiveTimeout(msg))
new TimerMsg(key, nextGen, this) with NotInfluenceReceiveTimeout
else
new TimerMsg(key, nextGen, this)
diff --git
a/actor/src/main/scala-2.13/org/apache/pekko/actor/dungeon/ReceiveTimeoutCompat.scala
b/actor/src/main/scala-2.13/org/apache/pekko/actor/dungeon/ReceiveTimeoutCompat.scala
new file mode 100644
index 0000000000..035ef7fae4
--- /dev/null
+++
b/actor/src/main/scala-2.13/org/apache/pekko/actor/dungeon/ReceiveTimeoutCompat.scala
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.actor.dungeon
+
+import org.apache.pekko.actor.Identify
+import org.apache.pekko.actor.NotInfluenceReceiveTimeout
+
+private[pekko] object ReceiveTimeoutCompat {
+
+ // Identify is also an AutoReceivedMessage. Checking its concrete class
first avoids polluting its secondary
+ // supertype cache on JDKs affected by JDK-8180450 when actor runtime code
also checks the AutoReceivedMessage marker.
+ @inline def isNotInfluenceReceiveTimeout(message: Any): Boolean =
+ message.isInstanceOf[Identify] ||
message.isInstanceOf[NotInfluenceReceiveTimeout]
+}
diff --git
a/actor/src/main/scala-3/org/apache/pekko/actor/dungeon/ReceiveTimeoutCompat.scala
b/actor/src/main/scala-3/org/apache/pekko/actor/dungeon/ReceiveTimeoutCompat.scala
new file mode 100644
index 0000000000..c10c616bfe
--- /dev/null
+++
b/actor/src/main/scala-3/org/apache/pekko/actor/dungeon/ReceiveTimeoutCompat.scala
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.actor.dungeon
+
+import org.apache.pekko.actor.Identify
+import org.apache.pekko.actor.NotInfluenceReceiveTimeout
+
+private[pekko] object ReceiveTimeoutCompat {
+
+ // Identify is also an AutoReceivedMessage. Checking its concrete class
first avoids polluting its secondary
+ // supertype cache on JDKs affected by JDK-8180450 when actor runtime code
also checks the AutoReceivedMessage marker.
+ inline def isNotInfluenceReceiveTimeout(message: Any): Boolean =
+ message.isInstanceOf[Identify] ||
message.isInstanceOf[NotInfluenceReceiveTimeout]
+}
diff --git a/actor/src/main/scala/org/apache/pekko/actor/ActorCell.scala
b/actor/src/main/scala/org/apache/pekko/actor/ActorCell.scala
index 7d36e05889..6bfbfedebd 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/ActorCell.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/ActorCell.scala
@@ -561,7 +561,7 @@ private[pekko] class ActorCell(
}
finally
// Schedule or reschedule receive timeout
- checkReceiveTimeoutIfNeeded(msg, timeoutBeforeReceive)
+ checkReceiveTimeoutIfNeeded(timeoutBeforeReceive)
}
def autoReceiveMessage(msg: Envelope): Unit = {
diff --git
a/actor/src/main/scala/org/apache/pekko/actor/dungeon/ReceiveTimeout.scala
b/actor/src/main/scala/org/apache/pekko/actor/dungeon/ReceiveTimeout.scala
index a8f4a67e65..3484337baa 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/dungeon/ReceiveTimeout.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/dungeon/ReceiveTimeout.scala
@@ -19,7 +19,7 @@ import scala.concurrent.duration.FiniteDuration
import org.apache.pekko
import pekko.actor.ActorCell
import pekko.actor.Cancellable
-import pekko.actor.NotInfluenceReceiveTimeout
+import pekko.actor.dungeon.ReceiveTimeoutCompat
private[pekko] object ReceiveTimeout {
final val emptyReceiveTimeoutData: (Duration, Cancellable) =
(Duration.Undefined, ActorCell.emptyCancellable)
@@ -37,9 +37,11 @@ private[pekko] trait ReceiveTimeout { this: ActorCell =>
final def setReceiveTimeout(timeout: Duration): Unit = receiveTimeoutData =
receiveTimeoutData.copy(_1 = timeout)
/** Called after `ActorCell.receiveMessage` or
`ActorCell.autoReceiveMessage`. */
- protected def checkReceiveTimeoutIfNeeded(message: Any, beforeReceive:
(Duration, Cancellable)): Unit =
- if (hasTimeoutData || receiveTimeoutChanged(beforeReceive))
- checkReceiveTimeout(!message.isInstanceOf[NotInfluenceReceiveTimeout] ||
receiveTimeoutChanged(beforeReceive))
+ protected def checkReceiveTimeoutIfNeeded(beforeReceive: (Duration,
Cancellable)): Unit = {
+ val timeoutChanged = receiveTimeoutChanged(beforeReceive)
+ if (hasTimeoutData || timeoutChanged)
+ checkReceiveTimeout(timeoutChanged)
+ }
final def checkReceiveTimeout(reschedule: Boolean): Unit = {
val (recvTimeout, task) = receiveTimeoutData
@@ -69,10 +71,13 @@ private[pekko] trait ReceiveTimeout { this: ActorCell =>
receiveTimeoutData ne beforeReceive
protected def cancelReceiveTimeoutIfNeeded(message: Any): (Duration,
Cancellable) = {
- if (hasTimeoutData && !message.isInstanceOf[NotInfluenceReceiveTimeout])
+ val beforeReceive = receiveTimeoutData
+ if ((beforeReceive ne emptyReceiveTimeoutData) &&
!ReceiveTimeoutCompat.isNotInfluenceReceiveTimeout(message))
cancelReceiveTimeoutTask()
- receiveTimeoutData
+ // Returning the state from before cancellation lets
checkReceiveTimeoutIfNeeded infer whether this message
+ // influenced the timeout without performing the type check a second time.
+ beforeReceive
}
private[pekko] def cancelReceiveTimeoutTask(): Unit =
diff --git
a/actor/src/main/scala/org/apache/pekko/actor/dungeon/TimerSchedulerImpl.scala
b/actor/src/main/scala/org/apache/pekko/actor/dungeon/TimerSchedulerImpl.scala
index e1ea980f3c..45e5b4d77a 100644
---
a/actor/src/main/scala/org/apache/pekko/actor/dungeon/TimerSchedulerImpl.scala
+++
b/actor/src/main/scala/org/apache/pekko/actor/dungeon/TimerSchedulerImpl.scala
@@ -90,7 +90,7 @@ import pekko.util.OptionVal
val nextGen = nextTimerGen()
val timerMsg =
- if (msg.isInstanceOf[NotInfluenceReceiveTimeout])
+ if (dungeon.ReceiveTimeoutCompat.isNotInfluenceReceiveTimeout(msg))
NotInfluenceReceiveTimeoutTimerMsg(key, nextGen, this)
else
InfluenceReceiveTimeoutTimerMsg(key, nextGen, this)
diff --git
a/bench-jmh/src/main/scala/org/apache/pekko/actor/ReceiveTimeoutTypePollutionBenchmark.scala
b/bench-jmh/src/main/scala/org/apache/pekko/actor/ReceiveTimeoutTypePollutionBenchmark.scala
new file mode 100644
index 0000000000..fd3a09cf90
--- /dev/null
+++
b/bench-jmh/src/main/scala/org/apache/pekko/actor/ReceiveTimeoutTypePollutionBenchmark.scala
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.actor
+
+import java.util.concurrent.TimeUnit
+
+import org.openjdk.jmh.annotations._
+
+@State(Scope.Thread)
+class ReceiveTimeoutTypePollutionInput {
+ private val messages = Array[AnyRef](Identify("id"), new Object)
+ private var index = 0
+
+ def next(): AnyRef = {
+ index = (index + 1) & 1
+ messages(index)
+ }
+}
+
+@State(Scope.Benchmark)
+@BenchmarkMode(Array(Mode.Throughput))
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@Fork(3)
+@Threads(12)
+@Warmup(iterations = 5, time = 1)
+@Measurement(iterations = 5, time = 1)
+class ReceiveTimeoutTypePollutionBenchmark {
+
+ @CompilerControl(CompilerControl.Mode.DONT_INLINE)
+ def isAutoReceived(message: AnyRef): Boolean =
+ message.isInstanceOf[AutoReceivedMessage]
+
+ @CompilerControl(CompilerControl.Mode.DONT_INLINE)
+ def isNotInfluenceReceiveTimeout(message: AnyRef): Boolean =
+ message.isInstanceOf[NotInfluenceReceiveTimeout]
+
+ @CompilerControl(CompilerControl.Mode.DONT_INLINE)
+ def isNotInfluenceReceiveTimeoutGuarded(message: AnyRef): Boolean =
+ dungeon.ReceiveTimeoutCompat.isNotInfluenceReceiveTimeout(message)
+
+ @Benchmark
+ def typePolluted(input: ReceiveTimeoutTypePollutionInput): Int = {
+ val message = input.next()
+ val before = isNotInfluenceReceiveTimeout(message)
+ val auto = isAutoReceived(message)
+ val after = isNotInfluenceReceiveTimeout(message)
+ (if (before) 1 else 0) | (if (auto) 2 else 0) | (if (after) 4 else 0)
+ }
+
+ @Benchmark
+ def concreteTypeGuard(input: ReceiveTimeoutTypePollutionInput): Int = {
+ val message = input.next()
+ val before = isNotInfluenceReceiveTimeoutGuarded(message)
+ val auto = isAutoReceived(message)
+ val after = isNotInfluenceReceiveTimeoutGuarded(message)
+ (if (before) 1 else 0) | (if (auto) 2 else 0) | (if (after) 4 else 0)
+ }
+
+ @Benchmark
+ def concreteTypeGuardAndStateChange(input:
ReceiveTimeoutTypePollutionInput): Int = {
+ val message = input.next()
+ val before = isNotInfluenceReceiveTimeoutGuarded(message)
+ val auto = isAutoReceived(message)
+ (if (before) 1 else 0) | (if (auto) 2 else 0)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
