+tck #284 support "demand when all downstreams demand" Processor in TCK #287

ktoso · 2015-07-12T20:58:20Z

Resolves #284 not being able to handle "processor that awaits both downstreams before pulling from upstream" in test case required_spec104_mustCallOnErrorOnAllItsSubscribersIfItEncountersANonRecoverableError as reported by @RuedigerMoeller.

The test-case required_mustRequestFromUpstreamForElementsThatHaveBeenRequestedLongAgo pointed out in that ticket will be addressed in a separate PR.

ktoso · 2015-07-12T20:58:40Z

Please review @reactive-streams/contributors @viktorklang :-)

ktoso · 2015-07-12T21:03:38Z

tck/src/main/java/org/reactivestreams/tck/IdentityProcessorVerification.java

          expectRequest();
          final T x = sendNextTFromUpstream();
          expectNextElement(sub1, x);
          sub1.request(1);

          // sub1 has received one element, and has one demand pending
-          // sub2 has not yet requested anything
+          // sub2 has received one element, and no more pending demand


The change is that:

we did not request from sub2, which made it impossible for "lazy, wait for both" processors to start requesting from upstream

now we request at least one element from each downstream, causing one element to be signalled down

we now check that the error signal overtakes the outstanding elements, which was the goal of the test

smaldini · 2015-07-14T13:12:30Z

very nice!

viktorklang · 2015-07-14T13:33:47Z

tck/src/test/java/org/reactivestreams/tck/IdentityProcessorVerificationTest.java

+
+            @Override
+            public void request(final long n) {
+              new Thread(new Runnable() {


Is it safe to create a new Thread for each request call? Is this the only solution?

Addressed by using the already present execution service here instead of new Thread

viktorklang · 2015-08-29T00:17:26Z

@ktoso I just noticed this one. Is this ready for merge? What do we need to proceed with it?

ktoso · 2015-08-29T00:18:28Z

Oh, and I seem to have missed your review comments in July hm.
I think it is an important fix, I'll address the comments and ping.

viktorklang · 2015-09-14T19:49:43Z

@ktoso let me know when the ping is coming :)

ktoso · 2015-09-14T19:50:57Z

Thanks for pinging me back, fell through the cracks due to vacation :) soon

viktorklang · 2015-09-14T19:53:23Z

No worries, just making sure it wasn't dropped on the floor :)

ktoso · 2015-09-15T23:08:35Z

@viktorklang addressed the comments, please have a look :)

viktorklang · 2015-09-16T13:10:25Z

tck/src/test/java/org/reactivestreams/tck/IdentityProcessorVerificationTest.java

+          @Override
+          public void subscribe(final Subscriber<? super Integer> s) {
+            int subscriberCount = subs.size();
+            if (subscriberCount == 0) s.onSubscribe(createSubscription(awaitLatch, s, demand1));


Stylistic comment but I think I'd prefer a switch here.

much more lines, I can convert if you'd prefer that :)

int subscriberCount = subs.size(); switch (subscriberCount) { case 0: s.onSubscribe(createSubscription(awaitLatch, s, demand1)); break; case 1: s.onSubscribe(createSubscription(awaitLatch, s, demand2)); break; default: throw new RuntimeException(String.format("This for-test-purposes-processor supports only 2 subscribers, yet got %s!", subscriberCount)); }

…" Processor in TCK

ktoso · 2015-09-16T23:57:44Z

Thanks for the last round of comments @viktorklang, addressed all of them (comments in line explain how)

ktoso · 2015-09-25T14:24:02Z

Now it's my turn to ping 😉 @viktorklang

viktorklang · 2015-09-25T15:02:06Z

Thanks @ktoso I'll have a look ASAP :)

ktoso · 2015-10-22T10:37:53Z

ping @viktorklang ;) merge here as well to celebrate the merge day? ;)

viktorklang · 2015-10-22T10:39:29Z

@ktoso Sorry, I've been swamped.

I'll have a look now.

