RxJava and race conditions

Question

I have seen some RxJava projects using multithreading with .subscribeOn(Schedulers.io()), and their .subscribe(() -> {...}) code is manipulating some data structures that are also processed by other Observable subscription handlers, and there are no synchronized() anywhere inside subscribe.

Coming from C/C++, I started to wonder - wait, if those subscribes happen to execute on different threads, then what about race conditions?

In C/C++ it would crash and burn with some access violation sooner or later. Of course, Java seems to be much more tolerant. But what if two subscribes happen to work with the same linked list in parallel? Won't they mess up the links between nodes? Shouldn't I protect them with synchronized()?

I searched for articles about RxJava and race conditions, but didn't find much. One article claims the following:

So, whenever you want to switch threads for any particular observable, all you need to do is specify the observeOn() operator just above it. Synchronization, state inconsistencies, race conditions, and all other threading edge cases are automatically handled under the hood.

Is it true and RxJava indeed somehow automagically protects its subscribes for race conditions with other subscribes working on the same data? Or is the author talking only about race conditions with RxJava itself or specifically about observeOn and not subscribeOn?

Answer 1

Unless someone can come along and explain how RxJava ensures that access to non-threadsafe structures is done in a threadsafe manner, I would assume that this is not threadsafe.

I'm not familiar with RxJava but absent any explicit guarantees, you should only depend on the language guarantees. There is another aspect that you need to consider here aside from race conditions: the JVM spec allows each thread to maintain it's own cache of the heap (or some subset of it.) That is if two threads have been passed references to a single object, they can both have completely independent copies of it. There are no requirements that changes to these local copies ever be communicated to other threads or shared memory by default. The program must force that through the use of synchronized blocks or other mechanisms.

However, as described here, final fields do not change and therefore do not need to be synchronized. This is one reason I strongly advocate the use of immutable objects as much as possible. This eliminates many opportunities for threading issues.

So in the specific example of a LinkedList, if you are accessing it from multiple threads, you should synchronize access to it. I'm not going to completely rule out the idea that there's some way that this is managed by RxJava but it doesn't seem likely to me.

It's important to understand that synchronized is not the only way to handle threadsafe access to memory in Java. That was true in early versions of Java but a number of other threading primitives and utilities were added quite a while ago. The synchronized keyword is fast and effective but has a major downside in that it is a bottleneck for thread execution and leads to contention. For this reason, you won't see it used much in Java code written after more threading control options were introduced.

If you have a significant number of threads accessing this structure concurrently, synchronizing on a common object will make the access safe but may defeat a lot of the point of the concurrent processing. You might want to look at using a threadsafe structure such as a BlockingQueue or CopyOnWriteArrayList. Depending on the situation, you may also want to consider using more advanced threading primitives such as Semaphores, CountDownLatches, and CyclicBarriers. I would recommend starting by looking at the JavaDocs for the 'java.util.concurrent' package and the sub-packages 'atomic' and 'locks'.

It's important as well to understand that using a threadsafe collection doesn't mean your algorithm automatically becomes threadsafe. The guarantees can vary between classes but in general, this just means that atomic operations will be done in a threadsafe manner. For example if you pull a value based on a key, increment it, and then write it back, a threadsafe map will not protect against dirty read/write scenarios.

Answer 2

Depending on the scheduler used and the scheduler's configuration regarding thread-pool size, the operators (map, filter and similar) may be executed concurrently. This is a useful feature.

To explicitly avoid parallel execution one may use the serialize operator above the thread-unsafe observer code.

Example:

Observable.fromArray(1, 2, 3)
        .subscribeOn(Schedulers.computation())
        .serialize()
        .map(i -> i * 2) // some thead-unsafe operation here
        .subscribe(i -> System.err.println(i));  // or here