Concurrency issues with observable.observeOn() and common resources

user_10092068 0
Viewed 143

I have an observable inside a function. The function happens in a certain queue, queueA, and the observable is subscribed to with observeOn(schedulerB). In onNext, I'm changing a class variable.

In another function, I'm changing the same class variable, from a different queue.

Here is some code to demonstrate my situation:

class SomeClass {

    var commonResource: [String: String] = [:]
    var queueA = DispatchQueue(label: "A")
    var queueB = DispatchQueue(label: "B")
    var schedulerB = ConcurrentDispatchQueueScheduler(queue: QueueB)       

    func writeToResourceInOnNext() {
        let obs: PublishSubject<String> = OtherClass.GetObservable()
        obs.observeOn(schedulerB)
           .subscribe(onNext: { [weak self] res in
            // this happens on queue B
            self.commonResource["key"] = res
        }
    }

    func writeToResource() {
        // this happens on queue A
        commonResource["key"] = "otherValue"
    }
}

My question is, is it likely to have concurrency issues, if commonResource is modified in both places at the same time?

What is the common practice for writing/reading from class/global variables inside onNext in an observable with observeOn?

Thanks all!

reactivex rx-swift swift concurrency
1 Answers

Since your SomeClass has no control over when these functions will be called or on what threads the answer is yes, you are setup to have concurrency issues in this code due to its passive nature.

The obvious solution here is to dispatch to queue B inside writeToResource() in order to avoid the race condition.

Another option would be to use an NSLock (or NSRecursiveLock) and lock it before you write to the resource and unlock it after.

The best practice is: when you have a side effect happening inside a subscribe function's closure (in this case writing to commonResource that the closure is the only place where the side effect occurs. This would mean doing away with the passive writeToResource() function and instead passing in an Observable that was generated by whatever code currently is calling the function.

user_5064410
Related