RunProgramAsync 支持取消、TaskUtils.ForEachAsync

Shared/General/Config.cs

@@ -104,12 +104,12 @@ public static class ConfigUtils {
 
 public class ConfigEntry<T>(string key, T? defaultValue, IConfigProvider? defaultProvider = null, bool encrypted = false) {
 
+    public IConfigProvider DefaultProvider => defaultProvider ?? ConfigUtils.AppData;
     /// <summary>
     /// 当设置项的值实际被改变时触发，参数为新值。
     /// </summary>
     public event Action<T?, IConfigProvider>? Changed;
 
-    public IConfigProvider DefaultProvider => defaultProvider ?? ConfigUtils.AppData;
     public void Edit(Func<T?, T?> editFunc, IConfigProvider? providerOverride = null) {
         var provider = providerOverride ?? DefaultProvider;
         Set(editFunc(Get(provider)), provider);
@@ -139,7 +139,5 @@ public class ConfigEntry<T>(string key, T? defaultValue, IConfigProvider? defaul
         => (providerOverride ?? DefaultProvider).Read(key, defaultValue, encrypted);
     public bool HasValue(IConfigProvider? providerOverride = null)
         => (providerOverride ?? DefaultProvider).HasValue(key);
-    public void Save(IConfigProvider? providerOverride = null) 
-        => (providerOverride ?? DefaultProvider).Save();
     public static implicit operator T?(ConfigEntry<T> entry) => entry.Get();
 }

Shared/Utils/TaskUtils.cs

@@ -11,7 +11,8 @@ public static class TaskUtils {
     /// 支持 notepad、git 等命令。
     /// 超时或启动失败会抛出异常。
     /// </summary>
-    public static async Task<(string Output, int ExitCode)> RunProgramAsync(string file, string arguments = "", int? timeoutMs = null, Encoding? encoding = null) {
+    public static async Task<(string Output, int ExitCode)> RunProgramAsync(string file, string arguments = "", int? timeoutMs = null, Encoding? encoding = null, CancellationToken cancellationToken = default) {
+        cancellationToken.ThrowIfCancellationRequested();
         var info = new ProcessStartInfo {
             Arguments = arguments,
             FileName = PathUtils.ToShortPath(file),
@@ -25,27 +26,54 @@ public static class TaskUtils {
         bool hasTimeout = timeoutMs.HasValue && timeoutMs > 0;
         Logger.Info($"运行程序，并返回其输出：{file} {arguments}{(hasTimeout ? $"，最长可等待 {timeoutMs}ms" : "")}");
 
-        // 读取输出和错误流
+        // 输出和错误流
         var outputTask = program.StandardOutput.ReadToEndAsync();
         var errorTask = program.StandardError.ReadToEndAsync();
+
         var completionSource = new TaskCompletionSource<object?>(TaskCreationOptions.RunContinuationsAsynchronously);
         program.Exited += (_, _) => completionSource.TrySetResult(null);
         if (program.HasExited) completionSource.TrySetResult(null);
         var task = completionSource.Task;
+        var timeoutTask = Task.Delay(hasTimeout ? timeoutMs!.Value : Timeout.Infinite);
+        var cancelTask = Task.Delay(Timeout.Infinite, cancellationToken);
+        Logger.Trace($"等待程序 {program.Id} 完成");
+        var completedTask = await Task.WhenAny(task, timeoutTask, cancelTask);
+        if (completedTask != task) {
+            try {
+                if (!program.HasExited) program.Kill();
+            } catch (InvalidOperationException) { // 进程已退出，无需处理
+            }
+            if (completedTask == cancelTask) throw new OperationCanceledException(cancellationToken);
+            throw new TimeoutException($"运行程序超时：{file} {arguments}");
+        }
+        return (await outputTask + await errorTask, program.ExitCode);
+    }
 
-        // 等待超时
-        if (hasTimeout) {
-            Logger.Trace($"等待程序 {program.Id} 完成");
-            if (await Task.WhenAny(task, Task.Delay(timeoutMs!.Value)) != task) {
+    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 {
-                    if (!program.HasExited) program.Kill();
-                } catch (InvalidOperationException) { // 进程已退出，无需处理
+                    linkedCancellationTokenSource.Cancel(); // 让尚未开始的 worker 尽快停止
+                } catch {
                 }
-                throw new TimeoutException($"运行程序超时：{file} {arguments}");
+                throw;
             }
         }
-        await task;
-        return (await outputTask + await errorTask, program.ExitCode);
+        await Task.WhenAll([.. Enumerable.Range(0, maxDegreeOfParallelism).Select(_ => Task.Run(WorkerAsync))]).ConfigureAwait(false);
     }
 
 }