Repository navigation
Description
RxJava version: 2.1.11
Java: 1.8.0_181
I'm encountering an intermittent deadlock in a rather long Flowable, and I believe I've pinpointed it to an observeOn(...) call. (I've reached this conclusion through a series of log statements.) I haven't been able to trace through the test when the deadlock occurs, as it only occurs about once every 30 - 40 executions, and each execution takes about a minute. I've managed to reproduce the deadlock about a dozen times (each time, I've been adding more logging to figure out where things are getting stuck).
Flowable<SomeType> flow = // Lots of stuff upstream
flow.doOnNext(x -> /* log 1 */)
.doOnComplete(() -> /* log 2 */)
.observeOn(Schedulers.io(), true)
.doOnNext(x -> /* log 3 */)
.doOnComplete(() -> /* log 4 */)
// Lots more downstream
In the test case where I experience the occasional deadlock, I expect only 1 item to be emitted through this part of the Flowable. I can see log statements 1 and 2 indicating that the item reaches the observeOn(...) and that the upstream is finished, but logs 3 and 4 are never reached. (I forgot to add a doOnError(...) to make sure an exception isn't sneaking through and holding things up else where, but I'm fairly confident there aren't any uncaught exceptions. I've added a doOnError(...) and am re-running my test now to make sure; I'll update my post once I have results.)
Because the logs are hit in this way, this leads me to believe the observeOn(...) is locking up somehow. What's really strange is that everything works fine most of the time.
None of the downstream operators should be attempting to dispose the Flowable early either. I believe my terminal operator is a blockingGet() on a Single, no timeout or anything.
I also logged the total number of threads in my JVM to see if I'm leaking threads somewhere, but I'm only at 49 when the deadlock occurs. (I'm using the IO scheduler, which I believe is backed by an unbounded pool, so I can't imagine I would be running out of worker threads.) I do have other Flowables doing unrelated tasks in the background. All of those Flowables are using the IO scheduler. Additionally, the up and down stream of the Flowable in my test also make use of the IO scheduler, but the deadlock always seems to happen here.
I realize that's not a lot of info to go off of, but I figured I'd ask the experts in case there's something glaring that I'm missing, or if there's something else I can do to figure out what's going on.
Thanks in advance!
Activity
akarnokd commentedon Aug 8, 2018
What you are describing should not happen unless you either call onNext concurrently in the upstrem or your downstream does not request. Please provide a minimal runnable code example demonstrating the problem.
jkarshin commentedon Aug 8, 2018
The sub-Flowable where I'm seeing the deadlock looks like this:
(I call it a "sub-Flowable" to distinguish it from a larger surrounding Flowable whose code I have omitted.)
That sub-Flowable is called in a loop, which is called in the lambda of a
map(...)operation on another larger Flowable. (I realize that callingblockingGet()within a lambda passed to a Flowable is bad, but could that cause the deadlock that I'm seeing?)In my test, the sub-Flowable that I pasted is successfully executed a few times, but very rarely, one of the later invocations will hang. For the iterations that do not hang, I can see all of the log statements being hit. For the iteration that does hang, I can see the logs before the
observeOn(...)being hit, but none of the 3 logs after it are hit.I was also able to use VisualVM to do a thread dump. Most of the threads have names like
RxCachedThreadScheduler-#andRxComputationThreadPool-#, but judging by their stack traces, most seem to be waiting in their respective pools:There are 5 threads that are blocked/waiting that aren't simply sitting in a pool. One of them seems to be waiting for the
blockingGet()call at the end of the sub-Flowable I described:The other 4 blocked threads seem unrelated, but I can briefly describe them.
The result of the sub-Flowable I pasted is being used to write to a cache; two of the other blocked threads are attempting to clear that cache, and it is known that the clear operation will block until any pending writes complete. The cache sits above this sub-Flowable, so the sub-Flowable can't reference it (which would lead to deadlock).
One of the other blocked threads is waiting for the cache clear operation from the other threads to succeed. The final blocked thread contains the terminal
blockingGet()at the end of the Flowable that surrounds the Flowable I pasted.The relevant logs look like this:
The only thing notable about these logs is that the logs before the
observeOn(...)always come from RxCachedThreadScheduler-21, which is the thread that seems blocked insideSingle.blockingGet(...)(whose thread dump I included above).The remaining memory in the JVM seems fine as well.
Unfortunately, I can't really give you the entire test to reproduce the problem, as it is quite large and part of a proprietary code base. (And, you'd have to run the test over and over again for about an hour on average to see the deadlock.)
I'll work on writing a smaller test that I can share with you that hopefully reproduces the deadlock more consistently, but so far I haven't had any luck.
akarnokd commentedon Aug 8, 2018
It is very likely
blockingGet()is at least part of the problem, especially if you call it from a simplemap. You could be blocking the thread servicing the parent flow that should trigger something that makes the inner flow complete and thus unblocking theblockingGet.davidmoten commentedon Aug 9, 2018
@jkarshin I'm interested in this, a unit test would be great 👍
jkarshin commentedon Aug 9, 2018
Minor update:
I've added a
doOnCancel(...),doOnRequest(...), anddoOnSubscribe(...)right below thecompose(...)call that triggers theobserveOn(...). When it hangs, I see:This seems to suggest that the downstream is successfully subscribing and requesting an item, and that it is not cancelling.
I'm still working on making the apparent deadlock easier to reproduce. I haven't been able to get a small unit test to show the same problem, but I have written a slightly smaller integration test that reproduces the deadlock more frequently. (Still too huge for me to share any useful code, but I'm working on it.)
jkarshin commentedon Aug 14, 2018
I converted the Flowable in my previous code snippet into a unit-testable Flowable in an attempt to reproduce the hang in a unit test, however, I can not get it to hang.
I then cloned RxJava to insert log statements into the internals of the
observeOn()logic and look for discrepancies between the logging of the unit test and the hanging production code.Long story short, I discovered that
NewThreadWorker.scheduleActual(...)is submitting a Runnable that never gets executed. (I decorated thedecoratedRunRunnable with some log statements and discovered that it does not get executed during the iteration of my production code that hangs.) The logs indicate that theRunnabledoes get submitted to the executor, but it almost seems as if the executor is blocked by some other task. (The executor only has a single thread in its pool.) I repeatedly logged the output of thetoString()method on the hanging executor in intervals of one second, and I get the same output every second:I don't know a lot about
ScheduledThreadPoolExecutor'stoString()method, but the fact that there is always an active thread and a queued task leads me to believe that something is hanging causing theobserveOn'sdecoratedRunnableto never be executed. (I believe if there were only a single submitted task, there would be either an active thread or a queued task, but not both.)This brings up a question:
For the IO Scheduler, the executors are cached so that they can be reused, correct? Is it possible that an executor can be recycled even if it hasn't completed its task? If so, the following situation might be causing my deadlock:
I'll try to capture the above situation in a unit test later this week.
akarnokd commentedon Aug 14, 2018
Try with NewThread so worker reuse is not in the picture. Also don't block and don't use any traditional wait-notify. Confine cache management into one thread.
jkarshin commentedon Aug 17, 2018
After much effort, I am unable to reproduce the deadlock in a unit test.
With more logging, I've discovered that when the deadlock occurs in my production code, a
ThreadWorkeris being reused. ThisThreadWorker'sScheduledExecutorService'stoString()output indicates that it is still executing a thread, even though it had been returned to theCachedWorkerPool.My initial thought was that the thread responsible for clearing one of my caches was somehow getting put back into the pool, but further logging indicates that this is not the case: I added logging to determine the name of the thread that was in the ScheduledExecutorService mentioned above, and surprisingly, it matched the name of the thread from which I was doing the logging. (Which was not the thread that would have been blocked clearing my cache.)
This leads me to believe that because I am nesting Flowables, Singles, Completables, etc, and doing
observeOn()s andblockingGet()s in those nested Publishers (for lack of a better collective term), it is possible that aThreadWorkercan be recycled even though it is still in use. And by random chance, if one of those workers is picked up within the flow that recycled it, a deadlock will occur. A thread dump verified that the thread that the recycled worker was supposed to use was waiting in ablockingGet()call.My next course of action is going to be to refactor my code base to try to eliminate these nested blocking calls and see if that eliminates the deadlock problem. That will probably take a few weeks, so feel free to close this ticket in the meantime; I can re-open it and renew my investigation if the problem persists.
Thanks again for the help.
akarnokd commentedon Aug 17, 2018
If a task executing on the standard schedulers does not respond to interruptions, it may lead to premature worker reuse.
blockingGetusesCountDownLatch.awaitwhich can be interrupted, thus when the parent worker is disposed, it should unblock theblockingGetand throw anInterruptedException.Unfortunately, waiting for a worker to run out of tasks may lead to memory leaks or excess pool growth (i.e., the self-release can't keep up with the request for more workers).
Without seeing into the actual code, I don't think we can help you much. I'd suggest using a dedicated, single-threaded scheduler to manage the cache and retrieve items non-blockingly:
jkarshin commentedon Aug 21, 2018
I was too stubborn to give up on this, so I pressed on and made some progress:
Here are some logs that correspond to what I was describing in my previous post:
Basically, the logs indicate that some work is done on thread 37, then it is returned to the pool, then more work is done on thread 37 (without taking 37 out of the pool), then finally thread 37 tries to take itself out of the pool, then the deadlock occurs. Additionally, with a thread dump, I can see that thread 37 is waiting in a blockingGet.
Normally, when a thread is put into the pool, it isn't used until a different thread takes it out of the pool. Ex:
Nowhere else in the logs does a thread attempt to check itself out of the pool, nor does a thread continue to do work after it has been returned to the pool (without being taken out of the pool).
This makes me think I've got some combination of RxJava operators that is causing a thread to be
used even after it has been returned to the pool. The thread eventually gets stuck in a blockingGet, waiting on the upstream to complete, while the upstream can randomly try to take that same thread out of the pool, leading to deadlock.
I was also able to get a stack trace where thread 37 is returned to the pool prematurely:
(The
RuntimeException: nullbit is from me callingnew RuntimeException().printStackTrace(...))Unfortunately, the stack trace doesn't reference my code at all, so I can't pinpoint what operators I am using. I did find it strange that there are a few calls to
cancel()in that trace (I don't think I'm explicitly cancelling anything). If I had to guess, one of the RxJava operators that I'm using is doing an implicitcancel()(causing the thread to be returned to the pool), and then continuing to use the thread.I'll try to reverse engineer the stack trace to find what combination of operators is causing that thread to get returned to the pool. If I can find the combination of operators, hopefully I can turn that into a unit test that consistently shows a thread being returned to the pool prematurely.
akarnokd commentedon Aug 21, 2018
I think I know what's going on.
observeOnsignals a completionconcatswitches to the next source, which starts with ajustoperatorSubscriptionArbitercancels the previous subscription, which is completingobserveOnreceives cancel, releases the workerconcatstarts a blocking operation still on the current thread (released)On your side, you could try that the
justthat ends up blocking also receives asubscribeOn(Schedulers.io()).I'll think about the
observeOnandSubscriptionArbiter's behaviors.jkarshin commentedon Aug 22, 2018
I believe you are correct. I found a piece of my code similar to what you described, and I was able to turn it into a unit test that consistently returns a worker thread to the pool prematurely. Using a
blockingGet(), I can make this deadlock consistently as well:RxJava: 2.1.11
The above hangs consistently on my machine and yields logs similar to the following:
(Replacing the
.map(... blockingGet ...)with aflatMapSingle(...)fixes the deadlock caused by re-using the prematurely recycled thread, but this isn't necessarily a viable solution for me, since that nested Single is hidden behind a sub-service whose interface I'd rather not change.)Running against the 2.x branch of RxJava after the changes you made in response to my issue (#6167) fixes the hanging unit test and seems to fix my hanging production code. So, I will wait for the next RxJava release =)
Thanks again!