quick/Explorer-Workbench
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity Open in Code-Server

Improve file operations, layout memory, and drive status.

Browse Source 
Add a sequential file-operations queue with pause, reorder, and optional auto-clear; persist window size and tree width; show free space; and clear leftover indexing status.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
netquick
2026-08-24 01:52:30 +02:00
parent d79605cde9
commit 9bf451932f
41 changed files with 2855 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files

471
src/Explorer.FileOperations/TransferQueue.cs
View File

@@ -1,4 +1,3 @@
using System.Threading.Channels;
using Explorer.Domain;
using Explorer.Domain.Abstractions;
using Microsoft.Extensions.Hosting;
@@ -12,9 +11,15 @@ public sealed class TransferQueue : BackgroundService
private readonly IFileSystemEnumerator _enumerator;
private readonly IIndexStore _store;
private readonly ILogger<TransferQueue> _logger;
private readonly Channel<Work> _channel = Channel.CreateUnbounded<Work>();
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;
public event EventHandler? Changed;
public event EventHandler<TransferJob>? JobFinished;
@@ -29,8 +34,11 @@ public sealed class TransferQueue : BackgroundService
_enumerator = enumerator;
_store = store;
_logger = logger;
_pauseRequested = () => _haltPause;
}
public bool IsPaused => _queuePaused;
public IReadOnlyList<TransferJob> Snapshot()
{
lock (_gate)
@@ -84,19 +92,106 @@ public sealed class TransferQueue : BackgroundService
}, cancellationToken).ConfigureAwait(false);
}
public void Cancel(long jobId)
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;
}
}
Pulse();
RaiseChanged();
}
public void Pause(long jobId)
{
lock (_gate)
{
var job = _jobs.FirstOrDefault(j => j.Id == jobId);
var job = Find(jobId);
if (job is null)
{
return;
}
if (job.Status is TransferStatus.Queued or TransferStatus.Running)
if (job.Status == TransferStatus.Queued)
{
job.Status = TransferStatus.Paused;
}
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;
}
_queuePaused = false;
_haltPause = false;
if (_holdJobId == jobId)
{
_holdJobId = null;
}
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)
{
job.Status = TransferStatus.Cancelled;
_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
{
@@ -104,17 +199,67 @@ public sealed class TransferQueue : BackgroundService
}
}
Changed?.Invoke(this, EventArgs.Empty);
running?.Cancel();
RaiseChanged();
}
public void Dismiss(long jobId)
{
lock (_gate)
{
_jobs.RemoveAll(j => j.Id == jobId);
_jobs.RemoveAll(j => j.Id == jobId && j.Status is TransferStatus.Done or TransferStatus.Cancelled or TransferStatus.Failed);
}
Changed?.Invoke(this, EventArgs.Empty);
RaiseChanged();
}
public void ClearFinished()
{
var removed = false;
lock (_gate)
{
removed = _jobs.RemoveAll(j => j.Status is TransferStatus.Done or TransferStatus.Cancelled) > 0;
}
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);
}
Pulse();
RaiseChanged();
return true;
}
private async Task EnqueueAsync(TransferJob job, CancellationToken cancellationToken)
@@ -122,76 +267,128 @@ public sealed class TransferQueue : BackgroundService
job.Id = await _store.Transfers.InsertAsync(job, cancellationToken).ConfigureAwait(false);
lock (_gate)
{
_jobs.Insert(0, job);
_jobs.Add(job);
}
_channel.Writer.TryWrite(new Work(job, new CancellationTokenSource()));
Changed?.Invoke(this, EventArgs.Empty);
Pulse();
RaiseChanged();
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await foreach (var work in _channel.Reader.ReadAllAsync(stoppingToken).ConfigureAwait(false))
while (!stoppingToken.IsCancellationRequested)
{
var job = work.Job;
if (job.Status == TransferStatus.Cancelling)
TransferJob? job;
lock (_gate)
{
job.Status = TransferStatus.Cancelled;
await Persist(job).ConfigureAwait(false);
job = CanStartNext() ? NextRunnable() : null;
}
if (job is null)
{
try
{
await _signal.WaitAsync(stoppingToken).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
break;
}
continue;
}
job.Status = TransferStatus.Running;
Changed?.Invoke(this, EventArgs.Empty);
try
{
switch (job.Op)
{
case TransferOp.Copy:
await CopyOrMove(job, move: false, stoppingToken).ConfigureAwait(false);
break;
case TransferOp.Move:
await CopyOrMove(job, move: true, stoppingToken).ConfigureAwait(false);
break;
case TransferOp.Delete:
Delete(job);
break;
}
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;
}
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Transfer failed {Op} {Src}", job.Op, job.SourcePath);
job.Status = TransferStatus.Failed;
job.Error = ex.Message;
}
await Persist(job).ConfigureAwait(false);
if (job.Status is TransferStatus.Done or TransferStatus.Cancelled)
{
lock (_gate)
{
_jobs.Remove(job);
}
}
Changed?.Invoke(this, EventArgs.Empty);
JobFinished?.Invoke(this, job);
await RunAsync(job, stoppingToken).ConfigureAwait(false);
}
}
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;
RaiseChanged();
try
{
switch (job.Op)
{
case TransferOp.Copy:
await CopyOrMove(job, move: false, linked.Token).ConfigureAwait(false);
break;
case TransferOp.Move:
await CopyOrMove(job, move: true, linked.Token).ConfigureAwait(false);
break;
case TransferOp.Delete:
Delete(job);
break;
}
if (job.Status == TransferStatus.Paused)
{
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 = 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 async Task CopyOrMove(TransferJob job, bool move, CancellationToken stoppingToken)
{
var src = job.SourcePath;
@@ -207,7 +404,7 @@ public sealed class TransferQueue : BackgroundService
if (item.IsDirectory)
{
await CopyDirectory(src, dst, move, job, stoppingToken).ConfigureAwait(false);
if (move && job.Status != TransferStatus.Failed && job.Status != TransferStatus.Cancelling)
if (move && job.Status is not TransferStatus.Failed and not TransferStatus.Cancelling and not TransferStatus.Paused)
{
try { Directory.Delete(PathRules.ToExtended(src), recursive: true); } catch { /* remaining files */ }
}
@@ -215,36 +412,36 @@ public sealed class TransferQueue : BackgroundService
return;
}
job.BytesTotal = item.SizeBytes;
var progress = new Progress<long>(b =>
job.FilesTotal = Math.Max(job.FilesTotal, 1);
job.CurrentPath = src;
var resume = File.Exists(PathRules.ToExtended(dst));
if (!TransferFile(src, dst, move, resume, job, committed: 0, stoppingToken, out var error))
{
job.BytesDone = b;
Changed?.Invoke(this, EventArgs.Empty);
});
var ok = move
? _shell.MoveFileWithProgress(src, dst, overwrite: false, progress, stoppingToken, out var error)
: _shell.CopyFileWithProgress(src, dst, overwrite: false, progress, stoppingToken, out error);
if (!ok)
{
job.Status = error == "Cancelled" ? TransferStatus.Cancelling : TransferStatus.Failed;
job.Error = error;
}
else
{
job.BytesDone = job.BytesTotal ?? job.BytesDone;
ApplyHalt(job, src, error);
return;
}
job.BytesDone = job.BytesTotal ?? job.BytesDone;
job.FilesDone = 1;
job.CurrentPath = null;
await Task.CompletedTask.ConfigureAwait(false);
}
private async Task CopyDirectory(string src, string dst, bool move, TransferJob job, CancellationToken stoppingToken)
{
Directory.CreateDirectory(PathRules.ToExtended(dst));
job.FilesDone = 0;
job.FilesTotal = 0;
job.BytesDone = 0;
job.BytesTotal = 0;
var stack = new Stack<(string From, string To)>();
stack.Push((src, dst));
while (stack.Count > 0)
{
stoppingToken.ThrowIfCancellationRequested();
if (job.Status == TransferStatus.Cancelling)
if (job.Status == TransferStatus.Cancelling || _haltPause)
{
ApplyHalt(job, job.CurrentPath, _haltPause ? "Paused" : "Cancelled");
return;
}
@@ -263,30 +460,82 @@ public sealed class TransferQueue : BackgroundService
{
Directory.CreateDirectory(PathRules.ToExtended(childDest));
stack.Push((child.FullPath, childDest));
continue;
}
else
{
job.BytesTotal = (job.BytesTotal ?? 0) + child.SizeBytes;
var ok = move
? _shell.MoveFileWithProgress(child.FullPath, childDest, false, null, stoppingToken, out var err)
: _shell.CopyFileWithProgress(child.FullPath, childDest, false, null, stoppingToken, out err);
if (ok)
{
job.BytesDone += child.SizeBytes;
}
else if (err != "Cancelled")
{
job.Error = err;
}
Changed?.Invoke(this, EventArgs.Empty);
job.FilesTotal++;
job.BytesTotal = (job.BytesTotal ?? 0) + child.SizeBytes;
var destExists = File.Exists(PathRules.ToExtended(childDest));
var destLen = destExists ? new FileInfo(PathRules.ToExtended(childDest)).Length : 0;
if (destExists && destLen == child.SizeBytes)
{
job.BytesDone += child.SizeBytes;
job.FilesDone++;
RaiseChanged(throttled: true);
continue;
}
job.CurrentPath = child.FullPath;
var committed = job.BytesDone;
if (!TransferFile(child.FullPath, childDest, move, destExists, job, committed, stoppingToken, out var err))
{
ApplyHalt(job, child.FullPath, err);
return;
}
job.BytesDone = committed + child.SizeBytes;
job.FilesDone++;
RaiseChanged(throttled: true);
}
}
job.CurrentPath = null;
await Task.CompletedTask.ConfigureAwait(false);
}
private bool TransferFile(
string src,
string dst,
bool move,
bool resumePartial,
TransferJob job,
long committed,
CancellationToken stoppingToken,
out string? error)
{
var progress = new Progress<long>(b =>
{
job.BytesDone = committed + b;
RaiseChanged(throttled: true);
});
var ok = move
? _shell.MoveFileWithProgress(src, dst, resumePartial, progress, stoppingToken, out error, _pauseRequested)
: _shell.CopyFileWithProgress(src, dst, resumePartial, progress, stoppingToken, out error, _pauseRequested);
return ok;
}
private void ApplyHalt(TransferJob job, string? path, string? error)
{
job.CurrentPath = path;
if (job.Status == TransferStatus.Cancelling || error == "Cancelled")
{
job.Status = TransferStatus.Cancelling;
job.Error = error == "Cancelled" ? null : error;
return;
}
if (_haltPause || error == "Paused")
{
job.Status = TransferStatus.Paused;
job.Error = null;
_haltPause = false;
return;
}
job.Status = TransferStatus.Failed;
job.Error = error;
}
private void Delete(TransferJob job)
{
var paths = job.AdditionalSources.Count > 0
@@ -300,7 +549,31 @@ public sealed class TransferQueue : BackgroundService
}
}
private Task Persist(TransferJob job) => _store.Transfers.UpdateAsync(job);
private TransferJob? Find(long jobId) => _jobs.FirstOrDefault(j => j.Id == jobId);
private readonly record struct Work(TransferJob Job, CancellationTokenSource Cts);
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 Task Persist(TransferJob job) => _store.Transfers.UpdateAsync(job);
}