Observing elements on main thread after calling dispose on main thread

[stephan7]

First of all many thanks for your incredible work regarding RxSwift.

Short description of the issue:

Setup scenario:

• subscribeOn on some background thread
• observeOn on main thread
• dispose called on main thread

-> sometimes emissions on main thread happens after dispose is called on main thread

Expected outcome:

Intuitively, I would expect that if I call dispose on the same thread where I also observe that after disposing no further elements are emitted then.

I looked through the issues section and found this:

just to make this clear, if you call dispose on one thread (like main), you won't observe any elements on that same thread. That is a guarantee.

see here
#38

What actually happens:

Sometimes emissions on main thread happens after dispose is called on main thread.

Self contained code example that reproduces the issue:

A self-contained test case of another user can be found on Stackoverflow: https://stackoverflow.com/questions/51856962/dispose-cancel-observable-subscribeon-and-observeon-different-schedulers

Just change the line if !strongSelf.isCancelled { there to if strongSelf.isCancelled { (without negation).

Check missing?:
Searching through the code I found that in the class ObserveOnSerialDispatchQueueSink there is cachedScheduleLambda. One could add a test there for !self.cancel.isDisposed. That would prevent above mentioned emissions after a dispose happened.

But I have to admit that I don't understand RxSwift's internal processes enough to judge whether this could be a proper fix.

Or maybe there are even good reasons for it and the behavior is so intentional.

RxSwift/RxCocoa/RxBlocking/RxTest version/commit

4.2.0

Platform/Environment

• [x ] iOS
• macOS
• tvOS
• watchOS
• playgrounds

How easy is to reproduce? (chances of successful reproduce after running the self contained code)

• [x ] easy, 100% repro (just let the program run for some time)
• sometimes, 10%-100%
• hard, 2% - 10%
• extremely hard, %0 - 2%

Xcode version:

  Xcode 10

Installation method:

• [ x] CocoaPods
• Carthage
• Git submodules

I have multiple versions of Xcode installed:
(so we can know if this is a potential cause of your issue)

• yes (which ones)
• [x ] no

[kzaher]

Hi @stephan7 ,

ObserverBase implements the cancellation logic, so it should work.

A self-contained test case of another user can be found on Stackoverflow: https://stackoverflow.com/questions/51856962/dispose-cancel-observable-subscribeon-and-observeon-different-schedulers

This is not a self contained case. Self contained case means that I can copy the code, run it, and it will demonstrate the issue.

If you can post self contained code sample, I can take a look.

[stephan7]

Hi @kzaher ,

Thanks for taking a look. You are right, some glue code is missing, sry.

Here a complete example:

• in Xcode use the 'Single View App' template
• add a Podfile with the following content:

target 'RxSelfContained' do
  use_frameworks!
  pod 'RxSwift', '~> 4'	
end

• run pod install
• afterwards open the xxx.xcworkspace in Xcode
• replace the content of ViewController.swift with the following code:

import UIKit
import RxSwift


class ViewController: UIViewController {
    
    override func viewDidLoad() {
        super.viewDidLoad()
        
        MainClass.shared.main()
    }

}

//code coming from user3449629
//from here: https://stackoverflow.com/questions/51856962/dispose-cancel-observable-subscribeon-and-observeon-different-schedulers
class TestClass {
    private var disposeBag = DisposeBag()
    
    private var isCancelled = false
    
    init(cancelAfter: TimeInterval, longRunningTaskDuration: TimeInterval) {
        assert(Thread.isMainThread)
        
        load(longRunningTaskDuration: longRunningTaskDuration)
        
        DispatchQueue.main.asyncAfter(deadline: .now() + cancelAfter) { [weak self] in
            self?.cancel()
        }
    }
    
    private func load(longRunningTaskDuration: TimeInterval) {
        assert(Thread.isMainThread)
        
        // We set task not cancelled
        isCancelled = false
        
        DataService
            .shared
            .longRunngingTaskEmulation(sleepFor: longRunningTaskDuration)
            // We want long running task to be executed in background thread
            .subscribeOn(ConcurrentDispatchQueueScheduler.init(queue: .global()))
            // We want to process result in Main thread
            .observeOn(MainScheduler.instance)
            .subscribe(onSuccess: { [weak self] (result) in
                assert(Thread.isMainThread)
                
                guard let strongSelf = self else {
                    return
                }
                
                if strongSelf.isCancelled {
                    print("Should not be called! Task is cancelled!")
                } else {
                    // Do something with result, set image to UIImageView, for instance
                    // But if task was cancelled, this method will set invalid (old) data
                    print(result)
                }
                
                }, onError: nil)
            .disposed(by: disposeBag)
    }
    
    // Cancel all tasks. Can be called in PreapreForReuse.
    private func cancel() {
        assert(Thread.isMainThread)
        
        // For test purposes. After cancel, old task should not make any changes.
        isCancelled = true
        
        // Cancel all tasks by creating new DisposeBag (and disposing old)
        disposeBag = DisposeBag()
    }
}

class DataService {
    static let shared = DataService()
    
    private init() { }
    
    func longRunngingTaskEmulation(sleepFor: TimeInterval) -> Single<String> {
        return Single
            .deferred {
                assert(!Thread.isMainThread)
                
                // Enulate long running task
                Thread.sleep(forTimeInterval: sleepFor)
                
                // Return dummy result for test purposes.
                return .just("Success")
        }
    }
}

class MainClass {
    static let shared = MainClass()
    
    private init() { }
    
    func main() {
        
        Timer.scheduledTimer(withTimeInterval: 0.150, repeats: true) { [weak self] (_) in
            assert(Thread.isMainThread)
            
            let longRunningTaskDuration: TimeInterval = 0.050
            
            let offset = TimeInterval(arc4random_uniform(20)) / 1000.0
            let cancelAfter = 0.040 + offset
            
            self?.executeTest(cancelAfter: cancelAfter, longRunningTaskDuration: longRunningTaskDuration)
        }
    }
    
    var items: [TestClass] = []
    func executeTest(cancelAfter: TimeInterval, longRunningTaskDuration: TimeInterval) {
        let item = TestClass(cancelAfter: cancelAfter, longRunningTaskDuration: longRunningTaskDuration)
        items.append(item)
    }
}

Run it on a device or on a simulator.

In the console of Xcode you should see something like this:

Success
Success
Success
Success
Success
Should not be called! Task is cancelled!
Success
Success
Should not be called! Task is cancelled!
Success
Success

I can also provide a complete Xcode project if that is helpful.

[kzaher]

Hi @stephan7 ,

yes, you are correct, this is a bug.

[stephan7]

Hi,

Very cool and thanks for the fast fix. I can confirm that the bug is now fixed.

(For testing I simply changed the Podfile to the corresponding commit:

target 'RxSelfContained' do
  use_frameworks!
  pod 'RxAtomic', :git => 'https://github.com/ReactiveX/RxSwift.git', :commit => 'bac86346087c7e267dd5a620eed90a7849fd54ff'
  pod 'RxSwift', :git => 'https://github.com/ReactiveX/RxSwift.git', :commit => 'bac86346087c7e267dd5a620eed90a7849fd54ff'
end

)