ktoso · 2015-10-22T10:39:59Z

I know, no worries :)

viktorklang · 2015-10-22T10:45:07Z

tck/src/test/java/org/reactivestreams/tck/IdentityProcessorVerificationTest.java

+                @Override
+                public void run() {
+                  if (demand.get() >= 0) {
+                    demand.addAndGet(n);


There's a race-condition here between addAndGet and cancel

This may work but can overflow demand into a negative value.

viktorklang · 2015-10-22T10:45:53Z

@ktoso Reviewed. There's a (to me) scary-looking custom Processor implementation.

ktoso · 2015-10-22T10:47:28Z

Thanks a lot for reviewing – Ouch, so got more work to do here it seems - perfect for the flight tomorrow though :)

viktorklang · 2015-10-22T10:50:39Z

@ktoso No worries. Processor is the trickiest thing to implement.

akarnokd · 2015-10-22T10:57:40Z

tck/src/test/java/org/reactivestreams/tck/IdentityProcessorVerificationTest.java

+
+          @Override
+          public void subscribe(final Subscriber<? super Integer> s) {
+            int subscriberCount = subs.size();


This won't work if subscribe is called from different threads concurrently.

viktorklang · 2016-04-06T08:46:56Z

@ktoso What's the plan here?

ktoso · 2016-04-06T08:48:26Z

Need to get back to it finally (adding to todolist), day job consuming entire awake-time recently.

ktoso · 2017-06-17T06:24:29Z

This is a pretty ancient PR - is there enough interest such that I should attempt reviving it?
It is not needed for the next minor release I think

viktorklang · 2017-06-17T08:56:37Z

@ktoso Well, it's 2 years old so no hurry, for sure. Since we have no idea if/when there'll be a 1.0.2 depending on how much work it is to fix it, it might be better to do it now.

viktorklang · 2017-06-26T12:15:20Z

@ktoso Did we want to make this into 1.0.1?

ktoso · 2017-06-26T12:44:31Z

I don't think it's critical to include this.
I will however give it another look during my flight home today (pretty long one).

viktorklang · 2017-06-26T12:49:31Z

@ktoso Cool. Thanks. Let me know if there's anything else I should include in the #379

viktorklang · 2017-11-08T09:18:25Z

@ktoso Perhaps we can work on this together? I'd love to be able to either merge or close this issue, so it doesn't linger on. Let me know what you need from me.

ktoso · 2017-11-10T02:42:50Z

I completely don't remember what this was about right now hm... If you'd be able to check was was missing here please do, currently in a time critical work thing sadly and can't look into this PR.

viktorklang · 2017-11-10T13:33:05Z

I don't know what is missing, I'm. afraid, otherwise I'd take a look at it :/

On Fri, 10 Nov 2017 at 03:42, ktoso ***@***.***> wrote: I completely don't remember what this was about right now hm... If you'd be able to check was was missing here please do, currently in a time critical work thing sadly and can't look into this PR. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#287 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAAqd92R1v7fWPm3BLMVoFa4kheN9EuYks5s07grgaJpZM4FXDWt> .

-- Cheers, √

ktoso · 2017-11-22T14:50:05Z

Yeah I honestly also don't remember... would need some detective work ~:

akarnokd · 2017-11-22T14:58:19Z

One can implement a Processor in a way that given a set of Subscribers, when all have expressed the number of items each want, the Processor picks the smallest amount and requests that from the upstream Subscription. Basically it paces the item production from the upstream to the slowest consumer (lockstep).

The TCK, however, expects that requesting from a Processor known to have data ready should be fulfilled immediately regardless the other Subscribers' state. I.e., the test subscribes with multiple test Subscribers and requests on one of them, then checks if items have arrived or not. Because the other Subscribers had no chance to request, a Processor won't request items from its upstream and won't emit to the first test Subscriber, causing test failures.

viktorklang · 2017-11-23T12:47:46Z

@akarnokd Hmmm, I see… Do you have any proposal for how we could fix this limitation in the TCK?

akarnokd · 2017-11-23T13:53:17Z

I see two options:

Introduce a method IndentityProcessorVerification.doesCoordinatedEmission() (returns boolean, default false) and then split the test body based on its value.
Introduce peeking method(s) on the test subscriber to detect the lack of emission to the first subscriber without failing the test, then issue the request for the second subscriber and repeat the assertion on the first subscriber.

viktorklang · 2017-11-23T14:55:05Z

@akarnokd It would seem as the first option would be the least involved/error-prone one. Would you like to take a stab at that?

akarnokd · 2017-11-23T14:58:08Z

Sure.

viktorklang · 2017-11-23T14:59:54Z

@akarnokd Awesome

akarnokd · 2017-12-04T10:01:36Z

Close this via #414?

viktorklang · 2017-12-04T10:02:32Z

@akarnokd Yes, thanks! :)

ktoso force-pushed the wip-284-emit-once-all-request-multi-ktoso branch 3 times, most recently from 89a4dc3 to 55b1140 Compare July 12, 2015 21:01

ktoso reviewed Jul 12, 2015
View reviewed changes

ktoso mentioned this pull request Jul 12, 2015

Clarification on multiple subscribers needed .. #284

Closed

viktorklang reviewed Jul 14, 2015
View reviewed changes

ktoso force-pushed the wip-284-emit-once-all-request-multi-ktoso branch 2 times, most recently from bc13273 to 4d09161 Compare September 15, 2015 23:07

viktorklang reviewed Sep 16, 2015
View reviewed changes

+tck reactive-streams#284 support "demand when all downstreams demand…

7fbd565

…" Processor in TCK

ktoso force-pushed the wip-284-emit-once-all-request-multi-ktoso branch from 4d09161 to 7fbd565 Compare September 16, 2015 23:57

viktorklang reviewed Oct 22, 2015
View reviewed changes

akarnokd reviewed Oct 22, 2015
View reviewed changes

akarnokd mentioned this pull request Nov 23, 2017

Allow request-coordinating Processor implementations to pass the TCK #414

Merged

viktorklang closed this Dec 4, 2017

+tck #284 support "demand when all downstreams demand" Processor in TCK #287

+tck #284 support "demand when all downstreams demand" Processor in TCK #287

Conversation

ktoso commented Jul 12, 2015

ktoso commented Jul 12, 2015

Choose a reason for hiding this comment

smaldini commented Jul 14, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viktorklang commented Aug 29, 2015

ktoso commented Aug 29, 2015

viktorklang commented Sep 14, 2015

ktoso commented Sep 14, 2015

viktorklang commented Sep 14, 2015

ktoso commented Sep 15, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ktoso commented Sep 16, 2015

ktoso commented Sep 25, 2015

viktorklang commented Sep 25, 2015

ktoso commented Oct 22, 2015

viktorklang commented Oct 22, 2015

ktoso commented Oct 22, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viktorklang commented Oct 22, 2015

ktoso commented Oct 22, 2015

viktorklang commented Oct 22, 2015

Choose a reason for hiding this comment

viktorklang commented Apr 6, 2016

ktoso commented Apr 6, 2016

ktoso commented Jun 17, 2017 • edited Loading

viktorklang commented Jun 17, 2017

viktorklang commented Jun 26, 2017

ktoso commented Jun 26, 2017

viktorklang commented Jun 26, 2017

viktorklang commented Nov 8, 2017

ktoso commented Nov 10, 2017

viktorklang commented Nov 10, 2017 via email

ktoso commented Nov 22, 2017

akarnokd commented Nov 22, 2017

viktorklang commented Nov 23, 2017

akarnokd commented Nov 23, 2017

viktorklang commented Nov 23, 2017

akarnokd commented Nov 23, 2017

viktorklang commented Nov 23, 2017

akarnokd commented Dec 4, 2017

viktorklang commented Dec 4, 2017

ktoso commented Jun 17, 2017 •

edited

Loading