Repository navigation
2.x: Surprising scheduler behaviour #6696
Description
Brief description
Given a shared upstream and two flows using .observeOn() before .filter() may change the original emission order.
RxJava Version
2.2.13
Code sample
private val subject = PublishSubject.create<Int>().toSerialized()
private val singleScheduler = Schedulers.single()
fun filterSubjectForIdOnScheduler(id: Int): Observable<Int> {
return subject
.observeOn(singleScheduler)
.filter {
Log.e("filter", "id = $id value = $it")
it == id
}
}
fun main() {
Observable.merge(
filterSubjectForIdOnScheduler(1),
filterSubjectForIdOnScheduler(2)
)
.subscribe { Log.e("subscribe", "value = $it") }
subject.onNext(2)
subject.onNext(1)
}
Actual (surprising) result
14491-14855 E/filter: id = 1 value = 2
14491-14855 E/filter: id = 1 value = 1
14491-14855 E/subscribe: value = 1
14491-14855 E/filter: id = 2 value = 2
14491-14855 E/subscribe: value = 2
14491-14855 E/filter: id = 2 value = 1
Expected result
14491-14855 E/filter: id = 1 value = 2
14491-14855 E/filter: id = 2 value = 2
14491-14855 E/subscribe: value = 2
14491-14855 E/filter: id = 1 value = 1
14491-14855 E/subscribe: value = 1
14491-14855 E/filter: id = 2 value = 1
Notes
- The code originally used a
Schedulers.from(Executors.newSingleThreadExecutor())— I thought that somehow the executors may be optimised in a way that batches individual runnables and cycles them before cycling the queue of observers. I foundSchedulers.single()which states (emphasis mine):
* Returns a default, shared, single-thread-backed {@link Scheduler} instance for work * requiring >>>strongly-sequential<<< execution on the same background thread.
- Instead of using
.observeOn()calling viascheduleDirectyields expected results:
singleScheduler.scheduleDirect { subject.onNext(2) }
singleScheduler.scheduleDirect { subject.onNext(1) }
// or
singleScheduler.scheduleDirect {
subject.onNext(2)
subject.onNext(1)
}
- It seems that the scheduler has two queues. One for emissions and one for observers. The emissions queue seems to be cycled before the queue of observers where sequential processing seemingly would need different priority of cycling (notify all observers about the first emission before proceeding to a next one).
Question
Is the actual result an expected one?
If so — could you explain why? Is it possible to alter the flow without reordering operators and achieve expected results?
Activity
akarnokd commentedon Oct 31, 2019
observeOnkeeps the single thread occupied as long as it sees work, in this case, the values 2 and 1 in quick succession. Trydelay(0)which schedules events individually.dariuszseweryn commentedon Oct 31, 2019
Do you mean
.delay(0, TimeUnit.SECONDS)? I have tried adding it before and after.observeOn()but it seems to work the same.akarnokd commentedon Oct 31, 2019
Replace
observeOnwithdelay(0, TimeUnit.SECONDS).dariuszseweryn commentedon Oct 31, 2019
Using
.delay(0, TimeUnit.SECONDS, singleScheduler)instead of.observeOn(singleScheduler)changed the behaviour to one that matches the expected result.For me this looks awkward — as a user I would like to observe on a specified scheduler in sequential order. Currently it seems to violate Law of least surprise. How do you think? Or maybe current implementation has some obvious performance advantage that I am unaware of?
akarnokd commentedon Oct 31, 2019
observeOnis designed to do as much work in a drain run as possible.Reactive concurrency is complicated and scheduling is an orthogonal concept (when vs. where), hence a perfectly sequential scheduler used with an operator can result in non-intuitive event signaling pattern downstream. Also your type of lockstepping and coordination is rare.
dariuszseweryn commentedon Oct 31, 2019
Is this well described somewhere so I could study the topic a bit more?
Should this be mentioned somewhere? It looks like an important gotcha.
I use this approach to relief the calling thread as quickly as possible since it is bound to native code and prone to break the underlying system that I do not control (here modelled as the
subject).akarnokd commentedon Oct 31, 2019
See my blog and the operator writing guide.
The
observeOnJavadoc can be updated to mention its eagerness.Fixed possibly incorrect order of emissions from RxBleGattCallback
dariuszseweryn commentedon Nov 4, 2019
A helper question — do you know why schedulers are designed this way by default (instead of being basically equivalent to
.delay(0, timeUnit, scheduler)?akarnokd commentedon Nov 4, 2019
Scheduler is an abstraction over crossing an asynchronous boundary by providing methods to run work somewhere. It doesn't know what or how much work it means.
observeOnis designed to be low overhead (both time and allocation) when moving data between threads.delayhas to schedule each item individually which adds a lot of allocation.dariuszseweryn commentedon Nov 19, 2019
I think this topic is cleared and appropriate Javadoc change has landed so I am closing. Thank you very much for the help.
I do have some other questions about surface'ing reactive and non-reactive world.
Observables are subscribed and disposed when using.subscribeOn()and without it — in the first situation theObservablemay or may not be subscribed depending on a race condition and in the second it will always be subscribed. Funny thing is thatTestSchedulerseems to work as no.subscribeOn()is present rather than other schedulers.UndeliverableException— this is something that is seemingly impossible to mitigate in a multi-threaded environment as there always may be a race condition under which theObserverwill already be unsubscribed.I would like to discuss those matters or find a way to learn on how to deal with them. Should I create separate issues or...?
akarnokd commentedon Nov 19, 2019
subscribeOnalways signalsonSubscribebut the subscription towards the upstream should be properly depend on whenTestScheduler::triggerActions()is invoked.dariuszseweryn commentedon Nov 20, 2019
observable.subscribeOn().subscribe().dispose()subscribing/not subscribing to theobservabledepending on a race condition. I see some weird behaviour in my tests but I assume I got something wrong. Will dig deeper and keep you posted if I find something funny.Fixed possibly incorrect order of emissions from RxBleGattCallback (#639