Reactive: StartAsync is blocking during the execution of the synchronous part of its delegate action

[davidnemeti]

The asynchronous StartAsync is blocking during the execution of the synchronous part of its delegate action, i.e. StartAsync blocks until the execution of its delegate action gets to the first await.

This causes a different (and worse) behavior than its synchronous counterpart (Start) has.

This means that the following code:

```c#
async Task Main()
{
IObservable slowObservable = Observable.Range(1, 5)
.Select(number => Observable.Start(() => Slow(number)))
.Merge();

IObservable<int> slowAsyncThatIsActuallySyncObservable = Observable.Range(1, 5)
    .Select(number => Observable.StartAsync(() => SlowAsyncThatIsActuallySync(number)))
    .Merge();

IObservable<int> slowAsyncThatIsActuallySyncWithTrickObservable = Observable.Range(1, 5)
    .Select(number => Observable.Start(() => SlowAsyncThatIsActuallySync(number)).Concat())
    .Merge();

IObservable<int> slowRealAsyncObservable = Observable.Range(1, 5)
    .Select(number => Observable.StartAsync(() => SlowRealAsync(number)))
    .Merge();

IObservable<int> slowSyncThenAsyncObservable = Observable.Range(1, 5)
    .Select(number => Observable.StartAsync(() => SlowSyncThenAsync(number)))
    .Merge();

IObservable<int> slowAsyncThenSyncObservable = Observable.Range(1, 5)
    .Select(number => Observable.StartAsync(() => SlowAsyncThenSync(number)))
    .Merge();

Console.WriteLine("Slow");
Console.WriteLine("====");
await slowObservable;
Console.WriteLine();

Console.WriteLine("SlowAsyncThatIsActuallySync");
Console.WriteLine("===========================");
await slowAsyncThatIsActuallySyncObservable;
Console.WriteLine();

Console.WriteLine("SlowAsyncThatIsActuallySyncWithTrick");
Console.WriteLine("====================================");
await slowAsyncThatIsActuallySyncWithTrickObservable;
Console.WriteLine();

Console.WriteLine("SlowRealAsync");
Console.WriteLine("=============");
await slowRealAsyncObservable;
Console.WriteLine();

Console.WriteLine("SlowSyncThenAsync");
Console.WriteLine("=================");
await slowSyncThenAsyncObservable;
Console.WriteLine();

Console.WriteLine("SlowAsyncThenSync");
Console.WriteLine("=================");
await slowAsyncThenSyncObservable;
Console.WriteLine();

}

int Slow(int number)
{
Console.WriteLine($"Slow BEGIN: {number}");
Thread.Sleep(TimeSpan.FromSeconds(1));
Console.WriteLine($"Slow END: {number}");
return number * 10;
}

async Task SlowAsyncThatIsActuallySync(int number)
{
Console.WriteLine($"SlowAsyncThatIsActuallySync BEGIN: {number}");
Thread.Sleep(TimeSpan.FromSeconds(1));
Console.WriteLine($"SlowAsyncThatIsActuallySync END: {number}");
return number * 10;
}

async Task SlowRealAsync(int number)
{
Console.WriteLine($"SlowRealAsync BEGIN: {number}");
await Task.Delay(TimeSpan.FromSeconds(1));
Console.WriteLine($"SlowRealAsync END: {number}");
return number * 10;
}

async Task SlowSyncThenAsync(int number)
{
Console.WriteLine($"SlowSyncThenAsync BEGIN: {number}");
Thread.Sleep(TimeSpan.FromSeconds(1));
await Task.Delay(TimeSpan.FromSeconds(1));
Console.WriteLine($"SlowSyncThenAsync END: {number}");
return number * 10;
}

async Task SlowAsyncThenSync(int number)
{
Console.WriteLine($"SlowAsyncThenSync BEGIN: {number}");
await Task.Delay(TimeSpan.FromSeconds(1));
Thread.Sleep(TimeSpan.FromSeconds(1));
Console.WriteLine($"SlowAsyncThenSync END: {number}");
return number * 10;
}

produces this output:

Slow

Slow BEGIN: 2
Slow BEGIN: 1
Slow BEGIN: 3
Slow BEGIN: 4
Slow BEGIN: 5
Slow END: 3
Slow END: 2
Slow END: 1
Slow END: 4
Slow END: 5

SlowAsyncThatIsActuallySync

SlowAsyncThatIsActuallySync BEGIN: 1
SlowAsyncThatIsActuallySync END: 1
SlowAsyncThatIsActuallySync BEGIN: 2
SlowAsyncThatIsActuallySync END: 2
SlowAsyncThatIsActuallySync BEGIN: 3
SlowAsyncThatIsActuallySync END: 3
SlowAsyncThatIsActuallySync BEGIN: 4
SlowAsyncThatIsActuallySync END: 4
SlowAsyncThatIsActuallySync BEGIN: 5
SlowAsyncThatIsActuallySync END: 5

SlowAsyncThatIsActuallySyncWithTrick

SlowAsyncThatIsActuallySync BEGIN: 1
SlowAsyncThatIsActuallySync BEGIN: 3
SlowAsyncThatIsActuallySync BEGIN: 2
SlowAsyncThatIsActuallySync BEGIN: 4
SlowAsyncThatIsActuallySync BEGIN: 5
SlowAsyncThatIsActuallySync END: 4
SlowAsyncThatIsActuallySync END: 2
SlowAsyncThatIsActuallySync END: 3
SlowAsyncThatIsActuallySync END: 5
SlowAsyncThatIsActuallySync END: 1

SlowRealAsync

SlowRealAsync BEGIN: 1
SlowRealAsync BEGIN: 2
SlowRealAsync BEGIN: 3
SlowRealAsync BEGIN: 4
SlowRealAsync BEGIN: 5
SlowRealAsync END: 5
SlowRealAsync END: 1
SlowRealAsync END: 3
SlowRealAsync END: 4
SlowRealAsync END: 2

SlowSyncThenAsync

SlowSyncThenAsync BEGIN: 1
SlowSyncThenAsync BEGIN: 2
SlowSyncThenAsync END: 1
SlowSyncThenAsync BEGIN: 3
SlowSyncThenAsync BEGIN: 4
SlowSyncThenAsync END: 2
SlowSyncThenAsync BEGIN: 5
SlowSyncThenAsync END: 3
SlowSyncThenAsync END: 4
SlowSyncThenAsync END: 5

SlowAsyncThenSync

SlowAsyncThenSync BEGIN: 1
SlowAsyncThenSync BEGIN: 2
SlowAsyncThenSync BEGIN: 3
SlowAsyncThenSync BEGIN: 4
SlowAsyncThenSync BEGIN: 5
SlowAsyncThenSync END: 1
SlowAsyncThenSync END: 3
SlowAsyncThenSync END: 5
SlowAsyncThenSync END: 4
SlowAsyncThenSync END: 2

The problematic part is the `SlowAsyncThatIsActuallySync` section in the output, where we can see that the `SlowAsyncThatIsActuallySync` method executions that are started with the `StartAsync` method run **sequentially**, while the `Slow` method executions started with the `Start` method run **concurrently**.

This is due to the fact that `SlowAsyncThatIsActuallySync` is actually a synchronous method.

However, note that the problem would still persist if we would try to start a mixed synchronous-asynchronous method (where by "mix" I mean that it has both slow synchronous and slow asynchronous part).

E.g. in case of `SlowSyncThenAsync` (where the slow *synchronous* part comes **before** the *asynchronous* part) the slow **synchronous part blocks**, while the slow **asynchronous part does not**.

However, in case of `SlowAsyncThenSync` (where the slow *synchronous* part comes **after** the *asynchronous* part) **neither the slow asynchronous part nor the slow synchronous part blocks**. So, if the asynchronous part comes first, **there is no blocking** at all.

This is because the `StartAsyncImpl` method (which is used by `StartAsync`) **does not use** its `scheduler` parameter to schedule the execution of the **synchronous** part of its `functionAsync` (thus it blocks); it only **uses** its `scheduler` parameter to schedule the execution of the **asynchronous** part of its `functionAsync` (by passing it to `ToObservable` which is called on `task`), thus making the execution of the "whole" `functionAsync` "half-scheduled":

```c#
        private IObservable<TSource> StartAsyncImpl<TSource>(Func<Task<TSource>> functionAsync, IScheduler scheduler)
        {
            var task = default(Task<TSource>);
            try
            {
                task = functionAsync();
            }
            catch (Exception exception)
            {
                return Throw<TSource>(exception);
            }

            if (scheduler != null)
            {
                return task.ToObservable(scheduler);
            }
            else
            {
                return task.ToObservable();
            }
        }

