From d6a1a074ebf462c3e511b30c1014926b97b25281 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=99=8E=E9=B8=A3?= Date: Sun, 12 Jul 2026 03:47:15 +0800 Subject: [PATCH] test: add stream type pollution benchmark Motivation: Affected JDKs can lose throughput when the same stream element crosses alternating interface-typed non-inlined operators. The initial issue analysis attributed this to the fused Connection slot without runtime evidence. Modification: Add a fused-stream JMH comparison for concrete, stable-interface, and alternating-interface chains. Add a directional Flow map test and document the verified JVM upgrade and profiling guidance. Result: The benchmark reproduces the issue on JDK 17 while showing that JDK 21.0.8 removes the penalty, avoiding an unnecessary change to the core Stream interpreter hot path. Tests: - stream-tests / Test / testOnly org.apache.pekko.stream.scaladsl.FlowMapSpec - bench-jmh / Jmh / compile - docs/paradox - +headerCheckAll - checkCodeStyle - validatePullRequest (Stream tests passed; pre-existing jdocs.stream.IntegrationDocTest mailbox class-name failure) - JMH on JDK 17.0.17 and JDK 21.0.8 - Qoder review: No must-fix findings References: Refs #1668 --- .../stream/StreamTypePollutionBenchmark.scala | 171 ++++++++++++++++++ .../paradox/stream/stream-flows-and-basics.md | 18 ++ .../pekko/stream/scaladsl/FlowMapSpec.scala | 26 +++ 3 files changed, 215 insertions(+) create mode 100644 bench-jmh/src/main/scala/org/apache/pekko/stream/StreamTypePollutionBenchmark.scala diff --git a/bench-jmh/src/main/scala/org/apache/pekko/stream/StreamTypePollutionBenchmark.scala b/bench-jmh/src/main/scala/org/apache/pekko/stream/StreamTypePollutionBenchmark.scala new file mode 100644 index 00000000000..e6f812f09a0 --- /dev/null +++ b/bench-jmh/src/main/scala/org/apache/pekko/stream/StreamTypePollutionBenchmark.scala @@ -0,0 +1,171 @@ +/* + * 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.stream + +import java.util.concurrent.TimeUnit + +import scala.concurrent.Await +import scala.concurrent.Future +import scala.concurrent.duration.Duration + +import com.typesafe.config.ConfigFactory +import org.openjdk.jmh.annotations._ + +import org.apache.pekko +import pekko.Done +import pekko.actor.ActorSystem +import pekko.stream.scaladsl.Keep +import pekko.stream.scaladsl.RunnableGraph +import pekko.stream.scaladsl.Sink +import pekko.stream.scaladsl.Source + +@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 StreamTypePollutionBenchmark { + import StreamTypePollutionBenchmark._ + + private var system: ActorSystem = _ + private var materializer: Materializer = _ + private var concreteGraph: RunnableGraph[Future[Done]] = _ + private var singleInterfaceGraph: RunnableGraph[Future[Done]] = _ + private var alternatingInterfacesGraph: RunnableGraph[Future[Done]] = _ + + @Setup(Level.Trial) + def setup(): Unit = { + system = ActorSystem( + "StreamTypePollutionBenchmark", + ConfigFactory.parseString(s""" + pekko { + log-dead-letters = off + actor.default-dispatcher.fork-join-executor { + parallelism-min = 12 + parallelism-factor = 1.0 + parallelism-max = 12 + } + stream.materializer.sync-processing-limit = ${Int.MaxValue} + } + """)) + materializer = SystemMaterializer(system).materializer + + val payloads = Vector.tabulate[Payload](ElementCount) { i => + (i & 3) match { + case 0 => new Payload0 + case 1 => new Payload1 + case 2 => new Payload2 + case _ => new Payload3 + } + } + val leftPayloads: Vector[Left] = payloads + + concreteGraph = + Source(payloads) + .map(KeepPayload) + .map(KeepPayload) + .map(KeepPayload) + .map(KeepPayload) + .map(KeepPayload) + .map(KeepPayload) + .toMat(Sink.ignore)(Keep.right) + + singleInterfaceGraph = + Source(leftPayloads) + .map(KeepLeft) + .map(KeepLeft) + .map(KeepLeft) + .map(KeepLeft) + .map(KeepLeft) + .map(KeepLeft) + .toMat(Sink.ignore)(Keep.right) + + alternatingInterfacesGraph = + Source(leftPayloads) + .map(ToRight) + .map(ToLeft) + .map(ToRight) + .map(ToLeft) + .map(ToRight) + .map(ToLeft) + .toMat(Sink.ignore)(Keep.right) + } + + @TearDown(Level.Trial) + def shutdown(): Unit = + system.close() + + @Benchmark + @OperationsPerInvocation(ElementCount) + def concreteElements(): Done = + Await.result(concreteGraph.run()(materializer), Duration.Inf) + + @Benchmark + @OperationsPerInvocation(ElementCount) + def singleInterfaceElements(): Done = + Await.result(singleInterfaceGraph.run()(materializer), Duration.Inf) + + @Benchmark + @OperationsPerInvocation(ElementCount) + def alternatingInterfaceElements(): Done = + Await.result(alternatingInterfacesGraph.run()(materializer), Duration.Inf) +} + +object StreamTypePollutionBenchmark { + final val ElementCount = 100000 + + trait Left { + def toRight: Right + } + + trait Right { + def toLeft: Left + } + + abstract class Payload extends Left with Right { + final override def toRight: Right = this + final override def toLeft: Left = this + } + + final class Payload0 extends Payload + final class Payload1 extends Payload + final class Payload2 extends Payload + final class Payload3 extends Payload + + object KeepPayload extends (Payload => Payload) { + @CompilerControl(CompilerControl.Mode.DONT_INLINE) + override def apply(payload: Payload): Payload = payload + } + + object KeepLeft extends (Left => Left) { + @CompilerControl(CompilerControl.Mode.DONT_INLINE) + override def apply(payload: Left): Left = payload + } + + object ToRight extends (Left => Right) { + @CompilerControl(CompilerControl.Mode.DONT_INLINE) + override def apply(payload: Left): Right = payload.toRight + } + + object ToLeft extends (Right => Left) { + @CompilerControl(CompilerControl.Mode.DONT_INLINE) + override def apply(payload: Right): Left = payload.toLeft + } +} diff --git a/docs/src/main/paradox/stream/stream-flows-and-basics.md b/docs/src/main/paradox/stream/stream-flows-and-basics.md index ef1c5662405..9fe89ae5802 100644 --- a/docs/src/main/paradox/stream/stream-flows-and-basics.md +++ b/docs/src/main/paradox/stream/stream-flows-and-basics.md @@ -313,6 +313,24 @@ This means that everything that is inside the red bubble will be executed by one by another. This scheme can be applied successively, always having one such boundary enclose the previous ones plus all operators that have been added since then. +#### Type-check scalability on older JDKs + +JDKs affected by [JDK-8180450](https://bugs.openjdk.org/browse/JDK-8180450) can lose multi-core scalability when the +same concrete class is repeatedly checked against different interfaces. In a stream this can happen when an element +class implements multiple interfaces and hot, non-inlined operator functions alternately accept those interface types. +The contention belongs to the concrete element class's JVM secondary-supertype cache; it is not caused merely by a +fused connection storing elements as `Any` or by a generic `grab[T]`. + +The preferred mitigation is to use JDK 21.0.8 or later in the JDK 21 update line, or JDK 23 or later. If an affected +JDK cannot be upgraded, keep the declared element type through a hot operator chain concrete, or use one stable +interface view, where practical. Avoid repeatedly changing the same objects between unrelated interface views across +operator functions that the JIT cannot inline. + +The JIT can inline operator functions and eliminate their bridge casts, so this is workload dependent. Before changing +an application, reproduce the issue with a representative benchmark or use the +[type-pollution agent](https://github.com/RedHatPerf/type-pollution-agent) to identify concrete classes whose secondary +supertype cache is actually changing at hot type-check sites. + @@@ warning Without fusing (i.e. up to version 2.0-M2) each stream operator had an implicit input buffer diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapSpec.scala index 701e35b0ff0..3235da5e499 100644 --- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapSpec.scala +++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapSpec.scala @@ -31,6 +31,32 @@ class FlowMapSpec extends StreamSpec(""" TestConfig.RandomTestRange.foreach(_ => runScript(script)(_.map(_.toString))) } + "preserve an element across alternating interface-typed maps" in { + trait Left { + def toRight: Right + } + trait Right { + def toLeft: Left + } + final class Payload extends Left with Right { + override def toRight: Right = this + override def toLeft: Left = this + } + + val payload = new Payload + val result = + Source + .single[Left](payload) + .map[Right](_.toRight) + .map[Left](_.toLeft) + .map[Right](_.toRight) + .map[Left](_.toLeft) + .runWith(Sink.head) + .futureValue + + result shouldBe theSameInstanceAs(payload) + } + "not blow up with high request counts" in { val probe = TestSubscriber.manualProbe[Int]() Source(List(1))