Skip to content

Avoid merge operator from over requesting upstream producer. #2765

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 1 commit into from
Closed

Avoid merge operator from over requesting upstream producer. #2765

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 12 additions & 2 deletions src/main/java/rx/internal/operators/OperatorMerge.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,7 +213,7 @@ private void handleNewSource(Observable<? extends T> t) {
InnerSubscriber<T> i = new InnerSubscriber<T>(this, producerIfNeeded);
i.sindex = childrenSubscribers.add(i);
t.unsafeSubscribe(i);
if (!isUnsubscribed()) {
if ((producerIfNeeded == null || producerIfNeeded.requested > 0) && !isUnsubscribed()) {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@akarnokd the original code calls request(1) regardless of the "requested" variable causing the operator to potentially buffer a dangerous amount of items and also a loop between producer.request() -> consumer.onNext() -> producer->request() ...

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Merge deliberately doesn't limit the source observable of observables; this is what MergeMaxConcurrent is for. Besides, I can't rule out a race for requested.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

/cc @benjchristensen

request(1);
}
}
Expand Down Expand Up @@ -523,6 +523,9 @@ private void drainAndComplete() {
}
}

public void requestMore(long n) {
request(n);
}
}

private static final class MergeProducer<T> implements Producer {
Expand All @@ -545,16 +548,23 @@ public void request(long n) {
if (n == Long.MAX_VALUE) {
requested = Long.MAX_VALUE;
} else {
BackpressureUtils.getAndAddRequest(REQUESTED, this, n);
final long count = BackpressureUtils.getAndAddRequest(REQUESTED, this, n);
if (ms.drainQueuesIfNeeded()) {
boolean sendComplete = false;
boolean requestMore = false;
synchronized (ms) {
if (ms.wip == 0 && ms.scalarValueQueue != null && ms.scalarValueQueue.isEmpty()) {
sendComplete = true;
}

if (count > 0 && ms.wip == 0 && ms.scalarValueQueue == null) {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is here to allow requesting from the upstream producer if there are no pending requests. Without these the request for more will never reach the source observable.

requestMore = true;
}
}
if (sendComplete) {
ms.drainAndComplete();
} else if (requestMore) {
ms.requestMore(n);
}
}
}
Expand Down
58 changes: 55 additions & 3 deletions src/test/java/rx/internal/operators/OperatorMergeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,9 +21,7 @@
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.*;

import java.util.ArrayList;
import java.util.Arrays;
Expand Down Expand Up @@ -667,6 +665,60 @@ public void onNext(Integer t) {
assertTrue(generated1.get() >= RxRingBuffer.SIZE * 2 && generated1.get() <= RxRingBuffer.SIZE * 3);
}

@Test
public void testWhenRequestCalledAndNotingPendingThenEmitsMore() throws InterruptedException {
final AtomicInteger generated1 = new AtomicInteger();
final TestScheduler scheduler = new TestScheduler();
TestSubscriber<Integer> nonScalartestSubscriber = spy(new TestSubscriber<Integer>() {
@Override
public void onStart() {
request(10);
}
});
Observable<Integer> o1 = createInfiniteObservable(generated1)
.flatMap(new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(final Integer integer) {
return Observable.create(new OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
subscriber.onNext(-integer);
subscriber.onCompleted();
}
});
}
})
.subscribeOn(scheduler);
o1.subscribe(nonScalartestSubscriber);

TestSubscriber<Integer> scalartestSubscriber = spy(new TestSubscriber<Integer>() {
@Override
public void onStart() {
request(10);
}
});
Observable<Integer> o2 = createInfiniteObservable(generated1)
.flatMap(new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(final Integer integer) {
return Observable.just(-integer);
}
})
.subscribeOn(scheduler);
o2.subscribe(scalartestSubscriber);

scheduler.triggerActions();
verify(nonScalartestSubscriber, times(10)).onNext(anyInt());
verify(scalartestSubscriber, times(10)).onNext(anyInt());

nonScalartestSubscriber.requestMore(100);
scalartestSubscriber.requestMore(100);
scheduler.triggerActions();

verify(nonScalartestSubscriber, times(110)).onNext(anyInt());
verify(scalartestSubscriber, times(110)).onNext(anyInt());
}

/**
* This is the same as the upstreams ones, but now adds the downstream as well by using observeOn.
*
Expand Down