This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 2610cda2af perf: batch async boundary elements (#3288)
2610cda2af is described below
commit 2610cda2af448c7fa23fa97fc937e711b74e03fd
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Sun Jul 5 23:18:20 2026 +0800
perf: batch async boundary elements (#3288)
Motivation:
Async stream islands send each element across the internal actor boundary
as a separate boundary event, which adds per-element allocation and mailbox
traffic.
Modification:
Batch elements produced during an interpreter run for internal stream
boundaries and flush them when the interpreter parks. Let the input boundary
implement the internal BoundarySubscriber directly, cap output batch size at
1024 elements, flush pending batch elements before completion or upstream
failure, and keep cancellation and already-terminated paths clearing pending
elements. Store initialized boundary lists in arrays, add a fused baseline to
the async-boundary benchmark, and ad [...]
Result:
Internal async boundary crossings allocate fewer boundary events and actor
messages while preserving element order, demand bounds, terminal ordering, and
supervision behavior.
Tests:
- rtk scalafmt --mode diff-ref=origin/main -- passed
- rtk scalafmt --list --mode diff-ref=origin/main -- passed
- rtk git diff --check -- passed
- rtk sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.FusingSpec -- -z \"drain asynchronous boundary batches
before failing\"" -- passed, 1 test
- rtk sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.FusingSpec
org.apache.pekko.stream.scaladsl.PublisherSinkSpec
org.apache.pekko.stream.io.FileSinkSpec
org.apache.pekko.stream.scaladsl.FlowMapWithResourceSpec" "stream /
mimaReportBinaryIssues" -- passed, 53 tests and MiMa
- rtk bash -lc 'set -o pipefail; sbt "bench-jmh/Jmh/run -wi 1 -i 3 -w 5s -r
5s -f 1 -prof gc .*AsyncBoundaryThroughputBenchmark.*" 2>&1 | tee
/tmp/pekko-async-boundary-final-squashed.log' -- passed
- rtk qodercli --help -- passed, confirmed -p/--print, --output-format,
--cwd, and --attachment
- qodercli stdout review of /tmp/project-review.diff -- passed, no must-fix
findings
- Independent subAgent review of /tmp/project-review.diff -- passed, no
must-fix findings
References:
None - internal stream boundary throughput optimization
---
.../stream/AsyncBoundaryThroughputBenchmark.scala | 2 +-
.../scala/org/apache/pekko/stream/FusingSpec.scala | 186 ++++++++++++++++
.../pr-2916-boundary-event-allocation.excludes | 6 +
.../stream/impl/fusing/ActorGraphInterpreter.scala | 245 +++++++++++++++++----
4 files changed, 398 insertions(+), 41 deletions(-)
diff --git
a/bench-jmh/src/main/scala/org/apache/pekko/stream/AsyncBoundaryThroughputBenchmark.scala
b/bench-jmh/src/main/scala/org/apache/pekko/stream/AsyncBoundaryThroughputBenchmark.scala
index d508d074c3..b93ec299ca 100644
---
a/bench-jmh/src/main/scala/org/apache/pekko/stream/AsyncBoundaryThroughputBenchmark.scala
+++
b/bench-jmh/src/main/scala/org/apache/pekko/stream/AsyncBoundaryThroughputBenchmark.scala
@@ -52,7 +52,7 @@ class AsyncBoundaryThroughputBenchmark {
implicit val system: ActorSystem =
ActorSystem("AsyncBoundaryThroughputBenchmark", config)
- @Param(Array("1", "3", "10"))
+ @Param(Array("0", "1", "3", "10"))
var asyncBoundaries = 0
var source: Source[Int, NotUsed] = _
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala
index 542dee5be8..03648f78c5 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala
@@ -19,15 +19,20 @@ import duration._
import org.apache.pekko
import pekko.Done
+import pekko.stream.QueueOfferResult
import pekko.stream.impl.UnfoldResourceSource
import pekko.stream.impl.fusing.GraphInterpreter
import pekko.stream.scaladsl._
import pekko.stream.stage.GraphStage
+import pekko.stream.testkit.TestPublisher
import pekko.stream.testkit.StreamSpec
import pekko.stream.testkit.Utils.TE
+import pekko.stream.testkit.scaladsl.TestSink
class FusingSpec extends StreamSpec {
+ val asyncBoundaryInputBuffer = Attributes.inputBuffer(16, 16)
+
def actorRunningStage = {
GraphInterpreter.currentInterpreter.context
}
@@ -47,6 +52,187 @@ class FusingSpec extends StreamSpec {
.sorted should ===(0 to 198 by 2)
}
+ "preserve elements across repeated asynchronous boundary batches" in {
+ val elements = 1 to 5000
+
+ Source(elements)
+ .map(identity)
+ .async
+ .addAttributes(asyncBoundaryInputBuffer)
+ .runWith(Sink.seq)
+ .futureValue should ===(elements)
+ }
+
+ "preserve elements across chained asynchronous boundary batches" in {
+ val elements = 1 to 5000
+
+ Source(elements)
+ .map(identity)
+ .async
+ .addAttributes(asyncBoundaryInputBuffer)
+ .map(identity)
+ .async
+ .addAttributes(asyncBoundaryInputBuffer)
+ .runWith(Sink.seq)
+ .futureValue should ===(elements)
+ }
+
+ "flush asynchronous boundary batches when async input suspends the
interpreter" in {
+ val elements = 1 to 64
+ val (queue, downstream) = Source
+ .fromGraph(Source.queue[Int](elements.size))
+ .async
+ .addAttributes(ActorAttributes.syncProcessingLimit(1) and
asyncBoundaryInputBuffer)
+ .toMat(TestSink[Int]())(Keep.both)
+ .run()
+
+ downstream.request(elements.size)
+ elements.foreach { elem =>
+ queue.offer(elem) should ===(QueueOfferResult.Enqueued)
+ }
+ downstream.expectNextN(elements)
+
+ queue.complete()
+ downstream.expectComplete()
+ }
+
+ "drain asynchronous boundary batches before completing" in {
+ val elements = 1 to 64
+
+ Source(elements)
+ .async
+ .addAttributes(asyncBoundaryInputBuffer)
+ .runWith(TestSink[Int]())
+ .request(elements.size)
+ .expectNextN(elements)
+ .expectComplete()
+ }
+
+ "drain asynchronous boundary batches before failing" in {
+ val elements = 1 to 64
+ val ex = TE("boom")
+ val upstream = TestPublisher.probe[Int]()
+ val downstream = Source
+ .fromPublisher(upstream)
+ .async
+ .addAttributes(asyncBoundaryInputBuffer)
+ .runWith(TestSink[Int]())
+
+ downstream.request(elements.size + 1)
+
+ var sent = 0
+ while (sent < elements.size) {
+ val request = upstream.expectRequest()
+ val send = math.min(request, elements.size - sent).toInt
+ elements.slice(sent, sent + send).foreach(upstream.sendNext)
+ sent += send
+ }
+
+ upstream.sendError(ex)
+ downstream.expectNextN(elements)
+ downstream.expectError(ex)
+ }
+
+ "not emit elements after an asynchronous boundary failure" in {
+ val ex = TE("boom")
+ val probe = Source(1 to 64)
+ .concat(Source.failed(ex))
+ .async
+ .addAttributes(asyncBoundaryInputBuffer)
+ .runWith(TestSink[Int]())
+
+ probe.request(65)
+
+ var expected = 1
+ var errorSignalled = false
+ while (!errorSignalled && expected <= 64) {
+ probe.expectNextOrError(expected, ex) match {
+ case Right(_) =>
+ expected += 1
+ case Left(_) =>
+ errorSignalled = true
+ }
+ }
+ if (!errorSignalled) probe.expectError(ex)
+ }
+
+ "propagate cancellation with asynchronous boundary elements in flight" in {
+ val upstream = TestPublisher.probe[Int]()
+ val downstream = Source
+ .fromPublisher(upstream)
+ .async
+ .addAttributes(asyncBoundaryInputBuffer)
+ .runWith(TestSink[Int]())
+
+ downstream.request(16)
+ upstream.expectRequest() should be >= 16L
+ (1 to 16).foreach(upstream.sendNext)
+
+ downstream.expectNext(1)
+ downstream.cancel()
+
+ upstream.expectCancellation()
+ }
+
+ "not exceed downstream demand across asynchronous boundary batches" in {
+ val downstream = Source(1 to 1000)
+ .async
+ .addAttributes(asyncBoundaryInputBuffer)
+ .runWith(TestSink[Int]())
+
+ downstream.request(1)
+ downstream.expectNext(1)
+ downstream.expectNoMessage(100.millis)
+
+ downstream.request(2)
+ downstream.expectNext(2, 3)
+ downstream.expectNoMessage(100.millis)
+
+ downstream.cancel()
+ }
+
+ "preserve resuming supervision across asynchronous boundary batches" in {
+ Source(List(1, 2, -1, 3, 4))
+ .async
+ .addAttributes(asyncBoundaryInputBuffer)
+ .map { elem =>
+ require(elem > 0)
+ elem
+ }
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
+ .runWith(Sink.seq)
+ .futureValue should ===(Seq(1, 2, 3, 4))
+ }
+
+ "preserve restarting supervision across asynchronous boundary batches" in {
+ Source(List(1, 3, -1, 5, 7))
+ .async
+ .addAttributes(asyncBoundaryInputBuffer)
+ .scan(0) { (old, current) =>
+ require(current > 0)
+ old + current
+ }
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.restartingDecider))
+ .runWith(Sink.seq)
+ .futureValue should ===(Seq(0, 1, 4, 0, 5, 12))
+ }
+
+ "preserve stopping supervision across asynchronous boundary batches" in {
+ val ex = TE("boom")
+ Source(List(1, 2, -1, 3, 4))
+ .async
+ .addAttributes(asyncBoundaryInputBuffer)
+ .map {
+ case -1 => throw ex
+ case elem => elem
+ }
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.stoppingDecider))
+ .runWith(TestSink[Int]())
+ .request(5)
+ .expectNext(1, 2)
+ .expectError(ex)
+ }
+
"use multiple actors when there are asynchronous boundaries in the
subflows (manual)" in {
val async = Flow[Int].map(x => { testActor ! actorRunningStage; x
}).async
Source(0 to 9)
diff --git
a/stream/src/main/mima-filters/2.0.x.backwards.excludes/pr-2916-boundary-event-allocation.excludes
b/stream/src/main/mima-filters/2.0.x.backwards.excludes/pr-2916-boundary-event-allocation.excludes
index d8a5fbb51a..4b4d0fc10d 100644
---
a/stream/src/main/mima-filters/2.0.x.backwards.excludes/pr-2916-boundary-event-allocation.excludes
+++
b/stream/src/main/mima-filters/2.0.x.backwards.excludes/pr-2916-boundary-event-allocation.excludes
@@ -28,3 +28,9 @@
ProblemFilters.exclude[Problem]("org.apache.pekko.stream.impl.fusing.ActorGraphI
ProblemFilters.exclude[Problem]("org.apache.pekko.stream.impl.fusing.GraphInterpreterShell*Abort*")
ProblemFilters.exclude[Problem]("org.apache.pekko.stream.impl.fusing.GraphInterpreterShell*AsyncInput*")
ProblemFilters.exclude[Problem]("org.apache.pekko.stream.impl.fusing.GraphInterpreterShell*ResumeShell*")
+
+# Optimize internal async boundary batching
+ProblemFilters.exclude[FinalMethodProblem]("org.apache.pekko.stream.impl.fusing.ActorGraphInterpreter#BatchingActorInputBoundary.onNext")
+ProblemFilters.exclude[FinalMethodProblem]("org.apache.pekko.stream.impl.fusing.ActorGraphInterpreter#BatchingActorInputBoundary.onError")
+ProblemFilters.exclude[FinalMethodProblem]("org.apache.pekko.stream.impl.fusing.ActorGraphInterpreter#BatchingActorInputBoundary.onComplete")
+ProblemFilters.exclude[FinalMethodProblem]("org.apache.pekko.stream.impl.fusing.ActorGraphInterpreter#BatchingActorInputBoundary.onSubscribe")
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala
index a7156ff29e..53dc14b6ca 100644
---
a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala
@@ -53,6 +53,11 @@ import org.reactivestreams.Subscription
object Resume extends DeadLetterSuppression with
NoSerializationVerificationNeeded
object Snapshot extends NoSerializationVerificationNeeded
+ private[stream] trait BoundarySubscriber extends Subscriber[Any] {
+ def next(elem: AnyRef): Unit
+ def batch(elements: Array[AnyRef]): Unit
+ }
+
sealed abstract class BoundaryEvent extends DeadLetterSuppression with
NoSerializationVerificationNeeded {
def shell: GraphInterpreterShell
@@ -83,7 +88,7 @@ import org.reactivestreams.Subscription
override def execute(): Unit = {
if (GraphInterpreter.Debug)
println(s"${boundary.shell.interpreter.Name} onError
port=${boundary.internalPortName}")
- boundary.onError(cause)
+ boundary.receiveError(cause)
}
override def shell: GraphInterpreterShell = boundary.shell
@@ -95,7 +100,7 @@ import org.reactivestreams.Subscription
override def execute(): Unit = {
if (GraphInterpreter.Debug)
println(s"${boundary.shell.interpreter.Name} onComplete
port=${boundary.internalPortName}")
- boundary.onComplete()
+ boundary.receiveComplete()
}
override def shell: GraphInterpreterShell = boundary.shell
@@ -107,7 +112,20 @@ import org.reactivestreams.Subscription
override def execute(): Unit = {
if (GraphInterpreter.Debug)
println(s"${boundary.shell.interpreter.Name} onNext $e
port=${boundary.internalPortName}")
- boundary.onNext(e)
+ boundary.receiveNext(e)
+ }
+
+ override def shell: GraphInterpreterShell = boundary.shell
+ override def logic: GraphStageLogic = boundary
+ override def cancel(): Unit = ()
+ }
+
+ final class OnNextBatch(boundary: BatchingActorInputBoundary, elements:
Array[AnyRef]) extends SimpleBoundaryEvent {
+ override def execute(): Unit = {
+ if (GraphInterpreter.Debug)
+ println(
+ s"${boundary.shell.interpreter.Name} onNextBatch
size=${elements.length} port=${boundary.internalPortName}")
+ boundary.receiveNextBatch(elements)
}
override def shell: GraphInterpreterShell = boundary.shell
@@ -121,7 +139,7 @@ import org.reactivestreams.Subscription
if (GraphInterpreter.Debug)
println(s"${boundary.shell.interpreter.Name} onSubscribe
port=${boundary.internalPortName}")
boundary.shell.subscribeArrived()
- boundary.onSubscribe(subscription)
+ boundary.receiveSubscribe(subscription)
}
override def shell: GraphInterpreterShell = boundary.shell
@@ -136,7 +154,8 @@ import org.reactivestreams.Subscription
publisher: Publisher[Any],
val internalPortName: String)
extends UpstreamBoundaryStageLogic[Any]
- with OutHandler {
+ with OutHandler
+ with BoundarySubscriber {
if (size <= 0) throw new IllegalArgumentException("buffer size cannot be
zero")
if ((size & (size - 1)) != 0) throw new IllegalArgumentException("buffer
size must be a power of two")
@@ -158,29 +177,32 @@ import org.reactivestreams.Subscription
def setActor(actor: ActorRef): Unit = this.actor = actor
- override def preStart(): Unit = {
- publisher.subscribe(new Subscriber[Any] {
- override def onError(t: Throwable): Unit = {
- ReactiveStreamsCompliance.requireNonNullException(t)
- actor ! new OnError(BatchingActorInputBoundary.this, t)
- }
+ override def preStart(): Unit = publisher.subscribe(this)
- override def onSubscribe(s: Subscription): Unit = {
- ReactiveStreamsCompliance.requireNonNullSubscription(s)
- actor ! new OnSubscribe(BatchingActorInputBoundary.this, s)
- }
+ final override def onError(t: Throwable): Unit = {
+ ReactiveStreamsCompliance.requireNonNullException(t)
+ actor ! new OnError(this, t)
+ }
- override def onComplete(): Unit = {
- actor ! new OnComplete(BatchingActorInputBoundary.this)
- }
+ final override def onSubscribe(s: Subscription): Unit = {
+ ReactiveStreamsCompliance.requireNonNullSubscription(s)
+ actor ! new OnSubscribe(this, s)
+ }
- override def onNext(t: Any): Unit = {
- ReactiveStreamsCompliance.requireNonNullElement(t)
- actor ! new OnNext(BatchingActorInputBoundary.this, t)
- }
- })
+ final override def onComplete(): Unit =
+ actor ! new OnComplete(this)
+
+ final override def onNext(t: Any): Unit = {
+ ReactiveStreamsCompliance.requireNonNullElement(t)
+ next(t.asInstanceOf[AnyRef])
}
+ final override def next(elem: AnyRef): Unit =
+ actor ! new OnNext(this, elem)
+
+ final override def batch(elements: Array[AnyRef]): Unit =
+ actor ! new OnNextBatch(this, elements)
+
@InternalStableApi
private def dequeue(): Any = {
val elem = inputBuffer(nextInputElementCursor)
@@ -214,7 +236,7 @@ import org.reactivestreams.Subscription
}
@InternalStableApi
- def onNext(elem: Any): Unit = {
+ def receiveNext(elem: Any): Unit = {
if (!upstreamCompleted) {
if (inputBufferElements == size) throw new
IllegalStateException("Input buffer overrun")
inputBuffer((nextInputElementCursor + inputBufferElements) &
IndexMask) = elem.asInstanceOf[AnyRef]
@@ -223,7 +245,20 @@ import org.reactivestreams.Subscription
}
}
- def onError(e: Throwable): Unit =
+ @InternalStableApi
+ def receiveNextBatch(elements: Array[AnyRef]): Unit = {
+ var i = 0
+ while (i < elements.length) {
+ receiveNext(elements(i))
+ i += 1
+ }
+ }
+
+ @InternalStableApi
+ def onNextBatch(elements: Array[AnyRef]): Unit =
+ receiveNextBatch(elements)
+
+ def receiveError(e: Throwable): Unit =
if (!upstreamCompleted || downstreamCanceled.isEmpty) {
upstreamCompleted = true
clear()
@@ -236,16 +271,16 @@ import org.reactivestreams.Subscription
if (!(upstreamCompleted || downstreamCanceled.isDefined) && (upstream ne
null)) {
upstream.cancel()
}
- if (!isClosed(out)) onError(e)
+ if (!isClosed(out)) receiveError(e)
}
- def onComplete(): Unit =
+ def receiveComplete(): Unit =
if (!upstreamCompleted) {
upstreamCompleted = true
if (inputBufferElements == 0) complete(out)
}
- def onSubscribe(subscription: Subscription): Unit = {
+ def receiveSubscribe(subscription: Subscription): Unit = {
ReactiveStreamsCompliance.requireNonNullSubscription(subscription)
if (downstreamCanceled.isDefined) {
upstreamCompleted = true
@@ -400,6 +435,7 @@ import org.reactivestreams.Subscription
def getActor: ActorRef = this.actor
private var subscriber: Subscriber[Any] = _
+ private var boundarySubscriber: BoundarySubscriber = _
private var downstreamDemand: Long = 0L
// This flag is only used if complete/fail is called externally since this
op turns into a Finished one inside the
// interpreter (i.e. inside this op this flag has no effects since if it
is completed the op will not be invoked)
@@ -409,15 +445,78 @@ import org.reactivestreams.Subscription
// when upstream failed before we got the exposed publisher
private var upstreamCompleted: Boolean = false
+ private var firstBatchedElement: AnyRef = _
+ private var batchedElements: Array[AnyRef] = _
+ private var batchedElementIndex = 0
+
+ private final val InitialBatchSize = 16
+ private final val MaxBatchSize = 1024
+
private def onNext(elem: Any): Unit = {
downstreamDemand -= 1
- tryOnNext(subscriber, elem)
+ if (boundarySubscriber eq null) tryOnNext(subscriber, elem)
+ else enqueueBatchElement(elem)
+ }
+
+ private def enqueueBatchElement(elem: Any): Unit = {
+ ReactiveStreamsCompliance.requireNonNullElement(elem)
+ val element = elem.asInstanceOf[AnyRef]
+ if (batchedElementIndex == MaxBatchSize) flushBatch()
+ val index = batchedElementIndex
+ if (index == 0) {
+ firstBatchedElement = element
+ } else {
+ var elements = batchedElements
+ if (elements eq null) {
+ elements = new Array[AnyRef](InitialBatchSize)
+ batchedElements = elements
+ } else if (index == elements.length) {
+ val newElements = new Array[AnyRef](index << 1)
+ System.arraycopy(elements, 0, newElements, 0, index)
+ elements = newElements
+ batchedElements = elements
+ }
+ if (index == 1) elements(0) = firstBatchedElement
+ elements(index) = element
+ }
+ batchedElementIndex = index + 1
+ }
+
+ def flushBatch(): Unit = {
+ val count = batchedElementIndex
+ if (count == 0) return
+
+ val boundary = boundarySubscriber
+ if (boundary eq null) {
+ clearBatch()
+ return
+ }
+
+ if (count == 1) boundary.next(firstBatchedElement)
+ else {
+ val elements = new Array[AnyRef](count)
+ System.arraycopy(batchedElements, 0, elements, 0, count)
+ boundary.batch(elements)
+ }
+
+ clearBatch()
+ }
+
+ private def clearBatch(): Unit = {
+ val count = batchedElementIndex
+ if (count != 0) {
+ firstBatchedElement = null
+ if (batchedElements ne null)
+ java.util.Arrays.fill(batchedElements, 0, count, null)
+ batchedElementIndex = 0
+ }
}
private def complete(): Unit = {
// No need to complete if had already been cancelled, or we closed
earlier
if (!(upstreamCompleted || downstreamCompleted)) {
upstreamCompleted = true
+ flushBatch()
publisher.shutdown(None)
if (subscriber ne null) tryOnComplete(subscriber)
}
@@ -426,10 +525,11 @@ import org.reactivestreams.Subscription
def fail(e: Throwable): Unit = {
// No need to fail if had already been cancelled, or we closed earlier
if (!(downstreamCompleted || upstreamCompleted)) {
+ flushBatch()
upstreamCompleted = true
publisher.shutdown(Some(e))
if ((subscriber ne null) && !e.isInstanceOf[SpecViolation])
tryOnError(subscriber, e)
- }
+ } else clearBatch()
}
setHandler(in, this)
@@ -460,6 +560,10 @@ import org.reactivestreams.Subscription
publisher.takePendingSubscribers().foreach { sub =>
if (subscriber eq null) {
subscriber = sub
+ boundarySubscriber = sub match {
+ case boundary: BoundarySubscriber => boundary
+ case _ => null
+ }
val subscription = new Subscription with
SubscriptionWithCancelException {
override def request(elements: Long): Unit = actor ! new
RequestMore(ActorOutputBoundary.this, elements)
override def cancel(cause: Throwable): Unit = actor ! new
Cancel(ActorOutputBoundary.this, cause)
@@ -487,8 +591,10 @@ import org.reactivestreams.Subscription
}
def cancel(cause: Throwable): Unit = {
+ clearBatch()
downstreamCompletionCause = Some(cause)
subscriber = null
+ boundarySubscriber = null
publisher.shutdown(Some(new ActorPublisher.NormalShutdownException))
cancel(in, cause)
}
@@ -503,7 +609,10 @@ import org.reactivestreams.Subscription
* INTERNAL API
*/
@InternalApi private[pekko] object GraphInterpreterShell {
- import ActorGraphInterpreter.BoundaryEvent
+ import ActorGraphInterpreter.{ ActorOutputBoundary,
BatchingActorInputBoundary, BoundaryEvent }
+
+ private val EmptyOutputBoundaries = new Array[ActorOutputBoundary](0)
+ private val EmptyInputBoundaries = new Array[BatchingActorInputBoundary](0)
/**
* @param promise Will be completed upon processing the event, or failed if
processing the event throws
@@ -522,6 +631,7 @@ import org.reactivestreams.Subscription
if (!shell.waitingForShutdown) {
shell.interpreter.runAsyncInput(logic, evt, promise, handler)
if (eventLimit == 1 && shell.interpreter.isSuspended) {
+ shell.flushOutputs()
shell.sendResume(true)
0
} else shell.runBatch(eventLimit - 1)
@@ -587,8 +697,8 @@ import org.reactivestreams.Subscription
// TODO: really needed?
private var subscribesPending = 0
- private var inputs: List[BatchingActorInputBoundary] = Nil
- private var outputs: List[ActorOutputBoundary] = Nil
+ private var inputs: Array[BatchingActorInputBoundary] = EmptyInputBoundaries
+ private var outputs: Array[ActorOutputBoundary] = EmptyOutputBoundaries
/*
* Limits the number of events processed by the interpreter before scheduling
@@ -618,21 +728,47 @@ import org.reactivestreams.Subscription
eventLimit: Int): Int = {
this.self = self
this.enqueueToShortCircuit = enqueueToShortCircuit
+ val logicCount = logics.length
var i = 0
- while (i < logics.length) {
- logics(i) match {
+ var inputBoundaryCount = 0
+ var outputBoundaryCount = 0
+ while (i < logicCount) {
+ val logic = logics(i)
+ logic match {
+ case _: BatchingActorInputBoundary => inputBoundaryCount += 1
+ case _: ActorOutputBoundary => outputBoundaryCount += 1
+ case _ =>
+ }
+ i += 1
+ }
+
+ val initializedInputs =
+ if (inputBoundaryCount == 0) EmptyInputBoundaries else new
Array[BatchingActorInputBoundary](inputBoundaryCount)
+ val initializedOutputs =
+ if (outputBoundaryCount == 0) EmptyOutputBoundaries else new
Array[ActorOutputBoundary](outputBoundaryCount)
+
+ i = 0
+ var inputIndex = 0
+ var outputIndex = 0
+ while (i < logicCount) {
+ val logic = logics(i)
+ logic match {
case in: BatchingActorInputBoundary =>
in.setActor(self)
subscribesPending += 1
- inputs ::= in
+ initializedInputs(inputIndex) = in
+ inputIndex += 1
case out: ActorOutputBoundary =>
out.setActor(self)
out.subscribePending()
- outputs ::= out
+ initializedOutputs(outputIndex) = out
+ outputIndex += 1
case _ =>
}
i += 1
}
+ inputs = initializedInputs
+ outputs = initializedOutputs
interpreter.init(subMat)
runBatch(eventLimit)
@@ -666,6 +802,7 @@ import org.reactivestreams.Subscription
try {
val usingShellLimit = shellEventLimit < actorEventLimit
val remainingQuota = interpreter.execute(Math.min(actorEventLimit,
shellEventLimit))
+ flushOutputs()
if (interpreter.isCompleted) {
// Cannot stop right away if not completely subscribed
if (canShutDown) interpreterCompleted = true
@@ -684,6 +821,16 @@ import org.reactivestreams.Subscription
}
}
+ private def flushOutputs(): Unit = {
+ val outputBoundaries = outputs
+ val count = outputBoundaries.length
+ var i = 0
+ while (i < count) {
+ outputBoundaries(i).flushBatch()
+ i += 1
+ }
+ }
+
/**
* Attempts to abort execution, by first propagating the reason given until
either
* - the interpreter successfully finishes
@@ -700,7 +847,13 @@ import org.reactivestreams.Subscription
// This should handle termination while interpreter is running. If the
upstream have been closed already this
// call has no effect and therefore does the right thing: nothing.
try {
- inputs.foreach(_.onInternalError(reason))
+ val inputBoundaries = inputs
+ val inputsCount = inputBoundaries.length
+ var inputIndex = 0
+ while (inputIndex < inputsCount) {
+ inputBoundaries(inputIndex).onInternalError(reason)
+ inputIndex += 1
+ }
interpreter.execute(abortLimit)
interpreter.finish()
} catch {
@@ -711,8 +864,20 @@ import org.reactivestreams.Subscription
interpreterCompleted = true
// Will only have an effect if the above call to the interpreter failed
to emit a proper failure to the downstream
// otherwise this will have no effect
- outputs.foreach(_.fail(reason))
- inputs.foreach(_.cancel(reason))
+ val outputBoundaries = outputs
+ val count = outputBoundaries.length
+ var i = 0
+ while (i < count) {
+ outputBoundaries(i).fail(reason)
+ i += 1
+ }
+ val inputBoundaries = inputs
+ val inputsCount = inputBoundaries.length
+ var inputIndex = 0
+ while (inputIndex < inputsCount) {
+ inputBoundaries(inputIndex).cancel(reason)
+ inputIndex += 1
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
