2.x: FlowableTimeoutTimed enhancement

[davidmoten]

I'm debugging a production problem where a timeout did not fire so having a close look at FlowableTimeoutTimed. I notice in TimeoutTimedSubscriber that multiple terminal events can be emitted to the downstream subscriber in race between the scheduled timeout runnable and the onError and onComplete methods. The volatile done flag doesn't provide complete protection unless we turn it into an AtomicBoolean and use compareAndSet.

The other thing I notice is that the timer field is accessed synchronously (only via onSubscribe and onNext) so rather than an AtomicReference can be a mutable field.

I can make PR if you think these are valid.

[davidmoten]

The onNext method would need protection as well.

[akarnokd]

See FlowableTimeoutTimed:51.

[davidmoten]

Ah good, thanks. And the timer field doesn't need to be AtomicReference?

BTW prod problem was external API not defined properly so all good.

[akarnokd]

You mean this and this?

[davidmoten]

Yep

…

On Tue, 4 Jul. 2017, 20:06 David Karnok, ***@***.***> wrote: You mean this <https://github.com/ReactiveX/RxJava/blob/2.x/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTimed.java#L191> and this <https://github.com/ReactiveX/RxJava/blob/2.x/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTimed.java#L232> ? — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#5461 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AATa60XJZUhH8eEth64QvWz2sD1DuFaFks5sKg6zgaJpZM4OMxK8> .

[akarnokd]

Yeah, doesn't really need AtomicReference in its current algorithm.

[davidmoten]

I'll fix

[akarnokd]

Btw, there is a more modern algorithm for this operator that doesn't need an additional layer of serialization.

[akarnokd]

Closing via #5466