Skip to content

OperatorObserveOn onComplete can be emitted despite onError being called #2929

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

Merged
merged 3 commits into from
May 13, 2015
Merged

OperatorObserveOn onComplete can be emitted despite onError being called #2929

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8eb6482
fix OperatorObserveOn race condition where onComplete could be emitte…
davidmoten May 1, 2015
18caf0e
use new pollQueue from @akarnokd
davidmoten May 9, 2015
c9d9324
don't reduce requested by produced if requested is Long.MAX_VALUE
davidmoten May 13, 2015
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
89 changes: 38 additions & 51 deletions src/main/java/rx/internal/operators/OperatorObserveOn.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,15 +75,19 @@ private static final class ObserveOnSubscriber<T> extends Subscriber<T> {
final NotificationLite<T> on = NotificationLite.instance();

final Queue<Object> queue;
volatile boolean completed = false;
volatile boolean failure = false;

// the status of the current stream
volatile boolean finished = false;

@SuppressWarnings("unused")
volatile long requested = 0;

@SuppressWarnings("rawtypes")
static final AtomicLongFieldUpdater<ObserveOnSubscriber> REQUESTED = AtomicLongFieldUpdater.newUpdater(ObserveOnSubscriber.class, "requested");

@SuppressWarnings("unused")
volatile long counter;

@SuppressWarnings("rawtypes")
static final AtomicLongFieldUpdater<ObserveOnSubscriber> COUNTER_UPDATER = AtomicLongFieldUpdater.newUpdater(ObserveOnSubscriber.class, "counter");

Expand Down Expand Up @@ -127,7 +131,7 @@ public void onStart() {

@Override
public void onNext(final T t) {
if (isUnsubscribed() || completed) {
if (isUnsubscribed()) {
return;
}
if (!queue.offer(on.next(t))) {
Expand All @@ -139,30 +143,23 @@ public void onNext(final T t) {

@Override
public void onCompleted() {
if (isUnsubscribed() || completed) {
if (isUnsubscribed() || finished) {
return;
}
if (error != null) {
return;
}
completed = true;
finished = true;
schedule();
}

@Override
public void onError(final Throwable e) {
if (isUnsubscribed() || completed) {
return;
}
if (error != null) {
if (isUnsubscribed() || finished) {
return;
}
error = e;
// unsubscribe eagerly since time will pass before the scheduled onError results in an unsubscribe event
unsubscribe();
// mark failure so the polling thread will skip onNext still in the queue
completed = true;
failure = true;
finished = true;
// polling thread should skip any onNext still in the queue
schedule();
}

Expand All @@ -185,52 +182,42 @@ protected void schedule() {
void pollQueue() {
int emitted = 0;
do {
/*
* Set to 1 otherwise it could have grown very large while in the last poll loop
* and then we can end up looping all those times again here before exiting even once we've drained
*/
counter = 1;

// middle:
while (!scheduledUnsubscribe.isUnsubscribed()) {
if (failure) {
child.onError(error);
return;
} else {
if (requested == 0 && completed && queue.isEmpty()) {
long produced = 0;
long r = requested;
while (!child.isUnsubscribed()) {
Throwable error;
if (finished) {
if ((error = this.error) != null) {
// errors shortcut the queue so
// release the elements in the queue for gc
queue.clear();
child.onError(error);
Copy link
Member

Choose a reason for hiding this comment

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

I meant error.

return;
} else
if (queue.isEmpty()) {
child.onCompleted();
return;
}
if (REQUESTED.getAndDecrement(this) != 0) {
Object o = queue.poll();
if (o == null) {
if (completed) {
if (failure) {
child.onError(error);
} else {
child.onCompleted();
}
return;
}
// nothing in queue
REQUESTED.incrementAndGet(this);
break;
} else {
if (!on.accept(child, o)) {
// non-terminal event so let's increment count
emitted++;
}
}
}
if (r > 0) {
Object o = queue.poll();
if (o != null) {
child.onNext(on.getValue(o));
r--;
emitted++;
produced++;
} else {
// we hit the end ... so increment back to 0 again
REQUESTED.incrementAndGet(this);
break;
}
} else {
break;
}
}
if (produced > 0 && requested != Long.MAX_VALUE) {
REQUESTED.addAndGet(this, -produced);
}
} while (COUNTER_UPDATER.decrementAndGet(this) > 0);

// request the number of items that we emitted in this poll loop
if (emitted > 0) {
request(emitted);
}
Expand Down
21 changes: 21 additions & 0 deletions src/perf/java/rx/operators/OperatorObserveOnPerf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,5 +66,26 @@ public void observeOnImmediate(Input input) throws InterruptedException {
input.observable.observeOn(Schedulers.immediate()).subscribe(o);
o.latch.await();
}

@Benchmark
public void observeOnComputationSubscribedOnComputation(Input input) throws InterruptedException {
LatchedObserver<Integer> o = input.newLatchedObserver();
input.observable.subscribeOn(Schedulers.computation()).observeOn(Schedulers.computation()).subscribe(o);
o.latch.await();
}

@Benchmark
public void observeOnNewThreadSubscribedOnComputation(Input input) throws InterruptedException {
LatchedObserver<Integer> o = input.newLatchedObserver();
input.observable.subscribeOn(Schedulers.computation()).observeOn(Schedulers.newThread()).subscribe(o);
o.latch.await();
}

@Benchmark
public void observeOnImmediateSubscribedOnComputation(Input input) throws InterruptedException {
LatchedObserver<Integer> o = input.newLatchedObserver();
input.observable.subscribeOn(Schedulers.computation()).observeOn(Schedulers.immediate()).subscribe(o);
o.latch.await();
}

}