namespace MeloongCore; /// /// 可合并和重启的工作器。 /// 在工作负载运行期间多次调用 会取消当前负载并使用最新令牌重新执行;多次调用也只重新执行一次。 /// public interface IRedoableWorker { /// 当前是否正在运行。 bool Running { get; } /// 是否曾经有过未被取消且未失败的运行。 bool HasSucceeded { get; } /// 标识上次进入空闲时,并非因为取消或失败,而是正常运行到结束。 bool LastSucceeded { get; } /// 在工作线程运行工作负载。若当前已在运行,则取消当前负载并使用最新令牌重启。 void Invoke(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 Invoke(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(_Invoke) { IsBackground = true }; th.Start(); } private void _Invoke() { 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(); } } 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(); } } return; } } /// public void Cancel() { lock (this) { pendingRedo = false; realCts?.Cancel(); } } /// public bool WaitIfRunning(int millisecondsTimeout = -1) => idleEvent.Wait(millisecondsTimeout); } /// public class RedoableWorker : RedoableWorkerBase { public RedoableWorker(Func workload) : base(workload) { } public RedoableWorker(Func workload) : base(_ => workload()) { } } /// public class RedoableWorker : RedoableWorkerBase { public RedoableWorker(Action workload) : base(ct => { workload(ct); return null; }) { } public RedoableWorker(Action workload) : base(_ => { workload(); return null; }) { } }