IAsyncEnumerable support in Dataflow

[manandre]

Contributes to #30863

cc @onionhammer @stephentoub

[Dotnet-GitSync-Bot]

Note regarding the new-api-needs-documentation label:

This serves as a reminder for when your PR is modifying a ref *.cs file and adding/modifying public APIs, to please make sure the API implementation in the src *.cs file is documented with triple slash comments, so the PR reviewers can sign off that change.

[manandre]

I am not proud of the ToEnumerable extension method. Do you see a way to enhance or even remove it?

[stephentoub]

Thanks for your interest in working on this.

I did a quick initial pass through and left some comments to be addressed. I'll do a more in-depth pass subsequently.

[manandre]

@stephentoub I have also reworked the test part but there is a test failure on Helix that I do not manage to analyze nor reproduce locally. Can you please help me?

[ghost]

Tagging subscribers to this area: @tarekgh
See info in area-owners.md if you want to be subscribed.

[tarekgh]

@manandre it looks this PR parked for long time. could you please force the CI rerun on it again just to run against the latest? you can do that by adding any trivial commit to it. Also, what was pending on this change?

@stephentoub do you have any more comments on this PR?

[manandre]

I have rebased on master branch but the build still fails on a unit test failure that I managed to reproduce locally (by running it in loop until failure):

E:\git\runtime\artifacts\bin\System.Threading.Tasks.Dataflow.Tests\net6.0-Debug>E:\git\runtime\artifacts\bin\testhost\net6.0-windows-Debug-x86\dotnet exec --runtimeconfig System.Threading.Tasks.Dataflow.Tests.runtimeconfig.json --depsfile System.Threading.Tasks.Dataflow.Tests.deps.json xunit.console.dll System.Threading.Tasks.Dataflow.Tests.dll -xml testResults.xml -nologo -nocolor -notrait category=IgnoreForCI -notrait category=OuterLoop -notrait category=failing -method System.Threading.Tasks.Dataflow.Tests.DataflowBlockExtensionTests.ReceiveAllAsync*
  Discovering: System.Threading.Tasks.Dataflow.Tests (method display = ClassAndMethod, method display options = None)
  Discovered:  System.Threading.Tasks.Dataflow.Tests (found 11 of 335 test cases)
  Starting:    System.Threading.Tasks.Dataflow.Tests (parallel test collections = on, max threads = 8)
    System.Threading.Tasks.Dataflow.Tests.DataflowBlockExtensionTests.ReceiveAllAsync_MultipleEnumerationsToEnd [FAIL]
      System.InvalidOperationException : An attempt was made to transition a task to a final state when it had already completed.
      Stack Trace:
        E:\git\runtime\src\libraries\System.Private.CoreLib\src\System\Runtime\CompilerServices\AsyncTaskMethodBuilderT.cs(443,0): at System.Runtime.CompilerServices.AsyncTaskMethodBuilder`1.SetExistingTaskResult(Task`1 task, TResult result)
        E:\git\runtime\src\libraries\System.Private.CoreLib\src\System\Runtime\CompilerServices\AsyncIteratorMethodBuilder.cs(62,0): at System.Runtime.CompilerServices.AsyncIteratorMethodBuilder.Complete()
        E:\git\runtime\src\libraries\System.Threading.Tasks.Dataflow\src\Base\DataflowBlock.IAsyncEnumerable.cs(33,0): at System.Threading.Tasks.Dataflow.DataflowBlock.<ReceiveAllAsync>g__ReceiveAllAsyncCore|43_0[TOutput](IReceivableSourceBlock`1 source, CancellationToken cancellationToken)+MoveNext()
        E:\git\runtime\src\libraries\System.Private.CoreLib\src\System\Runtime\CompilerServices\AsyncMethodBuilderCore.cs(42,0): at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)
           at System.Threading.Tasks.Dataflow.DataflowBlock.<ReceiveAllAsync>g__ReceiveAllAsyncCore|43_0[TOutput](IReceivableSourceBlock`1 source, CancellationToken cancellationToken)+System.Collections.Generic.IAsyncEnumerator<TOutput>.MoveNextAsync()
        E:\git\runtime\src\libraries\System.Threading.Tasks.Dataflow\tests\Dataflow\DataflowBlockExtensionTests.IAsyncEnumerable.cs(166,0): at System.Threading.Tasks.Dataflow.Tests.DataflowBlockExtensionTests.ReceiveAllAsync_MultipleEnumerationsToEnd()
        --- End of stack trace from previous location ---
  Finished:    System.Threading.Tasks.Dataflow.Tests
=== TEST EXECUTION SUMMARY ===
   System.Threading.Tasks.Dataflow.Tests  Total: 19, Errors: 0, Failed: 1, Skipped: 0, Time: 0,787s

Unfortunately, I do not understand what happens here. I would need some help.

[tarekgh]

@stephentoub could you please help @manandre with his question in #37876 (comment)?

[stephentoub]

I have rebased on master branch but the build still fails on a unit test failure that I managed to reproduce locally (by running it in loop until failure)
Unfortunately, I do not understand what happens here. I would need some help.

That error suggests that something is causing the continuations in the async iterator to be executed multiple times, likely due to a bug in the implementation of the blocks as part of the changes made in this PR.

In the name of moving this PR forward, I suggest scoping it back to the simpler ReceiveAllAsync and project updates, rather than also supporting the much more complicated and problematic new constructors.

[manandre]

@stephentoub Unfortunately, these random unit test failures only occur with the ReceiveAllAsync* tests, which rely on the already existing BufferBlock class (and not modified by this PR).

[stephentoub]

Unfortunately, these random unit test failures only occur with the ReceiveAllAsync* tests, which rely on the already existing BufferBlock class (and not modified by this PR).

Can you narrow the PR down to just ReceiveAllAsync, its tests, and the minimal project changes necessary to add the netstandard2.1 configuration? (i.e. leave out the rest of the TransformXx block tests). It's harder to prove that nothing else relevant has changed when there's so much other stuff here.

If it still repros after that, it should be easier to diagnose.

Thanks.

[manandre]

PR narrowed down to ReceiveAllAsync support. The random issue is still present. Any clue on how to diagnose such an issue?

[stephentoub]

@eerhardt, how should the project changes here interact with the project changes you made to dataflow?

[eerhardt]

how should the project changes here interact with the project changes you made to dataflow?

They are conflicting. I also added a $(NetCoreAppCurrent) build to this project in #48667.

[stephentoub]

I also added a $(NetCoreAppCurrent) build to this project in #48667.

I see that https://github.com/dotnet/runtime/pull/48667/files#diff-721cfca869eb5f5a10caa48506edc358e2edac219c8b5a2e123855e40fd9fd84R4 excludes it from the package, but that's just about the implementation assembly, and that doesn't matter because it's in box? What should be done for a ref here?

EDIT: I see your comment above. So the correct change for this PR is just to add the netstandard2.1 ref assembly?

[eerhardt]

So the correct change for this PR is just to add the netstandard2.1 ref assembly?

I believe the correct change for this PR is to add the netstandard2.1 build for both the ref and the implementation assembly.

You want it in the impl assembly as well, because that goes into the NuGet package.

[stephentoub]

You want it in the impl assembly as well, because that goes into the NuGet package.

Ah, of course.

[stephentoub]

Thanks!