Low level interpreter tests working without GraphAssembly #22423
This commit is contained in:
Johan Andrén
GitHub
parent
44260fe5d3
commit
c028b550f2
26 changed files with 354 additions and 575 deletions
|
|
@ -27,7 +27,7 @@ class FlatMapMergeBenchmark {
|
|||
|
||||
var graph: RunnableGraph[Future[Done]] = _
|
||||
|
||||
def createSource(count: Int): Graph[SourceShape[Int], NotUsed] = akka.stream.Fusing.aggressive(Source.repeat(1).take(count))
|
||||
def createSource(count: Int): Graph[SourceShape[Int], NotUsed] = Source.repeat(1).take(count)
|
||||
|
||||
@Setup
|
||||
def setup(): Unit = {
|
||||
|
|
|
|||
|
|
@ -122,7 +122,7 @@ class FusedGraphsBenchmark {
|
|||
val testSink = Sink.fromGraph(new JitSafeCompletionLatch)
|
||||
|
||||
def fuse(r: RunnableGraph[CountDownLatch]): RunnableGraph[CountDownLatch] = {
|
||||
RunnableGraph.fromGraph(Fusing.aggressive(r))
|
||||
RunnableGraph.fromGraph(r)
|
||||
}
|
||||
|
||||
val identityStage = new IdentityStage
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
package akka.stream
|
||||
|
||||
/*
|
||||
import java.util
|
||||
import java.util.concurrent.TimeUnit
|
||||
|
||||
|
|
@ -357,3 +358,4 @@ class NewLayoutBenchmark {
|
|||
}
|
||||
|
||||
}
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -276,17 +276,6 @@ public class FlowDocTest extends AbstractJavaTest {
|
|||
}
|
||||
|
||||
public void fusingAndAsync() {
|
||||
//#explicit-fusing
|
||||
Flow<Integer, Integer, NotUsed> flow =
|
||||
Flow.of(Integer.class).map(x -> x * 2).filter(x -> x > 500);
|
||||
Graph<FlowShape<Integer, Integer>, NotUsed> fused =
|
||||
akka.stream.Fusing.aggressive(flow);
|
||||
|
||||
Source.fromIterator(() -> Stream.iterate(0, x -> x + 1).iterator())
|
||||
.via(fused)
|
||||
.take(1000);
|
||||
//#explicit-fusing
|
||||
|
||||
//#flow-async
|
||||
Source.range(1, 3)
|
||||
.map(x -> x + 1).async()
|
||||
|
|
|
|||
|
|
@ -227,24 +227,17 @@ which will be running on the thread pools they have been configured to run on -
|
|||
Operator Fusion
|
||||
---------------
|
||||
|
||||
Akka Streams 2.0 contains an initial version of stream operator fusion support. This means that
|
||||
the processing steps of a flow or stream graph can be executed within the same Actor and has three
|
||||
consequences:
|
||||
By default Akka Streams will fuse the stream operators. This means that the processing steps of a flow or
|
||||
stream graph can be executed within the same Actor and has two consequences:
|
||||
|
||||
* starting up a stream may take longer than before due to executing the fusion algorithm
|
||||
* passing elements from one processing stage to the next is a lot faster between fused
|
||||
stages due to avoiding the asynchronous messaging overhead
|
||||
* fused stream processing stages no longer run in parallel to each other, meaning that
|
||||
* fused stream processing stages does not run in parallel to each other, meaning that
|
||||
only up to one CPU core is used for each fused part
|
||||
|
||||
The first point can be countered by pre-fusing and then reusing a stream blueprint as sketched below:
|
||||
|
||||
.. includecode:: ../code/docs/stream/FlowDocTest.java#explicit-fusing
|
||||
|
||||
In order to balance the effects of the second and third bullet points you will have to insert asynchronous
|
||||
boundaries manually into your flows and graphs by way of adding ``Attributes.asyncBoundary`` using the method
|
||||
``async`` on ``Source``, ``Sink`` and ``Flow`` to pieces that shall communicate with the rest of the graph in
|
||||
an asynchronous fashion.
|
||||
To allow for parallel processing you will have to insert asynchronous boundaries manually into your flows and
|
||||
graphs by way of adding ``Attributes.asyncBoundary`` using the method ``async`` on ``Source``, ``Sink`` and ``Flow``
|
||||
to pieces that shall communicate with the rest of the graph in an asynchronous fashion.
|
||||
|
||||
.. includecode:: ../code/docs/stream/FlowDocTest.java#flow-async
|
||||
|
||||
|
|
|
|||
|
|
@ -220,19 +220,6 @@ class FlowDocSpec extends AkkaSpec {
|
|||
//#flow-mat-combine
|
||||
}
|
||||
|
||||
"explicit fusing" in {
|
||||
//#explicit-fusing
|
||||
import akka.stream.Fusing
|
||||
|
||||
val flow = Flow[Int].map(_ * 2).filter(_ > 500)
|
||||
val fused = Fusing.aggressive(flow)
|
||||
|
||||
Source.fromIterator { () => Iterator from 0 }
|
||||
.via(fused)
|
||||
.take(1000)
|
||||
//#explicit-fusing
|
||||
}
|
||||
|
||||
"defining asynchronous boundaries" in {
|
||||
//#flow-async
|
||||
Source(List(1, 2, 3))
|
||||
|
|
|
|||
|
|
@ -115,15 +115,6 @@ class GraphDSLDocSpec extends AkkaSpec {
|
|||
priorityJobsIn.carbonCopy(),
|
||||
resultsOut.carbonCopy())
|
||||
|
||||
// A Shape must also be able to create itself from existing ports
|
||||
override def copyFromPorts(
|
||||
inlets: immutable.Seq[Inlet[_]],
|
||||
outlets: immutable.Seq[Outlet[_]]) = {
|
||||
assert(inlets.size == this.inlets.size)
|
||||
assert(outlets.size == this.outlets.size)
|
||||
// This is why order matters when overriding inlets and outlets.
|
||||
PriorityWorkerPoolShape[In, Out](inlets(0).as[In], inlets(1).as[In], outlets(0).as[Out])
|
||||
}
|
||||
}
|
||||
//#graph-dsl-components-shape
|
||||
|
||||
|
|
|
|||
|
|
@ -229,24 +229,17 @@ which will be running on the thread pools they have been configured to run on -
|
|||
Operator Fusion
|
||||
---------------
|
||||
|
||||
Akka Streams 2.0 contains an initial version of stream operator fusion support. This means that
|
||||
the processing steps of a flow or stream graph can be executed within the same Actor and has three
|
||||
consequences:
|
||||
By default Akka Streams will fuse the stream operators. This means that the processing steps of a flow or
|
||||
stream graph can be executed within the same Actor and has two consequences:
|
||||
|
||||
* starting up a stream may take longer than before due to executing the fusion algorithm
|
||||
* passing elements from one processing stage to the next is a lot faster between fused
|
||||
stages due to avoiding the asynchronous messaging overhead
|
||||
* fused stream processing stages no longer run in parallel to each other, meaning that
|
||||
* fused stream processing stages does not run in parallel to each other, meaning that
|
||||
only up to one CPU core is used for each fused part
|
||||
|
||||
The first point can be countered by pre-fusing and then reusing a stream blueprint as sketched below:
|
||||
|
||||
.. includecode:: ../code/docs/stream/FlowDocSpec.scala#explicit-fusing
|
||||
|
||||
In order to balance the effects of the second and third bullet points you will have to insert asynchronous
|
||||
boundaries manually into your flows and graphs by way of adding ``Attributes.asyncBoundary`` using the method
|
||||
``async`` on ``Source``, ``Sink`` and ``Flow`` to pieces that shall communicate with the rest of the graph in an
|
||||
asynchronous fashion.
|
||||
To allow for parallel processing you will have to insert asynchronous boundaries manually into your flows and
|
||||
graphs by way of adding ``Attributes.asyncBoundary`` using the method ``async`` on ``Source``, ``Sink`` and ``Flow``
|
||||
to pieces that shall communicate with the rest of the graph in an asynchronous fashion.
|
||||
|
||||
.. includecode:: ../code/docs/stream/FlowDocSpec.scala#flow-async
|
||||
|
||||
|
|
|
|||
|
|
@ -84,7 +84,10 @@ class StringSerializer(val system: ExtendedActorSystem) extends BaseSerializer w
|
|||
new String(bytes, "UTF-8")
|
||||
}
|
||||
|
||||
override def toBinary(o: AnyRef): Array[Byte] = o.asInstanceOf[String].getBytes("UTF-8")
|
||||
override def toBinary(o: AnyRef): Array[Byte] = {
|
||||
println("stringserializer")
|
||||
o.asInstanceOf[String].getBytes("UTF-8")
|
||||
}
|
||||
|
||||
override def fromBinary(bytes: Array[Byte], manifest: Option[Class[_]]): AnyRef = new String(bytes, "UTF-8")
|
||||
|
||||
|
|
|
|||
|
|
@ -91,6 +91,9 @@ class DaemonMsgCreateSerializerSpec extends AkkaSpec {
|
|||
DaemonMsgCreate(Props(classOf[MyActorWithParam], "a string"), Deploy.local, "/user/test", system.actorFor("/user")))
|
||||
println(String.valueOf(encodeHex(bytes)))
|
||||
*/
|
||||
import org.apache.commons.codec.binary.Hex.encodeHex
|
||||
|
||||
println(String.valueOf(encodeHex(SerializationExtension(system).serialize("a string").get)))
|
||||
|
||||
val oldBytesHex =
|
||||
"0a6a12020a001a48616b6b612e72656d6f74652e73657269616c697a6174696f" +
|
||||
|
|
|
|||
|
|
@ -1,13 +1,12 @@
|
|||
/**
|
||||
* Copyright (C) 2015-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
/* FIXME enable again
|
||||
|
||||
package akka.stream.impl
|
||||
|
||||
import akka.stream.stage.GraphStageLogic.{ EagerTerminateOutput, EagerTerminateInput }
|
||||
import akka.stream.testkit.StreamSpec
|
||||
import akka.stream._
|
||||
import akka.stream.Fusing.aggressive
|
||||
import akka.stream.scaladsl._
|
||||
import akka.stream.stage._
|
||||
import akka.stream.testkit.Utils.assertAllStagesStopped
|
||||
|
|
@ -60,11 +59,15 @@ class GraphStageLogicSpec extends StreamSpec with GraphInterpreterSpecKit {
|
|||
setHandler(in, new InHandler {
|
||||
override def onPush(): Unit = push(out, grab(in))
|
||||
override def onUpstreamFinish(): Unit = complete(out)
|
||||
override def toString = "InHandler"
|
||||
})
|
||||
setHandler(out, new OutHandler {
|
||||
override def onPull(): Unit = pull(in)
|
||||
override def toString = "OutHandler"
|
||||
})
|
||||
override def toString = "GraphStageLogicSpec.passthroughLogic"
|
||||
}
|
||||
override def toString = "GraphStageLogicSpec.passthrough"
|
||||
}
|
||||
|
||||
object emitEmptyIterable extends GraphStage[SourceShape[Int]] {
|
||||
|
|
@ -75,8 +78,8 @@ class GraphStageLogicSpec extends StreamSpec with GraphInterpreterSpecKit {
|
|||
setHandler(out, new OutHandler {
|
||||
override def onPull(): Unit = emitMultiple(out, Iterator.empty, () ⇒ emit(out, 42, () ⇒ completeStage()))
|
||||
})
|
||||
|
||||
}
|
||||
override def toString = "GraphStageLogicSpec.emitEmptyIterable"
|
||||
}
|
||||
|
||||
final case class ReadNEmitN(n: Int) extends GraphStage[FlowShape[Int, Int]] {
|
||||
|
|
@ -142,24 +145,6 @@ class GraphStageLogicSpec extends StreamSpec with GraphInterpreterSpecKit {
|
|||
|
||||
}
|
||||
|
||||
"emit all things before completing with two fused stages" in assertAllStagesStopped {
|
||||
val g = aggressive(Flow[Int].via(emit1234).via(emit5678))
|
||||
|
||||
Source.empty.via(g).runWith(TestSink.probe)
|
||||
.request(9)
|
||||
.expectNextN(1 to 8)
|
||||
.expectComplete()
|
||||
}
|
||||
|
||||
"emit all things before completing with three fused stages" in assertAllStagesStopped {
|
||||
val g = aggressive(Flow[Int].via(emit1234).via(passThrough).via(emit5678))
|
||||
|
||||
Source.empty.via(g).runWith(TestSink.probe)
|
||||
.request(9)
|
||||
.expectNextN(1 to 8)
|
||||
.expectComplete()
|
||||
}
|
||||
|
||||
"emit properly after empty iterable" in assertAllStagesStopped {
|
||||
|
||||
Source.fromGraph(emitEmptyIterable).runWith(Sink.seq).futureValue should ===(List(42))
|
||||
|
|
@ -207,6 +192,8 @@ class GraphStageLogicSpec extends StreamSpec with GraphInterpreterSpecKit {
|
|||
.connect(passThrough.out, Downstream)
|
||||
.init()
|
||||
|
||||
// note: a bit dangerous assumptions about connection and logic positions here
|
||||
// if anything around creating the logics and connections in the builder changes this may fail
|
||||
interpreter.complete(interpreter.connections(0))
|
||||
interpreter.cancel(interpreter.connections(1))
|
||||
interpreter.execute(2)
|
||||
|
|
@ -216,8 +203,8 @@ class GraphStageLogicSpec extends StreamSpec with GraphInterpreterSpecKit {
|
|||
|
||||
interpreter.isCompleted should ===(false)
|
||||
interpreter.isSuspended should ===(false)
|
||||
interpreter.isStageCompleted(interpreter.logics(0)) should ===(true)
|
||||
interpreter.isStageCompleted(interpreter.logics(1)) should ===(false)
|
||||
interpreter.isStageCompleted(interpreter.logics(1)) should ===(true)
|
||||
interpreter.isStageCompleted(interpreter.logics(2)) should ===(false)
|
||||
}
|
||||
|
||||
"not allow push from constructor" in {
|
||||
|
|
@ -249,4 +236,4 @@ class GraphStageLogicSpec extends StreamSpec with GraphInterpreterSpecKit {
|
|||
}
|
||||
|
||||
}
|
||||
*/
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
/**
|
||||
* Copyright (C) 2015-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
/* FIXME enable again
|
||||
|
||||
package akka.stream.impl.fusing
|
||||
|
||||
import java.util.concurrent.CountDownLatch
|
||||
|
|
@ -364,6 +364,7 @@ class ActorGraphInterpreterSpec extends StreamSpec {
|
|||
ise.getCause.getCause should (have message ("violating your spec"))
|
||||
}
|
||||
|
||||
/* TODO this one does not work
|
||||
"be able to handle Subscriber spec violations without leaking" in assertAllStagesStopped {
|
||||
val filthySubscriber = new Subscriber[Int] {
|
||||
override def onSubscribe(s: Subscription): Unit = s.request(1)
|
||||
|
|
@ -390,7 +391,8 @@ class ActorGraphInterpreterSpec extends StreamSpec {
|
|||
|
||||
upstream.expectCancellation()
|
||||
}
|
||||
*/
|
||||
|
||||
}
|
||||
}
|
||||
*/
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
/**
|
||||
* Copyright (C) 2015-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
/* FIXME enable again
|
||||
|
||||
package akka.stream.impl.fusing
|
||||
|
||||
import akka.stream.scaladsl.{ Sink, Source }
|
||||
|
|
@ -114,4 +114,3 @@ class ChasingEventsSpec extends AkkaSpec {
|
|||
}
|
||||
|
||||
}
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
/* FIXME enable again
|
||||
|
||||
package akka.stream.impl.fusing
|
||||
|
||||
import akka.stream.testkit.StreamSpec
|
||||
|
|
@ -10,17 +10,17 @@ class GraphInterpreterFailureModesSpec extends StreamSpec with GraphInterpreterS
|
|||
"GraphInterpreter" must {
|
||||
|
||||
"handle failure on onPull" in new FailingStageSetup {
|
||||
lastEvents() should be(Set(PreStart(stage)))
|
||||
lastEvents() should be(Set(PreStart(insideOutStage)))
|
||||
|
||||
downstream.pull()
|
||||
failOnNextEvent()
|
||||
stepAll()
|
||||
|
||||
lastEvents() should be(Set(Cancel(upstream), OnError(downstream, testException), PostStop(stage)))
|
||||
lastEvents() should be(Set(Cancel(upstream), OnError(downstream, testException), PostStop(insideOutStage)))
|
||||
}
|
||||
|
||||
"handle failure on onPush" in new FailingStageSetup {
|
||||
lastEvents() should be(Set(PreStart(stage)))
|
||||
lastEvents() should be(Set(PreStart(insideOutStage)))
|
||||
|
||||
downstream.pull()
|
||||
stepAll()
|
||||
|
|
@ -29,22 +29,22 @@ class GraphInterpreterFailureModesSpec extends StreamSpec with GraphInterpreterS
|
|||
failOnNextEvent()
|
||||
stepAll()
|
||||
|
||||
lastEvents() should be(Set(Cancel(upstream), OnError(downstream, testException), PostStop(stage)))
|
||||
lastEvents() should be(Set(Cancel(upstream), OnError(downstream, testException), PostStop(insideOutStage)))
|
||||
}
|
||||
|
||||
"handle failure on onPull while cancel is pending" in new FailingStageSetup {
|
||||
lastEvents() should be(Set(PreStart(stage)))
|
||||
lastEvents() should be(Set(PreStart(insideOutStage)))
|
||||
|
||||
downstream.pull()
|
||||
downstream.cancel()
|
||||
failOnNextEvent()
|
||||
stepAll()
|
||||
|
||||
lastEvents() should be(Set(Cancel(upstream), PostStop(stage)))
|
||||
lastEvents() should be(Set(Cancel(upstream), PostStop(insideOutStage)))
|
||||
}
|
||||
|
||||
"handle failure on onPush while complete is pending" in new FailingStageSetup {
|
||||
lastEvents() should be(Set(PreStart(stage)))
|
||||
lastEvents() should be(Set(PreStart(insideOutStage)))
|
||||
|
||||
downstream.pull()
|
||||
stepAll()
|
||||
|
|
@ -54,47 +54,47 @@ class GraphInterpreterFailureModesSpec extends StreamSpec with GraphInterpreterS
|
|||
failOnNextEvent()
|
||||
stepAll()
|
||||
|
||||
lastEvents() should be(Set(OnError(downstream, testException), PostStop(stage)))
|
||||
lastEvents() should be(Set(OnError(downstream, testException), PostStop(insideOutStage)))
|
||||
}
|
||||
|
||||
"handle failure on onUpstreamFinish" in new FailingStageSetup {
|
||||
lastEvents() should be(Set(PreStart(stage)))
|
||||
lastEvents() should be(Set(PreStart(insideOutStage)))
|
||||
|
||||
upstream.complete()
|
||||
failOnNextEvent()
|
||||
stepAll()
|
||||
|
||||
lastEvents() should be(Set(OnError(downstream, testException), PostStop(stage)))
|
||||
lastEvents() should be(Set(OnError(downstream, testException), PostStop(insideOutStage)))
|
||||
}
|
||||
|
||||
"handle failure on onUpstreamFailure" in new FailingStageSetup {
|
||||
lastEvents() should be(Set(PreStart(stage)))
|
||||
lastEvents() should be(Set(PreStart(insideOutStage)))
|
||||
|
||||
upstream.fail(TE("another exception")) // this is not the exception that will be propagated
|
||||
failOnNextEvent()
|
||||
stepAll()
|
||||
|
||||
lastEvents() should be(Set(OnError(downstream, testException), PostStop(stage)))
|
||||
lastEvents() should be(Set(OnError(downstream, testException), PostStop(insideOutStage)))
|
||||
}
|
||||
|
||||
"handle failure on onDownstreamFinish" in new FailingStageSetup {
|
||||
lastEvents() should be(Set(PreStart(stage)))
|
||||
lastEvents() should be(Set(PreStart(insideOutStage)))
|
||||
|
||||
downstream.cancel()
|
||||
failOnNextEvent()
|
||||
stepAll()
|
||||
|
||||
lastEvents() should be(Set(Cancel(upstream), PostStop(stage)))
|
||||
lastEvents() should be(Set(Cancel(upstream), PostStop(insideOutStage)))
|
||||
}
|
||||
|
||||
"handle failure in preStart" in new FailingStageSetup(initFailOnNextEvent = true) {
|
||||
stepAll()
|
||||
|
||||
lastEvents() should be(Set(Cancel(upstream), OnError(downstream, testException), PostStop(stage)))
|
||||
lastEvents() should be(Set(Cancel(upstream), OnError(downstream, testException), PostStop(insideOutStage)))
|
||||
}
|
||||
|
||||
"handle failure in postStop" in new FailingStageSetup {
|
||||
lastEvents() should be(Set(PreStart(stage)))
|
||||
lastEvents() should be(Set(PreStart(insideOutStage)))
|
||||
|
||||
upstream.complete()
|
||||
downstream.cancel()
|
||||
|
|
@ -110,4 +110,4 @@ class GraphInterpreterFailureModesSpec extends StreamSpec with GraphInterpreterS
|
|||
}
|
||||
|
||||
}
|
||||
*/
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
/**
|
||||
* Copyright (C) 2015-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
/* FIXME enable again
|
||||
|
||||
package akka.stream.impl.fusing
|
||||
|
||||
import akka.stream.testkit.StreamSpec
|
||||
|
|
@ -1181,4 +1181,3 @@ class GraphInterpreterPortsSpec extends StreamSpec with GraphInterpreterSpecKit
|
|||
}
|
||||
|
||||
}
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -1,14 +1,12 @@
|
|||
/**
|
||||
* Copyright (C) 2015-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
/* FIXME enable again
|
||||
|
||||
package akka.stream.impl.fusing
|
||||
|
||||
import akka.NotUsed
|
||||
import akka.stream.OverflowStrategy
|
||||
import akka.stream.scaladsl.{ Balance, Broadcast, Merge, Zip }
|
||||
import akka.stream.testkit.StreamSpec
|
||||
import akka.stream.{ OverflowStrategy, Attributes }
|
||||
import akka.stream.scaladsl.{ Merge, Broadcast, Balance, Zip }
|
||||
import GraphInterpreter._
|
||||
|
||||
class GraphInterpreterSpec extends StreamSpec with GraphInterpreterSpecKit {
|
||||
import GraphStages._
|
||||
|
|
@ -46,18 +44,10 @@ class GraphInterpreterSpec extends StreamSpec with GraphInterpreterSpecKit {
|
|||
val sink = new DownstreamProbe[Int]("sink")
|
||||
|
||||
// Constructing an assembly by hand and resolving ambiguities
|
||||
val assembly = new GraphAssembly(
|
||||
stages = Array(identity, identity),
|
||||
originalAttributes = Array(Attributes.none, Attributes.none),
|
||||
ins = Array(identity.in, identity.in, null),
|
||||
inOwners = Array(0, 1, -1),
|
||||
outs = Array(null, identity.out, identity.out),
|
||||
outOwners = Array(-1, 0, 1))
|
||||
val (logics, _, _) = GraphInterpreterSpecKit.createLogics(Array(identity), Array(source), Array(sink))
|
||||
val connections = GraphInterpreterSpecKit.createLinearFlowConnections(logics)
|
||||
|
||||
manualInit(assembly)
|
||||
interpreter.attachDownstreamBoundary(2, sink)
|
||||
interpreter.attachUpstreamBoundary(0, source)
|
||||
interpreter.init(null)
|
||||
manualInit(logics, connections)
|
||||
|
||||
lastEvents() should ===(Set.empty)
|
||||
|
||||
|
|
@ -67,7 +57,6 @@ class GraphInterpreterSpec extends StreamSpec with GraphInterpreterSpecKit {
|
|||
source.onNext(1)
|
||||
lastEvents() should ===(Set(OnNext(sink, 1)))
|
||||
}
|
||||
|
||||
"implement detacher stage" in new TestSetup {
|
||||
val source = new UpstreamProbe[Int]("source")
|
||||
val sink = new DownstreamProbe[Int]("sink")
|
||||
|
|
@ -372,4 +361,4 @@ class GraphInterpreterSpec extends StreamSpec with GraphInterpreterSpecKit {
|
|||
}
|
||||
|
||||
}
|
||||
*/
|
||||
|
||||
|
|
|
|||
|
|
@ -1,103 +1,268 @@
|
|||
/**
|
||||
* Copyright (C) 2015-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
/* FIXME enable again
|
||||
|
||||
package akka.stream.impl.fusing
|
||||
|
||||
import akka.event.Logging
|
||||
import akka.stream.ActorAttributes.SupervisionStrategy
|
||||
import akka.stream.Supervision.Decider
|
||||
import akka.stream._
|
||||
import akka.stream.impl.fusing.GraphInterpreter.{ DownstreamBoundaryStageLogic, Failed, GraphAssembly, UpstreamBoundaryStageLogic }
|
||||
import akka.stream.impl.fusing.GraphInterpreter.{ Connection, DownstreamBoundaryStageLogic, Failed, UpstreamBoundaryStageLogic }
|
||||
import akka.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler, _ }
|
||||
import akka.stream.testkit.StreamSpec
|
||||
import akka.stream.testkit.Utils.TE
|
||||
import akka.stream.impl.fusing.GraphInterpreter.GraphAssembly
|
||||
|
||||
import scala.collection.{ Map ⇒ SMap }
|
||||
import scala.language.existentials
|
||||
|
||||
object GraphInterpreterSpecKit {
|
||||
|
||||
/**
|
||||
* Create logics and enumerate stages and ports
|
||||
*
|
||||
* @param stages Stages to "materialize" into graph stage logic instances
|
||||
* @param upstreams Upstream boundary logics that are already instances of graph stage logic and should be
|
||||
* part of the graph, is placed before the rest of the stages
|
||||
* @param downstreams Downstream boundary logics, is placed after the other stages
|
||||
* @param attributes Optional set of attributes to pass to the stages when creating the logics
|
||||
* @return Created logics and the maps of all inlets respective outlets to those logics
|
||||
*/
|
||||
private[stream] def createLogics(
|
||||
stages: Array[GraphStageWithMaterializedValue[_ <: Shape, _]],
|
||||
upstreams: Array[UpstreamBoundaryStageLogic[_]],
|
||||
downstreams: Array[DownstreamBoundaryStageLogic[_]],
|
||||
attributes: Array[Attributes] = Array.empty): (Array[GraphStageLogic], SMap[Inlet[_], GraphStageLogic], SMap[Outlet[_], GraphStageLogic]) = {
|
||||
if (attributes.nonEmpty && attributes.length != stages.length)
|
||||
throw new IllegalArgumentException("Attributes must be either empty or one per stage")
|
||||
|
||||
var inOwners = SMap.empty[Inlet[_], GraphStageLogic]
|
||||
var outOwners = SMap.empty[Outlet[_], GraphStageLogic]
|
||||
|
||||
val logics = Array.ofDim[GraphStageLogic](upstreams.length + stages.length + downstreams.length)
|
||||
var idx = 0
|
||||
|
||||
while (idx < upstreams.length) {
|
||||
val upstream = upstreams(idx)
|
||||
upstream.stageId = idx
|
||||
logics(idx) = upstream
|
||||
upstream.out.id = 0
|
||||
outOwners = outOwners + (upstream.out → upstream)
|
||||
idx += 1
|
||||
}
|
||||
|
||||
var stageIdx = 0
|
||||
while (stageIdx < stages.length) {
|
||||
val stage = stages(stageIdx)
|
||||
setPortIds(stage.shape)
|
||||
|
||||
val stageAttributes =
|
||||
if (attributes.nonEmpty) stage.traversalBuilder.attributes and attributes(stageIdx)
|
||||
else stage.traversalBuilder.attributes
|
||||
|
||||
val logic = stage.createLogicAndMaterializedValue(stageAttributes)._1
|
||||
logic.stageId = idx
|
||||
|
||||
var inletIdx = 0
|
||||
while (inletIdx < stage.shape.inlets.length) {
|
||||
val inlet = stage.shape.inlets(inletIdx)
|
||||
inlet.id = inletIdx
|
||||
inOwners = inOwners + (inlet → logic)
|
||||
inletIdx += 1
|
||||
}
|
||||
|
||||
var outletIdx = 0
|
||||
while (outletIdx < stage.shape.outlets.length) {
|
||||
val outlet = stage.shape.outlets(outletIdx)
|
||||
outlet.id = outletIdx
|
||||
outOwners = outOwners + (outlet → logic)
|
||||
outletIdx += 1
|
||||
}
|
||||
logics(idx) = logic
|
||||
|
||||
idx += 1
|
||||
stageIdx += 1
|
||||
}
|
||||
|
||||
var downstreamIdx = 0
|
||||
while (downstreamIdx < downstreams.length) {
|
||||
val downstream = downstreams(downstreamIdx)
|
||||
downstream.stageId = idx
|
||||
logics(idx) = downstream
|
||||
downstream.in.id = 0
|
||||
inOwners = inOwners + (downstream.in → downstream)
|
||||
|
||||
idx += 1
|
||||
downstreamIdx += 1
|
||||
}
|
||||
|
||||
(logics, inOwners, outOwners)
|
||||
}
|
||||
|
||||
/**
|
||||
* Create connections given a list of flow logics where each one has one connection to the next one
|
||||
*/
|
||||
private[stream] def createLinearFlowConnections(logics: Seq[GraphStageLogic]): Array[Connection] = {
|
||||
require(logics.length >= 2, s"$logics is too short to create a linear flow")
|
||||
logics.sliding(2).zipWithIndex.map {
|
||||
case (window, idx) ⇒
|
||||
val outOwner = window(0)
|
||||
val inOwner = window(1)
|
||||
|
||||
val connection = new Connection(
|
||||
id = idx,
|
||||
outOwnerId = outOwner.stageId,
|
||||
outOwner = outOwner,
|
||||
outHandler = outOwner.outHandler(0),
|
||||
inOwner = inOwner,
|
||||
inOwnerId = inOwner.stageId,
|
||||
inHandler = inOwner.inHandler(0)
|
||||
)
|
||||
|
||||
outOwner.portToConn(outOwner.inCount) = connection
|
||||
inOwner.portToConn(0) = connection
|
||||
|
||||
connection
|
||||
}.toArray
|
||||
}
|
||||
|
||||
/**
|
||||
* Create interpreter connections for all the given `connectedPorts`.
|
||||
*/
|
||||
private[stream] def createConnections(
|
||||
logics: Seq[GraphStageLogic],
|
||||
connectedPorts: Seq[(Outlet[_], Inlet[_])],
|
||||
inOwners: SMap[Inlet[_], GraphStageLogic],
|
||||
outOwners: SMap[Outlet[_], GraphStageLogic]): Array[Connection] = {
|
||||
|
||||
val connections = Array.ofDim[Connection](connectedPorts.size)
|
||||
connectedPorts.zipWithIndex.foreach {
|
||||
case ((outlet, inlet), idx) ⇒
|
||||
|
||||
val outOwner = outOwners(outlet)
|
||||
val inOwner = inOwners(inlet)
|
||||
|
||||
val connection = new Connection(
|
||||
id = idx,
|
||||
outOwnerId = outOwner.stageId,
|
||||
outOwner = outOwner,
|
||||
outHandler = outOwner.outHandler(outlet.id),
|
||||
inOwnerId = inOwner.stageId,
|
||||
inOwner = inOwner,
|
||||
inHandler = inOwner.inHandler(inlet.id)
|
||||
)
|
||||
|
||||
connections(idx) = connection
|
||||
inOwner.portToConn(inlet.id) = connection
|
||||
outOwner.portToConn(outOwner.inCount + outlet.id) = connection
|
||||
}
|
||||
connections
|
||||
}
|
||||
|
||||
private def setPortIds(shape: Shape): Unit = {
|
||||
shape.inlets.zipWithIndex.foreach {
|
||||
case (inlet, idx) ⇒ inlet.id = idx
|
||||
}
|
||||
shape.outlets.zipWithIndex.foreach {
|
||||
case (outlet, idx) ⇒ outlet.id = idx
|
||||
}
|
||||
}
|
||||
|
||||
private def setPortIds(stage: GraphStageWithMaterializedValue[_ <: Shape, _]): Unit = {
|
||||
stage.shape.inlets.zipWithIndex.foreach { case (inlet, idx) ⇒ inlet.id = idx }
|
||||
stage.shape.outlets.zipWithIndex.foreach { case (inlet, idx) ⇒ inlet.id = idx }
|
||||
}
|
||||
|
||||
private def setLogicIds(logics: Array[GraphStageLogic]): Unit = {
|
||||
logics.zipWithIndex.foreach { case (logic, idx) ⇒ logic.stageId = idx }
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
trait GraphInterpreterSpecKit extends StreamSpec {
|
||||
|
||||
import GraphInterpreterSpecKit._
|
||||
val logger = Logging(system, "InterpreterSpecKit")
|
||||
|
||||
abstract class Builder {
|
||||
private var _interpreter: GraphInterpreter = _
|
||||
|
||||
protected def interpreter: GraphInterpreter = _interpreter
|
||||
|
||||
def stepAll(): Unit = interpreter.execute(eventLimit = Int.MaxValue)
|
||||
|
||||
def step(): Unit = interpreter.execute(eventLimit = 1)
|
||||
|
||||
object Upstream extends UpstreamBoundaryStageLogic[Int] {
|
||||
override val out = Outlet[Int]("up")
|
||||
out.id = 0
|
||||
override def toString = "Upstream"
|
||||
|
||||
setHandler(out, new OutHandler {
|
||||
override def onPull() = {
|
||||
// TODO handler needed but should it do anything?
|
||||
}
|
||||
|
||||
override def toString = "Upstream.OutHandler"
|
||||
})
|
||||
}
|
||||
|
||||
object Downstream extends DownstreamBoundaryStageLogic[Int] {
|
||||
override val in = Inlet[Int]("down")
|
||||
in.id = 0
|
||||
setHandler(in, new InHandler {
|
||||
override def onPush() = {
|
||||
// TODO handler needed but should it do anything?
|
||||
}
|
||||
|
||||
override def toString = "Downstream.InHandler"
|
||||
})
|
||||
|
||||
override def toString = "Downstream"
|
||||
}
|
||||
|
||||
class AssemblyBuilder(stages: Seq[GraphStageWithMaterializedValue[_ <: Shape, _]]) {
|
||||
var upstreams = Vector.empty[(UpstreamBoundaryStageLogic[_], Inlet[_])]
|
||||
var downstreams = Vector.empty[(Outlet[_], DownstreamBoundaryStageLogic[_])]
|
||||
var connections = Vector.empty[(Outlet[_], Inlet[_])]
|
||||
private var upstreams = Vector.empty[UpstreamBoundaryStageLogic[_]]
|
||||
private var downstreams = Vector.empty[DownstreamBoundaryStageLogic[_]]
|
||||
private var connectedPorts = Vector.empty[(Outlet[_], Inlet[_])]
|
||||
|
||||
def connect[T](upstream: UpstreamBoundaryStageLogic[T], in: Inlet[T]): AssemblyBuilder = {
|
||||
upstreams :+= upstream → in
|
||||
upstreams :+= upstream
|
||||
connectedPorts :+= upstream.out → in
|
||||
this
|
||||
}
|
||||
|
||||
def connect[T](out: Outlet[T], downstream: DownstreamBoundaryStageLogic[T]): AssemblyBuilder = {
|
||||
downstreams :+= out → downstream
|
||||
downstreams :+= downstream
|
||||
connectedPorts :+= out → downstream.in
|
||||
this
|
||||
}
|
||||
|
||||
def connect[T](out: Outlet[T], in: Inlet[T]): AssemblyBuilder = {
|
||||
connections :+= out → in
|
||||
connectedPorts :+= out → in
|
||||
this
|
||||
}
|
||||
|
||||
def buildAssembly(): GraphAssembly = {
|
||||
val ins = upstreams.map(_._2) ++ connections.map(_._2)
|
||||
val outs = connections.map(_._1) ++ downstreams.map(_._1)
|
||||
val inOwners = ins.map { in ⇒ stages.indexWhere(_.shape.inlets.contains(in)) }
|
||||
val outOwners = outs.map { out ⇒ stages.indexWhere(_.shape.outlets.contains(out)) }
|
||||
|
||||
new GraphAssembly(
|
||||
stages.toArray,
|
||||
Array.fill(stages.size)(Attributes.none),
|
||||
(ins ++ Vector.fill(downstreams.size)(null)).toArray,
|
||||
(inOwners ++ Vector.fill(downstreams.size)(-1)).toArray,
|
||||
(Vector.fill(upstreams.size)(null) ++ outs).toArray,
|
||||
(Vector.fill(upstreams.size)(-1) ++ outOwners).toArray)
|
||||
}
|
||||
|
||||
def init(): Unit = {
|
||||
val assembly = buildAssembly()
|
||||
val (logics, inOwners, outOwners) = createLogics(stages.toArray, upstreams.toArray, downstreams.toArray)
|
||||
val conns = createConnections(logics, connectedPorts, inOwners, outOwners)
|
||||
|
||||
val (conns, logics) =
|
||||
assembly.materialize(Attributes.none, assembly.stages.map(_.module), new java.util.HashMap, _ ⇒ ())
|
||||
_interpreter = new GraphInterpreter(assembly, NoMaterializer, logger, logics, conns,
|
||||
(_, _, _) ⇒ (), fuzzingMode = false, null)
|
||||
|
||||
for ((upstream, i) ← upstreams.zipWithIndex) {
|
||||
_interpreter.attachUpstreamBoundary(conns(i), upstream._1)
|
||||
}
|
||||
|
||||
for ((downstream, i) ← downstreams.zipWithIndex) {
|
||||
_interpreter.attachDownstreamBoundary(conns(i + upstreams.size + connections.size), downstream._2)
|
||||
}
|
||||
|
||||
_interpreter.init(null)
|
||||
manualInit(logics.toArray, conns)
|
||||
}
|
||||
}
|
||||
|
||||
def manualInit(assembly: GraphAssembly): Unit = {
|
||||
val (connections, logics) =
|
||||
assembly.materialize(Attributes.none, assembly.stages.map(_.module), new java.util.HashMap, _ ⇒ ())
|
||||
_interpreter = new GraphInterpreter(assembly, NoMaterializer, logger, logics, connections,
|
||||
(_, _, _) ⇒ (), fuzzingMode = false, null)
|
||||
def manualInit(logics: Array[GraphStageLogic], connections: Array[Connection]): Unit = {
|
||||
_interpreter = new GraphInterpreter(
|
||||
NoMaterializer,
|
||||
logger,
|
||||
logics,
|
||||
connections,
|
||||
onAsyncInput = (_, _, _) ⇒ (),
|
||||
fuzzingMode = false,
|
||||
context = null)
|
||||
_interpreter.init(null)
|
||||
}
|
||||
|
||||
def builder(stages: GraphStageWithMaterializedValue[_ <: Shape, _]*): AssemblyBuilder = new AssemblyBuilder(stages)
|
||||
def builder(stages: GraphStageWithMaterializedValue[_ <: Shape, _]*): AssemblyBuilder =
|
||||
new AssemblyBuilder(stages.toVector)
|
||||
|
||||
}
|
||||
|
||||
abstract class TestSetup extends Builder {
|
||||
|
|
@ -132,6 +297,7 @@ trait GraphInterpreterSpecKit extends StreamSpec {
|
|||
setHandler(out, new OutHandler {
|
||||
override def onPull(): Unit = lastEvent += RequestOne(UpstreamProbe.this)
|
||||
override def onDownstreamFinish(): Unit = lastEvent += Cancel(UpstreamProbe.this)
|
||||
override def toString = s"${UpstreamProbe.this.toString}.outHandler"
|
||||
})
|
||||
|
||||
def onNext(elem: T, eventLimit: Int = Int.MaxValue): Unit = {
|
||||
|
|
@ -161,6 +327,7 @@ trait GraphInterpreterSpecKit extends StreamSpec {
|
|||
override def onPush(): Unit = lastEvent += OnNext(DownstreamProbe.this, grab(in))
|
||||
override def onUpstreamFinish(): Unit = lastEvent += OnComplete(DownstreamProbe.this)
|
||||
override def onUpstreamFailure(ex: Throwable): Unit = lastEvent += OnError(DownstreamProbe.this, ex)
|
||||
override def toString = s"${DownstreamProbe.this.toString}.inHandler"
|
||||
})
|
||||
|
||||
def requestOne(eventLimit: Int = Int.MaxValue): Unit = {
|
||||
|
|
@ -185,6 +352,8 @@ trait GraphInterpreterSpecKit extends StreamSpec {
|
|||
class EventPropagateStage extends GraphStage[FlowShape[Int, Int]] {
|
||||
val in = Inlet[Int]("Propagate.in")
|
||||
val out = Outlet[Int]("Propagate.out")
|
||||
in.id = 0
|
||||
out.id = 0
|
||||
override val shape: FlowShape[Int, Int] = FlowShape(in, out)
|
||||
|
||||
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with InHandler with OutHandler {
|
||||
|
|
@ -196,6 +365,7 @@ trait GraphInterpreterSpecKit extends StreamSpec {
|
|||
|
||||
setHandlers(in, out, this)
|
||||
}
|
||||
override def toString = "EventPropagateStage"
|
||||
}
|
||||
|
||||
// step() means different depending whether we have a stage between the two probes or not
|
||||
|
|
@ -236,29 +406,22 @@ trait GraphInterpreterSpecKit extends StreamSpec {
|
|||
})
|
||||
}
|
||||
|
||||
private val assembly = if (!chasing) {
|
||||
new GraphAssembly(
|
||||
stages = Array.empty,
|
||||
originalAttributes = Array.empty,
|
||||
ins = Array(null),
|
||||
inOwners = Array(-1),
|
||||
outs = Array(null),
|
||||
outOwners = Array(-1))
|
||||
} else {
|
||||
val propagateStage = new EventPropagateStage
|
||||
new GraphAssembly(
|
||||
stages = Array(propagateStage),
|
||||
originalAttributes = Array(Attributes.none),
|
||||
ins = Array(propagateStage.in, null),
|
||||
inOwners = Array(0, -1),
|
||||
outs = Array(null, propagateStage.out),
|
||||
outOwners = Array(-1, 0))
|
||||
}
|
||||
val (logics, connections) =
|
||||
if (!chasing) {
|
||||
val logics = Array[GraphStageLogic](out, in)
|
||||
setLogicIds(logics)
|
||||
val connections = GraphInterpreterSpecKit.createLinearFlowConnections(logics)
|
||||
(logics, connections)
|
||||
} else {
|
||||
val propagateStage = new EventPropagateStage
|
||||
setPortIds(propagateStage)
|
||||
val logics = Array[GraphStageLogic](out, propagateStage.createLogic(Attributes.none), in)
|
||||
setLogicIds(logics)
|
||||
val connections = GraphInterpreterSpecKit.createLinearFlowConnections(logics)
|
||||
(logics, connections)
|
||||
}
|
||||
|
||||
manualInit(assembly)
|
||||
interpreter.attachDownstreamBoundary(interpreter.connections(if (chasing) 1 else 0), in)
|
||||
interpreter.attachUpstreamBoundary(interpreter.connections(0), out)
|
||||
interpreter.init(null)
|
||||
manualInit(logics, connections)
|
||||
}
|
||||
|
||||
abstract class FailingStageSetup(initFailOnNextEvent: Boolean = false) extends TestSetup {
|
||||
|
|
@ -280,7 +443,7 @@ trait GraphInterpreterSpecKit extends StreamSpec {
|
|||
|
||||
// Must be lazy because I turned this stage "inside-out" therefore changing initialization order
|
||||
// to make tests a bit more readable
|
||||
lazy val stage: GraphStageLogic = new GraphStageLogic(stageshape) {
|
||||
lazy val insideOutStage: GraphStageLogic = new GraphStageLogic(stageshape) {
|
||||
private def mayFail(task: ⇒ Unit): Unit = {
|
||||
if (!_failOnNextEvent) task
|
||||
else {
|
||||
|
|
@ -293,24 +456,27 @@ trait GraphInterpreterSpecKit extends StreamSpec {
|
|||
override def onPush(): Unit = mayFail(push(stageout, grab(stagein)))
|
||||
override def onUpstreamFinish(): Unit = mayFail(completeStage())
|
||||
override def onUpstreamFailure(ex: Throwable): Unit = mayFail(failStage(ex))
|
||||
override def toString = "insideOutStage.stagein"
|
||||
})
|
||||
|
||||
setHandler(stageout, new OutHandler {
|
||||
override def onPull(): Unit = mayFail(pull(stagein))
|
||||
override def onDownstreamFinish(): Unit = mayFail(completeStage())
|
||||
override def toString = "insideOutStage.stageout"
|
||||
})
|
||||
|
||||
override def preStart(): Unit = mayFail(lastEvent += PreStart(stage))
|
||||
override def preStart(): Unit = mayFail(lastEvent += PreStart(insideOutStage))
|
||||
override def postStop(): Unit =
|
||||
if (!_failOnPostStop) lastEvent += PostStop(stage)
|
||||
if (!_failOnPostStop) lastEvent += PostStop(insideOutStage)
|
||||
else throw testException
|
||||
|
||||
override def toString = "stage"
|
||||
override def toString = "insideOutStage"
|
||||
}
|
||||
|
||||
private val sandwitchStage = new GraphStage[FlowShape[Int, Int]] {
|
||||
override def shape = stageshape
|
||||
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = stage
|
||||
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = insideOutStage
|
||||
override def toString = "sandwitchStage"
|
||||
}
|
||||
|
||||
class UpstreamPortProbe[T] extends UpstreamProbe[T]("upstreamPort") {
|
||||
|
|
@ -330,8 +496,7 @@ trait GraphInterpreterSpecKit extends StreamSpec {
|
|||
.init()
|
||||
}
|
||||
|
||||
abstract class OneBoundedSetupWithDecider[T](decider: Decider, _ops: GraphStageWithMaterializedValue[Shape, Any]*) extends Builder {
|
||||
val ops = _ops.toArray
|
||||
abstract class OneBoundedSetupWithDecider[T](decider: Decider, ops: GraphStageWithMaterializedValue[Shape, Any]*) extends Builder {
|
||||
|
||||
val upstream = new UpstreamOneBoundedProbe[T]
|
||||
val downstream = new DownstreamOneBoundedPortProbe[T]
|
||||
|
|
@ -349,35 +514,15 @@ trait GraphInterpreterSpecKit extends StreamSpec {
|
|||
private def run() = interpreter.execute(Int.MaxValue)
|
||||
|
||||
private def initialize(): Unit = {
|
||||
import GraphInterpreter.Boundary
|
||||
|
||||
var i = 0
|
||||
val attributes = Array.fill[Attributes](ops.length)(ActorAttributes.supervisionStrategy(decider))
|
||||
val ins = Array.ofDim[Inlet[_]](ops.length + 1)
|
||||
val inOwners = Array.ofDim[Int](ops.length + 1)
|
||||
val outs = Array.ofDim[Outlet[_]](ops.length + 1)
|
||||
val outOwners = Array.ofDim[Int](ops.length + 1)
|
||||
|
||||
ins(ops.length) = null
|
||||
inOwners(ops.length) = Boundary
|
||||
outs(0) = null
|
||||
outOwners(0) = Boundary
|
||||
|
||||
while (i < ops.length) {
|
||||
val stage = ops(i).asInstanceOf[GraphStageWithMaterializedValue[FlowShape[_, _], _]]
|
||||
ins(i) = stage.shape.in
|
||||
inOwners(i) = i
|
||||
outs(i + 1) = stage.shape.out
|
||||
outOwners(i + 1) = i
|
||||
i += 1
|
||||
}
|
||||
|
||||
manualInit(new GraphAssembly(ops, attributes, ins, inOwners, outs, outOwners))
|
||||
interpreter.attachUpstreamBoundary(0, upstream)
|
||||
interpreter.attachDownstreamBoundary(ops.length, downstream)
|
||||
|
||||
interpreter.init(null)
|
||||
|
||||
val supervision = ActorAttributes.supervisionStrategy(decider)
|
||||
val attributes = Array.fill[Attributes](ops.length)(supervision)
|
||||
val (logics, _, _) = createLogics(
|
||||
ops.toArray,
|
||||
Array(upstream),
|
||||
Array(downstream),
|
||||
attributes)
|
||||
val connections = createLinearFlowConnections(logics)
|
||||
manualInit(logics, connections)
|
||||
}
|
||||
|
||||
initialize()
|
||||
|
|
@ -452,6 +597,7 @@ trait GraphInterpreterSpecKit extends StreamSpec {
|
|||
|
||||
}
|
||||
|
||||
abstract class OneBoundedSetup[T](_ops: GraphStageWithMaterializedValue[Shape, Any]*) extends OneBoundedSetupWithDecider[T](Supervision.stoppingDecider, _ops: _*)
|
||||
abstract class OneBoundedSetup[T](_ops: GraphStageWithMaterializedValue[Shape, Any]*)
|
||||
extends OneBoundedSetupWithDecider[T](Supervision.stoppingDecider, _ops: _*)
|
||||
}
|
||||
*/
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
/* FIXME enable again
|
||||
|
||||
package akka.stream.impl.fusing
|
||||
|
||||
import akka.stream.impl.ConstantFun
|
||||
|
|
@ -632,4 +632,3 @@ class InterpreterSpec extends StreamSpec with GraphInterpreterSpecKit {
|
|||
}
|
||||
|
||||
}
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
/* FIXME enable again
|
||||
|
||||
package akka.stream.impl.fusing
|
||||
|
||||
import akka.stream.impl.ConstantFun
|
||||
|
|
@ -128,4 +128,4 @@ class InterpreterStressSpec extends StreamSpec with GraphInterpreterSpecKit {
|
|||
}
|
||||
|
||||
}
|
||||
*/
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
/**
|
||||
* Copyright (C) 2015-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
/* FIXME enable again
|
||||
|
||||
package akka.stream.impl.fusing
|
||||
|
||||
import akka.stream.testkit.StreamSpec
|
||||
|
|
@ -191,4 +191,3 @@ class InterpreterSupervisionSpec extends StreamSpec with GraphInterpreterSpecKit
|
|||
}
|
||||
|
||||
}
|
||||
*/
|
||||
|
|
@ -1,165 +0,0 @@
|
|||
/**
|
||||
* Copyright (C) 2014-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
/* FIXME enable again
|
||||
package akka.stream.impl.fusing
|
||||
|
||||
import akka.stream.testkit.StreamSpec
|
||||
import akka.util.ByteString
|
||||
import akka.stream.stage._
|
||||
import akka.stream.{ Attributes, Supervision }
|
||||
import akka.stream.impl.fusing.GraphStages.SimpleLinearGraphStage
|
||||
|
||||
class IteratorInterpreterSpec extends StreamSpec with GraphInterpreterSpecKit {
|
||||
import Supervision.stoppingDecider
|
||||
|
||||
"IteratorInterpreter" must {
|
||||
|
||||
"work in the happy case" in {
|
||||
val itr = new IteratorInterpreter[Int, Int]((1 to 10).iterator, Seq(
|
||||
Map((x: Int) ⇒ x + 1))).iterator
|
||||
|
||||
itr.toSeq should be(2 to 11)
|
||||
}
|
||||
|
||||
"hasNext should not affect elements" in {
|
||||
val itr = new IteratorInterpreter[Int, Int]((1 to 10).iterator, Seq(
|
||||
Map((x: Int) ⇒ x))).iterator
|
||||
|
||||
itr.hasNext should be(true)
|
||||
itr.hasNext should be(true)
|
||||
itr.hasNext should be(true)
|
||||
itr.hasNext should be(true)
|
||||
itr.hasNext should be(true)
|
||||
|
||||
itr.toSeq should be(1 to 10)
|
||||
}
|
||||
|
||||
"work with ops that need extra pull for complete" in {
|
||||
val itr = new IteratorInterpreter[Int, Int]((1 to 10).iterator, Seq(NaiveTake(1))).iterator
|
||||
|
||||
itr.toSeq should be(Seq(1))
|
||||
}
|
||||
|
||||
"throw exceptions on empty iterator" in {
|
||||
val itr = new IteratorInterpreter[Int, Int](List(1).iterator, Seq(
|
||||
Map((x: Int) ⇒ x))).iterator
|
||||
|
||||
itr.next() should be(1)
|
||||
a[NoSuchElementException] should be thrownBy { itr.next() }
|
||||
}
|
||||
|
||||
"throw exceptions when op in chain throws" in {
|
||||
val itr = new IteratorInterpreter[Int, Int](List(1, 2, 3).iterator, Seq(
|
||||
Map((n: Int) ⇒ if (n == 2) throw new ArithmeticException() else n))).iterator
|
||||
|
||||
itr.next() should be(1)
|
||||
itr.hasNext should be(true)
|
||||
a[ArithmeticException] should be thrownBy { itr.next() }
|
||||
itr.hasNext should be(false)
|
||||
}
|
||||
|
||||
"work with an empty iterator" in {
|
||||
val itr = new IteratorInterpreter[Int, Int](Iterator.empty, Seq(
|
||||
Map((x: Int) ⇒ x + 1))).iterator
|
||||
|
||||
itr.hasNext should be(false)
|
||||
a[NoSuchElementException] should be thrownBy { itr.next() }
|
||||
}
|
||||
|
||||
"able to implement a ByteStringBatcher" in {
|
||||
val testBytes = (1 to 10).map(ByteString(_))
|
||||
|
||||
def newItr(threshold: Int) =
|
||||
new IteratorInterpreter[ByteString, ByteString](testBytes.iterator, Seq(
|
||||
ByteStringBatcher(threshold))).iterator
|
||||
|
||||
val itr1 = newItr(20)
|
||||
itr1.next() should be(ByteString(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
|
||||
itr1.hasNext should be(false)
|
||||
|
||||
val itr2 = newItr(10)
|
||||
itr2.next() should be(ByteString(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
|
||||
itr2.hasNext should be(false)
|
||||
|
||||
val itr3 = newItr(5)
|
||||
itr3.next() should be(ByteString(1, 2, 3, 4, 5))
|
||||
(6 to 10) foreach { i ⇒
|
||||
itr3.hasNext should be(true)
|
||||
itr3.next() should be(ByteString(i))
|
||||
}
|
||||
itr3.hasNext should be(false)
|
||||
|
||||
val itr4 =
|
||||
new IteratorInterpreter[ByteString, ByteString](Iterator.empty, Seq(
|
||||
ByteStringBatcher(10))).iterator
|
||||
|
||||
itr4.hasNext should be(false)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// This op needs an extra pull round to finish
|
||||
case class NaiveTake[T](count: Int) extends SimpleLinearGraphStage[T] {
|
||||
|
||||
override def createLogic(attributes: Attributes): GraphStageLogic =
|
||||
new GraphStageLogic(shape) with InHandler with OutHandler {
|
||||
private var left: Int = count
|
||||
|
||||
override def onPush(): Unit = {
|
||||
left -= 1
|
||||
push(out, grab(in))
|
||||
}
|
||||
|
||||
override def onPull(): Unit = {
|
||||
if (left == 0) completeStage()
|
||||
else pull(in)
|
||||
}
|
||||
|
||||
setHandlers(in, out, this)
|
||||
}
|
||||
|
||||
override def toString = "NaiveTake"
|
||||
}
|
||||
|
||||
case class ByteStringBatcher(threshold: Int, compact: Boolean = true) extends SimpleLinearGraphStage[ByteString] {
|
||||
require(threshold > 0, "Threshold must be positive")
|
||||
|
||||
override def createLogic(attributes: Attributes): GraphStageLogic =
|
||||
new GraphStageLogic(shape) with InHandler with OutHandler {
|
||||
private var buf: ByteString = ByteString.empty
|
||||
private var passthrough: Boolean = false
|
||||
|
||||
override def onPush(): Unit = {
|
||||
val elem = grab(in)
|
||||
if (passthrough) push(out, elem)
|
||||
else {
|
||||
buf = buf ++ elem
|
||||
if (buf.size >= threshold) {
|
||||
val batch = if (compact) buf.compact else buf
|
||||
passthrough = true
|
||||
buf = ByteString.empty
|
||||
push(out, batch)
|
||||
} else pull(in)
|
||||
}
|
||||
}
|
||||
|
||||
override def onPull(): Unit = {
|
||||
if (isClosed(in)) {
|
||||
push(out, buf)
|
||||
completeStage()
|
||||
} else pull(in)
|
||||
}
|
||||
|
||||
override def onUpstreamFinish(): Unit = {
|
||||
if (passthrough || buf.isEmpty) completeStage()
|
||||
else if (isAvailable(out)) onPull()
|
||||
}
|
||||
|
||||
setHandlers(in, out, this)
|
||||
}
|
||||
|
||||
override def toString = "ByteStringBatcher"
|
||||
}
|
||||
}
|
||||
*/
|
||||
|
|
@ -1,7 +1,7 @@
|
|||
/**
|
||||
* Copyright (C) 2015-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
/* FIXME enable again
|
||||
|
||||
package akka.stream.impl.fusing
|
||||
|
||||
import akka.actor.{ NoSerializationVerificationNeeded, ActorRef }
|
||||
|
|
@ -199,4 +199,3 @@ class KeepGoingStageSpec extends StreamSpec {
|
|||
}
|
||||
|
||||
}
|
||||
*/
|
||||
|
|
@ -1,7 +1,7 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
/* FIXME enable again
|
||||
|
||||
package akka.stream.impl.fusing
|
||||
|
||||
import akka.stream.Attributes
|
||||
|
|
@ -226,4 +226,4 @@ class LifecycleInterpreterSpec extends StreamSpec with GraphInterpreterSpecKit {
|
|||
override def toString = "PushFinish"
|
||||
}
|
||||
}
|
||||
*/
|
||||
|
||||
|
|
|
|||
|
|
@ -63,12 +63,10 @@ object GraphInterpreter {
|
|||
* Contains all the necessary information for the GraphInterpreter to be able to implement a connection
|
||||
* between an output and input ports.
|
||||
*
|
||||
* @param id Identifier of the connection. Corresponds to the array slot in the [[GraphAssembly]]
|
||||
* @param inOwnerId Identifier of the owner of the input side of the connection. Corresponds to the array slot in
|
||||
* the [[GraphAssembly]]
|
||||
* @param id Identifier of the connection.
|
||||
* @param inOwnerId Identifier of the owner of the input side of the connection.
|
||||
* @param inOwner The stage logic that corresponds to the input side of the connection.
|
||||
* @param outOwnerId Identifier of the owner of the output side of the connection. Corresponds to the array slot
|
||||
* in the [[GraphAssembly]]
|
||||
* @param outOwnerId Identifier of the owner of the output side of the connection.
|
||||
* @param outOwner The stage logic that corresponds to the output side of the connection.
|
||||
* @param inHandler The handler that contains the callback for input events.
|
||||
* @param outHandler The handler that contains the callback for output events.
|
||||
|
|
@ -85,7 +83,9 @@ object GraphInterpreter {
|
|||
var portState: Int = InReady
|
||||
var slot: Any = Empty
|
||||
|
||||
override def toString = s"Connection($id, $portState, $slot, $inHandler, $outHandler)"
|
||||
override def toString =
|
||||
if (GraphInterpreter.Debug) s"Connection($id, $inOwnerId, $inOwner, $outOwnerId, $outOwner, $inHandler, $outHandler, $portState, $slot)"
|
||||
else s"Connection($id, $portState, $slot, $inHandler, $outHandler)"
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -1,158 +0,0 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
|
||||
*/
|
||||
package akka.stream.impl.fusing
|
||||
|
||||
import akka.event.NoLogging
|
||||
import akka.stream._
|
||||
import akka.stream.impl.fusing.GraphInterpreter.{ DownstreamBoundaryStageLogic, UpstreamBoundaryStageLogic }
|
||||
import akka.stream.stage._
|
||||
import java.{ util ⇒ ju }
|
||||
|
||||
/**
|
||||
* INTERNAL API
|
||||
*/
|
||||
private[akka] object IteratorInterpreter {
|
||||
|
||||
final case class IteratorUpstream[T](input: Iterator[T]) extends UpstreamBoundaryStageLogic[T] with OutHandler {
|
||||
val out: Outlet[T] = Outlet[T]("IteratorUpstream.out")
|
||||
out.id = 0
|
||||
|
||||
private var hasNext = input.hasNext
|
||||
|
||||
def onPull(): Unit = {
|
||||
if (!hasNext) complete(out)
|
||||
else {
|
||||
val elem = input.next()
|
||||
hasNext = input.hasNext
|
||||
if (!hasNext) {
|
||||
push(out, elem)
|
||||
complete(out)
|
||||
} else push(out, elem)
|
||||
}
|
||||
}
|
||||
|
||||
override def onDownstreamFinish(): Unit = ()
|
||||
|
||||
setHandler(out, this)
|
||||
|
||||
override def toString = "IteratorUpstream"
|
||||
}
|
||||
|
||||
final case class IteratorDownstream[T]() extends DownstreamBoundaryStageLogic[T] with Iterator[T] with InHandler {
|
||||
val in: Inlet[T] = Inlet[T]("IteratorDownstream.in")
|
||||
in.id = 0
|
||||
|
||||
private var done = false
|
||||
private var nextElem: T = _
|
||||
private var needsPull = true
|
||||
private var lastFailure: Throwable = null
|
||||
|
||||
def onPush(): Unit = {
|
||||
nextElem = grab(in)
|
||||
needsPull = false
|
||||
}
|
||||
|
||||
override def onUpstreamFinish(): Unit = {
|
||||
done = true
|
||||
}
|
||||
|
||||
override def onUpstreamFailure(cause: Throwable): Unit = {
|
||||
done = true
|
||||
lastFailure = cause
|
||||
}
|
||||
|
||||
setHandler(in, this)
|
||||
|
||||
private def pullIfNeeded(): Unit = {
|
||||
if (needsPull) {
|
||||
pull(in)
|
||||
interpreter.execute(Int.MaxValue)
|
||||
}
|
||||
}
|
||||
|
||||
override def hasNext: Boolean = {
|
||||
if (!done) pullIfNeeded()
|
||||
!(done && needsPull) || (lastFailure ne null)
|
||||
}
|
||||
|
||||
override def next(): T = {
|
||||
if (lastFailure ne null) {
|
||||
val e = lastFailure
|
||||
lastFailure = null
|
||||
throw e
|
||||
} else if (!hasNext)
|
||||
Iterator.empty.next()
|
||||
else {
|
||||
needsPull = true
|
||||
nextElem
|
||||
}
|
||||
}
|
||||
|
||||
// don't let toString consume the iterator
|
||||
override def toString: String = "IteratorDownstream"
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* INTERNAL API
|
||||
*/
|
||||
private[akka] class IteratorInterpreter[I, O](
|
||||
val input: Iterator[I],
|
||||
val stages: Seq[GraphStageWithMaterializedValue[FlowShape[_, _], Any]]) {
|
||||
|
||||
import akka.stream.impl.fusing.IteratorInterpreter._
|
||||
|
||||
private val upstream = IteratorUpstream(input)
|
||||
private val downstream = IteratorDownstream[O]()
|
||||
|
||||
private def init(): Unit = {
|
||||
|
||||
var i = 0
|
||||
val length = stages.length
|
||||
val attributes = Array.fill[Attributes](length)(Attributes.none)
|
||||
val ins = Array.ofDim[Inlet[_]](length + 1)
|
||||
val inOwners = Array.ofDim[Int](length + 1)
|
||||
val outs = Array.ofDim[Outlet[_]](length + 1)
|
||||
val outOwners = Array.ofDim[Int](length + 1)
|
||||
val stagesArray = Array.ofDim[GraphStageWithMaterializedValue[Shape, Any]](length)
|
||||
|
||||
ins(length) = null
|
||||
inOwners(length) = length
|
||||
outs(0) = null
|
||||
outOwners(0) = 0
|
||||
|
||||
val stagesIterator = stages.iterator
|
||||
while (stagesIterator.hasNext) {
|
||||
val stage = stagesIterator.next()
|
||||
stagesArray(i) = stage
|
||||
ins(i) = stage.shape.in
|
||||
inOwners(i) = i
|
||||
outs(i + 1) = stage.shape.out
|
||||
outOwners(i + 1) = i
|
||||
i += 1
|
||||
}
|
||||
|
||||
// TODO: Fix this (assembly is gone)
|
||||
// val assembly = new GraphAssembly(stagesArray, attributes, ins, inOwners, outs, outOwners)
|
||||
//
|
||||
// val (connections, logics)
|
||||
//
|
||||
// val interpreter = new GraphInterpreter(
|
||||
// assembly,
|
||||
// NoMaterializer,
|
||||
// NoLogging,
|
||||
// logics,
|
||||
// connections,
|
||||
// (_, _, _) ⇒ throw new UnsupportedOperationException("IteratorInterpreter does not support asynchronous events."),
|
||||
// fuzzingMode = false,
|
||||
// null)
|
||||
// interpreter.attachUpstreamBoundary(connections(0), upstream)
|
||||
// interpreter.attachDownstreamBoundary(connections(length), downstream)
|
||||
// interpreter.init(null)
|
||||
}
|
||||
|
||||
init()
|
||||
|
||||
def iterator: Iterator[O] = downstream
|
||||
}
|
||||
|
|
@ -66,6 +66,7 @@ object GraphStageLogic {
|
|||
*/
|
||||
object EagerTerminateInput extends InHandler {
|
||||
override def onPush(): Unit = ()
|
||||
override def toString = "EagerTerminateInput"
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -75,6 +76,7 @@ object GraphStageLogic {
|
|||
object IgnoreTerminateInput extends InHandler {
|
||||
override def onPush(): Unit = ()
|
||||
override def onUpstreamFinish(): Unit = ()
|
||||
override def toString = "IgnoreTerminateInput"
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -102,6 +104,7 @@ object GraphStageLogic {
|
|||
*/
|
||||
object EagerTerminateOutput extends OutHandler {
|
||||
override def onPull(): Unit = ()
|
||||
override def toString = "EagerTerminateOutput"
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -110,6 +113,7 @@ object GraphStageLogic {
|
|||
object IgnoreTerminateOutput extends OutHandler {
|
||||
override def onPull(): Unit = ()
|
||||
override def onDownstreamFinish(): Unit = ()
|
||||
override def toString = "IgnoreTerminateOutput"
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -218,6 +222,9 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
|
|||
|
||||
/**
|
||||
* INTERNAL API
|
||||
*
|
||||
* Input handlers followed by output handlers, use `inHandler(id)` and `outHandler(id)` to access the respective
|
||||
* handlers.
|
||||
*/
|
||||
private[stream] var attributes: Attributes = Attributes.none
|
||||
|
||||
|
|
@ -227,6 +234,21 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
|
|||
// Using common array to reduce overhead for small port counts
|
||||
private[stream] val handlers = Array.ofDim[Any](inCount + outCount)
|
||||
|
||||
/**
|
||||
* INTERNAL API
|
||||
*/
|
||||
private[stream] def inHandler(id: Int): InHandler = {
|
||||
if (id > inCount) throw new IllegalArgumentException(s"$id not in inHandler range $inCount in $this")
|
||||
if (inCount < 1) throw new IllegalArgumentException(s"Tried to access inHandler $id but there are no in ports in $this")
|
||||
handlers(id).asInstanceOf[InHandler]
|
||||
}
|
||||
|
||||
private[stream] def outHandler(id: Int): OutHandler = {
|
||||
if (id > outCount) throw new IllegalArgumentException(s"$id not in outHandler range $outCount in $this")
|
||||
if (outCount < 1) throw new IllegalArgumentException(s"Tried to access outHandler $id but there are no out ports $this")
|
||||
handlers(inCount + id).asInstanceOf[OutHandler]
|
||||
}
|
||||
|
||||
/**
|
||||
* INTERNAL API
|
||||
*/
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue