Enable concurrent uploads to the storage service (#90)

* gh-89 Enable concurrent uploads to the storage service

Signed-off-by: Victor Chang <[email protected]>

Author: mocsharp

src/Configuration/StorageConfiguration.cs

@@ -53,9 +53,16 @@ public StorageConfiguration() : this(new FileSystem())
         public RetryConfiguration Retries { get; set; } = new RetryConfiguration();
 
         /// <summary>
-        /// Gets or set number of threads used for payload upload. Default is 1;
+        /// Gets or set number of payloads to be processed at a given time. Default is 1;
         /// </summary>
-        public int Concurrentcy { get; set; } = 1;
+        [ConfigurationKeyName("payloadProcessThreads")]
+        public int PayloadProcessThreads { get; set; } = 1;
+
+        /// <summary>
+        /// Gets or set the maximum number of concurrent uploads. Default is 2;
+        /// </summary>
+        [ConfigurationKeyName("concurrentUploads")]
+        public int ConcurrentUploads { get; set; } = 2;
 
         [JsonIgnore]
         public string TemporaryDataDirFullPath

src/InformaticsGateway/Logging/Log.700.PayloadService.cs

@@ -70,5 +70,8 @@ public static partial class Log
 
         [LoggerMessage(EventId = 720, Level = LogLevel.Debug, Message = "Payload {id} deleted.")]
         public static partial void PayloadDeleted(this ILogger logger, Guid id);
+
+        [LoggerMessage(EventId = 721, Level = LogLevel.Debug, Message = "Upload statistics: {threads} threads, {seconds} seconds.")]
+        public static partial void UploadStats(this ILogger logger, int threads, double seconds);
     }
 }

src/InformaticsGateway/Services/Connectors/PayloadNotificationService.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.IO;
 using System.IO.Abstractions;
 using System.Linq;
@@ -90,7 +91,7 @@ public Task StartAsync(CancellationToken cancellationToken)
                     async (task) => await UploadPayloadActionBlock(task, cancellationToken).ConfigureAwait(false),
                     new ExecutionDataflowBlockOptions
                     {
-                        MaxDegreeOfParallelism = _options.Value.Storage.Concurrentcy,
+                        MaxDegreeOfParallelism = _options.Value.Storage.PayloadProcessThreads,
                         MaxMessagesPerTask = 1,
                         CancellationToken = cancellationToken
                     });
@@ -155,8 +156,10 @@ private void BackgroundProcessing(CancellationToken cancellationToken)
         private async Task UploadPayloadActionBlock(Payload payload, CancellationToken cancellationToken)
         {
             Guard.Against.Null(payload, nameof(payload));
+            var stopwatch = new Stopwatch();
             try
             {
+                stopwatch.Start();
                 await Upload(payload, cancellationToken).ConfigureAwait(false);
 
                 if (payload.IsUploadComplete())
@@ -185,6 +188,11 @@ private async Task UploadPayloadActionBlock(Payload payload, CancellationToken c
                     }
                 }
             }
+            finally
+            {
+                stopwatch.Stop();
+                _logger.UploadStats(_options.Value.Storage.ConcurrentUploads, stopwatch.Elapsed.TotalSeconds);
+            }
         }
 
         private async Task Upload(Payload payload, CancellationToken cancellationToken)
@@ -193,10 +201,14 @@ private async Task Upload(Payload payload, CancellationToken cancellationToken)
 
             _logger.UploadingPayloadToBucket(payload.Id, _options.Value.Storage.StorageServiceBucketName);
 
-            for (var index = payload.Files.Count - 1; index >= 0; index--)
+            var options = new ParallelOptions
             {
-                var file = payload.Files[index];
+                CancellationToken = cancellationToken,
+                MaxDegreeOfParallelism = _options.Value.Storage.ConcurrentUploads
+            };
 
+            await Parallel.ForEachAsync(payload.Files, options, async (file, cancellationToke) =>
+            {
                 switch (file)
                 {
                     case DicomFileStorageInfo dicom:
@@ -206,10 +218,11 @@ private async Task Upload(Payload payload, CancellationToken cancellationToken)
                         }
                         break;
                 }
+
                 await UploadPayloadFile(payload.Id, file.UploadFilePath, file.FilePath, file.Source, file.Workflows, file.ContentType, cancellationToken).ConfigureAwait(false);
                 file.SetUploaded();
                 _instanceCleanupQueue.Queue(file);
-            }
+            }).ConfigureAwait(false);
         }
 
         private async Task UploadPayloadFile(Guid payloadId, string destinationPath, string sourcePath, string source, List<string> workflows, string contentType, CancellationToken cancellationToken)

src/InformaticsGateway/Services/DicomWeb/IStreamsWriter.cs

@@ -141,6 +141,7 @@ private async Task SaveInstance(Stream stream, string studyInstanceUid, string w
             if (!string.IsNullOrWhiteSpace(studyInstanceUid) && !studyInstanceUid.Equals(uids.StudyInstanceUid, StringComparison.OrdinalIgnoreCase))
             {
                 AddFailure(DicomStatus.StorageDataSetDoesNotMatchSOPClassWarning, uids);
+                return;
             }
 
             DicomStoragePaths storagePaths;

src/InformaticsGateway/Test/Monai.Deploy.InformaticsGateway.Test.csproj

@@ -23,12 +23,10 @@
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
     <PackageReference Include="Microsoft.AspNetCore.Mvc.WebApiCompatShim" Version="2.2.0" />
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.1.0" />
-    <PackageReference Include="NPOI" Version="2.5.6" />
-    <PackageReference Include="System.IO.Abstractions.TestingHelpers" Version="16.1.25" />
     <PackageReference Include="Microsoft.EntityFrameworkCore.InMemory" Version="6.0.6" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.2.0" />
     <PackageReference Include="Moq" Version="4.18.1" />
+    <PackageReference Include="NPOI" Version="2.5.6" />
     <PackageReference Include="System.IO.Abstractions.TestingHelpers" Version="17.0.18" />
     <PackageReference Include="xRetry" Version="1.8.0" />
     <PackageReference Include="xunit" Version="2.4.1" />

src/InformaticsGateway/Test/Services/DicomWeb/StreamsWriterTest.cs

@@ -222,7 +222,7 @@ public async Task Save_IgnoresInstancesWithWarning()
                 Assert.Equal(DicomStatus.StorageDataSetDoesNotMatchSOPClassWarning.Code, warningReason);
             }
 
-            _payloadAssembler.Verify(p => p.Queue(It.Is<string>(p => p == correlationId), It.IsAny<DicomFileStorageInfo>(), It.IsAny<uint>()));
+            _payloadAssembler.Verify(p => p.Queue(It.Is<string>(p => p == correlationId), It.IsAny<DicomFileStorageInfo>(), It.IsAny<uint>()), Times.Never());
         }
 
         [Fact(DisplayName = "Save - queues instances with Payload Assembler")]