Merge branch 'master' of ssh://git.swordlost.top:222/SikongJueluo/FPGA_WebLab
This commit is contained in:
alivender
147
server/src/Services/ProgressTracker.cs
Normal file
147
server/src/Services/ProgressTracker.cs
Normal file
@@ -0,0 +1,147 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Microsoft.AspNetCore.SignalR;
|
||||
using server.Hubs;
|
||||
|
||||
namespace server.Services;
|
||||
|
||||
public enum TaskState { Running, Completed, Failed, Cancelled }
|
||||
|
||||
public readonly struct TaskProgress
|
||||
{
|
||||
public string Id { get; }
|
||||
public int Current { get; }
|
||||
public int Total { get; }
|
||||
public TaskState State { get; }
|
||||
public long Timestamp { get; }
|
||||
public string? Error { get; }
|
||||
|
||||
public TaskProgress(string id, int current, int total, TaskState state, long timestamp, string? error = null)
|
||||
{
|
||||
Id = id;
|
||||
Current = current;
|
||||
Total = total;
|
||||
State = state;
|
||||
Timestamp = timestamp;
|
||||
Error = error;
|
||||
}
|
||||
|
||||
public TaskProgress WithUpdate(int? current = null, TaskState? state = null, string? error = null)
|
||||
{
|
||||
return new TaskProgress(
|
||||
Id,
|
||||
current ?? Current,
|
||||
Total,
|
||||
state ?? State,
|
||||
DateTimeOffset.UtcNow.ToUnixTimeSeconds(),
|
||||
error ?? Error
|
||||
);
|
||||
}
|
||||
|
||||
public ProgressInfo ToProgressInfo()
|
||||
{
|
||||
return new ProgressInfo
|
||||
{
|
||||
TaskId = Id,
|
||||
Status = State switch
|
||||
{
|
||||
TaskState.Running => ProgressStatus.Running,
|
||||
TaskState.Completed => ProgressStatus.Completed,
|
||||
TaskState.Failed => ProgressStatus.Failed,
|
||||
TaskState.Cancelled => ProgressStatus.Canceled,
|
||||
_ => ProgressStatus.Failed
|
||||
},
|
||||
ProgressPercent = Total > 0 ? ((double)Current * 100) / (double)Total : 0,
|
||||
ErrorMessage = Error ?? string.Empty
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
public sealed class ProgressTracker
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, TaskProgress> _tasks = new();
|
||||
private readonly Timer _cleaner;
|
||||
private readonly IHubContext<ProgressHub, IProgressReceiver> _hubContext;
|
||||
|
||||
// 构造器支持可选的Hub注入
|
||||
public ProgressTracker(IHubContext<ProgressHub, IProgressReceiver> hubContext)
|
||||
{
|
||||
_hubContext = hubContext;
|
||||
_cleaner = new Timer(CleanExpiredTasks, null,
|
||||
TimeSpan.FromMinutes(1), TimeSpan.FromMinutes(1));
|
||||
}
|
||||
|
||||
public void CleanExpiredTasks(object? obj)
|
||||
{
|
||||
var cutoff = DateTimeOffset.Now.AddMinutes(-3).ToUnixTimeSeconds();
|
||||
var expired = _tasks.Where(kvp => kvp.Value.Timestamp < cutoff).Select(kvp => kvp.Key).ToList();
|
||||
foreach (var id in expired)
|
||||
{
|
||||
_tasks.TryRemove(id, out _);
|
||||
}
|
||||
}
|
||||
|
||||
public string CreateTask(int total)
|
||||
{
|
||||
var id = Guid.NewGuid().ToString();
|
||||
var task = new TaskProgress(id, 0, total, TaskState.Running, DateTimeOffset.UtcNow.ToUnixTimeSeconds());
|
||||
_tasks[id] = task;
|
||||
NotifyIfNeeded(task);
|
||||
return id;
|
||||
}
|
||||
|
||||
// 核心更新方法，现在包含自动通知
|
||||
public bool UpdateTask(string id, Func<TaskProgress, TaskProgress> updater)
|
||||
{
|
||||
if (!_tasks.TryGetValue(id, out var current))
|
||||
return false;
|
||||
|
||||
var updated = updater(current);
|
||||
if (_tasks.TryUpdate(id, updated, current))
|
||||
{
|
||||
NotifyIfNeeded(updated);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// 自动通知逻辑 - 简单直接
|
||||
private void NotifyIfNeeded(TaskProgress task)
|
||||
{
|
||||
_hubContext.Clients.Group(task.Id).OnReceiveProgress(task.ToProgressInfo());
|
||||
}
|
||||
|
||||
public bool UpdateProgress(string id, int current)
|
||||
{
|
||||
return UpdateTask(id, p => p.WithUpdate(
|
||||
current: Math.Min(current, p.Total)));
|
||||
}
|
||||
|
||||
public bool AdvanceProgress(string id, int steps)
|
||||
{
|
||||
return UpdateTask(id, p => p.WithUpdate(
|
||||
current: Math.Min(p.Current + steps, p.Total)));
|
||||
}
|
||||
|
||||
public bool CancelProgress(string id)
|
||||
{
|
||||
return UpdateTask(id, p => p.WithUpdate(state: TaskState.Cancelled));
|
||||
}
|
||||
|
||||
public bool CompleteProgress(string id)
|
||||
{
|
||||
return UpdateTask(id, p => p.WithUpdate(
|
||||
current: p.Total, state: TaskState.Completed));
|
||||
}
|
||||
|
||||
public bool FailProgress(string id, string? error)
|
||||
{
|
||||
return UpdateTask(id, p => p.WithUpdate(
|
||||
state: TaskState.Failed, error: error));
|
||||
}
|
||||
|
||||
public TaskProgress? GetTask(string id)
|
||||
{
|
||||
_tasks.TryGetValue(id, out var task);
|
||||
return task.Id == null ? null : task;
|
||||
}
|
||||
}
|
||||
@@ -1,294 +0,0 @@
|
||||
using Microsoft.AspNetCore.SignalR;
|
||||
using System.Collections.Concurrent;
|
||||
using DotNext;
|
||||
using Common;
|
||||
using server.Hubs;
|
||||
|
||||
namespace server.Services;
|
||||
|
||||
public class ProgressReporter : ProgressInfo, IProgress<int>
|
||||
{
|
||||
private int _progress = 0;
|
||||
private int _stepProgress = 1;
|
||||
private int _expectedSteps = 100;
|
||||
private int _parentProportion = 100;
|
||||
|
||||
public int Progress => _progress;
|
||||
public int MaxProgress { get; set; } = 100;
|
||||
public int StepProgress
|
||||
{
|
||||
get => _stepProgress;
|
||||
set
|
||||
{
|
||||
_stepProgress = value;
|
||||
_expectedSteps = MaxProgress / value;
|
||||
}
|
||||
}
|
||||
public int ExpectedSteps
|
||||
{
|
||||
get => _expectedSteps;
|
||||
set
|
||||
{
|
||||
_expectedSteps = value;
|
||||
MaxProgress = Number.IntPow(10, Number.GetLength(value));
|
||||
_stepProgress = MaxProgress / value;
|
||||
}
|
||||
}
|
||||
public Func<int, Task>? ReporterFunc { get; set; } = null;
|
||||
public ProgressReporter? Parent { get; set; }
|
||||
public ProgressReporter? Child { get; set; }
|
||||
|
||||
private ProgressStatus _status = ProgressStatus.Pending;
|
||||
private string _errorMessage = string.Empty;
|
||||
|
||||
public override string TaskId { get; } = Guid.NewGuid().ToString();
|
||||
public override int ProgressPercent => _progress * 100 / MaxProgress;
|
||||
public override ProgressStatus Status => _status;
|
||||
public override string ErrorMessage => _errorMessage;
|
||||
|
||||
public ProgressReporter(Func<int, Task>? reporter = null, int initProgress = 0, int maxProgress = 100, int step = 1)
|
||||
{
|
||||
_progress = initProgress;
|
||||
MaxProgress = maxProgress;
|
||||
StepProgress = step;
|
||||
ReporterFunc = reporter;
|
||||
}
|
||||
|
||||
public ProgressReporter(int parentProportion, int expectedSteps = 100, Func<int, Task>? reporter = null)
|
||||
{
|
||||
this._parentProportion = parentProportion;
|
||||
MaxProgress = Number.IntPow(10, Number.GetLength(expectedSteps));
|
||||
StepProgress = MaxProgress / expectedSteps;
|
||||
ReporterFunc = reporter;
|
||||
}
|
||||
|
||||
private async void ForceReport(int value)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (ReporterFunc != null)
|
||||
await ReporterFunc(value);
|
||||
|
||||
if (Parent != null)
|
||||
Parent.Increase((value - _progress) / StepProgress * _parentProportion / (MaxProgress / StepProgress));
|
||||
|
||||
_progress = value;
|
||||
}
|
||||
catch (OperationCanceledException ex)
|
||||
{
|
||||
_errorMessage = ex.Message;
|
||||
this._status = ProgressStatus.Canceled;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_errorMessage = ex.Message;
|
||||
this._status = ProgressStatus.Failed;
|
||||
}
|
||||
}
|
||||
|
||||
public void Report(int value)
|
||||
{
|
||||
if (this._status == ProgressStatus.Pending)
|
||||
this._status = ProgressStatus.InProgress;
|
||||
else if (this.Status != ProgressStatus.InProgress)
|
||||
return;
|
||||
|
||||
if (value > MaxProgress) return;
|
||||
ForceReport(value);
|
||||
}
|
||||
|
||||
public void Increase(int? value = null)
|
||||
{
|
||||
if (this._status == ProgressStatus.Pending)
|
||||
this._status = ProgressStatus.InProgress;
|
||||
else if (this.Status != ProgressStatus.InProgress)
|
||||
return;
|
||||
|
||||
if (value.HasValue)
|
||||
{
|
||||
if (_progress + value.Value >= MaxProgress) return;
|
||||
this.Report(_progress + value.Value);
|
||||
}
|
||||
else
|
||||
{
|
||||
if (_progress + StepProgress >= MaxProgress) return;
|
||||
this.Report(_progress + StepProgress);
|
||||
}
|
||||
}
|
||||
|
||||
public void Finish()
|
||||
{
|
||||
this._status = ProgressStatus.Completed;
|
||||
this.ForceReport(MaxProgress);
|
||||
}
|
||||
|
||||
public void Cancel()
|
||||
{
|
||||
this._status = ProgressStatus.Canceled;
|
||||
this._errorMessage = "User Cancelled";
|
||||
this.ForceReport(_progress);
|
||||
}
|
||||
|
||||
public void Error(string message)
|
||||
{
|
||||
this._status = ProgressStatus.Failed;
|
||||
this._errorMessage = message;
|
||||
this.ForceReport(_progress);
|
||||
}
|
||||
|
||||
public ProgressReporter CreateChild(int proportion, int expectedSteps = 100)
|
||||
{
|
||||
var child = new ProgressReporter(proportion, expectedSteps);
|
||||
child.Parent = this;
|
||||
this.Child = child;
|
||||
return child;
|
||||
}
|
||||
}
|
||||
|
||||
public class ProgressTrackerService : BackgroundService
|
||||
{
|
||||
private static NLog.Logger logger = NLog.LogManager.GetCurrentClassLogger();
|
||||
private readonly ConcurrentDictionary<string, TaskProgressInfo> _taskMap = new();
|
||||
private readonly IHubContext<ProgressHub, IProgressReceiver> _hubContext;
|
||||
|
||||
private class TaskProgressInfo
|
||||
{
|
||||
public ProgressReporter? Reporter { get; set; }
|
||||
public string? ConnectionId { get; set; }
|
||||
public required CancellationToken CancellationToken { get; set; }
|
||||
public required CancellationTokenSource CancellationTokenSource { get; set; }
|
||||
public required DateTime UpdatedAt { get; set; }
|
||||
}
|
||||
|
||||
public ProgressTrackerService(IHubContext<ProgressHub, IProgressReceiver> hubContext)
|
||||
{
|
||||
_hubContext = hubContext;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
var now = DateTime.UtcNow;
|
||||
foreach (var kvp in _taskMap)
|
||||
{
|
||||
var info = kvp.Value;
|
||||
// 超过 1 分钟且任务已完成/失败/取消
|
||||
if (
|
||||
(now - info.UpdatedAt).TotalMinutes > 1 &&
|
||||
info.Reporter != null &&
|
||||
(
|
||||
info.Reporter.Status == ProgressStatus.Completed ||
|
||||
info.Reporter.Status == ProgressStatus.Failed ||
|
||||
info.Reporter.Status == ProgressStatus.Canceled
|
||||
)
|
||||
)
|
||||
{
|
||||
_taskMap.TryRemove(kvp.Key, out _);
|
||||
logger.Info($"Cleaned up task {kvp.Key}");
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.Error(ex, "Error during ProgressTracker cleanup");
|
||||
}
|
||||
await Task.Delay(TimeSpan.FromSeconds(30), stoppingToken);
|
||||
}
|
||||
}
|
||||
|
||||
public (string, ProgressReporter) CreateTask(CancellationToken? cancellationToken = null)
|
||||
{
|
||||
CancellationTokenSource? cancellationTokenSource;
|
||||
if (cancellationToken.HasValue)
|
||||
{
|
||||
cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken.Value);
|
||||
}
|
||||
else
|
||||
{
|
||||
cancellationTokenSource = new CancellationTokenSource();
|
||||
}
|
||||
|
||||
var progressInfo = new TaskProgressInfo
|
||||
{
|
||||
ConnectionId = null,
|
||||
UpdatedAt = DateTime.UtcNow,
|
||||
CancellationToken = cancellationTokenSource.Token,
|
||||
CancellationTokenSource = cancellationTokenSource,
|
||||
};
|
||||
|
||||
var progress = new ProgressReporter(async value =>
|
||||
{
|
||||
cancellationTokenSource.Token.ThrowIfCancellationRequested();
|
||||
|
||||
// 通过 SignalR 推送进度
|
||||
if (progressInfo.ConnectionId != null && progressInfo.Reporter != null)
|
||||
await _hubContext.Clients.Client(progressInfo.ConnectionId).OnReceiveProgress(progressInfo.Reporter);
|
||||
});
|
||||
|
||||
progressInfo.Reporter = progress;
|
||||
|
||||
_taskMap.TryAdd(progressInfo.Reporter.TaskId, progressInfo);
|
||||
|
||||
return (progressInfo.Reporter.TaskId, progress);
|
||||
}
|
||||
|
||||
public Optional<ProgressReporter> GetReporter(string taskId)
|
||||
{
|
||||
if (_taskMap.TryGetValue(taskId, out var info))
|
||||
{
|
||||
return info.Reporter;
|
||||
}
|
||||
return Optional<ProgressReporter>.None;
|
||||
}
|
||||
|
||||
public Optional<ProgressStatus> GetProgressStatus(string taskId)
|
||||
{
|
||||
if (_taskMap.TryGetValue(taskId, out var info))
|
||||
{
|
||||
if (info.Reporter != null)
|
||||
return info.Reporter.Status;
|
||||
}
|
||||
return Optional<ProgressStatus>.None;
|
||||
}
|
||||
|
||||
public bool BindTask(string taskId, string connectionId)
|
||||
{
|
||||
if (_taskMap.TryGetValue(taskId, out var info) && info != null)
|
||||
{
|
||||
lock (info)
|
||||
{
|
||||
info.ConnectionId = connectionId;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
public bool CancelTask(string taskId)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (_taskMap.TryGetValue(taskId, out var info) && info != null && info.Reporter != null)
|
||||
{
|
||||
lock (info)
|
||||
{
|
||||
info.CancellationTokenSource.Cancel();
|
||||
info.Reporter.Cancel();
|
||||
info.UpdatedAt = DateTime.UtcNow;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.Error(ex, $"Failed to cancel task {taskId}");
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user