Skip to content

Commit e3a463b

Browse files
mocsharpmocsharp
authored
Fix payload assembler not respecting user configured timeout window (#330)
* Fix payload assembler not respecting user configured timeout window Signed-off-by: Victor Chang <[email protected]> * Add test case for grouping over multiple associations Signed-off-by: Victor Chang <[email protected]> * Adjust pulse time to catch the previous regression Signed-off-by: Victor Chang <[email protected]> --------- Signed-off-by: Victor Chang <[email protected]>
1 parent 7ad457a commit e3a463b

14 files changed

+159
-72
lines changed

‎src/InformaticsGateway/Services/Connectors/PayloadAssembler.cs

Lines changed: 12 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -135,20 +135,21 @@ private async void OnTimedEvent(Object source, System.Timers.ElapsedEventArgs e)
135135
var payload = await _payloads[key].Task.ConfigureAwait(false);
136136
using var loggerScope = _logger.BeginScope(new LoggingDataDictionary<string, object> { { "CorrelationId", payload.CorrelationId } });
137137

138-
if (payload.IsUploadCompleted())
138+
// Wait for timer window closes before sending payload for processing
139+
if (payload.HasTimedOut)
139140
{
140-
if (_payloads.TryRemove(key, out _))
141+
if (payload.IsUploadCompleted())
141142
{
142-
await QueueBucketForNotification(key, payload).ConfigureAwait(false);
143+
if (_payloads.TryRemove(key, out _))
144+
{
145+
await QueueBucketForNotification(key, payload).ConfigureAwait(false);
146+
}
147+
else
148+
{
149+
_logger.BucketRemoveError(key);
150+
}
143151
}
144-
else
145-
{
146-
_logger.BucketRemoveError(key);
147-
}
148-
}
149-
else if (payload.HasTimedOut)
150-
{
151-
if (payload.AnyUploadFailures())
152+
else if (payload.AnyUploadFailures())
152153
{
153154
_payloads.TryRemove(key, out _);
154155
_logger.PayloadRemovedWithFailureUploads(key);

‎tests/Integration.Test/Common/Assertions.cs

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -199,7 +199,15 @@ internal void ShouldHaveCorrectNumberOfWorkflowRequestMessages(DataProvider data
199199
message.ApplicationId.Should().Be(MessageBrokerConfiguration.InformaticsGatewayApplicationId);
200200
var request = message.ConvertTo<WorkflowRequestEvent>();
201201
request.Should().NotBeNull();
202-
request.FileCount.Should().Be((dataProvider.DicomSpecs.NumberOfExpectedFiles(dataProvider.StudyGrouping)));
202+
203+
if (dataProvider.ClientSendOverAssociations == 1 || messages.Count == 1)
204+
{
205+
request.FileCount.Should().Be((dataProvider.DicomSpecs.NumberOfExpectedFiles(dataProvider.StudyGrouping)));
206+
}
207+
else
208+
{
209+
request.FileCount.Should().Be(dataProvider.DicomSpecs.FileCount / dataProvider.ClientSendOverAssociations);
210+
}
203211

204212
if (dataProvider.Workflows is not null)
205213
{

‎tests/Integration.Test/Common/DataProvider.cs

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,9 @@ internal class DataProvider
3838
public DicomStatus DimseRsponse { get; internal set; }
3939
public string StudyGrouping { get; internal set; }
4040
public string[] Workflows { get; internal set; } = null;
41+
public int ClientTimeout { get; internal set; }
42+
public int ClientAssociationPulseTime { get; internal set; } = 0;
43+
public int ClientSendOverAssociations { get; internal set; } = 1;
4144

4245
public DataProvider(Configurations configurations, ISpecFlowOutputHelper outputHelper)
4346
{

‎tests/Integration.Test/Common/DicomCStoreDataClient.cs

Lines changed: 51 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
using System.Diagnostics;
1818
using Ardalis.GuardClauses;
19+
using FellowOakDicom;
1920
using FellowOakDicom.Network;
2021
using FellowOakDicom.Network.Client;
2122
using Monai.Deploy.InformaticsGateway.Configuration;
@@ -48,15 +49,59 @@ public async Task SendAsync(DataProvider dataProvider, params object[] args)
4849
var host = args[1].ToString();
4950
var port = (int)args[2];
5051
var calledAeTitle = args[3].ToString();
51-
var timeout = (TimeSpan)args[4];
52+
var timeout = TimeSpan.FromSeconds(dataProvider.ClientTimeout);
53+
var associations = dataProvider.ClientSendOverAssociations;
54+
var pauseTime = TimeSpan.FromSeconds(dataProvider.ClientAssociationPulseTime);
5255

5356
_outputHelper.WriteLine($"C-STORE: {callingAeTitle} => {host}:{port}@{calledAeTitle}");
5457
var stopwatch = new Stopwatch();
5558
stopwatch.Start();
56-
var dicomClient = DicomClientFactory.Create(host, port, false, callingAeTitle, calledAeTitle);
57-
var countdownEvent = new CountdownEvent(dataProvider.DicomSpecs.Files.Count);
59+
60+
var filesPerAssociations = dataProvider.DicomSpecs.Files.Count / associations;
61+
5862
var failureStatus = new List<DicomStatus>();
59-
foreach (var file in dataProvider.DicomSpecs.Files)
63+
for (int i = 0; i < associations; i++)
64+
{
65+
var files = dataProvider.DicomSpecs.Files.Skip(i * filesPerAssociations).Take(filesPerAssociations).ToList();
66+
if (i + 1 == associations && dataProvider.DicomSpecs.Files.Count > (i + 1) * filesPerAssociations)
67+
{
68+
files.AddRange(dataProvider.DicomSpecs.Files.Skip(i * filesPerAssociations));
69+
}
70+
71+
try
72+
{
73+
await SendBatchAsync(
74+
files,
75+
callingAeTitle,
76+
host,
77+
port,
78+
calledAeTitle,
79+
timeout,
80+
stopwatch,
81+
failureStatus);
82+
await Task.Delay(pauseTime);
83+
}
84+
catch (DicomAssociationRejectedException ex)
85+
{
86+
_outputHelper.WriteLine($"Association Rejected: {ex.Message}");
87+
dataProvider.DimseRsponse = DicomStatus.Cancel;
88+
}
89+
}
90+
91+
stopwatch.Stop();
92+
lock (SyncRoot)
93+
{
94+
TotalTime += (int)stopwatch.Elapsed.TotalMilliseconds;
95+
}
96+
_outputHelper.WriteLine($"DICOMsend:{stopwatch.Elapsed.TotalSeconds}s");
97+
dataProvider.DimseRsponse = (failureStatus.Count == 0) ? DicomStatus.Success : failureStatus.First();
98+
}
99+
100+
private async Task SendBatchAsync(List<DicomFile> files, string callingAeTitle, string host, int port, string calledAeTitle, TimeSpan timeout, Stopwatch stopwatch, List<DicomStatus> failureStatus)
101+
{
102+
var dicomClient = DicomClientFactory.Create(host, port, false, callingAeTitle, calledAeTitle);
103+
var countdownEvent = new CountdownEvent(files.Count);
104+
foreach (var file in files)
60105
{
61106
var cStoreRequest = new DicomCStoreRequest(file);
62107
cStoreRequest.OnResponseReceived += (DicomCStoreRequest request, DicomCStoreResponse response) =>
@@ -67,24 +112,8 @@ public async Task SendAsync(DataProvider dataProvider, params object[] args)
67112
await dicomClient.AddRequestAsync(cStoreRequest);
68113
}
69114

70-
try
71-
{
72-
await dicomClient.SendAsync();
73-
countdownEvent.Wait(timeout);
74-
stopwatch.Stop();
75-
lock (SyncRoot)
76-
{
77-
TotalTime += (int)stopwatch.Elapsed.TotalMilliseconds;
78-
}
79-
_outputHelper.WriteLine($"DICOMsend:{stopwatch.Elapsed.TotalSeconds}s");
80-
}
81-
catch (DicomAssociationRejectedException ex)
82-
{
83-
_outputHelper.WriteLine($"Association Rejected: {ex.Message}");
84-
dataProvider.DimseRsponse = DicomStatus.Cancel;
85-
}
86-
87-
dataProvider.DimseRsponse = (failureStatus.Count == 0) ? DicomStatus.Success : failureStatus.First();
115+
await dicomClient.SendAsync();
116+
countdownEvent.Wait(timeout);
88117
}
89118
}
90119
}

‎tests/Integration.Test/Drivers/RabbitMqConsumer.cs

Lines changed: 16 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
*/
1717

1818
using System.Collections.Concurrent;
19+
using System.Diagnostics;
1920
using Monai.Deploy.Messaging.Messages;
2021
using Monai.Deploy.Messaging.RabbitMQ;
2122
using TechTalk.SpecFlow.Infrastructure;
@@ -32,7 +33,6 @@ internal class RabbitMqConsumer : IDisposable
3233

3334
public IReadOnlyList<Message> Messages
3435
{ get { return _messages.ToList(); } }
35-
public CountdownEvent MessageWaitHandle { get; private set; }
3636

3737
public RabbitMqConsumer(RabbitMQMessageSubscriberService subscriberService, string queueName, ISpecFlowOutputHelper outputHelper)
3838
{
@@ -54,15 +54,13 @@ public RabbitMqConsumer(RabbitMQMessageSubscriberService subscriberService, stri
5454
_messages.Add(eventArgs.Message);
5555
subscriberService.Acknowledge(eventArgs.Message);
5656
_outputHelper.WriteLine($"{DateTime.UtcNow} - {queueName} message received with correlation ID={eventArgs.Message.CorrelationId}, delivery tag={eventArgs.Message.DeliveryTag}");
57-
MessageWaitHandle?.Signal();
5857
});
5958
}
6059

61-
public void SetupMessageHandle(int count)
60+
public void ClearMessages()
6261
{
63-
_outputHelper.WriteLine($"Expecting {count} {_queueName} messages from RabbitMQ");
62+
_outputHelper.WriteLine($"Clearing messages received from RabbitMQ");
6463
_messages.Clear();
65-
MessageWaitHandle = new CountdownEvent(count);
6664
}
6765

6866
protected virtual void Dispose(bool disposing)
@@ -84,5 +82,18 @@ public void Dispose()
8482
Dispose(disposing: true);
8583
GC.SuppressFinalize(this);
8684
}
85+
86+
internal async Task<bool> WaitforAsync(int messageCount, TimeSpan messageWaitTimeSpan)
87+
{
88+
var stopwatch = new Stopwatch();
89+
stopwatch.Start();
90+
91+
while (messageCount > _messages.Count && stopwatch.Elapsed < messageWaitTimeSpan)
92+
{
93+
await Task.Delay(100);
94+
}
95+
96+
return messageCount >= _messages.Count;
97+
}
8798
}
8899
}

‎tests/Integration.Test/Features/AcrApi.feature

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ Feature: ACR API
2828
Given a DICOM study on a remote DICOMweb service
2929
And an ACR API request to query & retrieve by <requestType>
3030
When the ACR API request is sent
31-
Then a workflow requests sent to the message broker
31+
Then a single workflow request is sent to the message broker
3232
And a study is uploaded to the storage service
3333

3434
Examples:

‎tests/Integration.Test/Features/DicomDimseScp.feature

Lines changed: 24 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -31,14 +31,17 @@ Feature: DICOM DIMSE SCP Services
3131

3232
Scenario: Response to C-ECHO-RQ
3333
Given a called AE Title named 'C-ECHO-TEST' that groups by '0020,000D' for 5 seconds
34-
When a C-ECHO-RQ is sent to 'C-ECHO-TEST' from 'TEST-RUNNER' with timeout of 30 seconds
34+
And a DICOM client configured with 30 seconds timeout
35+
When a C-ECHO-RQ is sent to 'C-ECHO-TEST' from 'TEST-RUNNER'
3536
Then a successful response should be received
3637

3738
@messaging_workflow_request @messaging
3839
Scenario Outline: Respond to C-STORE-RQ and group data by Study Instance UID
3940
Given a called AE Title named 'C-STORE-STUDY' that groups by '0020,000D' for 3 seconds
41+
And a DICOM client configured with 300 seconds timeout
42+
And a DICOM client configured to send data over 1 associations and wait 0 between each association
4043
And <count> <modality> studies
41-
When a C-STORE-RQ is sent to 'Informatics Gateway' with AET 'C-STORE-STUDY' from 'TEST-RUNNER' with timeout of 300 seconds
44+
When a C-STORE-RQ is sent to 'Informatics Gateway' with AET 'C-STORE-STUDY' from 'TEST-RUNNER'
4245
Then a successful response should be received
4346
And <count> workflow requests sent to message broker
4447
And studies are uploaded to storage service
@@ -53,8 +56,10 @@ Feature: DICOM DIMSE SCP Services
5356
@messaging_workflow_request @messaging
5457
Scenario Outline: Respond to C-STORE-RQ and group data by Series Instance UID
5558
Given a called AE Title named 'C-STORE-SERIES' that groups by '0020,000E' for 3 seconds
59+
And a DICOM client configured with 300 seconds timeout
60+
And a DICOM client configured to send data over 1 associations and wait 0 between each association
5661
And <study_count> <modality> studies with <series_count> series per study
57-
When a C-STORE-RQ is sent to 'Informatics Gateway' with AET 'C-STORE-SERIES' from 'TEST-RUNNER' with timeout of 300 seconds
62+
When a C-STORE-RQ is sent to 'Informatics Gateway' with AET 'C-STORE-SERIES' from 'TEST-RUNNER'
5863
Then a successful response should be received
5964
And <series_count> workflow requests sent to message broker
6065
And studies are uploaded to storage service
@@ -65,3 +70,19 @@ Feature: DICOM DIMSE SCP Services
6570
| CT | 1 | 2 |
6671
| MG | 1 | 3 |
6772
| US | 1 | 2 |
73+
74+
@messaging_workflow_request @messaging
75+
Scenario Outline: Respond to C-STORE-RQ and group data by Study Instance UID over multiple associations
76+
Given a called AE Title named 'C-STORE-STUDY' that groups by '0020,000D' for 5 seconds
77+
And a DICOM client configured with 300 seconds timeout
78+
And a DICOM client configured to send data over <series_count> associations and wait <seconds> between each association
79+
And <study_count> <modality> studies with <series_count> series per study
80+
When C-STORE-RQ are sent to 'Informatics Gateway' with AET 'C-STORE-STUDY' from 'TEST-RUNNER'
81+
Then a successful response should be received
82+
And <workflow_requests> workflow requests sent to message broker
83+
And studies are uploaded to storage service
84+
85+
Examples:
86+
| modality | study_count | series_count | seconds | workflow_requests |
87+
| MG | 1 | 3 | 3 | 1 |
88+
| MG | 1 | 3 | 6 | 3 |

