worker 和进度相关

Shared/General/Progress.cs

@@ -7,10 +7,11 @@ namespace MeloongCore;
 public class ProgressProvider {
 
     /// <summary>
-    /// 当进度被改变时触发。
+    /// 当原始进度可能被改变时触发。
     /// </summary>
-    public event Action? ProgressChanged;
-    private void InvokeProgressChanged() => ProgressChanged?.Invoke();
+    public event Action<(double actual, double skiped)>? ProgressChanged;
+    private void InvokeProgressChanged((double actual, double skiped) raw) => ProgressChanged?.Invoke(raw);
+    private void InvokeProgressChanged() => InvokeProgressChanged(GetRaw());
 
     // ===================================== 主项进度 =====================================
 
@@ -122,7 +123,7 @@ public class ProgressProvider {
             if (childrens.Any(c => c.percentage > 0)) { // 将子项的进度加入主项
                 var mult = progressParts.splited / childrens.Sum(c => c.percentage);
                 childrens.ForEach(c => {
-                    var (subActual, subSkiped) = c.child.GetTotalProgress();
+                    var (subActual, subSkiped) = c.child.GetRaw();
                     progressParts.actual += mult * c.percentage * (action == ChildrenAction.Finished ? (1 - subSkiped) : subActual);
                     progressParts.skiped += mult * c.percentage * (action == ChildrenAction.Skiped ? (1 - subActual) : subSkiped);
                     c.child.ProgressChanged -= InvokeProgressChanged;
@@ -140,13 +141,13 @@ public class ProgressProvider {
     /// <summary>
     /// 获取包含子项进度的实际总进度值。
     /// </summary>
-    private (double actual, double skiped) GetTotalProgress() {
+    public (double actual, double skiped) GetRaw() {
         lock (this) {
             (double actual, double skiped) current = (progressParts.actual, progressParts.skiped);
             if (childrens.Any(c => c.percentage > 0)) { // 将子项的进度加入主项
                 var mult = progressParts.splited / childrens.Sum(c => c.percentage);
                 childrens.ForEach(c => {
-                    var (subActual, subSkiped) = c.child.GetTotalProgress();
+                    var (subActual, subSkiped) = c.child.GetRaw();
                     current.actual += mult * c.percentage * subActual;
                     current.skiped += mult * c.percentage * subSkiped;
                 });
@@ -160,9 +161,9 @@ public class ProgressProvider {
     /// <summary>
     /// 获取一个在多次观测之间必定单调递增的进度值，范围为 [0, 1]。
     /// </summary>
-    public double Observe() {
+    public double GetIncrement() {
         lock (this) {
-            (double actual, double skiped) current = GetTotalProgress();
+            (double actual, double skiped) current = GetRaw();
             if (current == observedProgress) return incrementProgress; // 未改变
             if (current.actual + current.skiped > 0.9999999) { // 已完成
                 incrementProgress = 1;
@@ -185,15 +186,15 @@ public class ProgressObserver : RateLimitedWorker {
     public ProgressObserver(ProgressProvider provider, Action<double> updateAction, double minimalIntervalMs = 70)
         : base(() => Observe(provider, updateAction), minimalIntervalMs, RateLimitMode.ImmediateThenMerge) {
         this.provider = provider;
-        provider.ProgressChanged += Invoke;
+        provider.ProgressChanged += StartObserver;
     }
     public override void Dispose() {
-        provider.ProgressChanged -= Invoke;
+        provider.ProgressChanged -= StartObserver;
         base.Dispose();
     }
-
+    private void StartObserver((double actual, double skiped) raw) => Start();
     private static void Observe(ProgressProvider provider, Action<double> updateAction) {
-        double progress = provider.Observe();
+        double progress = provider.GetIncrement();
         // TODO: 添加平缓过渡
         updateAction(progress);
     }

Shared/General/Workers/RateLimitedWorker.cs

@@ -21,7 +21,7 @@ public class RateLimitedWorker(Action workload, double minimalIntervalMs, RateLi
         RateLimitMode.ImmediateThenIgnore => Debouncer.Debounce(workload, TimeSpan.FromMilliseconds(minimalIntervalMs), leading: true, trailing: false)
     };
 
-    public void Invoke() {
+    public void Start() {
         if (!disposed) action.Invoke();
     }
 

Shared/General/Workers/RedoableWorker.cs

@@ -1,50 +1,63 @@
 namespace MeloongCore;
 
-/// <summary>
-/// 可合并和重启的工作器。
-/// <para/> 在工作负载运行期间多次调用 <see cref="Start(CancellationToken)"/> 会取消当前负载并使用最新令牌重新执行；多次调用也只重新执行一次。
-/// </summary>
-public interface IRedoableWorker {
-    /// <summary>当前是否正在运行。</summary>
-    bool Running { get; }
-    /// <summary>是否曾经有过未被取消且未失败的运行。</summary>
-    bool HasSucceeded { get; }
-    /// <summary>标识上次进入空闲时，并非因为取消或失败，而是正常运行到结束。</summary>
-    bool LastSucceeded { get; }
-    /// <summary>在工作线程运行工作负载。若当前已在运行，则取消当前负载并使用最新令牌重启。</summary>
-    void Start(CancellationToken cancellationToken = default);
+public interface IWorker {
+    /// <summary>在工作线程运行工作负载。</summary>
+    void Start(CancellationToken cancellationToken = default, ProgressProvider? progressOverride = null);
     /// <summary>取消当前运行。</summary>
     void Cancel();
-    /// <summary>仅在当前处于运行状态时，等待其完成。
+    /// <summary>若当前正在运行中，等待其完成。若未在运行中，则立即返回。</summary>
     /// <returns>若未超时则返回 true。</returns>
     bool WaitIfRunning(int millisecondsTimeout = -1, CancellationToken cancellationToken = default);
-    /// <summary>仅在当前处于运行状态时，异步等待其完成。
+    /// <summary>若当前正在运行中，异步等待其完成。若未在运行中，则立即返回。</summary>
     /// <returns>若未超时则返回 true。</returns>
     Task<bool> WaitIfRunningAsync(int millisecondsTimeout = -1, CancellationToken cancellationToken = default);
+
+    /// <summary>从空闲状态进入运行状态时触发。</summary>
+    event Action? Started;
+    /// <summary>运行结束并进入空闲状态时触发。</summary>
+    event Action? Stopped;
+    /// <summary>工作负载成功完成时触发。</summary>
+    public event Action? Succeeded;
+    /// <summary>工作负载执行失败时触发，参数为发生的异常。</summary>
+    public event Action<Exception>? Failed;
+    /// <summary>运行被取消时触发。</summary>
+    event Action? Canceled;
+
+    /// <summary>当前是否正在运行。</summary>
+    bool Running { get; }
+    /// <summary>是否曾经有过未被取消且未失败的运行。</summary>
+    bool HasSucceeded { get; }
+    /// <summary>标识上次进入空闲时，并非因为取消或失败，而是正常运行到结束。</summary>
+    bool LastSucceeded { get; }
+    /// <summary>当前工作负载的执行进度。</summary>
+    ProgressProvider Progress { get; }
 }
 
 // TODO: 现在的可观测性不佳，考虑增加调用方信息以及自身名称等，输出到日志，例如 CallerArgumentExpression
-/// <inheritdoc cref="IRedoableWorker"/>
-public abstract class RedoableWorkerBase<TOut>(Func<CancellationToken?, ProgressProvider?, TOut> workload, ProgressProvider? progress = null) : IRedoableWorker {
+/// <summary>
+/// 可合并和重启的工作器。
+/// <para/> 在工作负载运行期间多次调用 <see cref="Start(CancellationToken)"/> 会取消当前负载并使用最新令牌重新执行；多次调用也只重新执行一次。
+/// </summary>
+public abstract class RedoableWorkerBase<TOut>(Func<CancellationToken?, ProgressProvider?, TOut> workload, ProgressProvider? progress = null) : IWorker {
 
     // ============================================ 状态与事件 ============================================
 
     /// <inheritdoc/>
     public bool Running { get { lock (this) return running; } }
     private bool running;
-    /// <summary>从空闲状态进入运行状态时触发。</summary>
+    /// <inheritdoc/>
     public event Action? Started;
-    /// <summary>运行结束并进入空闲状态时触发。</summary>
+    /// <inheritdoc/>
     public event Action? Stopped;
 
     /// <inheritdoc/>
     public bool HasSucceeded { get { lock (this) return hasSucceeded; } }
     private bool hasSucceeded;
-    /// <summary>工作负载成功完成时触发，参数为返回值。<para/>这可能会在重启前触发。</summary>
-    public event Action<TOut>? Succeeded;
-    /// <summary>工作负载执行失败时触发，参数为发生的异常。<para/>这可能会在重启前触发。</summary>
+    /// <inheritdoc/>
+    public event Action? Succeeded;
+    /// <inheritdoc/>
     public event Action<Exception>? Failed;
-    /// <summary>运行被取消时触发。<para/>这可能会在重启前触发。</summary>
+    /// <inheritdoc/>
     public event Action? Canceled;
 
     /// <inheritdoc/>
@@ -62,7 +75,8 @@ public abstract class RedoableWorkerBase<TOut>(Func<CancellationToken?, Progress
     }
     private object? lastResult;
 
-    public readonly ProgressProvider Progress = progress ?? new ProgressProvider();
+    /// <inheritdoc/>
+    public ProgressProvider Progress { get; private set; } = progress ?? new ProgressProvider();
 
     private bool pendingRedo;
 
@@ -72,11 +86,12 @@ public abstract class RedoableWorkerBase<TOut>(Func<CancellationToken?, Progress
     private CancellationToken lastToken;
     private readonly ManualResetEventSlim idleEvent = new(initialState: true);
 
-    /// <inheritdoc/>
-    public void Start(CancellationToken cancellationToken = default) {
+    /// <summary>在工作线程运行工作负载。若当前已在运行，则取消当前负载并使用最新令牌重启。</summary>
+    public void Start(CancellationToken cancellationToken = default, ProgressProvider? progressOverride = null) {
         // 接取运行状态
         lock (this) {
             lastToken = cancellationToken;
+            if (progressOverride != null) Progress = progressOverride;
             if (running) {
                 pendingRedo = true;
                 realCts?.Cancel();
@@ -85,10 +100,10 @@ public abstract class RedoableWorkerBase<TOut>(Func<CancellationToken?, Progress
             running = true; idleEvent.Reset();
             realCts = CancellationTokenSource.CreateLinkedTokenSource(lastToken);
         }
-        var th = new Thread(_Invoke) { IsBackground = true };
+        var th = new Thread(_Start) { IsBackground = true, Name = nameof(RedoableWorker) };
         th.Start();
     }
-    private void _Invoke() {
+    private void _Start() {
         Started?.Invoke();
         while (true) {
             try {
@@ -96,7 +111,6 @@ public abstract class RedoableWorkerBase<TOut>(Func<CancellationToken?, Progress
                 Progress?.Reset();
                 TOut result = workload(realCts.Token, Progress); // 实际的执行
                 realCts.Token.ThrowIfCancellationRequested();
-                Succeeded?.Invoke(result);
                 lock (this) {
                     realCts?.Dispose();
                     if (pendingRedo) { // 接取重启请求
@@ -111,13 +125,10 @@ public abstract class RedoableWorkerBase<TOut>(Func<CancellationToken?, Progress
                     running = false; idleEvent.Set();
                     Progress?.Finish();
                 }
+                Succeeded?.Invoke();
             } catch (Exception ex) {
-                if (ex.IsCanceled()) {
-                    Canceled?.Invoke();
-                } else {
-                    Failed?.Invoke(ex);
+                if (!ex.IsCanceled())
                     Logger.Log(ex, $"工作线程执行失败{(pendingRedo ? "，但即将重启，或可忽略" : "")}", pendingRedo ? LogLevel.Info : LogLevel.Warn);
-                }
                 lock (this) {
                     realCts?.Dispose();
                     if (pendingRedo) { // 接取重启请求
@@ -130,6 +141,11 @@ public abstract class RedoableWorkerBase<TOut>(Func<CancellationToken?, Progress
                     running = false; idleEvent.Set();
                     Progress?.Skip();
                 }
+                if (ex.IsCanceled()) {
+                    Canceled?.Invoke();
+                } else {
+                    Failed?.Invoke(ex);
+                }
             }
             break;
         }

Tests/General/ProgressTest.cs

@@ -40,7 +40,7 @@ public class ProgressTest : TestBase {
         progress.Set(0.75, skiped: true);
 
         await Assert.That(changedCount).IsEqualTo(2);
-        await Assert.That(Math.Abs(progress.Observe() - 2.0 / 3) < 0.000001).IsTrue();
+        await Assert.That(Math.Abs(progress.GetIncrement() - 2.0 / 3) < 0.000001).IsTrue();
     }
 
     [Test]

Tests/Utils/TaskUtilsTest.cs

@@ -21,11 +21,11 @@ public class TaskUtilsTest : TestBase {
         ], progress);
 
         await Assert.That(ReferenceEquals(firstProgress, secondProgress)).IsFalse();
-        await Assert.That(progress.Observe()).IsEqualTo(0.5);
+        await Assert.That(progress.GetIncrement()).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);
+        await Assert.That(progress.GetIncrement()).IsEqualTo(1);
     }
 
     [Test]