Add UnfoldResource ValueTask Stage and DSL methods

docs/articles/streams/builtinstages.md

@@ -113,6 +113,14 @@ Can be used to implement many stateful sources without having to touch the more
 
 **completes** when the task returned by the unfold function completes with an null value
 
+### UnfoldValueTaskAsync
+
+Just like ``UnfoldAsync``, but the fold function returns a ``ValueTask``, with internal pooling to minimize allocation and improve latency.
+
+**emits** when there is demand and unfold state returned task completes with not null value
+
+**completes** when the task returned by the unfold function completes with an null value
+
 ### Empty
 
 Complete right away without ever emitting any elements. Useful when you have to provide a source to
@@ -196,6 +204,14 @@ Functions return ``Task`` to achieve asynchronous processing
 
 **completes** when ``Task`` from read function returns ``None``
 
+### UnfoldResourceValueTaskAsync
+
+Like ``UnfoldResourceAsync`` but takes ``ValueTask`` Functions instead, with amortization of allocations for the main read stage.
+
+**emits** when there is demand and ``ValueTask`` from read function returns value
+
+**completes** when ``ValueTask`` from read function returns ``None``
+
 ### Queue
 
 Materialize a ``SourceQueue`` onto which elements can be pushed for emitting from the source. The queue contains