‎tests/Integration.Test/StepDefinitions/AcrApiStepDefinitions.cs

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ public async Task GivenADICOMStudySentToAETFromWithTimeoutOfSeconds()
5959
{
6060
var modality = "US";
6161
_dataProvider.GenerateDicomData(modality, WorkflowStudyCount);
62-
_receivedMessages.SetupMessageHandle(WorkflowStudyCount);
62+
_receivedMessages.ClearMessages();
6363

6464
var storeScu = _objectContainer.Resolve<IDataClient>("StoreSCU");
6565
await storeScu.SendAsync(_dataProvider, "TEST-RUNNER", _configurations.OrthancOptions.Host, _configurations.OrthancOptions.DimsePort, "ORTHANC", TimeSpan.FromSeconds(300));
@@ -80,17 +80,16 @@ public async Task WhenTheACRAPIRequestIsSentTo()
8080
await _informaticsGatewayClient.Inference.NewInferenceRequest(_dataProvider.AcrRequest, CancellationToken.None);
8181
}
8282

83-
[Then(@"a workflow requests sent to the message broker")]
84-
public void ThenAWorkflowRequestsSentToTheMessageBroker()
83+
[Then(@"a single workflow request is sent to the message broker")]
84+
public async Task ThenAWorkflowRequestsSentToTheMessageBroker()
8585
{
86-
_receivedMessages.MessageWaitHandle.Wait(MessageWaitTimeSpan).Should().BeTrue();
86+
(await _receivedMessages.WaitforAsync(1, MessageWaitTimeSpan)).Should().BeTrue();
8787
_assertions.ShouldHaveCorrectNumberOfWorkflowRequestMessagesAndAcrRequest(_dataProvider, _receivedMessages.Messages, WorkflowStudyCount);
8888
}
8989

