Migrate from SynchronizedObserver to SerializedObserver #962
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,186 @@ | ||
| package rx.observers; | ||
|
|
||
| import java.util.ArrayList; | ||
|
|
||
| import rx.Observer; | ||
|
|
||
| /** | ||
| * Enforce single-threaded, serialized, ordered execution of onNext, onCompleted, onError. | ||
| * <p> | ||
| * When multiple threads are notifying they will be serialized by: | ||
| * <p> | ||
| * <li>Allowing only one thread at a time to emit</li> | ||
| * <li>Adding notifications to a queue if another thread is already emitting</li> | ||
| * <li>Not holding any locks or blocking any threads while emitting</li> | ||
| * <p> | ||
| * | ||
| * @param <T> | ||
| */ | ||
| public class SerializedObserver<T> implements Observer<T> { | ||
| private final Observer<? super T> actual; | ||
|
|
||
| private boolean emitting = false; | ||
| private boolean terminated = false; | ||
| private ArrayList<Object> queue = new ArrayList<Object>(); | ||
|
|
||
| private static Sentinel NULL_SENTINEL = new Sentinel(); | ||
| private static Sentinel COMPLETE_SENTINEL = new Sentinel(); | ||
|
|
||
| private static class Sentinel { | ||
|
|
||
| } | ||
|
|
||
| private static class ErrorSentinel extends Sentinel { | ||
| final Throwable e; | ||
|
|
||
| ErrorSentinel(Throwable e) { | ||
| this.e = e; | ||
| } | ||
| } | ||
|
|
||
| public SerializedObserver(Observer<? super T> s) { | ||
| this.actual = s; | ||
| } | ||
|
|
||
| @Override | ||
| public void onCompleted() { | ||
| boolean canEmit = false; | ||
| ArrayList<Object> list = null; | ||
| synchronized (this) { | ||
| if (terminated) { | ||
| return; | ||
| } | ||
| terminated = true; | ||
| if (!emitting) { | ||
| // emit immediately | ||
| emitting = true; | ||
| canEmit = true; | ||
| if (queue.size() > 0) { | ||
| list = queue; // copy reference | ||
| queue = new ArrayList<Object>(); // new version; | ||
| } | ||
| } else { | ||
| // someone else is already emitting so just queue it | ||
| queue.add(COMPLETE_SENTINEL); | ||
| } | ||
| } | ||
| if (canEmit) { | ||
| // we won the right to emit | ||
| try { | ||
| drainQueue(list); | ||
| actual.onCompleted(); | ||
| } finally { | ||
| synchronized (this) { | ||
| emitting = false; | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onError(final Throwable e) { | ||
| boolean canEmit = false; | ||
| ArrayList<Object> list = null; | ||
| synchronized (this) { | ||
| if (terminated) { | ||
| return; | ||
| } | ||
| terminated = true; | ||
| if (!emitting) { | ||
| // emit immediately | ||
| emitting = true; | ||
| canEmit = true; | ||
| if (queue.size() > 0) { | ||
| list = queue; // copy reference | ||
| queue = new ArrayList<Object>(); // new version; | ||
| } | ||
| } else { | ||
| // someone else is already emitting so just queue it ... after eliminating the queue to shortcut | ||
| queue.clear(); | ||
| queue.add(new ErrorSentinel(e)); | ||
| } | ||
| } | ||
| if (canEmit) { | ||
| // we won the right to emit | ||
| try { | ||
| drainQueue(list); | ||
| actual.onError(e); | ||
| } finally { | ||
| synchronized (this) { | ||
| emitting = false; | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onNext(T t) { | ||
| boolean canEmit = false; | ||
| ArrayList<Object> list = null; | ||
| synchronized (this) { | ||
| if (terminated) { | ||
| return; | ||
| } | ||
| if (!emitting) { | ||
| // emit immediately | ||
| emitting = true; | ||
| canEmit = true; | ||
| if (queue.size() > 0) { | ||
| list = queue; // copy reference | ||
| queue = new ArrayList<Object>(); // new version; | ||
| } | ||
| } else { | ||
| // someone else is already emitting so just queue it | ||
| if (t == null) { | ||
| queue.add(NULL_SENTINEL); | ||
| } else { | ||
| queue.add(t); | ||
| } | ||
| } | ||
| } | ||
| if (canEmit) { | ||
| // we won the right to emit | ||
| try { | ||
| drainQueue(list); | ||
|
There was a problem hiding this comment. Lets assume two concurrent onNext competes for canEmit, the first succeeds and drains an empty queue. The second comes in and enqueues its value. The two exit onNext. In this case, the second onNext's value sits in the queue until another event happens. (Btw., the queue/drain doesn't exhibit this.) Can we accept such delay? There was a problem hiding this comment. Yes, I'm aware of that problem and don't like it but do not have a better solution yet. The other two implementations fails the more critical performance tests (the state machine version took down our production server in under 2 minutes). There was a problem hiding this comment. I did some experimenting and came up with this rewrite. There was a problem hiding this comment. I'll take a look, does it solve both of the tradeoffs? There was a problem hiding this comment. It has the MAX_DRAIN_ITERATION which trades the potential event delay (1) with effectively continuous draining (MAX_VALUE). I can't think of any adaptive adjustment method, only a parameterized serialize() operator. There was a problem hiding this comment. The MAX_DRAIN_ITERATION still has the problem, just pushed back until after N iterations. So in this case are you just optimizing for the case when 1 item is in the queue? There was a problem hiding this comment. Yes, it doesn't allocate the queue if there is no overlapping, some unnecesary synchronization blocks were removed. It doesn't solve the tradeoff problem unfortunately; to avoid the delay and one threaded drain, one would need to have wait-notify which most likely causes poor performance and thread blocking. |
||
| actual.onNext(t); | ||
| } finally { | ||
| synchronized (this) { | ||
| if (terminated) { | ||
| list = queue; // copy reference | ||
| queue = new ArrayList<Object>(); // new version; | ||
| } else { | ||
| // release this thread | ||
| emitting = false; | ||
| canEmit = false; | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // if terminated this will still be true so let's drain the rest of the queue | ||
| if (canEmit) { | ||
| drainQueue(list); | ||
| } | ||
| } | ||
|
|
||
| public void drainQueue(ArrayList<Object> list) { | ||
| if (list == null || list.size() == 0) { | ||
| return; | ||
| } | ||
| for (Object v : list) { | ||
| if (v != null) { | ||
| if (v instanceof Sentinel) { | ||
| if (v == NULL_SENTINEL) { | ||
| actual.onNext(null); | ||
| } else if (v == COMPLETE_SENTINEL) { | ||
| actual.onCompleted(); | ||
| } else if (v instanceof ErrorSentinel) { | ||
| actual.onError(((ErrorSentinel) v).e); | ||
| } | ||
| } else { | ||
| actual.onNext((T) v); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,40 @@ | ||
| package rx.observers; | ||
|
|
||
| import rx.Observer; | ||
| import rx.Subscriber; | ||
|
|
||
| /** | ||
| * Enforce single-threaded, serialized, ordered execution of onNext, onCompleted, onError. | ||
| * <p> | ||
| * When multiple threads are notifying they will be serialized by: | ||
| * <p> | ||
| * <li>Allowing only one thread at a time to emit</li> | ||
| * <li>Adding notifications to a queue if another thread is already emitting</li> | ||
| * <li>Not holding any locks or blocking any threads while emitting</li> | ||
| * <p> | ||
| * | ||
| * @param <T> | ||
| */ | ||
| public class SerializedSubscriber<T> extends Subscriber<T> { | ||
|
|
||
| private final Observer<T> s; | ||
|
|
||
| public SerializedSubscriber(Subscriber<? super T> s) { | ||
| this.s = new SerializedObserver<T>(s); | ||
| } | ||
|
|
||
| @Override | ||
| public void onCompleted() { | ||
| s.onCompleted(); | ||
| } | ||
|
|
||
| @Override | ||
| public void onError(Throwable e) { | ||
| s.onError(e); | ||
| } | ||
|
|
||
| @Override | ||
| public void onNext(T t) { | ||
| s.onNext(t); | ||
| } | ||
| } |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
should onError be draining the queue before sending the error?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes we can, but if we're winning the right to emit immediately as it's doing here, it's highly unlikely there is anything in the queue.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Sorry, by "yes we can" I meant we can clear it ... which means skip draining it in the
onErrorcase.