Reactive: Being more async / Thread Pool friendly

[bnayae]

We should try to be more Thread Pool and async friendly by boosting
the shifting to non-blocking api.
for example:

• usage of lock can be replace by SemaphoreSlim and await gate.WaitAsync + gate.Release (try finally).
• Action / Action should be replaced with Func / Func or return of some task-like interface.

This means that we may need to introduce some newer interface for Rx (like Bart did with ITaskObservable).
Here some examples:

case 1:
void OnNext(T item)
may have to be changed to
Task OnNext(T item)
on this kind of API we can operate async operation and maintain the sequence without
having to block the current thread (which may happens to the a Thread Pool's thread).

case 2:
IScheduler may change the API from:
IDisposable Schedule(TState state, TimeSpan dueTime, Func action);
to something like:
Task Schedule(TState state, TimeSpan dueTime, CancellationToken, Func action);

[tamirdresher]

+1

[RxDave]

There may be another way to look at case 1, as follows:

void doesn't necessarily imply blocking, it implies fire-and-forget; therefore, Task only adds unnecessary information that is going in the wrong direction (observer -> observable). In other words, it's similar to the backpressure problem, because if we only need Task to ensure that notifications are pushed serially, that implies that we're worried about the producer being faster than the consumer.

Perhaps Task isn't the correct approach here. To avoid blocking a thread, can't we simply use ObserveOn to change the context?

Ultimately, observables aren't beholden to observers, it's the other way around, right?

And regarding backpressure, instead of defining an observable that must respect all asynchronous downstream computations in observers, create an observable __A__ that composes another observable __B__, whereby __B__ provides information to __A__ regarding when it should push notifications; e.g., sampling or rate throttling. That way, an observer of __A__ can also be the generator of __B__, and essentially control __A__ without breaking the "reactive" model; i.e., full-duplex channels.

[akarnokd]

Actually, it is possible to go lock free for most operators, but the general tradeoff is between allocation and holding off producers. The economic solution is the Reactive-Streams-like co-routines and integrated flow control.

Task OnNext(T item)

I think this requires allocation per item, even if there is no blocking needed or happening.

Task Schedule(TState state, TimeSpan dueTime, CancellationToken, Func action);

I can't think of any of the standard operator which would await a Task returned by the Schedule methods. For most of the time, only the cancellation of such "tasks" is necessary because the downstream disposed the stream.

[tamirdresher]

@RxDave i agree with what you're saying but the reality is that many libraries expose their API as asynchronous operations and using them inside the pipeline is not ideal

[RxDave]

I actually like the Task-returning IScheduler idea; although, perhaps some new extension methods are good enough?

I've defined my own async IScheduler extensions in the past and found them useful. Sometimes, I've even used them outside of Rx, as a simple asynchronous scheduler abstraction.

[RxDave]

@tamirdresher The Task-based overloads of SelectMany usually do the trick, no?

[tamirdresher]

@RxDave absolutely, but still, not ideal and less readable

[shiftkey]

case 1:
void OnNext(T item)
may have to be changed to
Task OnNext(T item)

Cross-referencing a related discussion on this: #141

[bartdesmet]

I'm freeing up cycles to tackle some design work in the next weeks/months after my current activity of porting some lessons learned from our internal service to Rx v4.0 (especially with regards to allocations), trying to converge implementation aspects on both ends slowly.

One thing we may want to look at is the IAsyncObserver<T> interface. For our service here, we ended up having this, but also got SyncToAsyncObserver<T> bridge code, because we kept the event processing pipeline itself synchronous. Prior to the introduction of ValueTask<T> and task-like types, we found a default implementation using async when used with practically 99.9% of callbacks being async wrappers for synchronous operations (e.g. Where(x => Task.FromResult(f(x)))) to be prohibitively expensive. I'm planning to revisit this analysis with newer async facilities.

We can also re-rationalize the schedulers in v4 given the move to .NET Core and the removal of legacy platforms, causing many facilities (such as "long running" and "stopwatch") to be available everywhere, thus leading to a far less common usage of the (academically satisfying but performance limiting) recursive signatures. Our service here has quite a different approach to scheduling and I'm looking at how we can converge matters.

[akarnokd]

How much are you willing to rearchitect?

For one, the Scheduler+Worker pair we use in RxJava could be worth investigating (it resembles the IEnumerable/IEnumerator design).

Second, injecting IDisposable upfront into an observer (via an OnSubscribe(IDisposable)) instead of just returning one after, allowing synchronous cancellation and avoid the need for scheduling just to have the IDisposable returned from IObservable.Subscribe(). Unfortunately, since IObserver is baked into the language, a new interface has to be introduced and dispatched on within Subscribe().

[RxDave]

@akarnokd Injecting an IDisposable doesn't help because it only has a Dispose method, without any mechanism to notify that it's been disposed. What you're proposing is similar to the pattern used for Task cancellation, thus CancellationToken could be used instead; however, there is already an overload for this pattern defined for Subscribe:

public static void Subscribe<T>(this IObservable<T> source, CancellationToken token)

Similarly, there is an overload of Observable.Create that accepts a CancellationToken in addition to an observer. And naturally, the subscribe function that you must supply returns Task, as this is the pattern for injecting cancellation; i.e., defining a coroutine in C#.

IObservable<T> Create<T>(Func<IObserver<T>, CancellationToken, Task> subscribeAsync)

These methods can be used together or separately. Do they satisfy your suggestion?

[akarnokd]

That might work for end consumers but I'm not sure it would be economic for intermediate operators. With IDisposable, you can create an operator class extending IObserver and IDisposable, have exactly one instance of it and send it downstream. Often a call to Dispose only has to set a bool indicating cancellation or call another (set of) IDisposable(s). I've tried it before and it works (see a source, operator, consumer example).

[RxDave]

Point taken about conflating the disposables and the observers to reduce the graph size; however, I think that may just be an implementation detail. You can do something similar without changing the pattern from IDisposable as a return value to a parameter. Perhaps the only catch when conflating observers with disposables is that all operators have to check if the observer that is passed to them implements IDisposable, and then simply dispose of it upon termination of their subscription. Therefore, each operator would composite the disposable of the former operator. Seems like a reasonable optimization considering that in Rx we shouldn't be implementing IObservable<T> ourselves anyway, and instead rely on the Rx-provided Create method, and in some edge cases the base classes, which would all behave properly.

As for synchronous cancellation, perhaps the parameterized pattern will be more efficient than inserting a trampoline, but I wonder whether it's significant enough of a difference to be worth changing the API?

[RxDave]

Here's a gist to prove my point about conflation not requiring an inversion of control:

https://gist.github.com/RxDave/6c18831a0014ab04806efacace022aea

[akarnokd]

There are 2 problems with that: dispose travels downstream instead of upstream and to implement Range & Take you still need trampolining & somehow getting the IDisposable at the level of Take.

[RxDave]

I don't see how it's a problem that the disposable is passed downstream, although as previous stated, I agree that the trampoline is still required if we care at all about synchronous disposal, but I wonder whether that's significant enough for an entire API rewrite.

[akarnokd]

I wonder whether that's significant enough for an entire API rewrite

It's a significant undertaking indeed (as it was for RxJava 1 & 2) so it is up to the designer(s)/contributor(s) of Rx.NET to do it for 4.x or not. I believe the reduced memory usage, less indirection and less need for asynchrony just for the sake of cancellation is worth the effort in C# as well.

[RxDave]

Just to be clear, I think there are two separate issues here:

1. Conflating an observer with its disposables relieves GC pressure.
2. The trampoline is currently necessary for synchronous cancellation.

2 is required because Subscribe returns IDisposable. To avoid this, IDisposable must be passed to Subscribe instead.

1 can be done either way. Whether we pass IDisposable to Subscribe or not.

I'd like to know what other people think about 1. Can it be done in Rx with minimal changes required?

I'm much less certain that 2 is worth the effort. I don't think that synchronous cancellation is all that common, and what is the real cost of the trampoline anyway? How much memory/performance will be gained by removing it?

[akarnokd]

Closing due to inactivity and possibly out-of-date given 4.x.