TaskUtils 的进度支持

Shared/General/Progress.cs

@@ -16,6 +16,8 @@ public class ProgressProvider {
 
     private (double actual, double skiped, double splited) progressParts = (0, 0, 0);
     private double progressSum => progressParts.actual + progressParts.skiped + progressParts.splited;
+    public bool Finished => progressParts.actual + progressParts.skiped > 0.9999999;
+
     private bool _Set(double value, bool skiped) {
         value = value.Clamp(0, 1);
         double delta = value - progressSum;

Shared/Utils/TaskUtils.cs

@@ -3,6 +3,82 @@ using System.Diagnostics;
 namespace MeloongCore;
 public static class TaskUtils {
 
+    /// <summary>
+    /// 并行执行所有任务。
+    /// </summary>
+    public static Task WhenAll(IEnumerable<Func<ProgressProvider?, Task>> tasks, ProgressProvider? progress = null) {
+        var taskList = tasks.ToList();
+        if (!taskList.Any()) {
+            progress?.Skip();
+            return Task.CompletedTask;
+        }
+        // 实际分割
+        var progressEach = 1d / taskList.Count;
+        return Task.WhenAll(taskList.Select(async (task, index) => {
+            var cp = progress?.SplitBy(progressEach)[0];
+            await task(cp).ConfigureAwait(false);
+            cp?.Finish();
+        }));
+    }
+    /// <summary>
+    /// 并行执行所有任务并返回结果。
+    /// </summary>
+    public static Task<TResult[]> WhenAll<TResult>(IEnumerable<Func<ProgressProvider?, Task<TResult>>> tasks, ProgressProvider? progress = null) {
+        var taskList = tasks.ToList();
+        if (!taskList.Any()) {
+            progress?.Skip();
+            return Task.FromResult(Array.Empty<TResult>());
+        }
+        // 实际分割
+        var progressEach = 1d / taskList.Count;
+        return Task.WhenAll(taskList.Select(async (task, index) => {
+            var cp = progress?.SplitBy(progressEach)[0];
+            var result = await task(cp).ConfigureAwait(false);
+            cp?.Finish();
+            return result;
+        }));
+    }
+    /// <summary>
+    /// 并行执行所有任务。
+    /// </summary>
+    public static Task WhenAll(IEnumerable<Task> tasks, ProgressProvider? progress = null)
+        => WhenAll(tasks.Select(task => new Func<ProgressProvider?, Task>(_ => task)), progress);
+    /// <summary>
+    /// 并行执行所有任务并返回结果。
+    /// </summary>
+    public static Task<TResult[]> WhenAll<TResult>(IEnumerable<Task<TResult>> tasks, ProgressProvider? progress = null)
+        => WhenAll(tasks.Select(task => new Func<ProgressProvider?, Task<TResult>>(_ => task)), progress);
+
+    public static void ForEach<T>(IEnumerable<T> source, Action<T> body)
+        => Parallel.ForEach(source, body);
+    public static async Task ForEachAsync<T>(IEnumerable<T> source, int maxDegreeOfParallelism, Func<T, ProgressProvider?, Task> body,
+        CancellationToken cancellationToken = default, ProgressProvider? progress = null) {
+        using var linkedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+        var token = linkedCancellationTokenSource.Token;
+        var enumeratorLock = new object();
+        using var enumerator = source.GetEnumerator();
+        async Task WorkerAsync(ProgressProvider? cp) {
+            try {
+                while (true) {
+                    T item;
+                    lock (enumeratorLock) {
+                        token.ThrowIfCancellationRequested();
+                        if (!enumerator.MoveNext()) return;
+                        item = enumerator.Current;
+                    }
+                    await body(item, cp).ConfigureAwait(false);
+                }
+            } catch {
+                try {
+                    linkedCancellationTokenSource.Cancel(); // 让尚未开始的 worker 尽快停止
+                } catch {
+                }
+                throw;
+            }
+        }
+        await TaskUtils.WhenAll(Enumerable.Repeat(WorkerAsync, maxDegreeOfParallelism), progress).ConfigureAwait(false);
+    }
+
     /// <summary>
     /// 静默运行程序并等待其结束，返回其输出和退出码。
     /// 支持 notepad、git 等命令。
@@ -46,33 +122,4 @@ public static class TaskUtils {
         return (await outputTask + await errorTask, program.ExitCode);
     }
 
-    public static void ForEach<T>(IEnumerable<T> source, Action<T> body)
-        => Parallel.ForEach(source, body);
-    public static async Task ForEachAsync<T>(IEnumerable<T> source, int maxDegreeOfParallelism, Func<T, CancellationToken, Task> body, CancellationToken cancellationToken = default) {
-        using var linkedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
-        var token = linkedCancellationTokenSource.Token;
-        var enumeratorLock = new object();
-        using var enumerator = source.GetEnumerator();
-        async Task WorkerAsync() {
-            try {
-                while (true) {
-                    T item;
-                    lock (enumeratorLock) {
-                        token.ThrowIfCancellationRequested();
-                        if (!enumerator.MoveNext()) return;
-                        item = enumerator.Current;
-                    }
-                    await body(item, token).ConfigureAwait(false);
-                }
-            } catch {
-                try {
-                    linkedCancellationTokenSource.Cancel(); // 让尚未开始的 worker 尽快停止
-                } catch {
-                }
-                throw;
-            }
-        }
-        await Task.WhenAll([.. Enumerable.Range(0, maxDegreeOfParallelism).Select(_ => Task.Run(WorkerAsync))]).ConfigureAwait(false);
-    }
-
 }

Tests/Utils/TaskUtilsTest.cs

@@ -0,0 +1,45 @@
+namespace MeloongCore.Tests;
+
+public class TaskUtilsTest : TestBase {
+
+    [Test]
+    public async Task WhenAll_汇总等权子进度并保持结果顺序() {
+        var progress = new ProgressProvider();
+        var continueSecond = new TaskCompletionSource<object?>(TaskCreationOptions.RunContinuationsAsynchronously);
+        ProgressProvider? firstProgress = null, secondProgress = null;
+
+        var whenAll = TaskUtils.WhenAll<int>([
+            child => {
+                firstProgress = child;
+                return Task.FromResult(1);
+            },
+            async child => {
+                secondProgress = child;
+                await continueSecond.Task;
+                return 2;
+            }
+        ], progress);
+
+        await Assert.That(ReferenceEquals(firstProgress, secondProgress)).IsFalse();
+        await Assert.That(progress.Observe()).IsEqualTo(0.5);
+        continueSecond.SetResult(null);
+        var results = await whenAll;
+        await Assert.That(string.Join(",", results)).IsEqualTo("1,2");
+        await Assert.That(progress.Observe()).IsEqualTo(1);
+    }
+
+    [Test]
+    public async Task WhenAll_省略父进度时仍传入独立进度() {
+        ProgressProvider? receivedProgress = null;
+
+        await TaskUtils.WhenAll([
+            child => {
+                receivedProgress = child;
+                return Task.CompletedTask;
+            }
+        ]);
+
+        await Assert.That(receivedProgress is not null).IsTrue();
+    }
+
+}