Change Async.Sequence to use StartImmediateAsTask (fsprojects#517)

Author: fcallejon

.gitignore

@@ -213,3 +213,6 @@ tests/FSharpPlusFable.Tests/node_modules/
 tests/FSharpPlusFable.Tests/FSharpPlusFable
 /node_modules
 *.fs.js
+
+tests/Benchmarks/BenchmarkDotNet.Artifacts
+tests/benchmarks/BenchmarkDotNet.Artifacts

FSharpPlus.sln

@@ -95,6 +95,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "CSharpLib", "tests\CSharpLi
 EndProject
 Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "FSharpPlusFable.Tests", "tests\FSharpPlusFable.Tests\FSharpPlusFable.Tests.fsproj", "{1CCD1BFB-60E4-40AA-B534-3C5EEE5E1E83}"
 EndProject
+Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "Benchmarks", "tests\benchmarks\Benchmarks.fsproj", "{EEFF08EB-8B0C-4F63-9425-4281EFF12087}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -158,6 +160,14 @@ Global
 		{1CCD1BFB-60E4-40AA-B534-3C5EEE5E1E83}.Fable|Any CPU.Build.0 = Fable|Any CPU
 		{1CCD1BFB-60E4-40AA-B534-3C5EEE5E1E83}.Fable3|Any CPU.ActiveCfg = Fable3|Any CPU
 		{1CCD1BFB-60E4-40AA-B534-3C5EEE5E1E83}.Fable3|Any CPU.Build.0 = Fable3|Any CPU
+		{EEFF08EB-8B0C-4F63-9425-4281EFF12087}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{EEFF08EB-8B0C-4F63-9425-4281EFF12087}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{EEFF08EB-8B0C-4F63-9425-4281EFF12087}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{EEFF08EB-8B0C-4F63-9425-4281EFF12087}.Release|Any CPU.Build.0 = Release|Any CPU
+		{EEFF08EB-8B0C-4F63-9425-4281EFF12087}.Fable|Any CPU.ActiveCfg = Debug|Any CPU
+		{EEFF08EB-8B0C-4F63-9425-4281EFF12087}.Fable|Any CPU.Build.0 = Debug|Any CPU
+		{EEFF08EB-8B0C-4F63-9425-4281EFF12087}.Fable3|Any CPU.ActiveCfg = Debug|Any CPU
+		{EEFF08EB-8B0C-4F63-9425-4281EFF12087}.Fable3|Any CPU.Build.0 = Debug|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -170,6 +180,7 @@ Global
 		{9B93F5E5-3D53-42F1-96E2-06E6A7B496A0} = {81F5F559-FD23-4E90-9EE6-3E2A2C1A7F96}
 		{7A5B766E-8141-4D8A-B3EB-91422FDBDF71} = {ED8079DD-2B06-4030-9F0F-DC548F98E1C4}
 		{1CCD1BFB-60E4-40AA-B534-3C5EEE5E1E83} = {ED8079DD-2B06-4030-9F0F-DC548F98E1C4}
+		{EEFF08EB-8B0C-4F63-9425-4281EFF12087} = {ED8079DD-2B06-4030-9F0F-DC548F98E1C4}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {789B5FFA-7891-4F60-831E-42C3C5ED2C51}

src/FSharpPlus/Extensions/Extensions.fs

@@ -67,11 +67,11 @@ module Extensions =
         #if !FABLE_COMPILER
         /// Combine all asyncs in one, chaining them in sequence order.
         static member Sequence (t:seq<Async<_>>) : Async<seq<_>> = async {
+            let startImmediateAsTask ct a =
+                Async.StartImmediateAsTask(a, ct).Result
+
             let! ct = Async.CancellationToken
-            return seq {
-                use enum = t.GetEnumerator ()
-                while enum.MoveNext() do
-                    yield Async.RunSynchronously (enum.Current, cancellationToken = ct) }}
+            return t |> Seq.map (startImmediateAsTask ct) }
         #endif
 
         /// Combine all asyncs in one, chaining them in sequence order.

tests/benchmarks/AsyncSequenceBenchmarks.fs

