672 lines
19 KiB
C#
672 lines
19 KiB
C#
using Explorer.Application;
|
|
using Explorer.Contracts;
|
|
using Explorer.Domain;
|
|
using Explorer.Domain.Abstractions;
|
|
using Microsoft.Extensions.Hosting;
|
|
using Microsoft.Extensions.Logging;
|
|
|
|
namespace Explorer.FileOperations;
|
|
|
|
public sealed class TransferQueue : BackgroundService, ITransferHost, IForegroundWorkSignal
|
|
{
|
|
private readonly IOperationExecutor _executor;
|
|
private readonly IIndexStore _store;
|
|
private readonly IVolumeService _volumes;
|
|
private readonly ILogger<TransferQueue> _logger;
|
|
private readonly List<TransferJob> _jobs = [];
|
|
private readonly object _gate = new();
|
|
private readonly SemaphoreSlim _signal = new(0);
|
|
private readonly Func<bool> _pauseRequested;
|
|
private volatile bool _queuePaused;
|
|
private volatile bool _haltPause;
|
|
private long? _holdJobId;
|
|
private CancellationTokenSource? _runningCts;
|
|
private DateTime _lastChangedUtc = DateTime.MinValue;
|
|
private int _restored;
|
|
|
|
public event EventHandler? Changed;
|
|
public event EventHandler<TransferJob>? JobFinished;
|
|
|
|
public TransferQueue(
|
|
IOperationExecutor executor,
|
|
IIndexStore store,
|
|
IVolumeService volumes,
|
|
ILogger<TransferQueue> logger)
|
|
{
|
|
_executor = executor;
|
|
_store = store;
|
|
_volumes = volumes;
|
|
_logger = logger;
|
|
_pauseRequested = () => _haltPause;
|
|
}
|
|
|
|
public bool IsPaused => _queuePaused;
|
|
|
|
public bool HasForegroundWork()
|
|
{
|
|
lock (_gate)
|
|
{
|
|
return _jobs.Any(j => j.Status is TransferStatus.Queued or TransferStatus.Running
|
|
or TransferStatus.Cancelling or TransferStatus.Waiting);
|
|
}
|
|
}
|
|
|
|
public IReadOnlyList<TransferJob> Snapshot()
|
|
{
|
|
lock (_gate)
|
|
{
|
|
return _jobs.ToList();
|
|
}
|
|
}
|
|
|
|
public async Task EnqueueCopyAsync(IReadOnlyList<string> sources, string destinationDirectory, CancellationToken cancellationToken = default)
|
|
{
|
|
foreach (var src in sources)
|
|
{
|
|
var dest = Path.Combine(destinationDirectory, PathRules.GetFileName(src));
|
|
await EnqueueAsync(new TransferJob
|
|
{
|
|
Op = TransferOp.Copy,
|
|
SourcePath = src,
|
|
DestinationPath = dest,
|
|
Status = TransferStatus.Queued,
|
|
CreatedUtc = DateTimeOffset.UtcNow
|
|
}, cancellationToken).ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
public async Task EnqueueMoveAsync(IReadOnlyList<string> sources, string destinationDirectory, CancellationToken cancellationToken = default)
|
|
{
|
|
foreach (var src in sources)
|
|
{
|
|
var dest = Path.Combine(destinationDirectory, PathRules.GetFileName(src));
|
|
await EnqueueAsync(new TransferJob
|
|
{
|
|
Op = TransferOp.Move,
|
|
SourcePath = src,
|
|
DestinationPath = dest,
|
|
Status = TransferStatus.Queued,
|
|
CreatedUtc = DateTimeOffset.UtcNow
|
|
}, cancellationToken).ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
public async Task EnqueueDeleteAsync(IReadOnlyList<string> paths, bool permanent = false, CancellationToken cancellationToken = default)
|
|
{
|
|
await EnqueueAsync(new TransferJob
|
|
{
|
|
Op = TransferOp.Delete,
|
|
SourcePath = string.Join("|", paths),
|
|
DestinationPath = permanent ? "permanent" : "recycle",
|
|
Status = TransferStatus.Queued,
|
|
CreatedUtc = DateTimeOffset.UtcNow,
|
|
AdditionalSources = paths.ToList()
|
|
}, cancellationToken).ConfigureAwait(false);
|
|
}
|
|
|
|
public Task EnqueueRenameAsync(string path, string newName, CancellationToken cancellationToken = default)
|
|
{
|
|
var dest = Path.Combine(PathRules.Parent(path), newName);
|
|
return EnqueueAsync(new TransferJob
|
|
{
|
|
Op = TransferOp.Rename,
|
|
SourcePath = path,
|
|
DestinationPath = dest,
|
|
Status = TransferStatus.Queued,
|
|
CreatedUtc = DateTimeOffset.UtcNow
|
|
}, cancellationToken);
|
|
}
|
|
|
|
public Task EnqueueEmptyRecycleBinAsync(CancellationToken cancellationToken = default)
|
|
=> EnqueueAsync(new TransferJob
|
|
{
|
|
Op = TransferOp.EmptyRecycleBin,
|
|
SourcePath = LocationRoots.RecycleBin,
|
|
Status = TransferStatus.Queued,
|
|
CreatedUtc = DateTimeOffset.UtcNow
|
|
}, cancellationToken);
|
|
|
|
public Task EnqueueExtractAsync(string archivePath, string destinationDirectory, CancellationToken cancellationToken = default)
|
|
=> EnqueueAsync(new TransferJob
|
|
{
|
|
Op = TransferOp.Extract,
|
|
SourcePath = archivePath,
|
|
DestinationPath = destinationDirectory,
|
|
Status = TransferStatus.Queued,
|
|
CreatedUtc = DateTimeOffset.UtcNow
|
|
}, cancellationToken);
|
|
|
|
public Task EnqueueCompressAsync(IReadOnlyList<string> sources, string archivePath, CancellationToken cancellationToken = default)
|
|
=> EnqueueAsync(new TransferJob
|
|
{
|
|
Op = TransferOp.Compress,
|
|
SourcePath = string.Join("|", sources),
|
|
DestinationPath = archivePath,
|
|
Status = TransferStatus.Queued,
|
|
CreatedUtc = DateTimeOffset.UtcNow,
|
|
AdditionalSources = sources.ToList()
|
|
}, cancellationToken);
|
|
|
|
public Task EnqueueAddToArchiveAsync(string archivePath, IReadOnlyList<string> sources, CancellationToken cancellationToken = default)
|
|
=> EnqueueAsync(new TransferJob
|
|
{
|
|
Op = TransferOp.AddToArchive,
|
|
SourcePath = string.Join("|", sources),
|
|
DestinationPath = archivePath,
|
|
Status = TransferStatus.Queued,
|
|
CreatedUtc = DateTimeOffset.UtcNow,
|
|
AdditionalSources = sources.ToList()
|
|
}, cancellationToken);
|
|
|
|
public Task EnqueueVerifyArchiveAsync(string archivePath, CancellationToken cancellationToken = default)
|
|
=> EnqueueAsync(new TransferJob
|
|
{
|
|
Op = TransferOp.VerifyArchive,
|
|
SourcePath = archivePath,
|
|
Status = TransferStatus.Queued,
|
|
CreatedUtc = DateTimeOffset.UtcNow
|
|
}, cancellationToken);
|
|
|
|
public Task EnqueueConvertAsync(string sourcePath, string destinationPath, ConversionKind kind, CancellationToken cancellationToken = default)
|
|
=> EnqueueAsync(new TransferJob
|
|
{
|
|
Op = TransferOp.Convert,
|
|
SourcePath = sourcePath,
|
|
DestinationPath = destinationPath,
|
|
Status = TransferStatus.Queued,
|
|
CreatedUtc = DateTimeOffset.UtcNow,
|
|
AdditionalSources = [kind.ToString()]
|
|
}, cancellationToken);
|
|
|
|
public void PauseAll()
|
|
{
|
|
_queuePaused = true;
|
|
_haltPause = true;
|
|
RaiseChanged();
|
|
}
|
|
|
|
public void ResumeAll()
|
|
{
|
|
_queuePaused = false;
|
|
_haltPause = false;
|
|
_holdJobId = null;
|
|
lock (_gate)
|
|
{
|
|
foreach (var job in _jobs.Where(j => j.Status == TransferStatus.Paused
|
|
&& (j.StartedUtc is not null || j.CurrentPath is not null)))
|
|
{
|
|
job.Status = TransferStatus.Queued;
|
|
QueuePersist(job);
|
|
}
|
|
}
|
|
|
|
Pulse();
|
|
RaiseChanged();
|
|
}
|
|
|
|
public void Pause(long jobId)
|
|
{
|
|
lock (_gate)
|
|
{
|
|
var job = Find(jobId);
|
|
if (job is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
if (job.Status == TransferStatus.Queued)
|
|
{
|
|
job.Status = TransferStatus.Paused;
|
|
QueuePersist(job);
|
|
}
|
|
else if (job.Status == TransferStatus.Running)
|
|
{
|
|
_haltPause = true;
|
|
_holdJobId = jobId;
|
|
}
|
|
}
|
|
|
|
RaiseChanged();
|
|
}
|
|
|
|
public void Resume(long jobId)
|
|
{
|
|
lock (_gate)
|
|
{
|
|
var job = Find(jobId);
|
|
if (job is null || job.Status != TransferStatus.Paused)
|
|
{
|
|
return;
|
|
}
|
|
|
|
job.Status = TransferStatus.Queued;
|
|
QueuePersist(job);
|
|
}
|
|
|
|
_queuePaused = false;
|
|
_haltPause = false;
|
|
if (_holdJobId == jobId)
|
|
{
|
|
_holdJobId = null;
|
|
}
|
|
|
|
Pulse();
|
|
RaiseChanged();
|
|
}
|
|
|
|
public void Retry(long jobId)
|
|
{
|
|
lock (_gate)
|
|
{
|
|
var job = Find(jobId);
|
|
if (job is null || job.Status != TransferStatus.Failed)
|
|
{
|
|
return;
|
|
}
|
|
|
|
job.RetryCount++;
|
|
job.Status = TransferStatus.Queued;
|
|
job.Error = null;
|
|
job.WaitReason = null;
|
|
QueuePersist(job);
|
|
}
|
|
|
|
Pulse();
|
|
RaiseChanged();
|
|
}
|
|
|
|
public void Cancel(long jobId)
|
|
{
|
|
CancellationTokenSource? running = null;
|
|
lock (_gate)
|
|
{
|
|
var job = Find(jobId);
|
|
if (job is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
if (job.Status is TransferStatus.Queued or TransferStatus.Paused or TransferStatus.Waiting)
|
|
{
|
|
job.Status = TransferStatus.Cancelled;
|
|
QueuePersist(job);
|
|
_jobs.Remove(job);
|
|
if (_holdJobId == jobId)
|
|
{
|
|
_holdJobId = null;
|
|
}
|
|
|
|
Pulse();
|
|
}
|
|
else if (job.Status is TransferStatus.Running or TransferStatus.Cancelling)
|
|
{
|
|
job.Status = TransferStatus.Cancelling;
|
|
_haltPause = false;
|
|
running = _runningCts;
|
|
}
|
|
else
|
|
{
|
|
job.Dismissed = true;
|
|
QueuePersist(job);
|
|
_jobs.Remove(job);
|
|
}
|
|
}
|
|
|
|
running?.Cancel();
|
|
RaiseChanged();
|
|
}
|
|
|
|
public void Dismiss(long jobId)
|
|
{
|
|
lock (_gate)
|
|
{
|
|
var job = Find(jobId);
|
|
if (job is null || job.Status is not (TransferStatus.Done or TransferStatus.Cancelled or TransferStatus.Failed))
|
|
{
|
|
return;
|
|
}
|
|
|
|
job.Dismissed = true;
|
|
QueuePersist(job);
|
|
_jobs.Remove(job);
|
|
}
|
|
|
|
RaiseChanged();
|
|
}
|
|
|
|
public void ClearFinished()
|
|
{
|
|
var removed = false;
|
|
lock (_gate)
|
|
{
|
|
foreach (var job in _jobs.Where(j => j.Status is TransferStatus.Done or TransferStatus.Cancelled).ToList())
|
|
{
|
|
job.Dismissed = true;
|
|
QueuePersist(job);
|
|
_jobs.Remove(job);
|
|
removed = true;
|
|
}
|
|
}
|
|
|
|
if (removed)
|
|
{
|
|
RaiseChanged();
|
|
}
|
|
}
|
|
|
|
public bool MoveUp(long jobId) => Move(jobId, -1);
|
|
|
|
public bool MoveDown(long jobId) => Move(jobId, 1);
|
|
|
|
public bool Move(long jobId, int delta)
|
|
{
|
|
if (delta == 0)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
lock (_gate)
|
|
{
|
|
var index = _jobs.FindIndex(j => j.Id == jobId);
|
|
var target = index + delta;
|
|
if (index < 0 || target < 0 || target >= _jobs.Count)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
var job = _jobs[index];
|
|
if (job.Status is TransferStatus.Running or TransferStatus.Cancelling)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
_jobs.RemoveAt(index);
|
|
_jobs.Insert(target, job);
|
|
PersistOrder();
|
|
}
|
|
|
|
Pulse();
|
|
RaiseChanged();
|
|
return true;
|
|
}
|
|
|
|
public void NotifyAvailability()
|
|
{
|
|
var resumed = false;
|
|
lock (_gate)
|
|
{
|
|
foreach (var job in _jobs.Where(j => j.Status == TransferStatus.Waiting))
|
|
{
|
|
if (!OperationAvailability.IsReady(_volumes, job))
|
|
{
|
|
continue;
|
|
}
|
|
|
|
job.Status = TransferStatus.Queued;
|
|
job.WaitReason = null;
|
|
QueuePersist(job);
|
|
resumed = true;
|
|
}
|
|
}
|
|
|
|
if (resumed)
|
|
{
|
|
Pulse();
|
|
RaiseChanged();
|
|
}
|
|
}
|
|
|
|
private async Task EnqueueAsync(TransferJob job, CancellationToken cancellationToken)
|
|
{
|
|
if (!OperationAvailability.IsReady(_volumes, job))
|
|
{
|
|
job.Status = TransferStatus.Waiting;
|
|
job.WaitReason = FileOperationErrors.DestinationUnavailable;
|
|
}
|
|
|
|
job.Id = await _store.Transfers.InsertAsync(job, cancellationToken).ConfigureAwait(false);
|
|
lock (_gate)
|
|
{
|
|
if (_jobs.All(j => j.Id != job.Id))
|
|
{
|
|
_jobs.Add(job);
|
|
}
|
|
}
|
|
|
|
Pulse();
|
|
RaiseChanged();
|
|
}
|
|
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
await RestoreAsync(stoppingToken).ConfigureAwait(false);
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
{
|
|
TransferJob? job;
|
|
lock (_gate)
|
|
{
|
|
job = CanStartNext() ? NextRunnable() : null;
|
|
}
|
|
|
|
if (job is null)
|
|
{
|
|
try
|
|
{
|
|
await _signal.WaitAsync(stoppingToken).ConfigureAwait(false);
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
break;
|
|
}
|
|
|
|
continue;
|
|
}
|
|
|
|
if (!OperationAvailability.IsReady(_volumes, job))
|
|
{
|
|
job.Status = TransferStatus.Waiting;
|
|
job.WaitReason = FileOperationErrors.DestinationUnavailable;
|
|
await Persist(job).ConfigureAwait(false);
|
|
RaiseChanged();
|
|
continue;
|
|
}
|
|
|
|
await RunAsync(job, stoppingToken).ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
private async Task RestoreAsync(CancellationToken cancellationToken)
|
|
{
|
|
if (Interlocked.Exchange(ref _restored, 1) == 1)
|
|
{
|
|
return;
|
|
}
|
|
|
|
IReadOnlyList<TransferJob> incomplete;
|
|
try
|
|
{
|
|
incomplete = await _store.Transfers.GetIncompleteAsync(cancellationToken).ConfigureAwait(false);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Failed to restore file operations queue");
|
|
return;
|
|
}
|
|
|
|
var interrupted = new List<TransferJob>();
|
|
lock (_gate)
|
|
{
|
|
var known = _jobs.Select(j => j.Id).ToHashSet();
|
|
foreach (var job in incomplete)
|
|
{
|
|
if (!known.Add(job.Id))
|
|
{
|
|
continue;
|
|
}
|
|
|
|
if (job.Status is TransferStatus.Running or TransferStatus.Cancelling)
|
|
{
|
|
job.Status = TransferStatus.Paused;
|
|
job.Error = null;
|
|
interrupted.Add(job);
|
|
}
|
|
|
|
_jobs.Add(job);
|
|
}
|
|
}
|
|
|
|
foreach (var job in interrupted)
|
|
{
|
|
await Persist(job).ConfigureAwait(false);
|
|
}
|
|
|
|
if (incomplete.Count > 0)
|
|
{
|
|
Pulse();
|
|
RaiseChanged();
|
|
}
|
|
}
|
|
|
|
private bool CanStartNext()
|
|
{
|
|
if (_queuePaused)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
return _holdJobId is not { } id
|
|
|| !_jobs.Any(j => j.Id == id && j.Status is TransferStatus.Paused or TransferStatus.Running or TransferStatus.Cancelling);
|
|
}
|
|
|
|
private TransferJob? NextRunnable()
|
|
=> _jobs.FirstOrDefault(j => j.Status == TransferStatus.Queued);
|
|
|
|
private async Task RunAsync(TransferJob job, CancellationToken stoppingToken)
|
|
{
|
|
using var linked = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
|
|
_runningCts = linked;
|
|
_haltPause = false;
|
|
job.Status = TransferStatus.Running;
|
|
job.StartedUtc ??= DateTimeOffset.UtcNow;
|
|
job.Error = null;
|
|
job.WaitReason = null;
|
|
await Persist(job).ConfigureAwait(false);
|
|
RaiseChanged();
|
|
try
|
|
{
|
|
await _executor.ExecuteAsync(job, _pauseRequested, () => RaiseChanged(throttled: true), linked.Token)
|
|
.ConfigureAwait(false);
|
|
|
|
if (_haltPause && job.Status == TransferStatus.Paused)
|
|
{
|
|
_haltPause = false;
|
|
}
|
|
|
|
if (job.Status == TransferStatus.Paused)
|
|
{
|
|
await Persist(job).ConfigureAwait(false);
|
|
RaiseChanged();
|
|
return;
|
|
}
|
|
|
|
if (job.Status == TransferStatus.Waiting)
|
|
{
|
|
await Persist(job).ConfigureAwait(false);
|
|
RaiseChanged();
|
|
return;
|
|
}
|
|
|
|
if (job.Status == TransferStatus.Cancelling)
|
|
{
|
|
job.Status = TransferStatus.Cancelled;
|
|
}
|
|
else if (job.Status != TransferStatus.Failed && !string.IsNullOrEmpty(job.Error))
|
|
{
|
|
job.Status = TransferStatus.Failed;
|
|
}
|
|
else if (job.Status != TransferStatus.Failed)
|
|
{
|
|
job.Status = TransferStatus.Done;
|
|
job.CurrentPath = null;
|
|
}
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
job.Status = TransferStatus.Cancelled;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Transfer failed {Op} {Src}", job.Op, job.SourcePath);
|
|
job.Status = TransferStatus.Failed;
|
|
job.Error = FileOperationErrors.IsLock(ex.Message) ? FileOperationErrors.FileInUse : ex.Message;
|
|
}
|
|
finally
|
|
{
|
|
if (ReferenceEquals(_runningCts, linked))
|
|
{
|
|
_runningCts = null;
|
|
}
|
|
}
|
|
|
|
if (job.Status == TransferStatus.Paused)
|
|
{
|
|
return;
|
|
}
|
|
|
|
await Persist(job).ConfigureAwait(false);
|
|
RaiseChanged();
|
|
JobFinished?.Invoke(this, job);
|
|
}
|
|
|
|
private TransferJob? Find(long jobId) => _jobs.FirstOrDefault(j => j.Id == jobId);
|
|
|
|
private void Pulse()
|
|
{
|
|
if (_signal.CurrentCount == 0)
|
|
{
|
|
_signal.Release();
|
|
}
|
|
}
|
|
|
|
private void RaiseChanged(bool throttled = false)
|
|
{
|
|
if (throttled)
|
|
{
|
|
var now = DateTime.UtcNow;
|
|
if ((now - _lastChangedUtc).TotalMilliseconds < 80)
|
|
{
|
|
return;
|
|
}
|
|
|
|
_lastChangedUtc = now;
|
|
}
|
|
|
|
Changed?.Invoke(this, EventArgs.Empty);
|
|
}
|
|
|
|
private void PersistOrder()
|
|
{
|
|
for (var i = 0; i < _jobs.Count; i++)
|
|
{
|
|
_jobs[i].SortOrder = i;
|
|
QueuePersist(_jobs[i]);
|
|
}
|
|
}
|
|
|
|
private void QueuePersist(TransferJob job) => _ = PersistSafe(job);
|
|
|
|
private async Task PersistSafe(TransferJob job)
|
|
{
|
|
try
|
|
{
|
|
await Persist(job).ConfigureAwait(false);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Failed to persist transfer job {Id}", job.Id);
|
|
}
|
|
}
|
|
|
|
private Task Persist(TransferJob job) => _store.Transfers.UpdateAsync(job);
|
|
}
|