AsyncCallbacks lost on finished stage by error by agolubev · Pull Request #23185 · akka/akka-core

agolubev · 2017-06-17T20:29:01Z

Ref #23111
Please review approach of adding all incoming requests to queue with acknowledge future.
In case the stream is finished but some futures stay not completed - we are failing them automatically.
There is a plan to substitute CallBackWapper with ConcurrentGraphStage everywhere (perhaps need a new name to explain that stage allows making callbacks not taking into account stream lifecycle.)

Let me know what do you think

There still some sync points in new GraphStage this is because we need to synchronize ideal Akka world with the not ideal external world. Initially, it was considered to introduce special actor with extended lifecycle to avoid sync points - but this was rejected.
Any ideas as to avoid synchronization are welcome.

akka-ci · 2017-06-17T21:12:59Z

Test FAILed.

akka-ci · 2017-06-17T22:15:08Z

Test FAILed.

akka-ci · 2017-06-17T22:25:36Z

Test FAILed.

agolubev · 2017-06-18T10:26:28Z

build failed due to binary incompatible - not sure how to fix this so far

akka-ci · 2017-06-22T19:30:23Z

Test PASSed.

agolubev · 2017-06-27T20:49:39Z

Ref #23078

patriknw

hmm, I need to think some more about this. So far only detailed comments without understanding if there is a better solution.

patriknw · 2017-06-28T11:00:21Z

