Skip to content

Blocking buffer until experiment #864

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 3 commits into from
Closed

Blocking buffer until experiment #864

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
dc4ee52
Proposed solution to the time gap, using unbounded buffering.
akarnokd Feb 12, 2014
dade7e1
Added bounded buffering capability to SubscribeOn
akarnokd Feb 12, 2014
5209ab1
Check child unsubscription status more eagerly.
akarnokd Feb 12, 2014
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
25 changes: 24 additions & 1 deletion rxjava-core/src/main/java/rx/Observable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7060,9 +7060,32 @@ public final Subscription subscribe(Subscriber<? super T> observer, Scheduler sc
* @return the source Observable modified so that its subscriptions and unsubscriptions happen
* on the specified {@link Scheduler}
* @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-subscribeon">RxJava Wiki: subscribeOn()</a>
* @see #subscribeOn(rx.Scheduler, int)
*/
public final Observable<T> subscribeOn(Scheduler scheduler) {
return nest().lift(new OperatorSubscribeOn<T>(scheduler));
return nest().lift(new OperatorSubscribeOn<T>(scheduler, false));
}
/**
* Asynchronously subscribes and unsubscribes Observers to this Observable on the specified {@link Scheduler}
* and allows buffering some events emitted from the source in the time gap between the original and
* actual subscription, and any excess events will block the source until the actual subscription happens.
* <p>
* This overload should help mitigate issues when subscribing to a PublishSubject (and derivatives
* such as GroupedObservable in operator groupBy) and events fired between the original and actual subscriptions
* are lost.
* <p>
* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/subscribeOn.png">
*
* @param scheduler
* the {@link Scheduler} to perform subscription and unsubscription actions on
* @param bufferSize the number of events to buffer before blocking the source while in the time gap,
* negative value indicates an unlimited buffer
* @return the source Observable modified so that its subscriptions and unsubscriptions happen
* on the specified {@link Scheduler}
* @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-subscribeon">RxJava Wiki: subscribeOn()</a>
*/
public final Observable<T> subscribeOn(Scheduler scheduler, int bufferSize) {
Copy link
Member

Choose a reason for hiding this comment

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

This overload should help mitigate issues when subscribing to a PublishSubject (and derivatives such as GroupedObservable in operator groupBy) and events fired between the original and actual subscriptions are lost.

That doesn't seem to need this overload since we special-case those two Observable instances even if bufferSize is not passed in and we say false for dontLoseEvents.

Copy link
Member

Choose a reason for hiding this comment

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

Ah, I see ... the automatic has an unbounded buffer so it never blocks. This overload allows for blocking as well.

Are you suggesting subscribeOn by itself never use the blocking form but any operator implementation that uses subscribeOn would choose to do so?

In that case, should subscribeOn do any buffering by default or only on demand?

Copy link
Member Author

Choose a reason for hiding this comment

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

My suggestion is that subscribeOn(Scheduler) never blocks with any source and events may be lost, and subscribeOn(Scheduler, int) does not lose events and may block depending on the buffer size.

return nest().lift(new OperatorSubscribeOn<T>(scheduler, true, bufferSize));
}

/**
Expand Down
193 changes: 193 additions & 0 deletions rxjava-core/src/main/java/rx/operators/BufferUntilSubscriber.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,193 @@
/**
* Copyright 2014 Netflix, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package rx.operators;

import java.util.LinkedList;
import java.util.Queue;
import rx.Subscriber;
import rx.subscriptions.CompositeSubscription;

/**
* Buffers the incoming events until notified, then replays the
* buffered events and continues as a simple pass-through subscriber.
* @param <T> the streamed value type
*/
public class BufferUntilSubscriber<T> extends Subscriber<T> {
/** The actual subscriber. */
private final Subscriber<? super T> actual;
/** Indicate the pass-through mode. */
private volatile boolean passthroughMode;
/** Protect mode transition. */
private final Object gate = new Object();
/** The buffered items. */
private final Queue<Object> queue = new LinkedList<Object>();
/** The queue capacity. */
private final int capacity;
/** Null sentinel (in case queue type is changed). */
private static final Object NULL_SENTINEL = new Object();
/** Complete sentinel. */
private static final Object COMPLETE_SENTINEL = new Object();
/**
* Container for an onError event.
*/
private static final class ErrorSentinel {
final Throwable t;

public ErrorSentinel(Throwable t) {
this.t = t;
}

}
/**
* Constructor that wraps the actual subscriber and shares its subscription.
* @param capacity the queue capacity to accept before blocking, negative value indicates an unbounded queue
* @param actual
*/
public BufferUntilSubscriber(int capacity, Subscriber<? super T> actual) {
super(actual);
this.actual = actual;
this.capacity = capacity;
}
/**
* Constructor that wraps the actual subscriber and uses the given composite
* subscription.
* @param capacity the queue capacity to accept before blocking, negative value indicates an unbounded queue
* @param actual
* @param cs
*/
public BufferUntilSubscriber(int capacity, Subscriber<? super T> actual, CompositeSubscription cs) {
super(cs);
this.actual = actual;
this.capacity = capacity;
}

/**
* Call this method to replay the buffered events and continue as a pass-through subscriber.
* If already in pass-through mode, this method is a no-op.
*/
public void enterPassthroughMode() {
if (!passthroughMode) {
synchronized (gate) {
if (!passthroughMode) {
while (!queue.isEmpty()) {
Object o = queue.poll();
if (!actual.isUnsubscribed()) {
if (o == NULL_SENTINEL) {
actual.onNext(null);
} else
if (o == COMPLETE_SENTINEL) {
actual.onCompleted();
} else
if (o instanceof ErrorSentinel) {
actual.onError(((ErrorSentinel)o).t);
} else
if (o != null) {
@SuppressWarnings("unchecked")
T v = (T)o;
actual.onNext(v);
} else {
throw new NullPointerException();
}
}
}
passthroughMode = true;
gate.notifyAll();
}
}
}
}
@Override
public void onNext(T t) {
if (!passthroughMode) {
synchronized (gate) {
if (!passthroughMode) {
if (capacity < 0 || queue.size() < capacity) {
queue.offer(t != null ? t : NULL_SENTINEL);
return;
}
try {
while (!passthroughMode) {
gate.wait();
}
if (actual.isUnsubscribed()) {
return;
}
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
actual.onError(ex);
return;
}
}
}
}
actual.onNext(t);
}

@Override
public void onError(Throwable e) {
if (!passthroughMode) {
synchronized (gate) {
if (!passthroughMode) {
if (capacity < 0 || queue.size() < capacity) {
queue.offer(new ErrorSentinel(e));
return;
}
try {
while (!passthroughMode) {
gate.wait();
}
if (actual.isUnsubscribed()) {
return;
}
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
actual.onError(ex);
return;
}
}
}
}
actual.onError(e);
}

@Override
public void onCompleted() {
if (!passthroughMode) {
synchronized (gate) {
if (!passthroughMode) {
if (capacity < 0 || queue.size() < capacity) {
queue.offer(COMPLETE_SENTINEL);
return;
}
try {
while (!passthroughMode) {
gate.wait();
}
if (actual.isUnsubscribed()) {
return;
}
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
actual.onError(ex);
return;
}
}
}
}
actual.onCompleted();
}

}
55 changes: 52 additions & 3 deletions rxjava-core/src/main/java/rx/operators/OperatorSubscribeOn.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@
import rx.Scheduler;
import rx.Scheduler.Inner;
import rx.Subscriber;
import rx.observables.GroupedObservable;
import rx.subjects.PublishSubject;
import rx.subscriptions.CompositeSubscription;
import rx.subscriptions.Subscriptions;
import rx.util.functions.Action0;
Expand All @@ -33,9 +35,27 @@
public class OperatorSubscribeOn<T> implements Operator<T, Observable<T>> {

private final Scheduler scheduler;

public OperatorSubscribeOn(Scheduler scheduler) {
/**
* Indicate that events fired between the original subscription time and
* the actual subscription time should not get lost.
*/
private final boolean dontLoseEvents;
/** The buffer size to avoid flooding. Negative value indicates an unbounded buffer. */
private final int bufferSize;
public OperatorSubscribeOn(Scheduler scheduler, boolean dontLoseEvents) {
this(scheduler, dontLoseEvents, -1);
}
/**
* Construct a SubscribeOn operator.
* @param scheduler the target scheduler
* @param dontLoseEvents indicate that events should be buffered until the actual subscription happens
* @param bufferSize if dontLoseEvents == true, this indicates the buffer size. Filling the buffer will
* block the source. -1 indicates an unbounded buffer
*/
public OperatorSubscribeOn(Scheduler scheduler, boolean dontLoseEvents, int bufferSize) {
this.scheduler = scheduler;
this.dontLoseEvents = dontLoseEvents;
this.bufferSize = bufferSize;
}

@Override
Expand All @@ -51,9 +71,38 @@ public void onCompleted() {
public void onError(Throwable e) {
subscriber.onError(e);
}

boolean checkNeedBuffer(Observable<?> o) {
return dontLoseEvents || ((o instanceof GroupedObservable<?, ?>)
|| (o instanceof PublishSubject<?>)
// || (o instanceof BehaviorSubject<?, ?>)
);
}
@Override
public void onNext(final Observable<T> o) {
if (checkNeedBuffer(o)) {
final CompositeSubscription cs = new CompositeSubscription();
subscriber.add(cs);
final BufferUntilSubscriber<T> bus = new BufferUntilSubscriber<T>(bufferSize, subscriber, new CompositeSubscription());
o.subscribe(bus);
scheduler.schedule(new Action1<Inner>() {
@Override
public void call(final Inner inner) {
cs.add(Subscriptions.create(new Action0() {
@Override
public void call() {
inner.schedule(new Action1<Inner>() {
@Override
public void call(final Inner inner) {
bus.unsubscribe();
}
});
}
}));
bus.enterPassthroughMode();
}
});
return;
}
Copy link
Member

Choose a reason for hiding this comment

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

For some reason the unsubscribe never propagates back up on the non-buffered path:

                            subscriber.add(Subscriptions.create(new Action0() {

                                @Override
                                public void call() {
                                    System.out.println("unsubscribe? " + inner.isUnsubscribed());
                                    inner.schedule(new Action1<Inner>() {

                                        @Override
                                        public void call(final Inner inner) {
                                            System.out.println("unsubscribe again?");
                                            cs.unsubscribe();
                                        }

                                    });
                                }

                            }));

The first println happens, the second doesn't.

I actually do not understand however why we need to schedule the unsubscribe, especially with our new lift model.

The unsubscribe does not happen from the parent, it happens from the children. And the children will most likely be on the same scheduler we're on, but even if it's not, why schedule it on the current thread? It's not the thread the parents are on, so why force it to propagate the unsubscribe to the parents on this thread?

Copy link
Member Author

Choose a reason for hiding this comment

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

inner is probably unsubscribed somewhere and the schedule() call does nothing.

I can't really explain why unsubscription is made this way. I would think it is for the single-threaded (or trampolined) scenarios where unsubscription has to happen after other events. Maybe @headinthebox knows.

scheduler.schedule(new Action1<Inner>() {

@Override
Expand Down
Loading