namespace MeloongCore; /// /// 可合并和重启的工作器。 /// 在工作负载运行期间多次调用 会取消当前负载并使用最新令牌重新执行;多次调用也只重新执行一次。 /// 若即将重启,则忽略当前负载的异常。 /// public interface IRedoableWorker { /// 当前是否正在运行。 bool Running { get; } /// 是否曾经有过未被取消且未失败的运行。 bool HasSucceeded { get; } /// 标识上次进入空闲时,并非因为取消或失败,而是正常运行到结束。 bool LastSucceeded { get; } /// 在工作线程运行工作负载。若当前已在运行,则取消当前负载并使用最新令牌重启。 void Run(CancellationToken cancellationToken = default); /// 取消当前运行。 void Cancel(); /// 仅在当前处于运行状态时,等待其完成。 /// 若未超时则返回 true。 bool WaitIfRunning(int millisecondsTimeout = -1); } /// public abstract class RedoableWorkerBase(Func workload) : IRedoableWorker { // ============================================ 状态 ============================================ /// public bool Running { get { lock (this) return running; } } private bool running; /// public bool HasSucceeded { get { lock (this) return hasSucceeded; } } private bool hasSucceeded; /// public bool LastSucceeded { get { lock (this) return lastSucceeded; } } private bool lastSucceeded; /// 上次未被取消且未失败的运行中,工作负载的返回值。 public TOut LastResult { get { lock (this) { if (!hasSucceeded) throw new InvalidOperationException("从未成功完成过。"); return lastResult == null ? default! : (TOut) lastResult; } } } private object? lastResult; private bool pendingRedo; // =========================================== 运行与取消 =========================================== private CancellationTokenSource? realCts; private CancellationToken lastToken; private readonly ManualResetEventSlim idleEvent = new(initialState: true); /// public void Run(CancellationToken cancellationToken = default) { // 接取运行状态 lock (this) { lastToken = cancellationToken; if (running) { pendingRedo = true; realCts?.Cancel(); return; } running = true; idleEvent.Reset(); realCts = CancellationTokenSource.CreateLinkedTokenSource(lastToken); } var th = new Thread(_Run); 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 { realCts = null; } 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 { realCts = null; } running = false; idleEvent.Set(); lastSucceeded = false; } if (!ex.IsCanceled()) Logger.Error(ex, "工作线程执行失败", LogBehavior.None); } } /// public void Cancel() { lock (this) { pendingRedo = false; realCts?.Cancel(); } } /// public bool WaitIfRunning(int millisecondsTimeout = -1) => idleEvent.Wait(millisecondsTimeout); } /// public sealed class RedoableWorker : RedoableWorkerBase { public RedoableWorker(Func workload) : base(workload) { } public RedoableWorker(Func workload) : base(_ => workload()) { } } /// public sealed class RedoableWorker : RedoableWorkerBase { public RedoableWorker(Action workload) : base(ct => { workload(ct); return null; }) { } public RedoableWorker(Action workload) : base(_ => { workload(); return null; }) { } }