(pekko) 05/06: test(stream): redesign ActorRefSource benchmark for optimization-sensitive measurement

[he...@apache.org]

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: 虎鸣 <he...@alibaba-inc.com>
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: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org