Observable.Interval().Take() hangs with ImmediateScheduler
- Dominant language
- C#
- Stars
- 7.2k
- Forks
- 798
- PR merge metrics
- No merged PRs in 30d
Description
Library version: 6.0.0
This is arguably just an unsupported scenario, since `Observable.Interval` is designed to work with schedulers suitable for time-based operation, and `ImmediateScheduler` is not, in general, suitable for that.
However, there's a scenario that people might reasonably expect to work:
```cs
IObservable source = Observable.Interval(TimeSpan.FromSeconds(1), ImmediateScheduler.Instance)
.Take(5);
source.Subscribe(i => Console.WriteLine($"OnNext({i}): {DateTime.Now}"));
Console.WriteLine($"Finished: {DateTime.Now}");
```
This initially seems to work, producing this output:
```
OnNext(0): 02/08/2023 08:56:22
OnNext(1): 02/08/2023 08:56:23
OnNext(2): 02/08/2023 08:56:24
OnNext(3): 02/08/2023 08:56:25
OnNext(4): 02/08/2023 08:56:26
```
But it then hangs. The call to `Subscribe` never returns. And internally, the interval continues merrily ticking away, delivering notifications into a `Take` operator that is no longer listening.
Although the `ImmediateScheduler` is not designed for timed operations, it does still implement the time-based `IScheduler` methods, and as the output above shows, they do work. They block, not returning until the specified time elapses, because that's what `ImmediateScheduler` does. (Well, a `DateTimeOffset`-based scheduler for more than 10 seconds into the future is not handled immediately, but that's not relevant to this issue.) But they do call you when you asked them to.
It's not totally unreasonable to expect this code to work. `Take` normally unsubscribes from its source once it has received the specified number of elements. Its source here is the `Interval`, so you'd expect that to stop trying to schedule any further work, meaning the `ImmediateScheduler` would be able to return, since all scheduled work has completed.
But it turns out that in this scenario, `Take` does _not_ unsubscribe from its source once it has received the 5th element. And that's because of this code here:
https://github.com/dotnet/reactive/blob/5903ac6ace7956fd550189af2d7620a479733c02/Rx.NET/Source/src/System.Reactive/Linq/Observable/Timer.cs#L134-L147
These calls to `SetUpstream` are how the `Timer` (which is what `Interval` is using internally) arranges to get shut down when its subscriber unsubscribes. Logically, it is setting itself as the upstream source for `Take` in this example. Normally, when `Take` hits its count, it calls `OnComplete` on its subscriber, which then triggers a tear down of the subscription, at which point `Take` disposes its upstream, which is what's supposed to notify `Take`'s source that it can stop.
That normally all works fine, but it doesn't work with `ImmediateScheduler`, because those calls to `SetUpstream` can't happen until the call to `Schedule` returns. The argument to `SetUpstream` in each case here is a call to the `parent._scheduler`. But if that's an `ImmediateScheduler`, it won't return until it's done. So those calls to `SetUpstream` can't occur until after the scheduler determines that it has no more work to do, but it can't discover that there's no more work to do if the calls to `SetUpstream` hasn't happened yet.
On the one hand, the basic problem here is that we're using a scheduler that doesn't work in the way `Interval` requires, so this could be dismissed as a non-bug.
On the other hand, this raises the question of whether a race condition exists in supported scenarios in which that `SetUpstream` might not occur before a downstream sink tries to shut down the timer, but fails to do so because it did that before the call to `Schedule` returns. E.g., if we're using a timer-friendly scheduler, might it determine that it can actually execute the timed work immediately? Or more subtly, what if it schedules work on a queue, then the system bogs down, and the thread that called `Schedule` doesn't return up the stack fast enough, and the scheduled work manages to run and then tries to shut down the subscription, but the timer runs forever because the call to `SetUpstream` happened _after_ the sink we're calling `SetUpstream` on has already stopped. We might be OK because it might be using a disposable that handles this race correctly, but we should at least check.
So it would be worth verifying that the possible race is handled correctly with non-immediate schedulers. It's also worth reviewing the rational for the way the `SchedulePeriodic` extension method emulates periodic scheduling on schedulers that don't inherently support it—it was changed at some point in the past, and there's a lengthy and complex explanation of the thinking behind that change. But I think that fixing this for the immediate scheduler would be tricky because in the current design for periodic scheduling, the only way to cancel a periodically scheduled work item is to `Dispose` the object returned by `SchedulePeriodic`, and that's fundamentally incompatible with immediate scheduling. This is why the `ImmediateScheduler` is, in general, considered unsuitable for use in timer-based sources, so this is probably just going to be resolved as "by design".
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.