9090
[Then(@"a study is uploaded to the storage service")]
9191
public async Task ThenAStudyIsUploadedToTheStorageService()
9292
{
93-
_receivedMessages.MessageWaitHandle.Wait(MessageWaitTimeSpan).Should().BeTrue();
9493
_receivedMessages.Messages.Should().NotBeNullOrEmpty();
9594
await _assertions.ShouldHaveUploadedDicomDataToMinio(_receivedMessages.Messages, _dataProvider.DicomSpecs.FileHashes);
9695
}

‎tests/Integration.Test/StepDefinitions/DicomDimseScpServicesStepDefinitions.cs

Lines changed: 26 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -88,7 +88,8 @@ public void GivenXStudiesWithYSeriesPerStudy(int studyCount, string modality, in
8888
Guard.Against.NegativeOrZero(seriesPerStudy);
8989

9090
_dataProvider.GenerateDicomData(modality, studyCount, seriesPerStudy);
91-
_receivedMessages.SetupMessageHandle(_dataProvider.DicomSpecs.NumberOfExpectedRequests(_dataProvider.StudyGrouping));
91+
92+
_receivedMessages.ClearMessages();
9293
}
9394

9495
[Given(@"a called AE Title named '([^']*)' that groups by '([^']*)' for (.*) seconds")]
@@ -125,12 +126,28 @@ await _informaticsGatewayClient.MonaiScpAeTitle.Create(new MonaiApplicationEntit
125126
}
126127
}
127128

