Fundamental waste of threads plus potential thread leaks

[Petikoch]

At the moment, there is a fundamental flaw in the examples...

Look e.g. at https://github.com/Petikoch/Java_MVVM_with_Swing_and_RxJava_Examples/blob/master/src/main/java/ch/petikoch/examples/mvvm_rxjava/example2/Example_2_ViewModel.java

Every

@Override
public void connectTo(final Example_2_Model model) {
    onEventFrom(vm2m_nameFirstname).execute(model::createAccount);
}

is basically just a typical RxJava

@Override
public void connectTo(final Example_2_Model model) {
    Subscription sub1 = vm2m_nameFirstname.observeOn(Schedulers.io()).subscribe(model::createAccount);
}

This immediately assigns 1 thread from the CachedThreadScheduler (Schedulers.io()). The 1 thread stays assigned until "someone" would call sub1.unsubscribe().

But nobody every calls sub1.unsubscribe() here in these examples. Even if the Example_2_ViewModel instance is not used anymore and subject to garbage collection, the assigned 1 thread stays.

Therefor exists two fundamental issues in the examples:

1. Waste of resources: Imagine a new item "only" every 10 seconds in the vm2m_nameFirstname subject. It doesn't make sense to keep 1 real thread for all the time here "ready". Imagine you have in total thousands of subjects in your hundreds of viewmodels. For every observeOn(Schedulers.io()) you pay 1 thread. Therefor you'll have thousands of threads. It would be better to just submit something like a Runnable on every arriving item to something like a cached thread pool. Do we see here the price of the "Rx is single threaded by default" design?

2. Thread leaks: If the Example_2_ViewModel instance is dereferenced and subject to garbage collection, nobody does the unsubscribe and "releases" the assigned thread(s).

[Petikoch]

Potential solution: Instead of Schedulers.io() use a Schedulers.from(cachedThreadPoolExecutor) and keep the cachedThreadPoolExecutor instance somewhere "central" as singleton due the lifetime of the application. Don't forget to either shutdown it explicitely at application stop request or use daemon threads in it (with it's consequences of JVM shutdown behaviour).

[Petikoch]

Other potential solution: Don't use observeOn(..) to decouple the GUI in terms of threading from the "slow" model (backed) calls, use something like RxJavaAsyncUtil (https://github.com/ReactiveX/RxJavaAsyncUtil/blob/0.x/src/main/java/rx/util/async/Async.java)

That's probably the better solution.

@mikebaum Do you have some insights on this?

[mikebaum]

@Petikoch Hello from Montreal.

Interesting stuff, need to look at your example more thoroughly to provide a good answer. I'll have a look on the weekend... My first impression is that AsyncUtil could help since it auto-unsubscribes via completing the stream when the background work is finished. Anyway, I'll get back to you soon.

[Petikoch]

Hi @mikebaum,

after sleeping two nights I came to the conclusion, that both of the above described solutions are needed in a typical RxJava project. For the particular use case of subscribing to a "ViewModel-Subject" and firing of backend (model) calls on events at the end of a pipeline/stream like in this project, I switched to the "AsyncUtil" approach.

After doing some search on this topic, I came across some very interesting discussions like ReactiveX/RxJava#2448.

I think the current implementation of the CachedThreadScheduler is "supersafe" regarding "Rx is single threaded by default". There will be always the same dedicated thread doing the work, and the same dedicated thread will be always there and ready, even if no events or very few events ever pass. The thread will be released and returned "back to the pool" not before the unsubscribe. In a use case like here, these threads (heavyweight resources) are "just hanging around for minutes or hours" and wasting resources. Wouldn't be a problem if it were fibers (lightweight threads).

The benefits of this "supersafety" is that you won't ever have concurrency issues within your code. But the price is high. In an application with a functional / immutable code style, you don't care about thread-hopping (different threads from a pool executing your code), but you care about thread-wasting like CachedThreadScheduler currently does IMO.

What do you think?

Best regards from Switzerland,
Peti

[mikebaum]

@Petikoch Looked at the example a little bit more. Here's my initial feedback:

1. One thing that immediately stood out, is there is no feedback to the UI when performing an async operation. So you have no way of blocking the UI in order to prevent the user from accidentally performing another model save with stale data.
2. If the operation fails it seems the failure is just swallowed somehow. I tested this by making the createAccount method fail every third time and I saw no logging or anything else to indicate something bad happened.

I have more feedback, but since I don't have a lot of time at this moment, it needs to wait. I'll offer a pull request shortly that has an example of how you could change the way you execute your model updates, to both provide feedback to the UI and handle exceptions.

Cheers,

Mike

[mikebaum]

@Petikoch Just to be clear, the pull request is not meant to be merged. I just used that mechanism to pass an example to you. You might also want to look at: reactive command and/or reactive command async. Another interesting technology to consider is Vert.x. Maybe you're already aware of that library, but if not it is something akin to a lightweight version of Akka that uses an EventBus to connect what they call Verticles (Actors). They offer some kind of Verticle that uses fibres, as you mentioned earlier. The really cool thing they have is a distributed, clustered EventBus. Using this event bus makes it dead simple to connect actors running in separate processes. I haven't played with it too much, but it looks great.

[Petikoch]

@mikebaum, thanks for the feedback!

I'll reflect on this next friday and give you feedback.