Bugfix in VirtualProcessor #24722

johanandren · 2018-03-13T15:47:31Z

johanandren · 2018-03-13T15:48:02Z

akka-stream-tests-tck/src/test/scala/akka/stream/tck/Reproducer.scala

+      this.required_spec109_mustIssueOnSubscribeForNonNullSubscriber()
+      i += 1
+    }
+  }


I'll remove this, but just to show that it passes the PR-validation (I got this to fail even locally before fix)

Ok please remove then :)

johanandren · 2018-03-13T15:50:00Z

akka-stream/src/main/scala/akka/stream/impl/StreamLayout.scala

          obj match {
            case subscription: Subscription ⇒
-              if (compareAndSet(subscriber, Both.create(subscriber))) establishSubscription(subscriber, subscription)
+              val establishing = Establishing.create(subscriber)
+              if (compareAndSet(subscriber, Both.create(subscriber))) establishSubscription(establishing, subscription)


What would happen before is that the state was set to Both here, but before establishSubscription would get to call subscriber.onSubscribe onComplete would execute and complete the sub, which would then get an onSubscribe call after being completed (that's the spec violation bug).

I fixed this by introducing an intermediate state Establishing which isn't changed to Both until after the onSubscribe call.

johanandren · 2018-03-13T15:51:56Z

akka-stream/src/main/scala/akka/stream/impl/StreamLayout.scala

-        case _ ⇒ // spec violation or cancellation race, but nothing we can do
+        case Establishing(s) ⇒
+          // keep trying until subscription established and can complete it
+          if (VirtualProcessor.Debug) println(s"VirtualPublisher#$hashCode($s).onError(${t.getMessage}), loop")


Seems I missed the recursion waiting for establish to complete here, but I'm not sure it should the done or my comment is wrong.

johanandren · 2018-03-13T15:52:25Z

akka-stream/src/main/scala/akka/stream/impl/StreamLayout.scala

+      case Establishing(s) ⇒
+        // keep trying until subscription established and can complete it
+        if (VirtualProcessor.Debug) println(s"VirtualPublisher#$hashCode($s).onComplete, loop")
+        onComplete()


So here we hot-spin to guarantee the onComplete happens after the subscription has happened.

are we sure that will complete?
is there a risk of live lock (starvation) if several threads ends up spinning here and there is no thread remaining to complete the subscription?

The only situation it wouldn't complete is if the establishing subscription never completes nor fails, I'm not sure what scenario that would be (or what we could possibly do about it).

I don't know the RS spec good enough to answer the part about several threads. A can several threads call try to complete the same subscriber? @ktoso ?

I think it's dangerous to spin and depend on that another thread unlocks the spinning. See for example this discussion https://groups.google.com/d/msg/akka-user/dRxTX8WmWlA/ku-nM0KcHJ8J and #17216 (comment)

Thanks for digging that up, had no idea.

I think the solution would be to buffer onComplete, if establishing, rather than spinning, and then having establish trigger the complete if needed when done.

It would be illegal to not complete establishing a sub, but it could happen; good to buffer onComplete instead

johanandren · 2018-03-13T15:53:18Z

build.sbt

@@ -344,7 +344,7 @@ lazy val streamTestkit = akkaModule("akka-stream-testkit")

 lazy val streamTests = akkaModule("akka-stream-tests")
  .dependsOn(streamTestkit % "test->test", remote % "test->test", stream)
-  .settings(Dependencies.streamTests)
+  .settings(Dependencies.streamTests ++ Dependencies.streamTestsTck)


leftover, I'll remove

akka-ci · 2018-03-13T16:05:39Z

Test FAILed.

This reverts commit ae5e41c.

akka-ci · 2018-03-13T17:04:07Z

Test FAILed.

johanandren · 2018-03-14T08:22:30Z

MiMa failure (added new internal classes)

akka-ci · 2018-03-16T17:19:34Z

Test PASSed.

ktoso · 2018-03-19T07:14:52Z

resolved conflict after Jame's fix in onError handler here; Stared long at impl and looks correct indeed

The resolution 85dc427

ktoso

LGTM, please see conflict resolution and merge

akka-ci · 2018-03-19T07:23:50Z

Test FAILed.

akka-ci · 2018-03-19T08:19:23Z

Test PASSed.

jroper · 2018-07-05T07:51:28Z

akka-stream/src/main/scala/akka/stream/impl/StreamLayout.scala

-        case _ ⇒ // spec violation or cancellation race, but nothing we can do
+        case est @ Establishing(_, false, OptionVal.None) ⇒
+          if (VirtualProcessor.Debug) println(s"VirtualPublisher#$hashCode($est).onError(${t.getMessage}), loop")
+          if (!compareAndSet(est, est.copy(onErrorBuffered = OptionVal.Some(ex)))) rec(ex)


Spec violation bug here - if the error is successfully buffered, and the exception was null, then this method doesn't throw an NPE, which it's required to do.

…nvoked This fix is similar to akka#24749, fixing a spec violation bug that was introduced in akka#24722.

…nvoked (#25311) * Ensure NPE is always through when VirtualProcessor.onError(null) is invoked This fix is similar to #24749, fixing a spec violation bug that was introduced in #24722.

* [PORT[ Bugfix in VirtualProcessor (akka/akka#24722) * Update API Verify list * Fix errors reported by TCK * Remove all recursions

johanandren commented Mar 13, 2018

View reviewed changes

akka-ci added the validating PR is currently being validated by Jenkins label Mar 13, 2018

johanandren commented Mar 13, 2018

View reviewed changes

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

johanandren added 11 commits March 13, 2018 17:08

Single-runnable repeater (hopefully)

484dc85

Repeat like no-one is watching

743f113

Fix plus noisy printlns

c3943b9

Toggleable debug logging for VirtualProcessor

c0626a2

Updated docs graph

c96dc4e

Re-enable debug logging to pinpoint another potential bug

c21bae5

Covered some more cases

9f2713f

Revert "Repeat like no-one is watching"

0adfbe2

This reverts commit ae5e41c.

Removed duplicate repeater

de457ed

Fixed some minor things

08c6b6b

copyright header

0bb6779

johanandren force-pushed the wip-24581-repeater-johanandren branch from 956b756 to 0bb6779 Compare March 13, 2018 16:13

MiMa filter

a28efac

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 Mar 14, 2018

akka-ci removed the validating PR is currently being validated by Jenkins label Mar 16, 2018

Merge branch 'master' into wip-24581-repeater-johanandren

85dc427

ktoso approved these changes Mar 19, 2018

View reviewed changes

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 validating PR is currently being validated by Jenkins labels Mar 19, 2018

Update StreamLayout.scala

1474a0e

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 Mar 19, 2018

ktoso merged commit 770b3a3 into akka:master Mar 19, 2018

johanandren deleted the wip-24581-repeater-johanandren branch March 26, 2018 10:18

jroper reviewed Jul 5, 2018

View reviewed changes

jroper mentioned this pull request Jul 5, 2018

Small TCK fix in cancelled sink #24749

Merged

jroper added a commit to jroper/akka that referenced this pull request Jul 5, 2018

Ensure NPE is always through when VirtualProcessor.onError(null) is i…

10cccc9

…nvoked This fix is similar to akka#24749, fixing a spec violation bug that was introduced in akka#24722.

jroper mentioned this pull request Jul 5, 2018

Ensure NPE is always through when VirtualProcessor.onError(null) is invoked #25311

Merged

jroper added a commit to jroper/akka that referenced this pull request Jul 5, 2018

Ensure NPE is always through when VirtualProcessor.onError(null) is i…

ca00286

…nvoked This fix is similar to akka#24749, fixing a spec violation bug that was introduced in akka#24722.

jroper added a commit to jroper/akka that referenced this pull request Jul 5, 2018

Ensure NPE is always through when VirtualProcessor.onError(null) is i…

eb0bcbe

…nvoked This fix is similar to akka#24749, fixing a spec violation bug that was introduced in akka#24722.

jroper added a commit to jroper/akka that referenced this pull request Jul 10, 2018

Ensure NPE is always through when VirtualProcessor.onError(null) is i…

36eee09

…nvoked This fix is similar to akka#24749, fixing a spec violation bug that was introduced in akka#24722.

johanandren mentioned this pull request Jul 11, 2018

Ensure NPE is always through when VirtualProcessor.onError(null) is invoked #25339

Closed

Arkatufus added a commit to Arkatufus/akka.net that referenced this pull request May 18, 2022

[PORT[ Bugfix in VirtualProcessor (akka/akka#24722)

1b1158c

Arkatufus mentioned this pull request May 18, 2022

[PORT] Akka/Akka#24722: Bugfix in VirtualProcessor akkadotnet/akka.net#5950

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Bugfix in VirtualProcessor #24722

Bugfix in VirtualProcessor #24722

johanandren commented Mar 13, 2018

johanandren Mar 13, 2018

ktoso Mar 15, 2018

johanandren Mar 13, 2018 •

edited

johanandren Mar 13, 2018

johanandren Mar 13, 2018

patriknw Mar 15, 2018

johanandren Mar 15, 2018

patriknw Mar 15, 2018

johanandren Mar 15, 2018

ktoso Mar 19, 2018

johanandren Mar 13, 2018

akka-ci commented Mar 13, 2018

akka-ci commented Mar 13, 2018

johanandren commented Mar 14, 2018 •

edited

akka-ci commented Mar 16, 2018

ktoso commented Mar 19, 2018 •

edited

ktoso left a comment

akka-ci commented Mar 19, 2018

akka-ci commented Mar 19, 2018

jroper Jul 5, 2018

Bugfix in VirtualProcessor #24722

Bugfix in VirtualProcessor #24722

Conversation

johanandren commented Mar 13, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

johanandren Mar 13, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

akka-ci commented Mar 13, 2018

akka-ci commented Mar 13, 2018

johanandren commented Mar 14, 2018 • edited

akka-ci commented Mar 16, 2018

ktoso commented Mar 19, 2018 • edited

ktoso left a comment

Choose a reason for hiding this comment

akka-ci commented Mar 19, 2018

akka-ci commented Mar 19, 2018

Choose a reason for hiding this comment

johanandren Mar 13, 2018 •

edited

johanandren commented Mar 14, 2018 •

edited

ktoso commented Mar 19, 2018 •

edited