@@ -846,6 +862,16 @@ If a Task fails, the stream also fails (unless a different supervision strategy
 
 **completes** when upstream completes and all tasks has been completed and all elements has been emitted
 
+### SelectValueTaskAsync
+
+Version of ``SelectAsync`` that is optimized for ValueTask returns. Prefer this over ``SelectAsync`` if your work may be synchronus or is primarily waiting on ``ValueTask``
+
+**emits** when the ``ValueTask`` returned by the provided function finishes for the next element in sequence
+
+**backpressures** when the number of tasks reaches the configured parallelism and the downstream backpressures
+
+**completes** when upstream completes and all tasks has been completed and all elements has been emitted
+
 ### SelectAsyncUnordered
 
 Like ``SelectAsync`` but ``Task`` results are passed downstream as they arrive regardless of the order of the elements

src/Directory.Build.props

@@ -22,6 +22,7 @@
     <XunitRunnerVersion>2.5.3</XunitRunnerVersion>
     <TestSdkVersion>17.9.0</TestSdkVersion>
     <HyperionVersion>0.12.2</HyperionVersion>
+    <SystemThreadingTasksExtensionsVersion>4.5.4</SystemThreadingTasksExtensionsVersion>
     <NewtonsoftJsonVersion>[13.0.1,)</NewtonsoftJsonVersion>
     <NBenchVersion>2.0.1</NBenchVersion>
     <ProtobufVersion>3.25.3</ProtobufVersion>

src/benchmark/Akka.Benchmarks/Streams/SelectAsyncBenchmarks.cs

@@ -0,0 +1,189 @@
+// //-----------------------------------------------------------------------
+// // <copyright file="SelectAsyncBenchmarks.cs" company="Akka.NET Project">
+// //     Copyright (C) 2009-2024 Lightbend Inc. <http://www.lightbend.com>
+// //     Copyright (C) 2013-2024 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// // </copyright>
+// //-----------------------------------------------------------------------
+
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Akka.Actor;
+using Akka.Benchmarks.Configurations;
+using Akka.Streams;
+using Akka.Streams.Dsl;
+using BenchmarkDotNet.Attributes;
+
+namespace Akka.Benchmarks.Streams;
+
+[Config(typeof(MicroBenchmarkConfig))]
+public class SelectAsyncBenchmarks
+{
+    public struct IntOrCompletion
+    {
+        public readonly int IntValue;
+        public readonly TaskCompletionSource? Completion;
+
+        public IntOrCompletion(int intValue, TaskCompletionSource? completion)
+        {
+            IntValue = intValue;
+            Completion = completion;
+        }
+    }
+    private ActorSystem system;
+    private ActorMaterializer materializer;
+
+    private IRunnableGraph<Task> simpleGraph;
+    private Task<Done> selectAsyncStub;
+    private Channel<IntOrCompletion> asyncCh;
+    private Task<Done> selectValueTaskAsyncStub;
+    private Channel<IntOrCompletion> vtAsyncCh;
+    private Task<Done> selectAsyncSyncStub;
+    private Task<Done> selectAsyncValueTaskSyncStub;
+    private Channel<IntOrCompletion> asyncChSync;
+    private Channel<IntOrCompletion> vtAsyncChSync;
+
+    [GlobalSetup]
+    public void Setup()
+    {
+        system = ActorSystem.Create("system");
+        materializer = system.Materializer();
+        asyncCh = Channel.CreateUnbounded<IntOrCompletion>();
+
+        asyncChSync = Channel.CreateUnbounded<IntOrCompletion>();
+
+        vtAsyncChSync = Channel.CreateUnbounded<IntOrCompletion>();
+
+        selectAsyncSyncStub = Source.ChannelReader(asyncChSync.Reader)
+            .SelectAsync(4, a =>
+            {
+                if (a.Completion != null)
+                {
+                    a.Completion.TrySetResult();
+                }
+                else
+                {
+                }
+
+                return Task.FromResult(NotUsed.Instance);
+            }).RunWith(Sink.Ignore<NotUsed>(), materializer);
+
+        selectAsyncValueTaskSyncStub = Source.ChannelReader(vtAsyncChSync.Reader)
+            .SelectValueTaskAsync(4, a =>
+            {
+                if (a.Completion != null)
+                {
+                    a.Completion.TrySetResult();
+                }
+                else
+                {
+                }
+
+                return ValueTask.FromResult(NotUsed.Instance);
+            }).RunWith(Sink.Ignore<NotUsed>(), materializer);
+        selectAsyncStub = Source.ChannelReader(asyncCh.Reader)
+            .SelectAsync(4, async a =>
+            {
+                if (a.Completion != null)
+                {
+                    a.Completion.TrySetResult();
+                }
+                else
+                {
+                    //await Task.Yield();
+                    await Task.Delay(0);
+                }
+
+                return NotUsed.Instance;
+            }).RunWith(Sink.Ignore<NotUsed>(), materializer);
+        vtAsyncCh = Channel.CreateUnbounded<IntOrCompletion>();
+        int vta = 0;
+        selectValueTaskAsyncStub = Source.ChannelReader(vtAsyncCh.Reader)
+            .SelectValueTaskAsync(4, async a =>
+            {
+                if (a.Completion != null)
+                {
+                    a.Completion.TrySetResult();
+                    //return NotUsed.Instance;
+                }
+                else
+                {
+                    //await Task.Yield();
+                    await Task.Delay(0);
+                    //return NotUsed.Instance;
+                    //Console.WriteLine(++vta);
+                    //return vta;
+                }
+
+                return NotUsed.Instance;
+            }).RunWith(Sink.Ignore<NotUsed>(), materializer);
+    }
+
+    [GlobalCleanup]
+    public void Cleanup()
+    {
+        materializer.Dispose();
+        system.Dispose();
+    }
+
+    [Benchmark]
+    public async Task RunSelectAsync()
+    {
+        var completion = new TaskCompletionSource(TaskCreationOptions
+            .RunContinuationsAsynchronously);
+        for (int i = 0; i < 100; i++)
+        {
+            asyncCh.Writer.TryWrite(new IntOrCompletion(i, null));
+        }
+
+        asyncCh.Writer.TryWrite(new IntOrCompletion(0, completion));
+        await completion.Task;
+
+    }
+
+
+    [Benchmark]
+    public async Task RunSelectValueTaskAsync()
+    {
+        var completion = new TaskCompletionSource(TaskCreationOptions
+            .RunContinuationsAsynchronously);
+        for (int i = 0; i < 100; i++)
+        {
+            vtAsyncCh.Writer.TryWrite(new IntOrCompletion(i, null));
+        }
+
+        vtAsyncCh.Writer.TryWrite(new IntOrCompletion(0, completion));
+        await completion.Task;
+
+    }
+
+    [Benchmark]
+    public async Task RunSelectAsyncSync()
+    {
+        var completion = new TaskCompletionSource(TaskCreationOptions
+            .RunContinuationsAsynchronously);
+        for (int i = 0; i < 100; i++)
+        {
+            asyncChSync.Writer.TryWrite(new IntOrCompletion(i, null));
+        }
+
+        asyncChSync.Writer.TryWrite(new IntOrCompletion(0, completion));
+        await completion.Task;
+
+    }
+
+
+    [Benchmark]
+    public async Task RunSelectValueTaskAsyncSync()
+    {
+        var completion = new TaskCompletionSource(TaskCreationOptions
+            .RunContinuationsAsynchronously);
+        for (int i = 0; i < 100; i++)
+        {
+            vtAsyncChSync.Writer.TryWrite(new IntOrCompletion(i, null));
+        }
+
+        vtAsyncChSync.Writer.TryWrite(new IntOrCompletion(0, completion));
+        await completion.Task;
+
+    }
+}