2.x: distinguish between sync and async dispose in ScheduledRunnable

[akarnokd]

This PR adds logic to distinguish between synchronous and asynchronous dispose calls when setFuture is executing. It should prevent interrupting the currently running task body if it requested cancellation indirectly before the setFuture was executed by the Thread which scheduled the task.

Fixes #5711

[codecov]

Codecov Report

Merging #5715 into 2.x will decrease coverage by 0.04%.
The diff coverage is 88.88%.

@@             Coverage Diff              @@
##                2.x    #5715      +/-   ##
============================================
- Coverage     96.28%   96.24%   -0.05%     
- Complexity     5822     5825       +3     
============================================
  Files           634      634              
  Lines         41604    41609       +5     
  Branches       5761     5761              
============================================
- Hits          40060    40048      -12     
- Misses          616      622       +6     
- Partials        928      939      +11

Impacted Files	Coverage Δ	Complexity Δ
...activex/internal/schedulers/ScheduledRunnable.java	96.22% <88.88%> (+0.48%)	29 <4> (+2)	⬆️
...nternal/operators/observable/ObservableCreate.java	94.87% <0%> (-3.42%)	2% <0%> (ø)
...a/io/reactivex/internal/util/QueueDrainHelper.java	60.28% <0%> (-2.84%)	32% <0%> (-2%)
...l/operators/observable/ObservableFlatMapMaybe.java	90.19% <0%> (-2.62%)	2% <0%> (ø)
...rnal/operators/flowable/FlowableFlatMapSingle.java	94.56% <0%> (-2.18%)	2% <0%> (ø)
.../io/reactivex/internal/schedulers/IoScheduler.java	92.47% <0%> (-2.16%)	9% <0%> (ø)
...rnal/operators/flowable/FlowableSkipLastTimed.java	95.91% <0%> (-2.05%)	2% <0%> (ø)
...rnal/subscriptions/DeferredScalarSubscription.java	98.46% <0%> (-1.54%)	28% <0%> (-1%)
.../operators/observable/ObservableFlatMapSingle.java	91.04% <0%> (-1.5%)	2% <0%> (ø)
...ternal/operators/completable/CompletableMerge.java	96.42% <0%> (-1.2%)	2% <0%> (ø)
... and 20 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 3f9ffae...049f6f9. Read the comment docs.

[artem-zinnatullin]

LGTM, but comments

[artem-zinnatullin]

supernit: ASYNC doesn't mean another thread :)

But for name conciseness I guess it's fine

[artem-zinnatullin]

Change of compareAndSet position in if statement seems to be a behavior change…

Though it doesn't affect ScheduledRunnableTest on my machine

Do we need that change?

[akarnokd]

Yes. With the previous order, if o was null because of lack of a parent, the CAS didn't happen and the task would partially appear to be still active. With the sync-async marker changes, isDisposed was switched to check the parent reference for indication of being disposed instead of the future reference, which required now 3 comparisons instead of 2.

[artem-zinnatullin]

Makes sense, can you please add a test to cover that then?

When I changed order back to original one on your branch, whole ScheduledRunnableTest passed

[akarnokd]

Sure.

[akarnokd]

Unit tests added.

[artem-zinnatullin]

Hm, what worries me here is that we base boolean async on get(THREAD_INDEX) != Thread.currentThread(), but in finally of run() we set Thread to null which always gonna give async == false

RxJava/src/main/java/io/reactivex/internal/schedulers/ScheduledRunnable.java

Line 67 in b99efbf

lazySet(THREAD_INDEX, null);

Yes, it's lazySet(), but aren't we creating unwanted race condition here between run() and dispose()?

The worst scenario is that if dispose() indeed was called from another thread but lazySet() memory write becomes visible to it, we'll have async == false which is actually wrong in this scenario…

[akarnokd]

Cancelling from the current thread could only happen while the run callback is executing. Any other cancellation that comes after this can only be asynchronous. It doesn't matter what value is in THREAD_INDEX at this time because that will not be the same thread as the caller's.

Since we are running with single-threaded thread pools, any subsequent task on the same pool having the same Thread will find the references marked as DONE and not cancel the previous task. This is guaranteed by the sequential task processing of the thread pool.

[artem-zinnatullin]

It doesn't matter what value is in THREAD_INDEX at this time because that will not be the same thread as the caller's.

Aha, I see now, 👍