4.x: Reimplement Timeout(TimeSpan) with lock-free methods #546
Conversation
|
Can we first have these Disposable-Helper-methods with SetSingle, SetSerial, SetMultiple, GetIsDisposed, Dispose taking those ref-parameters? I am currently working on this, it will reduce a lot of repetitive code, especially those volatile fields everywhere and lots of Interlocked stuff. This will be a lot cleaner afterwards and more intention revealing wrt. the disposable fields. |
|
Yes, I was planning on doing that. We can have just the helper merged, then a separate PR can go over all classes and change to them. |
|
Ok, I am working on the helper currently. |
|
Helper was merged. |
| public override void OnNext(TSource value) | ||
| { | ||
| var idx = Volatile.Read(ref _index); | ||
| if (idx != long.MaxValue && Interlocked.CompareExchange(ref _index, idx + 1, idx) == idx) |
There was a problem hiding this comment.
How many messages would it have to produce per second for how many years to actually break this?
There was a problem hiding this comment.
If you could produce 1 message per nanosecond, this would overflow in 500+ years.
|
|
||
| Interlocked.Exchange(ref _cancel, null)?.Dispose(); | ||
| } | ||
|
|
||
| protected void ClearObserver() |
There was a problem hiding this comment.
Because _cancel holds a SingleAssignmentDisposable which holds this in itself and would call Dispose again on the implementing instance. In fact, any newer Run implementation returning this is prone to this wasteful call.
There was a problem hiding this comment.
That's an issue I have been trying to wrap my head around for some time now. Can we defer this change for later, going with the wasteful call for now, because it would add more complexity to the Sink and I am thinking around ways to lower complexity of Sink.
This PR reimplements the
Timeout.Relativeoperator with a lock-free set of methods. The atomic index change by one ensures that the main source won; when the index atomically changes tolong.MaxValuethe timeout won. Emissions after these transitions are guaranteed to be thread-safe thus no lock is needed.The PR also inlined the two
SingleAssignmentDisposables and theStableCompositeDisposablereturn. However, to avoid a cyclicDisposedue to thecancelinput saved inSink, I have exposed just the clearing of the downstream observable and otherwise theDispose(bool)disposes the inlinedIDisposables.If this style is acceptable, the other variants will be updated to lock-free as well in a separate PR.