In contrast, ToAsync (which is used by Start) uses its scheduler parameter to schedule the execution of its function parameter (which, of course, does not have separate synchronous and asynchronous parts, since the whole function is synchronous; thus the "whole" function is "fully-scheduled"):

c# public virtual Func<IObservable<TResult>> ToAsync<TResult>(Func<TResult> function, IScheduler scheduler) { return () => { var subject = new AsyncSubject<TResult>(); scheduler.Schedule(() => { var result = default(TResult); try { result = function(); } catch (Exception exception) { subject.OnError(exception); return; } subject.OnNext(result); subject.OnCompleted(); }); return subject.AsObservable(); }; }

I propose that the StartAsyncImpl method should use its scheduler parameter to schedule the execution of the synchronous part of its functionAsync.

[akarnokd]

Looks like you could use Defer + SubscribeOn to move the initial function call onto the background thread:

static IObservable<T> CreateAndStartAsync<T>(Func<Task<T>> func, IScheduler scheduler) {
    return Observable.Defer(() => Observable.StartAsync(func, scheduler)).SubscribeOn(scheduler);
}

With the upcoming extension anything feature of C#, you will be able to "patch" Observable yourself with this static method.

[davidnemeti]

Of course, I can implement workarounds until the real fix is done for StartAsync.

E.g. I can use the trick that I wrote in my previous comment (hot version):

```c#
Observable.Range(1, 5)
.Select(number => Observable.Start(() => SlowAsyncThatIsActuallySync(number)).Concat())
.Merge();

Or I can use yours (cold version, which, of course, is better since I can control the concurrency by passing the `maxConcurrent` parameter to the `Merge` method):

```c#
    Observable.Range(1, 5)
        .Select(number => CreateAndStartAsync(() => SlowAsyncThatIsActuallySync(number), Scheduler.Default))
        .Merge();

But, actually, this CreateAndStartAsync "patch":

```c#
static IObservable CreateAndStartAsync(Func> func, IScheduler scheduler) {
return Observable.Defer(() => Observable.StartAsync(func, scheduler)).SubscribeOn(scheduler);
}

is just a workaround (and would not be needed) for the improperly behaving `FromAsync`:

```c#
        public virtual IObservable<Unit> FromAsync(Func<Task> actionAsync, IScheduler scheduler)
        {
            return Defer(() => StartAsync(actionAsync, scheduler));
        }

which behaves improperly because the used StartAsync behaves improperly.

But StartAsync should be fixed anyway.

[akarnokd]

As far as I can remember, StartAsync behaves like this for at least 8 years by now and your request is a drastic behavior change. What if there is code out there where the function execution is expecting the current thread to be of the caller's so it can access some thread-confined resource before the processing task is started for it?

[davidnemeti]

Yes, I realize that it behaves like this for a long time, and honestly I find it quite shocking that nobody has complained about it. It caused me a lot of confusion till I realized this inconsistency.

Until then, I had been desperately trying to add several schedulers as parameter for StartAsync/FromAsync, add ObserveOn at different places with several schedulers as parameter (while the Start works fine without explicitly specifying a scheduler). Then I looked into the source code of Rx, and realized that no wonder StartAsync/FromAsync do not work as expected.

I also realize that it would be a drastic change in behavior, just as any bugfix is that fixes a fundamentally wrong behavior.

However, I think my arguments are strong enough:

• The behavior of StartAsync is inconsistent with the behavior of Start regarding the execution of the synchronous part of StartAsync's delegate action.
• The behavior of StartAsync is inconsistent even with itself, since the execution of a long-running synchronous code in the synchronous part of StartAsync's delegate action behaves differently from the execution of a long-running synchronous (or asynchronous) code in the asynchronous part of StartAsync's delegate action, because the former blocks while the latter does not.

However, if introducing a breaking change like this is not acceptable, then at least we could introduce overloads for StartAsync/FromAsync which can control the scheduling of both the synchronous and asynchronous part:

```c#
public static IObservable StartAsync(Func actionAsync, IScheduler schedulerSyncPart, IScheduler schedulerAsyncPart)

```c#
public static IObservable<Unit> FromAsync(Func<Task> actionAsync, IScheduler schedulerSyncPart, IScheduler schedulerAsyncPart)

and rename the scheduler parameter of the existing methods so that they would emphasize that they control the scheduling only for the asynchronous part:

```c#
public static IObservable StartAsync(Func actionAsync, IScheduler schedulerAsyncPart)

```c#
public static IObservable<Unit> FromAsync(Func<Task> actionAsync, IScheduler schedulerAsyncPart)

Although, this solution would not provide us the fixed behavior for the already existing StartAsync/FromAsync methods, however, by looking at the parameters the developer at least can get a hint about the behavior of the scheduling.

How does that sound?

BTW, am I really the only one who finds this behavior of StartAsync strange/wrong? @bartdesmet, could you please share your thoughts on this topic? Thanks!

[davidnemeti]

@onovotny, @bartdesmet, could you please join into the conversion? I would really like to hear your opinions about this topic. Thanks!

[davidnemeti]

So, by pure chance, I came across the newest version of WithLatestFrom method. There is a new comment on it which says:

```c#
/// Starting from Rx.NET 4.0, this will subscribe to before subscribing to to have a latest element readily available
/// in case emits an element right away.

Indeed, the following code:

```c#
var xs = Observable.Return(1);
var ys = Observable.Return("bar");
var zs = xs.WithLatestFrom(ys, (x, y) => x + y);

produces different results for zs in Rx.NET 4.0 than in Rx.NET 3.0.

In Rx.NET 3.0 zs would be an empty observable, while in Rx.NET 4.0 zs would be an observable with a single item 1bar. I would call it a "drastic behavior change".

The change has been implemented in pull request #152, and now is part of Rx.NET 4.0.

Now, it was not a bugfix, rather the reason for this change was something like "most of the time, the expected behavior is..." (BTW, I totally agree both with the change and with the argument), and it has been acknowledged in the pull request that this was a breaking change.

Thus, I feel the counterargument to my proposed change of StartAsyncImpl so that it would be a drastic behavior change is not satisfactory anymore. (I mean, it surely is a drastic change, but so is the change in WithLatestFrom. So, a change being a drastic change should not be a counterargument on its own to implementing the change.)

@onovotny, @bartdesmet, any thoughts on this topic?

[DaRosenberg]

I just also spent a good deal of time in confusion over this behavior.

I agree with @davidnemeti that this should be fixed. The fact that it's been broken for a long time is not a valid reason to let it stay broken. Nor is the fact that workarounds exist - devs will spend hours before they even realize a workaround is needed, because this is simply counter-intuitive.

As @davidnemeti mentions, much more drastic changes than this have happened since. :) And it's not like this necessarily has to go into a patch release either.