diff --git a/src/Cmdlets/NewMinIOObjectCmdlet.cs b/src/Cmdlets/NewMinIOObjectCmdlet.cs index 53fe0a3..6818255 100644 --- a/src/Cmdlets/NewMinIOObjectCmdlet.cs +++ b/src/Cmdlets/NewMinIOObjectCmdlet.cs @@ -302,7 +302,7 @@ namespace PSMinIO.Cmdlets } /// - /// Uploads a collection of files + /// Uploads a collection of files with multi-layer progress tracking /// /// Files to upload private void UploadFileCollection(FileInfo[] files) @@ -324,38 +324,48 @@ namespace PSMinIO.Cmdlets WriteVerboseMessage("Uploading {0} files to bucket '{1}'", validFiles.Length, BucketName); - // Calculate total size for overall progress - var totalSize = validFiles.Sum(f => f.Length); - var totalProcessed = 0L; - + // Create multi-layer progress reporter + var progressReporter = new MultiLayerProgressReporter(this, validFiles.Length, "Uploading Files"); var uploadedObjects = new List(); - for (int i = 0; i < validFiles.Length; i++) + try { - var fileInfo = validFiles[i]; - var objectName = GetObjectName(fileInfo); - - try + for (int i = 0; i < validFiles.Length; i++) { - WriteVerboseMessage("Uploading file {0}/{1}: {2} -> {3}", - i + 1, validFiles.Length, fileInfo.Name, objectName); + var fileInfo = validFiles[i]; + var objectName = GetObjectName(fileInfo); - var uploadResult = UploadSingleFile(fileInfo, objectName); - if (uploadResult != null) + try { - uploadedObjects.Add(uploadResult); - totalProcessed += fileInfo.Length; + // Start new file in progress reporter + progressReporter.StartNewFile(fileInfo.Name, fileInfo.Length); + progressReporter.QueueVerboseMessage("Uploading file {0}/{1}: {2} -> {3}", + i + 1, validFiles.Length, fileInfo.Name, objectName); + + var uploadResult = UploadSingleFileWithProgress(fileInfo, objectName, progressReporter); + if (uploadResult != null) + { + uploadedObjects.Add(uploadResult); + progressReporter.CompleteCurrentFile(); + progressReporter.QueueVerboseMessage("Successfully uploaded: {0}", fileInfo.Name); + } + } + catch (Exception ex) + { + var errorRecord = new ErrorRecord(ex, "FileUploadFailed", ErrorCategory.WriteError, fileInfo); + WriteError(errorRecord); + progressReporter.QueueVerboseMessage("Failed to upload {0}: {1}", fileInfo.Name, ex.Message); } - WriteVerboseMessage("Successfully uploaded: {0}", fileInfo.Name); - } - catch (Exception ex) - { - var errorRecord = new ErrorRecord(ex, "FileUploadFailed", ErrorCategory.WriteError, fileInfo); - WriteError(errorRecord); - WriteVerboseMessage("Failed to upload {0}: {1}", fileInfo.Name, ex.Message); + // Process queued progress updates + progressReporter.ProcessQueuedUpdates(); } } + finally + { + // Complete all progress reporting + progressReporter.Complete(); + } // Always return uploaded objects if PassThru is specified if (PassThru.IsPresent) @@ -501,7 +511,73 @@ namespace PSMinIO.Cmdlets } /// - /// Uploads a single file and returns the result + /// Uploads a single file with progress tracking and returns the result + /// + /// File to upload + /// Object name in bucket + /// Multi-layer progress reporter + /// Upload result + private MinIOUploadResult? UploadSingleFileWithProgress(FileInfo fileInfo, string objectName, MultiLayerProgressReporter progressReporter) + { + // Create upload result + var uploadResult = new MinIOUploadResult(BucketName, objectName, string.Empty, fileInfo.Length) + { + SourceFilePath = fileInfo.FullName + }; + uploadResult.MarkStarted(); + + try + { + // Determine content type + var fileContentType = ContentType ?? GetContentType(fileInfo.Extension); + uploadResult.ContentType = fileContentType; + + // Convert metadata + Dictionary? metadata = null; + if (Metadata != null && Metadata.Count > 0) + { + metadata = new Dictionary(); + foreach (var key in Metadata.Keys) + { + if (key != null && Metadata[key] != null) + { + metadata[key.ToString()!] = Metadata[key]!.ToString()!; + } + } + } + + // Upload the file with progress tracking + string etag; + using (var fileStream = fileInfo.OpenRead()) + { + etag = S3Client.PutObject( + BucketName, + objectName, + fileStream, + fileContentType, + metadata, + bytesTransferred => + { + // Update progress reporter from background thread + progressReporter.UpdateTransferProgress(bytesTransferred); + uploadResult.BytesTransferred = bytesTransferred; + }); + } + + uploadResult.ETag = etag; + uploadResult.MarkCompleted(); + + return uploadResult; + } + catch (Exception ex) + { + uploadResult.MarkFailed(ex.Message); + throw; + } + } + + /// + /// Uploads a single file and returns the result (legacy method for backward compatibility) /// /// File to upload /// Object name in bucket diff --git a/src/Utils/MultiLayerProgressReporter.cs b/src/Utils/MultiLayerProgressReporter.cs new file mode 100644 index 0000000..7ce0cba --- /dev/null +++ b/src/Utils/MultiLayerProgressReporter.cs @@ -0,0 +1,199 @@ +using System; +using System.Management.Automation; + +namespace PSMinIO.Utils +{ + /// + /// Multi-layer progress reporter for 3-level progress tracking: + /// Collection progress (overall files) > File progress (current file) > Transfer progress (bytes) + /// + public class MultiLayerProgressReporter + { + private readonly PSCmdlet _cmdlet; + private readonly ThreadSafeProgressCollector _progressCollector; + + // Activity IDs for the three layers + private const int CollectionActivityId = 1; + private const int FileActivityId = 2; + private const int TransferActivityId = 3; + + // Collection-level tracking + private int _totalFiles; + private int _currentFileIndex; + private string _collectionOperation = "Processing Files"; + + // File-level tracking + private string _currentFileName = string.Empty; + private long _currentFileSize; + private long _currentFileBytesTransferred; + + // Transfer-level tracking + private long _totalBytesTransferred; + private DateTime _startTime; + + public MultiLayerProgressReporter(PSCmdlet cmdlet, int totalFiles, string operation = "Processing Files") + { + _cmdlet = cmdlet ?? throw new ArgumentNullException(nameof(cmdlet)); + _progressCollector = new ThreadSafeProgressCollector(cmdlet); + _totalFiles = totalFiles; + _collectionOperation = operation; + _startTime = DateTime.UtcNow; + + // Initialize collection progress + UpdateCollectionProgress(); + } + + /// + /// Starts processing a new file + /// + /// Name of the file being processed + /// Size of the file in bytes + public void StartNewFile(string fileName, long fileSize) + { + _currentFileIndex++; + _currentFileName = fileName; + _currentFileSize = fileSize; + _currentFileBytesTransferred = 0; + + // Update collection progress + UpdateCollectionProgress(); + + // Initialize file progress + UpdateFileProgress(); + + // Initialize transfer progress + UpdateTransferProgress(); + + // Process any queued updates + _progressCollector.ProcessQueuedUpdates(); + } + + /// + /// Updates transfer progress for the current file + /// + /// Bytes transferred for current file + public void UpdateTransferProgress(long bytesTransferred) + { + _currentFileBytesTransferred = bytesTransferred; + UpdateTransferProgress(); + } + + /// + /// Completes the current file and updates overall progress + /// + public void CompleteCurrentFile() + { + _totalBytesTransferred += _currentFileSize; + + // Complete transfer progress + _progressCollector.QueueProgressCompletion(TransferActivityId, "Transfer Progress", FileActivityId); + + // Complete file progress + _progressCollector.QueueProgressCompletion(FileActivityId, "File Progress", CollectionActivityId); + + // Update collection progress + UpdateCollectionProgress(); + + // Process updates + _progressCollector.ProcessQueuedUpdates(); + } + + /// + /// Completes all progress reporting + /// + public void Complete() + { + // Complete collection progress + _progressCollector.QueueProgressCompletion(CollectionActivityId, _collectionOperation); + + // Process final updates and complete + _progressCollector.Complete(); + } + + /// + /// Processes any queued progress updates + /// + public void ProcessQueuedUpdates() + { + _progressCollector.ProcessQueuedUpdates(); + } + + /// + /// Updates collection-level progress + /// + private void UpdateCollectionProgress() + { + var percentage = _totalFiles > 0 ? (int)((double)_currentFileIndex / _totalFiles * 100) : 0; + var status = $"Processing file {_currentFileIndex} of {_totalFiles}"; + + if (_currentFileIndex > 0) + { + var elapsed = DateTime.UtcNow - _startTime; + var avgTimePerFile = elapsed.TotalSeconds / _currentFileIndex; + var remainingFiles = _totalFiles - _currentFileIndex; + var estimatedTimeRemaining = TimeSpan.FromSeconds(avgTimePerFile * remainingFiles); + + status += $" (ETA: {SizeFormatter.FormatDuration(estimatedTimeRemaining)})"; + } + + _progressCollector.QueueProgressUpdate(CollectionActivityId, _collectionOperation, status, percentage); + } + + /// + /// Updates file-level progress + /// + private void UpdateFileProgress() + { + var percentage = _currentFileSize > 0 ? (int)((double)_currentFileBytesTransferred / _currentFileSize * 100) : 0; + var status = $"Processing: {_currentFileName} ({SizeFormatter.FormatBytes(_currentFileBytesTransferred)}/{SizeFormatter.FormatBytes(_currentFileSize)})"; + + _progressCollector.QueueProgressUpdate(FileActivityId, "File Progress", status, percentage, CollectionActivityId); + } + + /// + /// Updates transfer-level progress + /// + private void UpdateTransferProgress() + { + var percentage = _currentFileSize > 0 ? (int)((double)_currentFileBytesTransferred / _currentFileSize * 100) : 0; + var status = $"Transferring: {SizeFormatter.FormatBytes(_currentFileBytesTransferred)}/{SizeFormatter.FormatBytes(_currentFileSize)}"; + + if (_currentFileBytesTransferred > 0) + { + var elapsed = DateTime.UtcNow - _startTime; + if (elapsed.TotalSeconds > 0) + { + var speed = _currentFileBytesTransferred / elapsed.TotalSeconds; + status += $" at {SizeFormatter.FormatSpeed(speed)}"; + } + } + + _progressCollector.QueueProgressUpdate(TransferActivityId, "Transfer Progress", status, percentage, FileActivityId); + } + + /// + /// Queues a verbose message + /// + /// Message to log + /// Message arguments + public void QueueVerboseMessage(string message, params object[] args) + { + _progressCollector.QueueVerboseMessage(message, args); + } + + /// + /// Gets the current collection progress percentage + /// + public int CollectionProgressPercentage => _totalFiles > 0 ? (int)((double)_currentFileIndex / _totalFiles * 100) : 0; + + /// + /// Gets the current file progress percentage + /// + public int FileProgressPercentage => _currentFileSize > 0 ? (int)((double)_currentFileBytesTransferred / _currentFileSize * 100) : 0; + + /// + /// Gets the total bytes transferred across all files + /// + public long TotalBytesTransferred => _totalBytesTransferred + _currentFileBytesTransferred; + } +} diff --git a/src/Utils/ThreadSafeProgressCollector.cs b/src/Utils/ThreadSafeProgressCollector.cs new file mode 100644 index 0000000..bd7ebc9 --- /dev/null +++ b/src/Utils/ThreadSafeProgressCollector.cs @@ -0,0 +1,155 @@ +using System; +using System.Collections.Concurrent; +using System.Management.Automation; +using System.Threading; + +namespace PSMinIO.Utils +{ + /// + /// Thread-safe progress data collector that accumulates progress updates from background threads + /// and allows the main thread to safely report them to PowerShell + /// + public class ThreadSafeProgressCollector + { + private readonly PSCmdlet _cmdlet; + private readonly ConcurrentQueue _progressQueue = new(); + private readonly ConcurrentQueue _verboseQueue = new(); + private readonly object _lockObject = new(); + private volatile bool _isCompleted = false; + + public ThreadSafeProgressCollector(PSCmdlet cmdlet) + { + _cmdlet = cmdlet ?? throw new ArgumentNullException(nameof(cmdlet)); + } + + /// + /// Queues a progress update from a background thread + /// + public void QueueProgressUpdate(int activityId, string activity, string statusDescription, int percentComplete, int parentActivityId = -1) + { + if (_isCompleted) return; + + _progressQueue.Enqueue(new ProgressUpdate + { + ActivityId = activityId, + Activity = activity, + StatusDescription = statusDescription, + PercentComplete = percentComplete, + ParentActivityId = parentActivityId, + Timestamp = DateTime.UtcNow + }); + } + + /// + /// Queues a progress completion from a background thread + /// + public void QueueProgressCompletion(int activityId, string activity, int parentActivityId = -1) + { + if (_isCompleted) return; + + _progressQueue.Enqueue(new ProgressUpdate + { + ActivityId = activityId, + Activity = activity, + StatusDescription = "Completed", + PercentComplete = 100, + ParentActivityId = parentActivityId, + IsCompleted = true, + Timestamp = DateTime.UtcNow + }); + } + + /// + /// Queues a verbose message from a background thread + /// + public void QueueVerboseMessage(string message, params object[] args) + { + if (_isCompleted) return; + + _verboseQueue.Enqueue(new VerboseMessage + { + Message = args.Length > 0 ? string.Format(message, args) : message, + Timestamp = DateTime.UtcNow + }); + } + + /// + /// Processes all queued updates from the main thread (safe to call PowerShell methods) + /// + public void ProcessQueuedUpdates() + { + // Process verbose messages first + while (_verboseQueue.TryDequeue(out var verboseMessage)) + { + MinIOLogger.WriteVerbose(_cmdlet, verboseMessage.Message); + } + + // Process progress updates + while (_progressQueue.TryDequeue(out var progressUpdate)) + { + var progressRecord = new ProgressRecord( + progressUpdate.ActivityId, + progressUpdate.Activity, + progressUpdate.StatusDescription) + { + PercentComplete = progressUpdate.PercentComplete + }; + + if (progressUpdate.ParentActivityId >= 0) + { + progressRecord.ParentActivityId = progressUpdate.ParentActivityId; + } + + if (progressUpdate.IsCompleted) + { + progressRecord.RecordType = ProgressRecordType.Completed; + } + + _cmdlet.WriteProgress(progressRecord); + } + } + + /// + /// Marks the collector as completed (no more updates will be accepted) + /// + public void Complete() + { + _isCompleted = true; + // Process any remaining updates + ProcessQueuedUpdates(); + } + + /// + /// Gets the number of pending progress updates + /// + public int PendingProgressUpdates => _progressQueue.Count; + + /// + /// Gets the number of pending verbose messages + /// + public int PendingVerboseMessages => _verboseQueue.Count; + + /// + /// Progress update data structure + /// + private class ProgressUpdate + { + public int ActivityId { get; set; } + public string Activity { get; set; } = string.Empty; + public string StatusDescription { get; set; } = string.Empty; + public int PercentComplete { get; set; } + public int ParentActivityId { get; set; } = -1; + public bool IsCompleted { get; set; } + public DateTime Timestamp { get; set; } + } + + /// + /// Verbose message data structure + /// + private class VerboseMessage + { + public string Message { get; set; } = string.Empty; + public DateTime Timestamp { get; set; } + } + } +}