fix Amb backpressure bug #2961

davidmoten · 2015-05-19T05:48:55Z

The Amb operator has a backpressure bug:

if m requests were made before the first emission then m subscriptions were started on each source observable. We only want once subscription on each.

This PR adds fix code to OnSubscribeAmb and two unit tests.

akarnokd · 2015-05-19T07:35:37Z

src/main/java/rx/internal/operators/OnSubscribeAmb.java

+                } else {
+                    //subscriptions already happened so propagate the request to all the  
+                    //amb subscribers
+                    for (AmbSubscriber<T> ambSubscriber: selection.ambSubscribers) {


This may still run concurrently with the subscription loop and some subscribers won't get the newer requests during that time. The simplest fix is to subscribe all AmbSubscriber in the call() of the operator with 0 initial requests so all subsequent request() can target all or the winner without any additional race. The other way is to get into some complicated serialization logic.

davidmoten · 2015-05-19T07:39:19Z

yeah I like your first suggestion, I'll look at that

davidmoten · 2015-05-19T08:01:21Z

I've moved the subscription to sources loop into the call method as suggested.

akarnokd · 2015-05-19T08:30:22Z

src/main/java/rx/internal/operators/OnSubscribeAmb.java

+                        if (!ambSubscriber.isUnsubscribed()) {
+                            // make a best endeavours check to not waste requests 
+                            // if first emission has already occurred
+                            if (selection.choice.get() == ambSubscriber) {


I'd move selection.choice into a final variable at the beginning of the method so we don't re-read two fields unnecessarily with all those memory barriers around.

davidmoten · 2015-05-19T11:12:04Z

I've done that and reduced volatile reads a bit too. Given that selection and selection.choice are both final fields what is the interaction with the memory barriers? (sorry, a bit of a general question this for my education)

akarnokd · 2015-05-19T11:21:49Z

Variables tend to be re-read around volatiles and the JIT may be smart enough to optimize it and pull them upfront, but you can't be sure.

akarnokd · 2015-05-19T11:23:35Z

src/main/java/rx/internal/operators/OnSubscribeAmb.java

                    // there is a single winner so we unsubscribe it
-                    selection.choice.get().unsubscribe();
+                    c.unsubscribe();
                } 
                // if we are racing with others still existing, we'll also unsubscribe them
                if(!selection.ambSubscribers.isEmpty()) {


I'd read selection.ambSubscribers into a local variable as well.

davidmoten · 2015-05-19T11:41:31Z

added ambSubscribers final variable

akarnokd · 2015-05-19T11:46:10Z

src/main/java/rx/internal/operators/OnSubscribeAmb.java

+    final Iterable<? extends Observable<? extends T>> sources;
+    final Selection<T> selection = new Selection<T>();
+    final AtomicReference<AmbSubscriber<T>> choice = selection.choice;
+    final Collection<AmbSubscriber<T>> ambSubscribers = selection.ambSubscribers;


I meant read them into a method-local variable.

davidmoten · 2015-05-19T12:15:34Z

ok, done

akarnokd · 2015-05-19T14:49:46Z

I'd like to review the whole operator but I don't have the time and the focus to do it now. I'll come back to this in a few hours.

zsxwing · 2015-05-19T16:56:10Z

if m requests were made before the first emission then m subscriptions were started on each source observable. We only want once subscription on each.

Maybe I missed some discussion about this case. So now we support calling request in onStart multiple times?

davidmoten · 2015-05-19T20:59:00Z

Maybe I missed some discussion about this case. So now we support calling request in onStart multiple times?

Yes we do, multiple requests in onStart are additive but this is about what happens between subscriptions starting and first emission occurring. The new test testSubscriptionOnlyHappensOnce demos the problem in that by subscribing to an amb of async sources with an onStart request of one and then immediately (post subscription) calling request(1) we get a bunch of extra subscriptions.

zsxwing · 2015-05-20T00:03:56Z

src/test/java/rx/internal/operators/OnSubscribeAmbTest.java

+                .delay(100, TimeUnit.MILLISECONDS).subscribeOn(Schedulers.computation());
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        Observable.amb(o1, o2).subscribe(ts);
+        ts.requestMore(1);


This should be forbidden in the real case. Right? request should be called in Subscriber.

So testSubscriptionOnlyHappensOnce is a test case that requesting twice before first emission. Right? The other request is in subscribe.

davidmoten · 2015-05-20T00:31:21Z

Interesting point but I assume async operators downstream of the amb could do what we are trying to simulate with TestSubscriber.

Saying that requests between between onStart and first emission are forbidden is not something I see documented anywhere, let me know if you see it.

zsxwing · 2015-05-20T00:36:13Z

Saying that requests between between onStart and first emission are forbidden is not something I see documented anywhere, let me know if you see it.

I don't mean it. I mean request is protected to avoid calling out of Subscriber.

zsxwing · 2015-05-20T00:59:18Z

src/main/java/rx/internal/operators/OnSubscribeAmb.java

+
+    //give default access instead of private as a micro-optimization 
+    //for access from anonymous classes below
+    final Iterable<? extends Observable<? extends T>> sources;


Cool. Just learnt that accessing a private field of the outer class uses invokestatic.

And generates an accessor method in that outer class, the count of which (methods) is a big deal on Android. So 👍.

zsxwing · 2015-05-20T01:11:51Z

LGTM for the fix. Just some minor comments for the tests.

davidmoten · 2015-05-20T02:06:31Z

@zsxwing I've updated the test that called ts.requestMore(1) before subscription so that it uses the onStart method of TestSubscriber instead. Thanks for the review.

zsxwing · 2015-05-20T02:11:51Z

LGTM. Let's wait for @akarnokd to take a final look.

akarnokd · 2015-05-20T07:27:44Z

I found a few places for possible optimizations:

in AmbSubscriber, you can have a plain boolean chosen field to avoid the reference comparison:

boolean chosen;
...
    private boolean isSelected() {
        if (chosen) {
            return true;
        }
        if (selection.choice.get() == this) {
            // fast-path
            chosen = true;
            return true;
        } else {
            if (selection.choice.compareAndSet(null, this)) {
                selection.unsubscribeOthers(this);
                chosen = true;
                return true;
            } else {
                // we lost so unsubscribe ... and force cleanup again due to possible race conditions
                selection.unsubscribeLosers();
                return false;
            }
        }
    }

There is a race between the unsubscription logic at L360 and the loop at L381: while the loop is busy creating and subscribing AmbSubscribers, a concurrent unsubscribe will start cancelling that have reached the queue but any subsequent. Possible solutions:
- Move the creation and enqueueing of AmbSubscribers at the beginning, then create the unsubscription logic and finally loop and subscribe the AmbSubscribers. This way, the queue will contain all that need to be unsubscribed.
- Instead of quitting after the subscriber.isUnsubscribed(), drain the queue and unsubscribe its contents, do a similar check after the loop as well to clean up any stragglers.
- Have a custom subscription manager, similar to what the subjects use that can unsubscribe latecommers and keep the current one-loop approach (I have plans to cover such custom container in my blog soon.)

davidmoten · 2015-05-21T00:03:24Z

I like the optimization for isSelected. I've also committed the first approach for the unsub race but I think we can do better as you imply. Might go for the second.

davidmoten · 2015-05-24T05:33:33Z

Instead of quitting after the subscriber.isUnsubscribed(), drain the queue and unsubscribe its contents, do a similar check after the loop as well to clean up any stragglers.

I've implemented this suggestion @akarnokd. Ready for (another) review, thanks.

akarnokd · 2015-05-26T07:30:06Z

Looks good. Thanks!

fix Amb backpressure bug

davidmoten force-pushed the amb-bug branch from aa50d51 to 1dd1192 Compare May 19, 2015 06:01

akarnokd reviewed May 19, 2015
View reviewed changes

davidmoten force-pushed the amb-bug branch from 1dd1192 to ef6a222 Compare May 19, 2015 07:58

akarnokd reviewed May 19, 2015
View reviewed changes

davidmoten force-pushed the amb-bug branch from ef6a222 to bd17175 Compare May 19, 2015 11:03

akarnokd reviewed May 19, 2015
View reviewed changes

davidmoten force-pushed the amb-bug branch from bd17175 to 0236d2f Compare May 19, 2015 11:39

akarnokd reviewed May 19, 2015
View reviewed changes

davidmoten force-pushed the amb-bug branch from 0236d2f to 13092e1 Compare May 19, 2015 12:14

akarnokd added the Bug label May 19, 2015

zsxwing reviewed May 20, 2015
View reviewed changes

davidmoten force-pushed the amb-bug branch from a5bf766 to 975df61 Compare May 20, 2015 01:54

davidmoten force-pushed the amb-bug branch 2 times, most recently from 47ebbbf to bb659b6 Compare May 20, 2015 23:54

davidmoten force-pushed the amb-bug branch 2 times, most recently from 5d5d929 to b937fdd Compare May 24, 2015 05:28

fix Amb backpressure problems

9585c6e

davidmoten force-pushed the amb-bug branch from b937fdd to 9585c6e Compare May 24, 2015 05:35

akarnokd added a commit that referenced this pull request May 26, 2015

Merge pull request #2961 from davidmoten/amb-bug

4329fed

fix Amb backpressure bug

akarnokd merged commit 4329fed into ReactiveX:1.x May 26, 2015

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix Amb backpressure bug #2961

fix Amb backpressure bug #2961

davidmoten commented May 19, 2015

akarnokd May 19, 2015

davidmoten commented May 19, 2015

davidmoten commented May 19, 2015

akarnokd May 19, 2015

davidmoten commented May 19, 2015

akarnokd commented May 19, 2015

akarnokd May 19, 2015

davidmoten commented May 19, 2015

akarnokd May 19, 2015

davidmoten commented May 19, 2015

akarnokd commented May 19, 2015

zsxwing commented May 19, 2015

davidmoten commented May 19, 2015

zsxwing May 20, 2015

zsxwing May 20, 2015

davidmoten commented May 20, 2015

zsxwing commented May 20, 2015

zsxwing May 20, 2015

JakeWharton May 20, 2015

zsxwing commented May 20, 2015

davidmoten commented May 20, 2015

zsxwing commented May 20, 2015

akarnokd commented May 20, 2015

davidmoten commented May 21, 2015

davidmoten commented May 24, 2015

akarnokd commented May 26, 2015

fix Amb backpressure bug #2961

fix Amb backpressure bug #2961

Conversation

davidmoten commented May 19, 2015

Choose a reason for hiding this comment

davidmoten commented May 19, 2015

davidmoten commented May 19, 2015

Choose a reason for hiding this comment

davidmoten commented May 19, 2015

akarnokd commented May 19, 2015

Choose a reason for hiding this comment

davidmoten commented May 19, 2015

Choose a reason for hiding this comment

davidmoten commented May 19, 2015

akarnokd commented May 19, 2015

zsxwing commented May 19, 2015

davidmoten commented May 19, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davidmoten commented May 20, 2015

zsxwing commented May 20, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zsxwing commented May 20, 2015

davidmoten commented May 20, 2015

zsxwing commented May 20, 2015

akarnokd commented May 20, 2015

davidmoten commented May 21, 2015

davidmoten commented May 24, 2015

akarnokd commented May 26, 2015