Merge pull request #2666 from FirelyTeam/feature/fix-vsexpand-race-co…

…ndition Fix race condition around expansing a valueset + unit test.
FirelyTeam · Jan 15, 2024 · 4ad46ee · 4ad46ee
2 parents 8b0d266 + 673477c
commit 4ad46ee
Show file tree

Hide file tree

Showing 3 changed files with 70 additions and 8 deletions.
diff --git a/Hl7.Fhir.sln b/Hl7.Fhir.sln
@@ -101,7 +101,7 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Hl7.Fhir.Specification.STU3
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Hl7.Fhir.STU3.Tests", "src\Hl7.Fhir.STU3.Tests\Hl7.Fhir.STU3.Tests.csproj", "{CFF0DDA5-5155-4144-B426-91C7623D08E7}"
 EndProject
-Project("{D954291E-2A0B-460D-934E-DC6B0785DB48}") = "Hl7.Fhir.Shims.Conformance", "src\Hl7.Fhir.Shims.Base\Hl7.Fhir.Shims.Conformance.shproj", "{150A59A2-371D-4747-8B08-C8E6340EC962}"
+Project("{D954291E-2A0B-460D-934E-DC6B0785DB48}") = "Hl7.Fhir.Shims.Base", "src\Hl7.Fhir.Shims.Base\Hl7.Fhir.Shims.Base.shproj", "{150A59A2-371D-4747-8B08-C8E6340EC962}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution

diff --git a/src/Hl7.Fhir.Shims.Base/Specification/Terminology/LocalTerminologyService.cs b/src/Hl7.Fhir.Shims.Base/Specification/Terminology/LocalTerminologyService.cs
@@ -118,17 +118,21 @@ private async Task<ValueSet> getExpandedValueSet(ValueSet vs, string operation)
         {
             try
             {
-                if (!vs.HasExpansion)
+                await _semaphore.WaitAsync().ConfigureAwait(false);
+
+                try
                 {
-                    try
+                    // We might have a cached or pre-expanded version brought to us by the _source
+                    if (!vs.HasExpansion)
                     {
-                        await _semaphore.WaitAsync().ConfigureAwait(false);
+                        // This will expand te vs - since we do not deepcopy() it, it will change the instance
+                        // as it was passed to us from the source
                         await _expander.ExpandAsync(vs).ConfigureAwait(false);
                     }
-                    finally
-                    {
-                        _semaphore.Release();
-                    }
+                }
+                finally
+                {
+                    _semaphore.Release();
                 }
             }
             catch (TerminologyServiceException e)

diff --git a/src/Hl7.Fhir.Specification.Shared.Tests/Source/TerminologyTests.cs b/src/Hl7.Fhir.Specification.Shared.Tests/Source/TerminologyTests.cs
@@ -4,9 +4,11 @@
 using Hl7.Fhir.Specification.Source;
 using Hl7.Fhir.Specification.Terminology;
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
+using System.Threading.Tasks.Dataflow;
 using Xunit;
 using T = System.Threading.Tasks;
 
@@ -700,6 +702,62 @@ public async T.Task FallbackServiceValidateCodeTestWithVS()
             isSuccess(result).Should().BeTrue();
         }
 
+
+
+        /// <summary>
+        /// Test for issue 556 (https://github.com/FirelyTeam/firely-net-sdk/issues/556) 
+        /// </summary>
+        [Fact, Trait("Category", "LongRunner")]
+        public async T.Task RunValueSetExpanderMultiThreaded()
+        {
+            var nrOfParrallelTasks = 50;
+            var results = new ConcurrentBag<(string uri, int total)>();
+            var valuesetSource = new CachedResolver(ZipSource.CreateValidationSource());
+
+            var valuesets = new string[] {
+                "http://hl7.org/fhir/ValueSet/request-status",
+                "http://hl7.org/fhir/ValueSet/care-plan-intent",
+                "http://hl7.org/fhir/ValueSet/administrative-gender",
+               };
+
+            void expandAction(string url)
+            {
+                var ts = new LocalTerminologyService(valuesetSource);
+                var parameters = new ExpandParameters().WithValueSet(url);
+
+#pragma warning disable xUnit1031 // Do not use blocking task operations in test method
+                var expansion = (ValueSet)ts.Expand(parameters).Result;
+#pragma warning restore xUnit1031 // Do not use blocking task operations in test method
+                results.Add((url, expansion.Expansion.Total ?? 0));
+            }
+
+            var processor = new ActionBlock<string>(expandAction, new ExecutionDataflowBlockOptions
+            {
+                MaxDegreeOfParallelism = 100
+            });
+
+            var buffer = new BufferBlock<string>();
+            buffer.LinkTo(processor, new DataflowLinkOptions { PropagateCompletion = true });
+
+            for (int i = 0; i < nrOfParrallelTasks; i++)
+            {
+                buffer.Post(valuesets[i % valuesets.Length]);
+            }
+
+            buffer.Complete();
+            await processor.Completion;
+
+            var groupsPerUri = results.GroupBy(results => results.uri);
+
+            // Verify that within each group, the number of items in the expansion is the same.
+            // Race conditions will cause these nubmers to differ per group.
+            foreach (var group in groupsPerUri)
+            {
+                group.Aggregate(group.First().total, (total, next) => { Assert.Equal(total, next.total); return total; });
+            }
+        }
+
+
         #region helper functions
         private static T.Task<Parameters> validateCodedValue(ITerminologyService service, string url = null, string context = null, string code = null,
             string system = null, string version = null, string display = null,