Synchronization Context Rx.Net Async

[fedeAlterio]

Hi!
Will Rx.Net Async consider Synchronization Context?
For example, if I apply a Where operator with an async filter, exists a way to guarantee that the filters are posted on a Synchronization Context. Basically is it possible to decide wether or not to use ConfigureAwait(false) internally?
Thank you

[idg10]

Historically, Rx has never made any attempt to flow context of any kind. Sometimes calls are processed immediately (e.g., Where does all work for each item within OnNext on whatever thread you called that on, so the predicate is invoked on that thread, and if the predicate returns true, the downstream OnNext is also called on that thread), but even this is an implementation detail. In general there's no guarantee of this, and in any case where calls occur through a scheduler, there's no passing of context at all.

I believe previous maintainers of Rx have looked at this from time to time, and there would be non-trivial costs to flowing context that would have a negative impact on existing Rx users. We don't want to impose that overhead across the board because it would be supplying a feature that existing Rx users were definitely coping without.

We do have an open work item to look at this again in a future version of Rx to see if anything can be done to address the scenarios that lead to people asking for this. I think any solution would need to be an "edge" type thing, where context would be optionally reestablished as events leave the world of Rx, but I don't know if we could do it without explicitly flowing context through. E.g. you could imagine a ValueWithSynchronizationContext<T> that combined a value with a context, and then perhaps some specialized subscription mechanism. But we've not done any work on that yet.

Regarding Async Rx.NET, you said:

guarantee that the filters are posted on a Synchronization Context

What do you mean by "filters" here? Are you thinking specifically predicates for things like Where and TakeUntil?

Since Async Rx.NET offers overloads of those sorts of operators where the callbacks can be async (they can return a ValueTask<bool>) that does provide a way to force things to go via a synchronization context if you need them to, and you wouldn't need Async Rx.NET's help for that. Something like this

public static class AsyncEx
{
    public static Func<TIn, ValueTask<TResult>> OnContext<TIn, TResult>(
        this Func<TIn, ValueTask<TResult>> m,
        SynchronizationContext context)
    {
        return async arg =>
        {
            TaskCompletionSource<TResult> tcs = new();

            context.Post(
                async o =>
                {
                    (TIn a, Func<TIn, ValueTask<TResult>> m, TaskCompletionSource<TResult> c) = ((TIn, Func<TIn, ValueTask<TResult>>, TaskCompletionSource<TResult>))o!;
                    try
                    {
                        tcs.SetResult(await m(a));
                    }
                    catch (Exception x)
                    {
                        tcs.SetException(x);
                    }
                },
                (arg, m, tcs));

            return await tcs.Task;
        };
    }
}

Then if you have some predicate like this:

Func<int, ValueTask<bool>> isEven = async i =>
{
    Console.WriteLine($"Predicate invoked on {Environment.CurrentManagedThreadId}");
    return i % 2 == 0;
};

then given some SynchronizationContext called ctx you can use that extension method thus:

Func<int, ValueTask<bool>> isEvenOnCtx = isEven.OnContext(ctx);

and now isEvenOnContext invokes IsEven via the context. You can then pass that to Async RX.NET operators such as Where.

[fedeAlterio]

Thank you for your detailed response. Really appreciated. I'm a bit worried if the solution you proposed can suffer by frequent context switching between (for example) thread pool and the Synchronization Context used.
However if we use .ConfigureAwait(false) on the awaits, and we check if we are already in the current SC before calling post, we should fix even that problem, am I correct?
More generally we can imagine an overload called Where(filter, bool continueOnCallerContext) with an implementation similar to yours, considering the caller context at the same way the TaskAwaiter does (so, SC if not null, then TaskScheduler.Current if not default, otherwise threadPool). We can extend all other operators with the same technique, am I right?

I'm only a bit worried that this solution is a bit like reinventing what . ConfigureAwait already does