+          Source.queue[Unit](10, OverflowStrategy.fail).toMat(TestSink.probe)(Keep.both).run()
+        intercept[IllegalStateException] {
+          Await.result(
+            (1 to 15).map(_ ⇒ queue.offer(())).last, 3.seconds)


is this guaranteed to throw?
I think you have to use a TestSink that is not requesting elements to be sure?

actually, I use TestSink.probe in line 297. So I believe this is what you are expecting.

patriknw · 2017-06-28T11:01:28Z

-  val out = Outlet[T]("queueSource.out")
+  val out = Outlet[T]("queueSource.offerout")
  override val shape: SourceShape[T] = SourceShape.of(out)
+  println(QueueSource)


remember to remove

patriknw · 2017-06-28T11:04:34Z

+abstract class ConcurrentGraphStageLogic[T] private[stream] (override val inCount: Int, override val outCount: Int) extends GraphStageLogic(inCount, outCount) {
+  def this(shape: Shape) = this(shape.inlets.size, shape.outlets.size)
+
+  val concurrentSet = ConcurrentHashMap.newKeySet[Promise[_]]()


patriknw · 2017-06-28T11:11:24Z

+   */
+  private[this] final val lock = new ReentrantLock
+
+  private[this] val callbackState = new AtomicReference[CallbackState](NotInitialized(Nil))


This is only accessed inside the lock so then there shouldn't be any need for an AtomicReference.

patriknw · 2017-06-28T11:16:50Z

+
+    ac.ack.map(ack ⇒ {
+      ack.future.andThen { case _ ⇒ concurrentSet.remove(ack) }(ec)
+      concurrentSet.add(ack)


place this above previous line, so that things are not optimized and remove happens before add

patriknw · 2017-06-28T11:18:47Z

+  }
+
+  @throws(classOf[Exception])
+  override def postStop(): Unit = {


easy to forget calling super.postStop from subclass

patriknw · 2017-06-28T11:30:53Z


-      override def postStop(): Unit = stopCallback {
+      override def postStop(): Unit = stopWithStoppedCallbackLogic {
        case Pull(promise) ⇒ promise.failure(new IllegalStateException("Stream is terminated. QueueSink is detached"))


need for tryFailure?

patriknw · 2017-06-28T11:33:03Z

+   * To preserve message order when switching between not initialized / initialized states
+   * lock is used. Case is similar to RepointableActorRef
+   */
+  private[this] final val lock = new ReentrantLock


I'd prefer synchronized if we don't use any specific features from ReentrantLock

patriknw · 2017-06-28T11:40:41Z

+
+  @throws(classOf[Exception])
+  override def postStop(): Unit = {
+    concurrentSet.asScala.foreach(_.failure(new IllegalStateException("Stream is stopped. Callback is detached.")))


need tryFailure?

patriknw · 2017-06-28T12:07:23Z

Looking at how ordinary AsyncCallback is executed

onAsyncInput in GraphInterpreter (ActorGraphInterpreter)
new AsyncInput
runAsyncInput

runAsyncInput seems to be the place where the event (callback parameter) is applied to the handler, and that is skipped if isStageCompleted.

Could we introduce a

trait HasCallbackPromise[T] {
  def done: Promise[T]
}

and in runAsyncInput fail that promise if isStageCompleted and the event implements that trait ?

What am I missing?

agolubev · 2017-06-30T20:19:56Z

It's a little bit more complex. Source.queue returns futures for each element. Future succeeded when an element is pulling by a stream. The issue here is that some data elements already went through runAsyncInput and waiting to be processed by the actor. At this particular time actor finishes because of stream completion. In this situation, some Futures will not be completed at all. This is what I'm trying to solve here.
So we need some buffer of such "in transit" elements. Callback that will remove element once it consumed by actor and some graceful completion that fails all "in transit" futures.

agolubev · 2017-06-30T20:25:55Z

Still I can try "HasCallbackPromise as trait" approach instead of "AckedCallback abstract class" it you think this will give some benefits.

patriknw · 2017-07-02T08:18:42Z

Isn't another issue that it's not safe to return a callback as materialized value, because it can be inoked "too early". That is what the calback wrapper is trying to solve.

agolubev · 2017-07-02T13:04:08Z

@patriknw , callback wrapper solves the issue of invoking materialized value too early - current PR is enhancing wrapper by adding tracking of what messages being processed by stream to complete all futures

patriknw · 2017-07-02T17:24:17Z

Right, Optimal solution would be to have callbacks that "just works", and not a special one for when you use them as materialized value. This is difficult and it's great that we start thinking about how to solve this.

agolubev · 2017-10-31T04:22:01Z

rebased and seems like added all feedback changes.

akka-ci · 2017-10-31T05:03:31Z

Test PASSed.
5779 tests run, 458 skipped, 0 failed.
Test PASSed.

akka-ci · 2017-10-31T05:03:32Z

Test PASSed.

patriknw

LGTM, with 2 small things

patriknw · 2017-10-31T10:00:41Z

+ProblemFilters.exclude[MissingClassProblem]("akka.stream.stage.CallbackWrapper$Initialized$")
+ProblemFilters.exclude[MissingClassProblem]("akka.stream.stage.CallbackWrapper$NotInitialized$")
+ProblemFilters.exclude[MissingClassProblem]("akka.stream.stage.CallbackWrapper$CallbackState")
+ProblemFilters.exclude[MissingClassProblem]("akka.stream.stage.CallbackWrapper")


The file should now be for 2.5.6, and something was added there. Should this be removed?

patriknw · 2017-10-31T10:04:18Z

+        // not started yet
+        case list @ Pending(l) ⇒ if (!currentState.compareAndSet(list, Pending(Event(event, OptionVal.None) :: l))) internalInvoke(event)
+        // started - can just send message to stream
+        case Initialized       ⇒ onAsyncInput(event)


move this case to the top

agolubev · 2017-10-31T14:26:39Z

pushed changes per recent feedback

akka-ci · 2017-10-31T15:08:32Z

Test PASSed.
5780 tests run, 458 skipped, 0 failed.
Test PASSed.

akka-ci · 2017-10-31T15:08:33Z

Test PASSed.

johanandren

LGTM, excellent work @agolubev !

…golubev

patriknw · 2017-11-09T13:49:46Z

Yay, great work @agolubev
I think this unlocks several other tickets.

jrudolph

Seems I missed the train on reviewing soon enough...

It seems that I seem to be missing something. Where are the promises ever completed? I think we also need tests for the new features. A few smaller comments below.

jrudolph · 2017-11-09T13:52:23Z

+  /**
+   * Dispatch an asynchronous notification.
+   * This method is thread-safe and may be invoked from external execution contexts.
+   * Promise in `HasCallbackPromise` will fail if stream is already closed or closed before


Seems, the description is based on an outdated version that still had HasCallbackPromise.

jrudolph · 2017-11-09T13:54:29Z

+      (currentState.getAndSet(Initializing): @unchecked) match {
+        case Pending(l) ⇒ l.reverse.foreach(ack ⇒ {
+          onAsyncInput(ack.e)
+        })


Instead of adding @unchecked, we usually add a catch-all clause that throws an IllegalStageException to make it explicit that missing a state was not an oversight.

jrudolph · 2017-11-09T13:56:47Z

-        interpreter.onAsyncInput(GraphStageLogic.this, event, handler.asInstanceOf[Any ⇒ Unit])
+    val result = new ConcurrentAsyncCallback[T](handler)
+    asyncCallbacksInProgress.add(result)
+    if (_interpreter != null) result.onStart()


Is it prevented (or does it need to be prevented) that onStart is called twice, once here, and once below in beforePreStart? Is it the assumption, that if _interpreter != null we currently are executing preStart? But wouldn't then the onStart triggered in beforePreStart already be called?

jrudolph · 2017-11-09T14:02:26Z

+        case list @ Pending(l) ⇒ if (!currentState.compareAndSet(list, Pending(Event(event, OptionVal.None) :: l))) internalInvoke(event)
+        // initializing is in progress in another thread (initializing thread is managed by akka)
+        case Initializing ⇒ if (!currentState.compareAndSet(Initializing, Pending(Event(event, OptionVal.None) :: Nil))) {
+          (currentState.get(): @unchecked) match {


What's the benefit of not just recursing directly? Isn't it now missing matching for Completed?

jrudolph · 2017-11-09T14:04:02Z

+        // started - can just send message to stream
+        case Initialized ⇒ sendEvent(event, promise)
+        // initializing is in progress in another thread (initializing thread is managed by Akka)
+        case Initializing ⇒ if (!currentState.compareAndSet(Initializing, Pending(Event(event, OptionVal(promise)) :: Nil))) {


See below, why not recurse directly?

jrudolph · 2017-11-09T14:09:32Z

+
+    // external call
+    override def invokeWithFeedback(event: T): Future[Done] = {
+      val promise: Promise[Done] = Promise[Done]()


Where is the promise ever completed? Isn't that missing?

* Add reproduction unit test * Port akka/akka-core#23970 * Update API Approval list * Update API Approval list * Add async callback unit tests * skip non-implemented feature * Fix TaskCompletionSource allocation problem * Code cleanup * Implement early callback from akka/akka-core#23185 * Make sure early callbacks also get cancelled on stop * Fix naming and copyright header * Update API Approval list * Fix missing feedback callback * rerun tests * Implement locks * cleanup lock code * extend locking to `_callbackWaitingForInterpreter` * Fix unit test * rerun tests * xml-doc and TBD cleanup * more TBD and XML-DOC cleanup * defined `ConcurrentAsyncCallback` * introduced `ConcurrentAsyncCallback` into `GraphStage` * fixed all compilation errors * cleanup hub code * added API approvals * cleaned up reverse --------- Co-authored-by: Gregorius Soedharmo <arkatufus@yahoo.com>

akka-ci added validating PR is currently being validated by Jenkins needs-attention Indicates a PR validation failure (set by CI infrastructure) and removed validating PR is currently being validated by Jenkins labels Jun 17, 2017

akka-ci added the needs-attention Indicates a PR validation failure (set by CI infrastructure) label Jun 17, 2017

akka-ci added validating PR is currently being validated by Jenkins tested PR that was successfully built and tested by Jenkins and removed needs-attention Indicates a PR validation failure (set by CI infrastructure) validating PR is currently being validated by Jenkins labels Jun 22, 2017

patriknw reviewed Jun 28, 2017

View reviewed changes

agolubev force-pushed the 23111-AsyncCallbacks_lost_finished_stage-agolubev branch from ca8285c to 1c5fe18 Compare August 7, 2017 22:00

akka-ci added validating PR is currently being validated by Jenkins needs-attention Indicates a PR validation failure (set by CI infrastructure) and removed tested PR that was successfully built and tested by Jenkins labels Aug 7, 2017

akka-ci added validating PR is currently being validated by Jenkins and removed needs-attention Indicates a PR validation failure (set by CI infrastructure) labels Oct 31, 2017

akka-ci added tested PR that was successfully built and tested by Jenkins and removed validating PR is currently being validated by Jenkins labels Oct 31, 2017

patriknw approved these changes Oct 31, 2017

View reviewed changes

+str AsyncCallbacks lost on finished stage by error

9b43ce7

agolubev force-pushed the 23111-AsyncCallbacks_lost_finished_stage-agolubev branch from e672e45 to 9b43ce7 Compare October 31, 2017 14:25

akka-ci added validating PR is currently being validated by Jenkins and removed tested PR that was successfully built and tested by Jenkins labels Oct 31, 2017

akka-ci added tested PR that was successfully built and tested by Jenkins and removed validating PR is currently being validated by Jenkins labels Oct 31, 2017

johanandren approved these changes Nov 1, 2017

View reviewed changes

Merge branch 'master' into 23111-AsyncCallbacks_lost_finished_stage-a…

26f0f2c

…golubev

patriknw merged commit 3c7d40b into akka:master Nov 9, 2017

jrudolph reviewed Nov 9, 2017

View reviewed changes

jrudolph mentioned this pull request Nov 9, 2017

Follow up work for AsyncCallback.invokeWithFeedback #23953

Closed

4 tasks

johanandren mentioned this pull request Nov 13, 2017

Akka stream Source.queue hangs when using fail overflow strategy #23078

Closed

patriknw mentioned this pull request Nov 17, 2017

Remove usage of CallbackWrapper (Akka 2.5.7) akka/alpakka-kafka#368

Closed

Philippus mentioned this pull request Nov 21, 2017

Update Scala to 2.11.12 / 2.12.4 akka/alpakka-kafka#370

Merged

This was referenced Mar 24, 2025

Cancelled sinks are blocking other BroadcastHub consumers akkadotnet/akka.net#7578

Closed

Fix #7578, Cancelled sinks are blocking other BroadcastHub consumers akkadotnet/akka.net#7579

Closed

Arkatufus added a commit to Arkatufus/akka.net that referenced this pull request Mar 27, 2025

Implement early callback from akka/akka-core#23185

2a1ded1

Conversation

agolubev commented Jun 17, 2017

Uh oh!

akka-ci commented Jun 17, 2017

Uh oh!

akka-ci commented Jun 17, 2017

Uh oh!

akka-ci commented Jun 17, 2017

Uh oh!

agolubev commented Jun 18, 2017

Uh oh!

akka-ci commented Jun 22, 2017

Uh oh!

agolubev commented Jun 27, 2017

Uh oh!

patriknw left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

patriknw commented Jun 28, 2017

Uh oh!

agolubev commented Jun 30, 2017

Uh oh!

agolubev commented Jun 30, 2017

Uh oh!

patriknw commented Jul 2, 2017

Uh oh!

agolubev commented Jul 2, 2017

Uh oh!

patriknw commented Jul 2, 2017

Uh oh!

agolubev commented Oct 31, 2017

Uh oh!

akka-ci commented Oct 31, 2017

Uh oh!

akka-ci commented Oct 31, 2017

Uh oh!

patriknw left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

agolubev commented Oct 31, 2017

Uh oh!

akka-ci commented Oct 31, 2017

Uh oh!

akka-ci commented Oct 31, 2017

Uh oh!

johanandren left a comment

Choose a reason for hiding this comment

Uh oh!

patriknw commented Nov 9, 2017

Uh oh!

jrudolph left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment