Reactive: IAsyncEnumerable<T> converted to IObservable<T> throws StackOverflowException

[chyczewski-maciej]

Hi,
An IObservable created from IAsyncEnumberable throws System.StackOverflowException
when there's a subscriber and enough items in the sequence.

What are the platform(s), environment(s) and related component version(s)?

• .net core 3.1 (sdk 3.1.100, runtime 3.1.0)
• Windows 10 x64
• Both debug and release builds

Which library version?

System.Reactive version 4.3.2
System.Linq.Async version 4.0.0

What is the expected outcome?

I'd expect that a sequence of an unlimited length would be supported.

What is the actual outcome?

After ~2780 items processed System.StackOverflowException is thrown

What is the stacktrace of the exception(s) if any?

I didn't manage to capture a stack trace before the exception gets thrown.

Do you have a code snippet or project that reproduces the problem?

Here's the simplest example I managed to come up with

Program.cs:

using System;
using System.Linq;
using System.Reactive.Linq;

namespace StackOverflowRepro
{
    public class Program
    {
        public static void Main(string[] args)
        {
            Enumerable
                .Range(0, 20_000)
                .ToAsyncEnumerable()
                .ToObservable()
                .Subscribe(Console.WriteLine);
            Console.ReadLine();
        }
    }
}

csproj:

<Project Sdk="Microsoft.NET.Sdk.Web">

  <PropertyGroup>
    <TargetFramework>netcoreapp3.1</TargetFramework>
    <OutputType>Exe</OutputType>
  </PropertyGroup>
  <ItemGroup>
    <PackageReference Include="System.Linq.Async" Version="4.0.0" />
    <PackageReference Include="System.Reactive" Version="4.3.2" />
  </ItemGroup>
</Project>

[akarnokd]

I can't reproduce it in an unit test but with the settings above, it does indeed fail due to some odd recursion:

   at System.Linq.AsyncEnumerable.ToObservableObservable`1.<>c__DisplayClass2_0.<Subscribe>g__Core|0()
   at System.Linq.AsyncEnumerable.ToObservableObservable`1.<>c__DisplayClass2_0.<<Subscribe>g__Core|0>d.MoveNext() in d:\a\1\s\Ix.NET\Source\System.Linq.Async\System\Linq\Operators\ToObservable.cs:line 57
   at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)
   at System.Linq.AsyncEnumerable.ToObservableObservable`1.<>c__DisplayClass2_0.<Subscribe>g__Core|0()
   at System.Linq.AsyncEnumerable.ToObservableObservable`1.<>c__DisplayClass2_0.<<Subscribe>g__Core|0>d.MoveNext() in d:\a\1\s\Ix.NET\Source\System.Linq.Async\System\Linq\Operators\ToObservable.cs:line 57
   at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)
   at System.Linq.AsyncEnumerable.ToObservableObservable`1.<>c__DisplayClass2_0.<Subscribe>g__Core|0()
   at System.Linq.AsyncEnumerable.ToObservableObservable`1.<>c__DisplayClass2_0.<<Subscribe>g__Core|0>d.MoveNext() in d:\a\1\s\Ix.NET\Source\System.Linq.Async\System\Linq\Operators\ToObservable.cs:line 57
   at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)
   at System.Linq.AsyncEnumerable.ToObservableObservable`1.<>c__DisplayClass2_0.<Subscribe>g__Core|0()
   at System.Linq.AsyncEnumerable.ToObservableObservable`1.<>c__DisplayClass2_0.<<Subscribe>g__Core|0>d.MoveNext() in d:\a\1\s\Ix.NET\Source\System.Linq.Async\System\Linq\Operators\ToObservable.cs:line 57
   at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)
   at System.Linq.AsyncEnumerable.ToObservableObservable`1.<>c__DisplayClass2_0.<Subscribe>g__Core|0()
   at System.Linq.AsyncEnumerable.ToObservableObservable`1.<>c__DisplayClass2_0.<<Subscribe>g__Core|0>d.MoveNext() in d:\a\1\s\Ix.NET\Source\System.Linq.Async\System\Linq\Operators\ToObservable.cs:line 57
   at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)
   at System.Linq.AsyncEnumerable.ToObservableObservable`1.<>c__DisplayClass2_0.<Subscribe>g__Core|0()
   at System.Linq.AsyncEnumerable.ToObservableObservable`1.<>c__DisplayClass2_0.<<Subscribe>g__Core|0>d.MoveNext() in d:\a\1\s\Ix.NET\Source\System.Linq.Async\System\Linq\Operators\ToObservable.cs:line 57
   at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)
   at System.Linq.AsyncEnumerable.ToObservableObservable`1.<>c__DisplayClass2_0.<Subscribe>g__Core|0()
   at System.Linq.AsyncEnumerable.ToObservableObservable`1.<>c__DisplayClass2_0.<<Subscribe>g__Core|0>d.MoveNext() in d:\a\1\s\Ix.NET\Source\System.Linq.Async\System\Linq\Operators\ToObservable.cs:line 57
   at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)
   at System.Linq.AsyncEnumerable.ToObservableObservable`1.<>c__DisplayClass2_0.<Subscribe>g__Core|0()
   at System.Linq.AsyncEnumerable.ToObservableObservable`1.Subscribe(IObserver`1 observer) in d:\a\1\s\Ix.NET\Source\System.Linq.Async\System\Linq\Operators\ToObservable.cs:line 75
   at System.ObservableExtensions.Subscribe[T](IObservable`1 source, Action`1 onNext, Action`1 onError) in d:\a\1\s\Rx.NET\Source\src\System.Reactive\Observable.Extensions.cs:line 95
   at ConsoleApp1.Program.Main(String[] args) in C:\Users\akarnokd\source\repos\ConsoleApp1\ConsoleApp1\Program.cs:line 30

The source has shifted since the release so this is the failure line actually.

[akarnokd]

The example works with another ToObservable implementation:

async_enumerable_dotnet.AsyncEnumerable.ToObservable(
                    Enumerable.Range(0, 20_000)
                    .ToAsyncEnumerable()
                )
                .Subscribe(v =>
                {
                    if (v == 500)
                    {
                        Console.WriteLine(Environment.StackTrace);
                    }
                    Console.WriteLine(v);
                }, e => Console.WriteLine(e));

            Console.ReadLine();

[akarnokd]

Okay, looks like 4.0.0 uses an outdated implementation for ToObservable which recurses. Somehow the changes were not cherry-picked back in September.

[danielcweber]

There's other benefits in the newer implementation of ToObservable as well that unfortunately didn't see a preview release yet. @onovotny Could you please trigger a preview build for Ix.Async ?

[clairernovotny]

I plan on getting a new stable build out of both Ix/Rx out very soon. Is this fix in master?

[danielcweber]

The changed implementation of ToObservable is from #915 and should fix this issue as well. Also, #915 is on master.

[clairernovotny]

Perfect, thanks for confirming. I'll get a release out today.

[bartdesmet]

This was released as far as I can tell, so closing the issue.