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 1110ee198ea7ec02483fd9e59df2bd9e9f29132f Author: 虎鸣 <[email protected]> AuthorDate: Fri Jun 19 18:02:57 2026 +0800 perf(stream): add direct-push fast path in ActorRefSource Motivation: When the buffer is empty and downstream has demand, ActorRefSource was enqueuing elements into the buffer and immediately dequeuing them in tryPush — an unnecessary enqueue/dequeue round-trip on every element. Modification: Check buffer.isEmpty && isAvailable(out) before the buffer path; when both hold, push directly to the outlet, skipping the buffer entirely. Result: Each element in the low-latency case (consumer keeps up with producer) avoids two buffer operations (~5-10ns per element). No behavioral change — ordering is preserved since the fast path only fires when the buffer is empty. Tests: - sbt "stream-tests / Test / testOnly ...ActorRefSourceSpec" — 15/15 pass - New test sends elements one at a time with continuous demand and verifies FIFO ordering through the direct-push path. References: Refs akkadotnet/akka.net#8263 --- .../pekko/stream/scaladsl/ActorRefSourceSpec.scala | 17 +++++++++++++++++ .../org/apache/pekko/stream/impl/ActorRefSource.scala | 2 ++ 2 files changed, 19 insertions(+) diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/ActorRefSourceSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/ActorRefSourceSpec.scala index c7538c7539..ac350d9c56 100644 --- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/ActorRefSourceSpec.scala +++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/ActorRefSourceSpec.scala @@ -248,5 +248,22 @@ class ActorRefSourceSpec extends StreamSpec { } } } + "push directly without buffer round-trip when buffer is empty and demand exists" in { + val s = TestSubscriber.manualProbe[Int]() + val ref = Source + .actorRef({ case "ok" => CompletionStrategy.draining }, PartialFunction.empty, 100, OverflowStrategy.fail) + .to(Sink.fromSubscriber(s)) + .run() + val sub = s.expectSubscription() + // Request enough demand upfront so every element hits the direct-push fast path + sub.request(10) + // Send elements one at a time — each should push directly without buffering + for (n <- 1 to 10) { + ref ! n + s.expectNext(n) + } + ref ! "ok" + s.expectComplete() + } } } diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/ActorRefSource.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/ActorRefSource.scala index 1a76f23bf0..1b36f4fd20 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/ActorRefSource.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/ActorRefSource.scala @@ -85,6 +85,8 @@ private object ActorRefSource { m, buf.used, name) + } else if (buf.isEmpty && isAvailable(out)) { + push(out, m) } else if (!buf.isFull) { buf.enqueue(m) tryPush() --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
