dotnet / dotnet/reactive

Potential race condition in RefCount when using it with source that supports resubscription after error

Open
#2,297 6 comments 0 reactions 1 assignee Claimed by @idg10 View on GitHub
[area] Rx
Dominant language
C#
Stars
7.2k
Forks
798
PR merge metrics
No merged PRs in 30d

Description

#### Bug

> Which library version?
6.1.0, 7.0.0-preview.1

> What are the platform(s), environment(s) and related component version(s)?
I've tested on Linux x64 and arm64

> What is the use case or problem?

There appears to be race condition within `RefCount` implementation when using it like this:
```csharp
src.Multicast().RefCount().ObserveOn(TaskPoolScheduler.Default).Retry()
```
When error occurs, retry attempts will race with each other and potentially lead to a state when subscriptions exist but connection to source observable is not recreated. More complete example:

```csharp
// Simulate polling some remote device
var src = Observable.Create(async (observer, ct) =>
{
Console.WriteLine("START");
Int32 i = 0;
while (!ct.IsCancellationRequested)
{
Console.WriteLine(++i);
observer.OnNext(i);
await Task.Delay(500);

// Simulate connection failure
if (i > 5)
throw new InvalidOperationException();
}
})
.SubscribeOn(TaskPoolScheduler.Default);

src = src.Multicast(new StatelessSubject())
.RefCount();

const Int32 parallelCount = 12;
for (Int32 i = 0; i < parallelCount; i++)
{
Int32 i1 = i;
src
.ObserveOn(TaskPoolScheduler.Default)
.Do(_ => {},
ex => { Console.WriteLine($"Error {i1}"); }, // Simulate handling error in downstream consumer
() => Console.WriteLine("Completed"))
.Retry() // This retry can't just be moved before .Multicast since downstream consumers might need to know when errors occur (to reset it's own internal business logic, etc)
.Subscribe();
}

Thread.Sleep(-1);

/// "Dumb" subject that doesn't remember error state that I've taken from https://stackoverflow.com/a/64991229
public class StatelessSubject : ISubject, ISubject, IObserver, IObservable
{
private IImmutableList> _observers = ImmutableArray>.Empty;

public IDisposable Subscribe(IObserver observer)
{
ImmutableInterlocked.Update(ref _observers, x => x.Add(observer));
return Disposable.Create(() => ImmutableInterlocked.Update(ref _observers, (Func>, IImmutableList>>)(x => x.Remove(observer))));
}

public void OnNext(T value)
{
foreach (var observer in Volatile.Read(ref _observers))
observer.OnNext(value);
}

public void OnError(Exception error)
{
foreach (var observer in Interlocked.Exchange(ref _observers, []))
observer.OnError(error);
}

public void OnCompleted()
{
foreach (var observer in Interlocked.Exchange(ref _observers, []))
observer.OnCompleted();
}
}
```

With large enough `parallelCount` it will stuck after a few iterations.

> What is the expected outcome?

When error occurs in the source, I'd expect `RefCount` to atomically reset it's state (clear internal subscription to source and all observers), propagate error to "old" observers and be ready to immediately accept subscriptions from new observers, restarting source subscription anew.

> What is the actual outcome?

`RefCount` operator entering incorrect state, when it has active downstream subscriptions, but doesn't create upstream subscription.

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.