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 10a518a9598ecdf57b5d0080670aac975b2dc72d Author: θιΈ£ <[email protected]> AuthorDate: Fri Jun 19 19:08:30 2026 +0800 test(stream): redesign ActorRefSource benchmark for optimization-sensitive measurement The original buffer=1024 saturated-push benchmark didn't exercise the direct-push fast path (buffer stays full, so elements always enqueue). Replace with two targeted benchmarks: - no_buffer (buffer=0): measures FunctionRef+AsyncCallback path cost without buffer noise, isolating the tuple boxing elimination benefit - pingpong (buffer=1, sync send-wait): ensures buffer is always empty when next element arrives, guaranteeing 100% direct-path hits Results (10 iter, 5 warmup, same machine): no_buffer: 18.0M β 19.5M ops/s (+8.3%) pingpong: 883K β 988K ops/s (+11.9%) --- .../pekko/stream/ActorRefSourceBenchmark.scala | 37 ++++++++++++++++------ 1 file changed, 28 insertions(+), 9 deletions(-) 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 index aa8af59634..11f9e2ef4e 100644 --- a/bench-jmh/src/main/scala/org/apache/pekko/stream/ActorRefSourceBenchmark.scala +++ b/bench-jmh/src/main/scala/org/apache/pekko/stream/ActorRefSourceBenchmark.scala @@ -19,6 +19,7 @@ package org.apache.pekko.stream import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicLong import scala.concurrent.Await import scala.concurrent.duration._ @@ -26,7 +27,6 @@ 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 @@ -44,9 +44,6 @@ class ActorRefSourceBenchmark { implicit val system: ActorSystem = ActorSystem("ActorRefSourceBenchmark") - private var sourceRef: ActorRef = _ - private var doneLatch: CountDownLatch = _ - @Setup def setup(): Unit = { SystemMaterializer(system).materializer @@ -59,17 +56,16 @@ class ActorRefSourceBenchmark { @Benchmark @OperationsPerInvocation(OperationsPerInvocation) - def actorRef_source_push_100k(): Unit = { - doneLatch = new CountDownLatch(1) - val mat = Source + def actorRef_source_no_buffer_100k(): Unit = { + val doneLatch = new CountDownLatch(1) + val sourceRef = Source .actorRef[Any]( completionMatcher = { case "done" => CompletionStrategy.draining }, failureMatcher = PartialFunction.empty, - bufferSize = 1024, + bufferSize = 0, overflowStrategy = OverflowStrategy.dropHead) .toMat(Sink.ignore)(Keep.left) .run() - sourceRef = mat val sender = new Thread(() => { var i = 0 @@ -85,4 +81,27 @@ class ActorRefSourceBenchmark { throw new RuntimeException("ActorRefSource benchmark timed out") sender.join() } + + @Benchmark + @OperationsPerInvocation(OperationsPerInvocation) + def actorRef_source_pingpong_100k(): Unit = { + val counter = new AtomicLong(0) + val sourceRef = Source + .actorRef[Long]( + completionMatcher = { case -1L => CompletionStrategy.draining }, + failureMatcher = PartialFunction.empty, + bufferSize = 1, + overflowStrategy = OverflowStrategy.dropHead) + .toMat(Sink.foreach[Long](_ => counter.incrementAndGet()))(Keep.left) + .run() + + var i = 0L + while (i < OperationsPerInvocation) { + sourceRef ! i + val expected = i + 1 + while (counter.get() < expected) () // spin-wait for consumption + i += 1 + } + sourceRef ! -1L + } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
