Release of DotNext.Threading 5.8.0

CHANGELOG.md

@@ -1,6 +1,11 @@
 Release Notes
 ====
 
+# 06-25-2024
+<a href="https://www.nuget.org/packages/dotnext.threading/5.8.0">DotNext.Threading 5.8.0</a>
+* Introduced `WaitAnyAsync` method to wait on a group of cancellation tokens
+* Added cancellation support for `WaitAsync` extension method for [WaitHandle](https://learn.microsoft.com/en-us/dotnet/api/system.threading.waithandle) class
+
 # 06-20-2024
 <a href="https://www.nuget.org/packages/dotnext/5.7.0">DotNext 5.7.0</a>
 * `Timestamp.ElapsedTicks` returns a value that is always greater than zero

README.md

@@ -44,31 +44,11 @@ All these things are implemented in 100% managed code on top of existing .NET AP
 * [NuGet Packages](https://www.nuget.org/profiles/rvsakno)
 
 # What's new
-Release Date: 06-20-2024
+Release Date: 06-25-2024
 
-<a href="https://www.nuget.org/packages/dotnext/5.7.0">DotNext 5.7.0</a>
-* `Timestamp.ElapsedTicks` returns a value that is always greater than zero
-* Fixed hidden copies of value types caused by **in** modifier
-* Added support of [TimeProvider](https://learn.microsoft.com/en-us/dotnet/api/system.timeprovider) to `Timestamp` and `Timeout` types
-
-<a href="https://www.nuget.org/packages/dotnext.metaprogramming/5.7.0">DotNext.Metaprogramming 5.7.0</a>
-* Updated dependencies
-
-<a href="https://www.nuget.org/packages/dotnext.unsafe/5.7.0">DotNext.Unsafe 5.7.0</a>
-* Updated dependencies
-
-<a href="https://www.nuget.org/packages/dotnext.threading/5.7.0">DotNext.Threading 5.7.0</a>
-* Fixed [241](https://github.com/dotnet/dotNext/issues/241)
-* Introduced API for implementing leases, see `DotNext.Threading.Leases` namespace
-
-<a href="https://www.nuget.org/packages/dotnext.io/5.7.0">DotNext.IO 5.7.0</a>
-* Various performance improvements
-
-<a href="https://www.nuget.org/packages/dotnext.net.cluster/5.7.0">DotNext.Net.Cluster 5.7.0</a>
-* Fixed [242](https://github.com/dotnet/dotNext/issues/242)
-
-<a href="https://www.nuget.org/packages/dotnext.aspnetcore.cluster/5.7.0">DotNext.AspNetCore.Cluster 5.7.0</a>
-* Fixed [242](https://github.com/dotnet/dotNext/issues/242)
+<a href="https://www.nuget.org/packages/dotnext.threading/5.8.0">DotNext.Threading 5.8.0</a>
+* Introduced `WaitAnyAsync` method to wait on a group of cancellation tokens
+* Added cancellation support for `WaitAsync` extension method for [WaitHandle](https://learn.microsoft.com/en-us/dotnet/api/system.threading.waithandle) class
 
 Changelog for previous versions located [here](./CHANGELOG.md).
 

src/DotNext.Tests/Threading/AsyncBridgeTests.cs

@@ -26,11 +26,25 @@ public static async Task WaitForSignal()
         await ev.WaitAsync(DefaultTimeout);
     }
 
+    [Fact]
+    public static async Task CancelWaitForSignal()
+    {
+        using var ev = new ManualResetEvent(false);
+        using var cts = new CancellationTokenSource();
+
+        var task = ev.WaitAsync(cts.Token).AsTask();
+        cts.Cancel();
+
+        var e = await ThrowsAsync<OperationCanceledException>(Func.Constant(task));
+        Equal(cts.Token, e.CancellationToken);
+    }
+
     [Fact]
     public static async Task AlreadySignaled()
     {
         using var ev = new ManualResetEvent(true);
         True(await ev.WaitAsync(DefaultTimeout));
+        True(ev.WaitAsync().IsCompletedSuccessfully);
     }
 
     [Fact]
@@ -68,4 +82,43 @@ public static async Task CancellationTokenAwaitCornerCases()
         await new CancellationToken(true).WaitAsync();
         await ThrowsAsync<ArgumentException>(new CancellationToken(false).WaitAsync().AsTask);
     }
+
+    [Fact]
+    public static async Task WaitForCancellationSingleToken()
+    {
+        using var cts = new CancellationTokenSource();
+        var task = AsyncBridge.WaitAnyAsync([cts.Token]);
+        False(task.IsCompletedSuccessfully);
+
+        cts.Cancel();
+        Equal(cts.Token, await task);
+    }
+
+    [Fact]
+    public static async Task WaitForCancellationTwoTokens()
+    {
+        using var cts1 = new CancellationTokenSource();
+        using var cts2 = new CancellationTokenSource();
+        var task = AsyncBridge.WaitAnyAsync([cts1.Token, cts2.Token]);
+        False(task.IsCompletedSuccessfully);
+
+        cts2.Cancel();
+        cts1.Cancel();
+        Equal(cts2.Token, await task);
+    }
+
+    [Fact]
+    public static async Task WaitForCancellationMultipleTokens()
+    {
+        using var cts1 = new CancellationTokenSource();
+        using var cts2 = new CancellationTokenSource();
+        using var cts3 = new CancellationTokenSource();
+        var task = AsyncBridge.WaitAnyAsync([cts1.Token, cts2.Token, cts3.Token]);
+        False(task.IsCompletedSuccessfully);
+
+        cts3.Cancel();
+        cts2.Cancel();
+        cts1.Cancel();
+        Equal(cts3.Token, await task);
+    }
 }

src/DotNext.Threading/DotNext.Threading.csproj

@@ -7,7 +7,7 @@
     <ImplicitUsings>true</ImplicitUsings>
     <IsTrimmable>true</IsTrimmable>
     <Features>nullablePublicOnly</Features>
-    <VersionPrefix>5.7.0</VersionPrefix>
+    <VersionPrefix>5.8.0</VersionPrefix>
     <VersionSuffix></VersionSuffix>
     <Authors>.NET Foundation and Contributors</Authors>
     <Product>.NEXT Family of Libraries</Product>

src/DotNext.Threading/Threading/AsyncBridge.CancellationToken.cs

@@ -1,4 +1,6 @@
+using System.Buffers;
 using System.Collections.Concurrent;
+using System.Runtime.InteropServices;
 using Debug = System.Diagnostics.Debug;
 using Unsafe = System.Runtime.CompilerServices.Unsafe;
 
@@ -38,6 +40,127 @@ internal void Return(CancellationTokenValueTask vt)
                 Add(vt);
         }
     }
+
+    private abstract class CancellationTokenCompletionSource : TaskCompletionSource<CancellationToken>
+    {
+        protected static readonly Action<object?, CancellationToken> Callback = OnCanceled;
+
+        private bool initialized; // volatile
+
+        protected CancellationTokenCompletionSource(out InitializationFlag flag)
+            : base(TaskCreationOptions.RunContinuationsAsynchronously)
+            => flag = new(ref initialized);
+
+        private static void OnCanceled(object? source, CancellationToken token)
+        {
+            Debug.Assert(source is CancellationTokenCompletionSource);
+
+            Unsafe.As<CancellationTokenCompletionSource>(source).OnCanceled(token);
+        }
+
+        private void OnCanceled(CancellationToken token)
+        {
+            if (Volatile.Read(ref initialized) && TrySetResult(token))
+            {
+                Cleanup();
+            }
+        }
+
+        private static void Unregister(ReadOnlySpan<CancellationTokenRegistration> registrations)
+        {
+            foreach (ref readonly var registration in registrations)
+            {
+                registration.Unregister();
+            }
+        }
+
+        private protected virtual void Cleanup() => Unregister(Registrations);
+
+        private protected abstract ReadOnlySpan<CancellationTokenRegistration> Registrations { get; }
+
+        [StructLayout(LayoutKind.Auto)]
+        protected readonly ref struct InitializationFlag
+        {
+            private readonly ref bool flag;
+
+            internal InitializationFlag(ref bool flag) => this.flag = ref flag;
+
+            internal CancellationToken InitializationCompleted(ReadOnlySpan<CancellationTokenRegistration> registrations)
+            {
+                Volatile.Write(ref flag, true);
+
+                foreach (ref readonly var registration in registrations)
+                {
+                    if (registration.Token.IsCancellationRequested)
+                    {
+                        return registration.Token;
+                    }
+                }
+
+                return new(canceled: false);
+            }
+        }
+    }
+
+    private sealed class CancellationTokenCompletionSource1 : CancellationTokenCompletionSource
+    {
+        private readonly CancellationTokenRegistration registration;
+
+        internal CancellationTokenCompletionSource1(CancellationToken token)
+            : base(out var flag)
+        {
+            registration = token.UnsafeRegister(Callback, this);
+
+            if (flag.InitializationCompleted(new(in registration)) is { IsCancellationRequested: true } canceledToken)
+                Callback(this, canceledToken);
+        }
+
+        private protected override ReadOnlySpan<CancellationTokenRegistration> Registrations => new(in registration);
+    }
+
+    private sealed class CancellationTokenCompletionSource2 : CancellationTokenCompletionSource
+    {
+        private readonly (CancellationTokenRegistration, CancellationTokenRegistration) registrations;
+
+        internal CancellationTokenCompletionSource2(CancellationToken token1, CancellationToken token2)
+            : base(out var flag)
+        {
+            registrations.Item1 = token1.UnsafeRegister(Callback, this);
+            registrations.Item2 = token2.UnsafeRegister(Callback, this);
+
+            if (flag.InitializationCompleted(registrations.AsReadOnlySpan()) is { IsCancellationRequested: true } canceledToken)
+                Callback(this, canceledToken);
+        }
+
+        private protected override ReadOnlySpan<CancellationTokenRegistration> Registrations => registrations.AsReadOnlySpan();
+    }
+
+    private sealed class CancellationTokenCompletionSourceN : CancellationTokenCompletionSource
+    {
+        private readonly CancellationTokenRegistration[] registrations;
+
+        internal CancellationTokenCompletionSourceN(ReadOnlySpan<CancellationToken> tokens)
+            : base(out var flag)
+        {
+            registrations = ArrayPool<CancellationTokenRegistration>.Shared.Rent(tokens.Length);
+
+            for (var i = 0; i < tokens.Length; i++)
+            {
+                registrations[i] = tokens[i].UnsafeRegister(Callback, this);
+            }
+
+            if (flag.InitializationCompleted(registrations) is { IsCancellationRequested: true } canceledToken)
+                Callback(this, canceledToken);
+        }
+
+        private protected override ReadOnlySpan<CancellationTokenRegistration> Registrations => new(registrations);
+
+        private protected override void Cleanup()
+        {
+            ArrayPool<CancellationTokenRegistration>.Shared.Return(registrations, clearArray: true);
+            base.Cleanup();
+        }
+    }
 
     private static readonly Action<CancellationTokenValueTask> CancellationTokenValueTaskCompletionCallback = new CancellationTokenValueTaskPool().Return;
 

src/DotNext.Threading/Threading/AsyncBridge.WaitHandle.cs

@@ -30,13 +30,6 @@ protected override void AfterConsumed()
             Interlocked.Decrement(ref instantiatedTasks);
             backToPool(this);
         }
-
-        internal void Complete(object? token, bool timedOut)
-        {
-            Debug.Assert(token is short);
-
-            TrySetResult(Unsafe.Unbox<short>(token), !timedOut);
-        }
     }
 
     private sealed class WaitHandleValueTaskPool : ConcurrentBag<WaitHandleValueTask>