Using publish/share/etc leads to upstream operator running on unexpected scheduler #6144

Activity

1. akarnokd

   added

   Bug

   2.x

   on Aug 7, 2018
2. akarnokd

   added this to the 2.2 backlog milestone on Aug 7, 2018
3. akarnokd commented on Aug 7, 2018

   

   akarnokd

   on Aug 7, 2018

   Member

   More actions

   This a bug due to operator fusion breaking the thread-confinement expected after observeOn. You can work around it by applying hide() before share(). I'll post a fix shortly.
4. akarnokd

   closed this as completedin #6145on Aug 8, 2018
5. damianw commented on Aug 13, 2018

   

   damianw

   on Aug 13, 2018

   Author

   More actions

   @akarnokd I've noticed that .hide() does not always solve the problem - adding a second map (and/or probably other operators) after the share causes the problem to reappear:

       val scheduler1 = Schedulers.from(Executors.newSingleThreadExecutor { runnable ->
           Thread(runnable, "scheduler1")
       })
   
       val scheduler2 = Schedulers.from(Executors.newSingleThreadExecutor { runnable ->
           Thread(runnable, "scheduler2")
       })
   
       val flowable: Flowable<Long> = Flowable.interval(1L, TimeUnit.MILLISECONDS)
               .onBackpressureLatest()
               .take(10_000)
               .observeOn(scheduler1)
               .map {
                   val threadName = Thread.currentThread().name
                   if (threadName != "scheduler1") {
                       throw AssertionError("Wrong thread on map #1: $threadName")
                   }
                   it
               }
               .hide()
               .share()
               .map {
                   val threadName = Thread.currentThread().name
                   if (threadName != "scheduler1") {
                       throw AssertionError("Wrong thread on map #2: $threadName")
                   }
                   it
               }
   
       flowable
               .observeOn(scheduler2)
               .doOnNext { Thread.sleep(10L) }
               .subscribe {
                   println("onNext on ${Thread.currentThread().name}: $it")
               }

   Output:

   onNext on scheduler2: 0
   onNext on scheduler2: 1
   ...
   onNext on scheduler2: 94
   onNext on scheduler2: 95
   io.reactivex.exceptions.OnErrorNotImplementedException: Wrong thread on map #2: scheduler2
   	at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
   	at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
   	at io.reactivex.internal.subscribers.LambdaSubscriber.onError(LambdaSubscriber.java:79)
   	at io.reactivex.internal.operators.flowable.FlowableDoOnEach$DoOnEachSubscriber.onError(FlowableDoOnEach.java:111)
   	at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.checkTerminated(FlowableObserveOn.java:207)
   	at io.reactivex.internal.operators.flowable.FlowableObserveOn$ObserveOnSubscriber.runAsync(FlowableObserveOn.java:392)
   	at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.run(FlowableObserveOn.java:176)
   	at io.reactivex.internal.schedulers.ExecutorScheduler$ExecutorWorker$BooleanRunnable.run(ExecutorScheduler.java:261)
   	at io.reactivex.internal.schedulers.ExecutorScheduler$ExecutorWorker.run(ExecutorScheduler.java:226)
   	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
   	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
   	at java.lang.Thread.run(Thread.java:748)
   Caused by: java.lang.AssertionError: Wrong thread on map #2: scheduler2
   	at RxTestKt$main$flowable$2.apply(RxTest.kt:31)
   	at RxTestKt$main$flowable$2.apply(RxTest.kt)
   	at io.reactivex.internal.operators.flowable.FlowableMap$MapSubscriber.onNext(FlowableMap.java:64)
   	at io.reactivex.internal.operators.flowable.FlowableRefCount$RefCountSubscriber.onNext(FlowableRefCount.java:193)
   	at io.reactivex.internal.operators.flowable.FlowablePublish$PublishSubscriber.dispatch(FlowablePublish.java:545)
   	at io.reactivex.internal.operators.flowable.FlowablePublish$InnerSubscriber.request(FlowablePublish.java:615)
   	at io.reactivex.internal.operators.flowable.FlowableRefCount$RefCountSubscriber.request(FlowableRefCount.java:216)
   	at io.reactivex.internal.subscribers.BasicFuseableSubscriber.request(BasicFuseableSubscriber.java:153)
   	at io.reactivex.internal.operators.flowable.FlowableObserveOn$ObserveOnSubscriber.runAsync(FlowableObserveOn.java:407)
   	... 6 more
   Exception in thread "scheduler2" io.reactivex.exceptions.OnErrorNotImplementedException: Wrong thread on map #2: scheduler2
   	at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
   	at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
   	at io.reactivex.internal.subscribers.LambdaSubscriber.onError(LambdaSubscriber.java:79)
   	at io.reactivex.internal.operators.flowable.FlowableDoOnEach$DoOnEachSubscriber.onError(FlowableDoOnEach.java:111)
   	at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.checkTerminated(FlowableObserveOn.java:207)
   	at io.reactivex.internal.operators.flowable.FlowableObserveOn$ObserveOnSubscriber.runAsync(FlowableObserveOn.java:392)
   	at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.run(FlowableObserveOn.java:176)
   	at io.reactivex.internal.schedulers.ExecutorScheduler$ExecutorWorker$BooleanRunnable.run(ExecutorScheduler.java:261)
   	at io.reactivex.internal.schedulers.ExecutorScheduler$ExecutorWorker.run(ExecutorScheduler.java:226)
   	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
   	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
   	at java.lang.Thread.run(Thread.java:748)
   Caused by: java.lang.AssertionError: Wrong thread on map #2: scheduler2
   	at RxTestKt$main$flowable$2.apply(RxTest.kt:31)
   	at RxTestKt$main$flowable$2.apply(RxTest.kt)
   	at io.reactivex.internal.operators.flowable.FlowableMap$MapSubscriber.onNext(FlowableMap.java:64)
   	at io.reactivex.internal.operators.flowable.FlowableRefCount$RefCountSubscriber.onNext(FlowableRefCount.java:193)
   	at io.reactivex.internal.operators.flowable.FlowablePublish$PublishSubscriber.dispatch(FlowablePublish.java:545)
   	at io.reactivex.internal.operators.flowable.FlowablePublish$InnerSubscriber.request(FlowablePublish.java:615)
   	at io.reactivex.internal.operators.flowable.FlowableRefCount$RefCountSubscriber.request(FlowableRefCount.java:216)
   	at io.reactivex.internal.subscribers.BasicFuseableSubscriber.request(BasicFuseableSubscriber.java:153)
   	at io.reactivex.internal.operators.flowable.FlowableObserveOn$ObserveOnSubscriber.runAsync(FlowableObserveOn.java:407)
   	... 6 more
   

   I have two questions about this:

   1. Does 2.x: Fix boundary fusion of concatMap and publish operator #6145 also fix this situation?
   2. Is there any alternative workaround that can also fix this situation?
6. akarnokd commented on Aug 13, 2018

   

   akarnokd

   on Aug 13, 2018

   Member

   More actions

   Does #6145 also fix this situation?

   No, this is a different effect. Due to backpressure and buffering, the publish may service its consumer on a different thread than the generator upstream to it.

   Is there any alternative workaround that can also fix this situation?

   If you want that map to execute on the desired thread, you have to apply observeOn(scheduler1) before it.
7. damianw commented on Aug 13, 2018

   

   damianw

   on Aug 13, 2018

   Author

   More actions

   Thanks for the reply, @akarnokd. That makes sense - I will keep that in mind. Are there any existing mechanisms that allow something like publish to preserve the observation scheduling?
8. akarnokd commented on Aug 13, 2018

   

   akarnokd

   on Aug 13, 2018

   Member

   More actions

   No. It is an async boudary in which case all bets are off. You have to ensure the proper thread via observeOn.

   👍React with 👍1Reacted by Damian Wieczorek