Worker、Progress 优化

Shared/General/Progress.cs

@@ -1,33 +1,83 @@
 namespace MeloongCore;
 public class ProgressProvider {
 
-    // 设置 workerProgress
-    private (double actual, double skiped) workerProgress = (0, 0);
+    // 主项进度
+    private (double actual, double skiped, double splited) progressParts = (0, 0, 0);
+    private double progressSum => progressParts.actual + progressParts.skiped + progressParts.splited;
+    /// <summary>
+    /// 将当前进度设置为指定值。
+    /// 若 <paramref name="skiped"/> 为 true，这段进度的增量值将被计为跳过：当前观测进度不会增加，后续观测进度将增加地更快。
+    /// </summary>
     public void Set(double value, bool skiped = false) {
         value = value.Clamp(0, 1);
-        double delta = value - workerProgress.actual - workerProgress.skiped;
+        double delta = value - progressSum;
         if (delta < 0) { // 等比减少
-            workerProgress = (workerProgress.actual * value / (workerProgress.actual + workerProgress.skiped), 
-                              workerProgress.skiped * value / (workerProgress.actual + workerProgress.skiped));
+            progressParts = (progressParts.actual * value / progressSum,
+                              progressParts.skiped * value / progressSum,
+                              progressParts.splited * value / progressSum);
         } else if (skiped) {
-            workerProgress.skiped += delta;
+            progressParts.skiped += delta;
         } else {
-            workerProgress.actual += delta;
+            progressParts.actual += delta;
         }
     }
+    /// <summary>
+    /// 将当前进度增加指定值。
+    /// 若 <paramref name="skiped"/> 为 true，这段进度的增量值将被计为跳过：当前观测进度不会增加，后续观测进度将增加地更快。
+    /// </summary>
     public void Add(double value, bool skiped = false) 
-        => Set(value + workerProgress.actual + workerProgress.skiped, skiped);
+        => Set(value + progressSum, skiped);
 
-    // 观察
-    private (double actual, double skiped) observedWorkerProgress = (0, 0);
+    // 子项进度
+    private readonly List<(double weight, ProgressProvider sub)> childrens = [];
+    /// <summary>
+    /// 拆分一个子项：当该子项完成时，当前进度将增至 <paramref name="value"/>。
+    /// </summary>
+    public ProgressProvider SplitTo(double value) 
+        => SplitBy(value - progressSum).First();
+    /// <summary>
+    /// 拆分多个子项，每个子项都占据对应指定的进度量。
+    /// <para/>例如，调用 <c>SplitBy(0.2, 0.3)</c> 将返回两个子项，分别占据总进度的 20%、30%。
+    /// </summary>
+    public List<ProgressProvider> SplitBy(params double[] percentages) {
+        var totalWeight = percentages.Sum().Clamp(0, 1);
+        var totalPercentage = (totalWeight + progressSum).Clamp(0, 1) - progressSum;
+        progressParts.splited += totalPercentage;
+        return percentages.Select(percentage => {
+            if (percentage <= 0) throw new ArgumentException("子项进度必须为正数");
+            var weight = percentage / totalWeight * totalPercentage;
+            var sub = new ProgressProvider();
+            childrens.Add((weight, sub));
+            return sub;
+        }).ToList();
+    }
+    private (double actual, double skiped) GetTotalProgress() {
+        (double actual, double skiped) current = (progressParts.actual, progressParts.skiped);
+        if (childrens.Any()) { // 将子项的进度加入主项
+            var perWeight = progressParts.splited / childrens.Sum(c => c.weight);
+            childrens.ForEach(c => {
+                var (subActual, subSkiped) = c.sub.GetTotalProgress();
+                current.actual += subActual * perWeight * c.weight;
+                current.skiped += subSkiped * perWeight * c.weight;
+            });
+        }
+        return current;
+    }
+
+    // 观测
+    private (double actual, double skiped) observedProgress = (0, 0);
     private double incrementProgress = 0;
+    /// <summary>
+    /// 获取一个在多次观测之间必定单调递增的进度值，范围为 [0, 1]。
+    /// </summary>
     public double Observe() {
-        if (workerProgress == observedWorkerProgress) return incrementProgress; // 未改变
-        if (workerProgress.actual >= observedWorkerProgress.actual) { // 进度增加
+        (double actual, double skiped) current = GetTotalProgress();
+        if (current == observedProgress) return incrementProgress; // 未改变
+        if (current.actual >= observedProgress.actual) { // 进度增加
             incrementProgress += (1 - incrementProgress) *
-                (workerProgress.actual - observedWorkerProgress.actual) / (1 - observedWorkerProgress.actual - workerProgress.skiped);
+                (current.actual - observedProgress.actual) / (1 - observedProgress.actual - current.skiped);
         }
-        observedWorkerProgress = workerProgress;
+        observedProgress = current;
         return incrementProgress;
     }
 

Shared/General/Worker.cs

@@ -3,7 +3,6 @@ namespace MeloongCore;
 /// <summary>
 /// 可合并和重启的工作器。
 /// <para/> 在工作负载运行期间多次调用 <see cref="Run()"/> 会取消当前负载并使用最新令牌重新执行；多次调用也只重新执行一次。
-/// <para/> 若即将重启，则忽略当前负载的异常。
 /// </summary>
 public interface IRedoableWorker {
     /// <summary>当前是否正在运行。</summary>
@@ -70,42 +69,44 @@ public abstract class RedoableWorkerBase<TOut>(Func<CancellationToken, TOut> wor
             running = true; idleEvent.Reset();
             realCts = CancellationTokenSource.CreateLinkedTokenSource(lastToken);
         }
-        var th = new Thread(_Run);
+        var th = new Thread(_Run) { IsBackground = true };
         th.Start();
     }
     private void _Run() {
-        try {
-            TOut result = workload(realCts!.Token); // 实际的执行
-            realCts.Token.ThrowIfCancellationRequested();
-            lock (this) {
-                realCts?.Dispose();
-                if (pendingRedo) { // 接取重启请求
-                    pendingRedo = false;
-                    realCts = CancellationTokenSource.CreateLinkedTokenSource(lastToken);
-                    _Run(); return;
-                } else {
+        while (true) {
+            try {
+                realCts!.Token.ThrowIfCancellationRequested();
+                TOut result = workload(realCts.Token); // 实际的执行
+                realCts.Token.ThrowIfCancellationRequested();
+                lock (this) {
+                    realCts?.Dispose();
+                    if (pendingRedo) { // 接取重启请求
+                        pendingRedo = false;
+                        realCts = CancellationTokenSource.CreateLinkedTokenSource(lastToken);
+                        continue;
+                    }
                     realCts = null;
+                    lastSucceeded = true;
+                    lastResult = result;
+                    hasSucceeded = true;
+                    running = false; idleEvent.Set();
                 }
-                running = false; idleEvent.Set();
-                lastSucceeded = true;
-                lastResult = result;
-                hasSucceeded = true;
-            }
-        } catch (Exception ex) {
-            lock (this) {
-                realCts?.Dispose();
-                if (pendingRedo) { // 接取重启请求
-                    pendingRedo = false;
-                    realCts = CancellationTokenSource.CreateLinkedTokenSource(lastToken);
-                    if (!ex.IsCanceled()) Logger.Info(ex, "在重启前出现了异常");
-                    _Run(); return;
-                } else {
+            } catch (Exception ex) {
+                if (!ex.IsCanceled()) 
+                    Logger.Log(ex, $"工作线程执行失败{(pendingRedo ? "，但即将重启，或可忽略" : "")}", pendingRedo ? LogLevel.Info : LogLevel.Warn);
+                lock (this) {
+                    realCts?.Dispose();
+                    if (pendingRedo) { // 接取重启请求
+                        pendingRedo = false;
+                        realCts = CancellationTokenSource.CreateLinkedTokenSource(lastToken);
+                        continue;
+                    }
                     realCts = null;
+                    lastSucceeded = false;
+                    running = false; idleEvent.Set();
                 }
-                running = false; idleEvent.Set();
-                lastSucceeded = false;
             }
-            if (!ex.IsCanceled()) Logger.Error(ex, "工作线程执行失败", LogBehavior.None);
+            return;
         }
     }