128-
[When(@"a C-ECHO-RQ is sent to '([^']*)' from '([^']*)' with timeout of (.*) seconds")]
129-
public async Task WhenAC_ECHO_RQIsSentToFromWithTimeoutOfSeconds(string calledAeTitle, string callingAeTitle, int clientTimeoutSeconds)
129+
[Given(@"a DICOM client configured with (.*) seconds timeout")]
130+
public void GivenADICOMClientConfiguredWithSecondsTimeout(int timeout)
131+
{
132+
Guard.Against.NegativeOrZero(timeout);
133+
_dataProvider.ClientTimeout = timeout;
134+
}
135+
136+
[Given(@"a DICOM client configured to send data over (.*) associations and wait (.*) between each association")]
137+
public void GivenADICOMClientConfiguredToSendDataOverAssociationsAndWaitSecondsBetweenEachAssociation(int associations, int pulseTime)
138+
{
139+
Guard.Against.NegativeOrZero(associations);
140+
Guard.Against.Negative(pulseTime);
141+
142+
_dataProvider.ClientSendOverAssociations = associations;
143+
_dataProvider.ClientAssociationPulseTime = pulseTime;
144+
}
145+
146+
[When(@"a C-ECHO-RQ is sent to '([^']*)' from '([^']*)'")]
147+
public async Task WhenAC_ECHO_RQIsSentToFromWithTimeoutOfSeconds(string calledAeTitle, string callingAeTitle)
130148
{
131149
Guard.Against.NullOrWhiteSpace(calledAeTitle);
132150
Guard.Against.NullOrWhiteSpace(callingAeTitle);
133-
Guard.Against.NegativeOrZero(clientTimeoutSeconds);
134151

135152
var echoScu = _objectContainer.Resolve<IDataClient>("EchoSCU");
136153
await echoScu.SendAsync(
@@ -139,7 +156,7 @@ await echoScu.SendAsync(
139156
_configuration.InformaticsGatewayOptions.Host,
140157
_informaticsGatewayConfiguration.Dicom.Scp.Port,
141158
calledAeTitle,
142-
TimeSpan.FromSeconds(clientTimeoutSeconds));
159+
TimeSpan.FromSeconds(_dataProvider.ClientTimeout));
143160
}
144161

145162
[Then(@"a successful response should be received")]
@@ -148,13 +165,13 @@ public void ThenASuccessfulResponseShouldBeReceived()
148165
_dataProvider.DimseRsponse.Should().Be(DicomStatus.Success);
149166
}
150167

151-
[When(@"a C-STORE-RQ is sent to '([^']*)' with AET '([^']*)' from '([^']*)' with timeout of (.*) seconds")]
152-
public async Task WhenAC_STORE_RQIsSentToWithAETFromWithTimeoutOfSeconds(string application, string calledAeTitle, string callingAeTitle, int clientTimeoutSeconds)
168+
[When(@"a C-STORE-RQ is sent to '([^']*)' with AET '([^']*)' from '([^']*)'")]
169+
[When(@"C-STORE-RQ are sent to '([^']*)' with AET '([^']*)' from '([^']*)'")]
170+
public async Task WhenAC_STORE_RQIsSentToWithAETFromWithTimeoutOfSeconds(string application, string calledAeTitle, string callingAeTitle)
153171
{
154172
Guard.Against.NullOrWhiteSpace(application);
155173
Guard.Against.NullOrWhiteSpace(calledAeTitle);
156174
Guard.Against.NullOrWhiteSpace(callingAeTitle);
157-
Guard.Against.NegativeOrZero(clientTimeoutSeconds);
158175

159176
var storeScu = _objectContainer.Resolve<IDataClient>("StoreSCU");
160177

@@ -168,8 +185,7 @@ await storeScu.SendAsync(
168185
callingAeTitle,
169186
host,
170187
port,
171-
calledAeTitle,
172-
TimeSpan.FromSeconds(clientTimeoutSeconds));
188+
calledAeTitle);
173189

174190
_dataProvider.ReplaceGeneratedDicomDataWithHashes();
175191
}

‎tests/Integration.Test/StepDefinitions/DicomWebStowServiceStepDefinitions.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@ public void GivenNStudies(int studyCount, string modality, string grouping)
5151

5252
_dataProvider.GenerateDicomData(modality, studyCount);
5353
_dataProvider.StudyGrouping = grouping;
54-
_receivedMessages.SetupMessageHandle(_dataProvider.DicomSpecs.NumberOfExpectedRequests(grouping));
54+
_receivedMessages.ClearMessages();
5555
}
5656

5757
[Given(@"a workflow named '(.*)'")]

0 commit comments

Comments
 (0)