Avoid receive timeout type pollution

actor-tests/src/test/scala/org/apache/pekko/actor/ReceiveTimeoutSpec.scala

@@ -20,6 +20,7 @@ import java.util.concurrent.atomic.AtomicInteger
 import scala.concurrent.Await
 import scala.concurrent.duration._
 
+import org.apache.pekko.actor.dungeon.{ ReceiveTimeout => ReceiveTimeoutSupport }
 import org.apache.pekko.testkit._
 
 object ReceiveTimeoutSpec {
@@ -79,6 +80,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()
 

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.ReceiveTimeout.isNotInfluenceReceiveTimeout(msg))
         new TimerMsg(key, nextGen, this) with NotInfluenceReceiveTimeout
       else
         new TimerMsg(key, nextGen, this)

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 = {

actor/src/main/scala/org/apache/pekko/actor/dungeon/ReceiveTimeout.scala

@@ -19,10 +19,16 @@ import scala.concurrent.duration.FiniteDuration
 import org.apache.pekko
 import pekko.actor.ActorCell
 import pekko.actor.Cancellable
+import pekko.actor.Identify
 import pekko.actor.NotInfluenceReceiveTimeout
 
 private[pekko] object ReceiveTimeout {
   final val emptyReceiveTimeoutData: (Duration, Cancellable) = (Duration.Undefined, ActorCell.emptyCancellable)
+
+  // 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]
 }
 
 private[pekko] trait ReceiveTimeout { this: ActorCell =>
@@ -37,9 +43,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 +77,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) && !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 =

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.ReceiveTimeout.isNotInfluenceReceiveTimeout(msg))
         NotInfluenceReceiveTimeoutTimerMsg(key, nextGen, this)
       else
         InfluenceReceiveTimeoutTimerMsg(key, nextGen, this)

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.ReceiveTimeout.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)
+  }
+}