This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch perf/actorref-source-optimization in repository https://gitbox.apache.org/repos/asf/pekko.git
commit 58115f2ea983412025235492da43cad9b8669e76 Author: 虎鸣 <[email protected]> AuthorDate: Fri Jun 19 18:13:16 2026 +0800 test(stream): add JMH benchmark for Source.actorRef push throughput Motivation: No existing benchmark measured Source.actorRef throughput specifically. The StageActorRefBenchmark covers StageActor tell throughput via a custom sink, but not the Source.actorRef ingress path which is the target of the recent performance optimizations. Modification: Add ActorRefSourceBenchmark that creates a Source.actorRef with a 1024-element buffer and pushes 100k Int elements from a dedicated sender thread, measuring throughput in elements/second. Result: Provides a reproducible benchmark for tracking Source.actorRef performance across the direct-push fast path and message-only FunctionRef optimizations. Tests: - Not run - benchmark only (compile verified via sbt bench-jmh/Compile/compile) References: Refs akkadotnet/akka.net#8263 --- .../pekko/stream/ActorRefSourceBenchmark.scala | 88 ++++++++++++++++++++++ 1 file changed, 88 insertions(+) diff --git a/bench-jmh/src/main/scala/org/apache/pekko/stream/ActorRefSourceBenchmark.scala b/bench-jmh/src/main/scala/org/apache/pekko/stream/ActorRefSourceBenchmark.scala new file mode 100644 index 0000000000..aa8af59634 --- /dev/null +++ b/bench-jmh/src/main/scala/org/apache/pekko/stream/ActorRefSourceBenchmark.scala @@ -0,0 +1,88 @@ +/* + * 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.CountDownLatch +import java.util.concurrent.TimeUnit + +import scala.concurrent.Await +import scala.concurrent.duration._ + +import org.openjdk.jmh.annotations._ + +import org.apache.pekko +import pekko.actor.ActorRef +import pekko.actor.ActorSystem +import pekko.stream.scaladsl.Keep +import pekko.stream.scaladsl.Sink +import pekko.stream.scaladsl.Source + +object ActorRefSourceBenchmark { + final val OperationsPerInvocation = 100000 +} + +@State(Scope.Benchmark) +@OutputTimeUnit(TimeUnit.SECONDS) +@BenchmarkMode(Array(Mode.Throughput)) +class ActorRefSourceBenchmark { + import ActorRefSourceBenchmark._ + + implicit val system: ActorSystem = ActorSystem("ActorRefSourceBenchmark") + + private var sourceRef: ActorRef = _ + private var doneLatch: CountDownLatch = _ + + @Setup + def setup(): Unit = { + SystemMaterializer(system).materializer + } + + @TearDown + def shutdown(): Unit = { + Await.result(system.terminate(), 5.seconds) + } + + @Benchmark + @OperationsPerInvocation(OperationsPerInvocation) + def actorRef_source_push_100k(): Unit = { + doneLatch = new CountDownLatch(1) + val mat = Source + .actorRef[Any]( + completionMatcher = { case "done" => CompletionStrategy.draining }, + failureMatcher = PartialFunction.empty, + bufferSize = 1024, + overflowStrategy = OverflowStrategy.dropHead) + .toMat(Sink.ignore)(Keep.left) + .run() + sourceRef = mat + + val sender = new Thread(() => { + var i = 0 + while (i < OperationsPerInvocation) { + sourceRef ! i + i += 1 + } + sourceRef ! "done" + doneLatch.countDown() + }) + sender.start() + if (!doneLatch.await(30, TimeUnit.SECONDS)) + throw new RuntimeException("ActorRefSource benchmark timed out") + sender.join() + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
