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