Cannot push port (Balance.out8) twice

[jarandaf]

I'm getting this error using Akka Streams 2.4.8. I am not sure where it comes from, the stacktrace does not tell much:

[ERROR] [07/12/2016 14:02:27.093] [default-akka.actor.default-dispatcher-248] [akka://default/user/StreamSupervisor-18/flow-18-21-unknown-operation] Error in stage [Balance]: requirement failed: Cannot push port (Balance.out8) twice
java.lang.IllegalArgumentException: requirement failed: Cannot push port (Balance.out8) twice
        at scala.Predef$.require(Predef.scala:219)
        at akka.stream.stage.GraphStageLogic.push(GraphStage.scala:436)
        at akka.stream.scaladsl.Balance$$anon$17.akka$stream$scaladsl$Balance$$anon$$dequeueAndDispatch(Graph.scala:613)
        at akka.stream.scaladsl.Balance$$anon$17$$anon$18.onPush(Graph.scala:618)
        at akka.stream.impl.fusing.GraphInterpreter.processElement$1(GraphInterpreter.scala:590)
        at akka.stream.impl.fusing.GraphInterpreter.processEvent(GraphInterpreter.scala:601)
        at akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:542)
        at akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:471)
        at akka.stream.impl.fusing.GraphInterpreterShell.receive(ActorGraphInterpreter.scala:410)
        at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$processEvent(ActorGraphInterpreter.scala:603)
        at akka.stream.impl.fusing.ActorGraphInterpreter$$anonfun$receive$1.applyOrElse(ActorGraphInterpreter.scala:618)
        at akka.actor.Actor$class.aroundReceive(Actor.scala:484)
        at akka.stream.impl.fusing.ActorGraphInterpreter.aroundReceive(ActorGraphInterpreter.scala:529)
        at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526)
        at akka.actor.ActorCell.invoke(ActorCell.scala:495)
        at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)
        at akka.dispatch.Mailbox.run(Mailbox.scala:224)
        at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
        at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
        at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
        at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
        at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

I am using a balancer to distribute lines processing from a file source.

[ktoso]

Thanks for reporting, it could be a bug in Balance itself.
You mean the cookbook itself is exposing this issue?

[jarandaf]

Not really, the balancer works fine in some other scenarios. I am using dropWhile in combination with Sink.headOption to stop the stream on some input data condition, something like the following:

val path = Paths.get(s"/tmp/foo.txt")
val sink = Sink.headOption[Boolean]

val g = RunnableGraph fromGraph GraphDSL.create(sink) { implicit b => sink =>
  val source = FileIO.fromPath(path)
    .via(Framing.delimiter(ByteString(System.lineSeparator), maximumFrameLength=8192, allowTruncation=true))
    .map(_.utf8String)
  // Generates a 'JobResult'
  val compute = Flow.fromFunction(Producer produce).withAttributes(ActorAttributes dispatcher "compute-dispatcher")
  // Using same example balancer from cookbook
  val balanced = b add balancer(compute, numWorkers).async
  // Publishes boolean values
  val publish = b add Flow[JobResult].mapAsyncUnordered(1024) { Publisher.publish(_, system) }
        .dropWhile(_.result.result == true)
        .map(_.result.result)

  source ~> balanced ~> publish ~> sink
  ClosedShape
}

So basically I want to check if there is some file line which does not match some condition. Since I expect this not to happen very often, I am using dropWhile. It should be very common not to find any line matching this condition, that is why I am using headOption. Changing the sink to Sink.seq[T] does not throw any error but I don't want neither to wait to process the whole file nor to buffer results.

[drewhk]

This is almost surely a bug. Thanks for the reproducer, I will try to reproduce the issue. We have a nice compile-time flag that allows me to look at the actual events that happen, maybe that will shed some light on what is happening.

[johanandren]

@jarandaf there is some details missing in the reproducer, how are you creating your Balance instance (how many outputs, and waitForAllDownstreams true or false)?

[jarandaf]

Hi @johanandren, this is the balancer I used:

// Deals with balancing workload over the given number of workers.
private def balancer[In, Out](worker: Flow[In, Out, Any], workerCount: Int): Flow[In, Out, NotUsed] = {
  Flow fromGraph GraphDSL.create() { implicit b =>
    val balancer = b add Balance[In](workerCount, waitForAllDownstreams = false)
    val merge    = b add Merge[Out](workerCount)

    1 to workerCount foreach { _ => balancer ~> worker.async ~> merge }

    FlowShape(balancer.in, merge.out)
  }
}

As per number of outputs, I did use the number of cores available in my box.

Hope it helps!

[drewhk]

Ah you took it @johanandren . Then I look for something else.

[johanandren]

You know me @drewhk, I just love these pull-push-twice tickets. ;)

[drewhk]

It builds character, so they say.

[johanandren]

Figured it out:

• one of the balance downstreams pulls
• but then cancels before getting a push
• the cancelled out is enqueued, have triggered an upstream pull and cannot be pushed when the element arrives (so it is actually pulling a closed port, order of assertions in GraphStage makes it log as push-twice)

If we filter the closed out when the upstream element arrives, there may not be any out to receive the pulled element, so we'd need to buffer it. But then the rest of the outs may close before ever pulling and the element is lost. Not sure what to do here.

[drewhk]

If we filter the closed out when the upstream element arrives, there may not be any out to receive the pulled element, so we'd need to buffer it.

No need to do an extra buffer, just don't grab it from the port.

But then the rest of the outs may close before ever pulling and the element is lost.

What do you mean by lost? If all the outputs have been closed then there is no "loss". I mean, we could have sent just before the "cancel" arrived and it will be "lost", too.

[johanandren]

Thanks @drewhk, went down a rabbit hole with my first try solving it. Your comment made it obvious what to do.