-
Notifications
You must be signed in to change notification settings - Fork 12
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
11 changed files
with
242 additions
and
28 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,97 @@ | ||
| // ReSharper disable UnusedType.Global | ||
| // ReSharper disable UnusedMember.Global | ||
|
|
||
| namespace EasilyNET.Core.System; | ||
|
|
||
| /// <summary> | ||
| /// 表示一个异步屏障,它会阻塞一组任务,直到所有任务都到达屏障,通常用于需要协调多个异步任务的场景,确保所有任务在某个同步点之前都已完成特定的工作,然后再继续执行 | ||
| /// </summary> | ||
| /// <example> | ||
| /// <code> | ||
| /// <![CDATA[ | ||
| /// var barrier = new AsyncBarrier(3); | ||
| /// var cts = new CancellationTokenSource(); | ||
| /// | ||
| /// // 启动三个任务，每个任务都会在屏障处等待 | ||
| /// var tasks = new List<Task> | ||
| /// { | ||
| /// Task.Run(async () => | ||
| /// { | ||
| /// Console.WriteLine("任务1到达屏障"); | ||
| /// await barrier.SignalAndWait(cts.Token); | ||
| /// Console.WriteLine("任务1继续执行"); | ||
| /// }), | ||
| /// Task.Run(async () => | ||
| /// { | ||
| /// Console.WriteLine("任务2到达屏障"); | ||
| /// await barrier.SignalAndWait(cts.Token); | ||
| /// Console.WriteLine("任务2继续执行"); | ||
| /// }), | ||
| /// Task.Run(async () => | ||
| /// { | ||
| /// Console.WriteLine("任务3到达屏障"); | ||
| /// await barrier.SignalAndWait(cts.Token); | ||
| /// Console.WriteLine("任务3继续执行"); | ||
| /// }) | ||
| /// }; | ||
| /// | ||
| /// // 等待所有任务完成 | ||
| /// await Task.WhenAll(tasks); | ||
| /// ]]> | ||
| /// </code> | ||
| /// </example> | ||
| public sealed class AsyncBarrier | ||
| { | ||
| private readonly int participantCount; | ||
| private readonly Stack<Waiter> waiters; | ||
|
|
||
| /// <summary> | ||
| /// 使用指定数量的参与者初始化 <see cref="AsyncBarrier" /> 类的新实例 | ||
| /// </summary> | ||
| /// <param name="participants">屏障的参与者数量</param> | ||
| /// <exception cref="ArgumentOutOfRangeException">当参与者数量小于或等于零时抛出</exception> | ||
| public AsyncBarrier(int participants) | ||
| { | ||
| if (participants <= 0) | ||
| throw new ArgumentOutOfRangeException(nameof(participants), $"参数 {nameof(participants)} 必须是一个正数。"); | ||
| participantCount = participants; | ||
| waiters = new(participants - 1); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// 表示一个参与者已到达屏障,并等待所有其他参与者到达屏障 | ||
| /// </summary> | ||
| /// <param name="cancellationToken">用于监视取消请求的令牌</param> | ||
| /// <returns>表示异步操作的任务</returns> | ||
| public ValueTask SignalAndWait(CancellationToken cancellationToken) | ||
| { | ||
| lock (waiters) | ||
| { | ||
| if (waiters.Count + 1 == participantCount) | ||
| { | ||
| while (waiters.Count > 0) | ||
| { | ||
| var waiter = waiters.Pop(); | ||
| waiter.CompletionSource.TrySetResult(default); | ||
| waiter.CancellationRegistration.Dispose(); | ||
| } | ||
| return new(cancellationToken.IsCancellationRequested | ||
| ? Task.FromCanceled(cancellationToken) | ||
| : Task.CompletedTask); | ||
| } | ||
| TaskCompletionSource<EmptyStruct> tcs = new(TaskCreationOptions.RunContinuationsAsynchronously); | ||
| var ctr = cancellationToken.CanBeCanceled | ||
| ? cancellationToken.Register(static (tcs, ct) => ((TaskCompletionSource<EmptyStruct>)tcs!).TrySetCanceled(ct), tcs) | ||
| : default; | ||
| waiters.Push(new(tcs, ctr)); | ||
| return new(tcs.Task); | ||
| } | ||
| } | ||
|
|
||
| private readonly struct Waiter(TaskCompletionSource<EmptyStruct> completionSource, CancellationTokenRegistration cancellationRegistration) | ||
| { | ||
| internal TaskCompletionSource<EmptyStruct> CompletionSource => completionSource; | ||
|
|
||
| internal CancellationTokenRegistration CancellationRegistration => cancellationRegistration; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,20 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT license. See LICENSE file in the project root for full license information. | ||
|
|
||
| // ReSharper disable UnusedMember.Global | ||
|
|
||
| namespace EasilyNET.Core.System; | ||
|
|
||
| /// <summary> | ||
| /// 一个空结构体 | ||
| /// </summary> | ||
| /// <remarks> | ||
| /// 当泛型类型需要一个类型参数但完全不使用时,这可以节省 4 个字节,相对于 System.Object | ||
| /// </remarks> | ||
| internal readonly struct EmptyStruct | ||
| { | ||
| /// <summary> | ||
| /// 获取空结构体的一个实例 | ||
| /// </summary> | ||
| internal static EmptyStruct Instance => default; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,64 @@ | ||
| using EasilyNET.Core.System; | ||
|
|
||
| // ReSharper disable MethodSupportsCancellation | ||
|
|
||
| namespace EasilyNET.Test.Unit.System; | ||
|
|
||
| [TestClass] | ||
| public class AsyncBarrierTest | ||
| { | ||
| [TestMethod] | ||
| public async Task AllParticipantsReachBarrier_ShouldContinueExecution() | ||
| { | ||
| var barrier = new AsyncBarrier(3); | ||
| var cts = new CancellationTokenSource(); | ||
| var tasks = new List<Task> | ||
| { | ||
| Task.Run(async () => { await barrier.SignalAndWait(cts.Token); }), | ||
| Task.Run(async () => { await barrier.SignalAndWait(cts.Token); }), | ||
| Task.Run(async () => { await barrier.SignalAndWait(cts.Token); }) | ||
| }; | ||
| await Task.WhenAll(tasks); | ||
| } | ||
|
|
||
| [TestMethod] | ||
| public async Task CancellationToken_ShouldCancelTask() | ||
| { | ||
| var barrier = new AsyncBarrier(3); | ||
| var cts = new CancellationTokenSource(); | ||
| var tasks = new List<Task> | ||
| { | ||
| Task.Run(async () => { await barrier.SignalAndWait(cts.Token); }), | ||
| Task.Run(async () => { await barrier.SignalAndWait(cts.Token); }), | ||
| Task.Run(async () => | ||
| { | ||
| await cts.CancelAsync(); // 取消令牌 | ||
| await Assert.ThrowsExceptionAsync<TaskCanceledException>(async () => { await barrier.SignalAndWait(cts.Token); }); | ||
| }) | ||
| }; | ||
| try | ||
| { | ||
| await Task.WhenAll(tasks); | ||
| } | ||
| catch (TaskCanceledException) | ||
| { | ||
| // 预期的异常，不需要处理 | ||
| } | ||
| } | ||
|
|
||
| [TestMethod] | ||
| public async Task LessParticipants_ShouldWaitIndefinitely() | ||
| { | ||
| var barrier = new AsyncBarrier(3); | ||
| var cts = new CancellationTokenSource(); | ||
| var tasks = new List<Task> | ||
| { | ||
| Task.Run(async () => { await barrier.SignalAndWait(cts.Token); }), | ||
| Task.Run(async () => { await barrier.SignalAndWait(cts.Token); }) | ||
| }; | ||
| var delayTask = Task.Delay(1000, cts.Token); | ||
| var allTasks = Task.WhenAll(tasks); | ||
| var completedTask = await Task.WhenAny(allTasks, delayTask); | ||
| Assert.AreEqual(delayTask, completedTask); // 确保任务在等待 | ||
| } | ||
| } |