@@ -0,0 +1,129 @@
+module AsyncSequences
+
+open BenchmarkDotNet.Attributes
+open System.Threading
+open System.Threading.Tasks
+
+type AsyncSeqTaskState<'t> =
+    | Idle
+    | Ok of 't
+    | Error of exn
+    | Cancelled
+
+let sequence0 (t: seq<Async<_>>) : Async<seq<_>> =
+    async {
+        let! ct = Async.CancellationToken
+
+        return
+            seq {
+                use enum = t.GetEnumerator()
+
+                while enum.MoveNext() do
+                    yield Async.RunSynchronously(enum.Current, cancellationToken = ct)
+            }
+    }
+
+let sequence_StartImmediateAsTask (t: seq<Async<_>>) : Async<seq<_>> =
+    async {
+        let startImmediateAsTask ct a =
+            Async.StartImmediateAsTask(a, ct).Result
+
+        let! ct = Async.CancellationToken
+        return t |> Seq.map (startImmediateAsTask ct)
+    }
+
+let sequence_ManualResetEventSlim (t: seq<Async<_>>) : Async<seq<_>> =
+    async {
+        let startImmediateAsTask ct (a: Async<'t>) : 't =
+            let mutable state =
+                AsyncSeqTaskState<'t>.Idle
+
+            let mutex = new ManualResetEventSlim(false)
+
+            let setState newState =
+                try
+                    state <- newState
+                finally
+                    mutex.Set()
+
+            Async.StartWithContinuations(
+                a,
+                (fun k -> AsyncSeqTaskState<'t>.Ok k |> setState),
+                (fun e -> AsyncSeqTaskState<'t>.Error e |> setState),
+                (fun _ -> setState AsyncSeqTaskState<'t>.Cancelled),
+                ct
+            )
+
+            mutex.Wait()
+
+            match state with
+            | Idle
+            | Cancelled -> TaskCanceledException() |> raise
+            | Ok res -> res
+            | Error e -> raise e
+
+        let! ct = Async.CancellationToken
+        return t |> Seq.map (startImmediateAsTask ct)
+    }
+
+let SyncSum = async {
+    return 1 + 1
+}
+
+let AsyncWith1SecSleep = async {
+    do! Async.Sleep 1
+    return 1 + 1
+}
+
+let SyncAsyncSleepOverAsync = async {
+    Async.RunSynchronously (Async.Sleep 5)
+    return 1 + 1
+}
+
+type Benchmarks() =
+    [<Params(10, 100, 1000)>]
+    member val public times = 0 with get, set
+    
+    [<Params(1)>]
+    member val public threads = 0 with get, set
+
+    [<GlobalSetup>]
+    member self.GlobalSetup() =
+        if self.threads > 0 then
+            ThreadPool.SetMinThreads (self.threads, self.threads) |> ignore
+            ThreadPool.SetMaxThreads (self.threads, self.threads) |> ignore
+
+    [<Benchmark(Baseline = true)>]
+    member this.Base() =
+        if this.threads = 1 then
+            failwith "This function will fail with just one available thread."
+        seq {
+            for _ = 1 to this.times do
+                yield SyncSum
+                yield AsyncWith1SecSleep
+                yield SyncAsyncSleepOverAsync }
+        |> sequence0
+        |> Async.RunSynchronously
+        |> Seq.toArray
+
+    [<Benchmark>]
+    member this.StartImmediateAsTask() =        
+        seq {
+            for _ = 1 to this.times do
+                yield SyncSum
+                yield AsyncWith1SecSleep
+                yield SyncAsyncSleepOverAsync }
+        |> sequence_StartImmediateAsTask
+        |> Async.RunSynchronously
+        |> Seq.toArray
+
+    [<Benchmark>]
+    member this.ManualResetEventSlim() =
+        seq {
+            for _ = 1 to this.times do
+                yield SyncSum
+                yield AsyncWith1SecSleep
+                yield SyncAsyncSleepOverAsync }
+        |> sequence_ManualResetEventSlim
+        |> Async.RunSynchronously
+        |> Seq.toArray

tests/benchmarks/Benchmarks.fsproj

@@ -0,0 +1,21 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <TargetFramework>net6.0</TargetFramework>
+    <OutputType>Exe</OutputType>
+    <PlatformTarget>AnyCPU</PlatformTarget>
+    <Optimize>true</Optimize>
+    <Configuration>Release</Configuration>
+    <IsPackable>false</IsPackable>
+  </PropertyGroup>
+  <PropertyGroup Condition="'$(Configuration)'=='Release'">
+    <Tailcalls>true</Tailcalls>
+  </PropertyGroup>
+  <ItemGroup>
+    <PackageReference Include="BenchmarkDotNet" Version="0.12.1" />
+    <PackageReference Include="BenchmarkDotNet.Diagnostics.Windows" Version="0.12.1" Condition="'$(OS)' == 'Windows_NT'" />
+  </ItemGroup>
+  <ItemGroup>
+    <Compile Include="AsyncSequenceBenchmarks.fs" />
+    <Compile Include="Program.fs" />
+  </ItemGroup>
+</Project>

tests/benchmarks/Program.fs

@@ -0,0 +1,6 @@
+open AsyncSequences
+
+[<EntryPoint>]
+let main _ =
+    do BenchmarkDotNet.Running.BenchmarkRunner.Run<Benchmarks>() |> ignore
+    0