Merge pull request #74 from hudl/MARVEL-2040-FixConcurencyProblem

Replace non threaded safe List with a ConcurrentDictionary in bulkhead factory.

Author: codypeck

src/Hudl.Mjolnir/Bulkhead/BulkheadFactory.cs

@@ -97,7 +97,7 @@ internal class SemaphoreBulkheadHolder
             private readonly GroupKey _key;
 
             private ISemaphoreBulkhead _bulkhead;
-            public ISemaphoreBulkhead Bulkhead { get { return _bulkhead; } }
+            public ISemaphoreBulkhead Bulkhead => _bulkhead;
 
             private readonly IMetricEvents _metricEvents;
             private readonly MjolnirConfiguration _config;

src/Hudl.Mjolnir/Config/MjolnirConfiguration.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 
 namespace Hudl.Mjolnir.Config
@@ -141,7 +142,7 @@ public virtual BreakerConfiguration GetBreakerConfiguration(string key)
         /// </summary>
         public void NotifyAfterConfigUpdate()
         {
-            foreach (var observer in _observers)
+            foreach (var observer in _observers.Values)
             {
                 observer.OnNext(this);
             }
@@ -161,13 +162,14 @@ public void Dispose()
                 _onDispose();
             }
         }
-
-        private readonly List<IObserver<MjolnirConfiguration>> _observers = new List<IObserver<MjolnirConfiguration>>();
-
+        
+        private readonly ConcurrentDictionary<IObserver<MjolnirConfiguration>, IObserver<MjolnirConfiguration>>
+            _observers = new ConcurrentDictionary<IObserver<MjolnirConfiguration>, IObserver<MjolnirConfiguration>>();
+        
         public IDisposable Subscribe(IObserver<MjolnirConfiguration> observer)
         {
-            var subscription = new Subscription(() => _observers.Remove(observer));
-            _observers.Add(observer);
+            var subscription = new Subscription(() => _observers.TryRemove(observer, out var unused));
+            _observers.TryAdd(observer, observer);
             return subscription;
         }
     }

src/Hudl.Mjolnir/Hudl.Mjolnir.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
-    <VersionPrefix>3.2.0</VersionPrefix>
-    <FileVersion>3.2.0</FileVersion>
+    <VersionPrefix>3.2.1</VersionPrefix>
+    <FileVersion>3.2.1</FileVersion>
     <AssemblyVersion>3.0.0.0</AssemblyVersion>
     <TargetFrameworks>netstandard1.2</TargetFrameworks>
     <GenerateAssemblyInfo>false</GenerateAssemblyInfo>

tests/unit/Hudl.Mjolnir.Tests/Bulkhead/BulkheadFactoryTests.cs

@@ -5,8 +5,10 @@
 using Hudl.Mjolnir.Tests.Helper;
 using Moq;
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Threading;
+using System.Threading.Tasks;
 using Hudl.Mjolnir.Config;
 using Xunit;
 using static Hudl.Mjolnir.Bulkhead.BulkheadFactory;
@@ -335,5 +337,61 @@ public void GetBulkhead_WhenInitializingBulkheadAndMaxConcurrentConfigIsInvalid_
 
             Assert.Equal(validMaxConcurrent, bulkhead.CountAvailable);
         }
+
+
+        /// <summary>
+        /// This test may past even if there is a concurrency problem. It will always pass if the concurrency is working
+        /// fine. Do not ignore FAILURES!!
+        /// </summary>
+        [Fact]
+        public void MultiThreadedInitialization_DoesNotThrow()
+        {
+            var mockMetricEvents = new Mock<IMetricEvents>(MockBehavior.Strict);
+
+            var key = AnyGroupKey;
+            const int validMaxConcurrent = 10;
+
+            var mockConfig = new MjolnirConfiguration
+            {
+                BulkheadConfigurations = new Dictionary<string, BulkheadConfiguration>
+                {
+                    {
+                        key.Name,
+                        new BulkheadConfiguration
+                        {
+                            MaxConcurrent = validMaxConcurrent
+                        }
+                    }
+                }
+            };
+
+            var mockLogFactory = new Mock<IMjolnirLogFactory>(MockBehavior.Strict);
+            mockLogFactory.Setup(m => m.CreateLog<BulkheadFactory>()).Returns(new DefaultMjolnirLog<BulkheadFactory>());
+            mockLogFactory.Setup(m => m.CreateLog<SemaphoreBulkheadHolder>())
+                .Returns(new DefaultMjolnirLog<SemaphoreBulkheadHolder>());
+            var factory = new BulkheadFactory(mockMetricEvents.Object, mockConfig, mockLogFactory.Object);
+
+            var groupKeys = new List<GroupKey>();
+            for (var i = 0; i < 50000000; ++i)
+            {
+                groupKeys.Add(GroupKey.Named(1.ToString()));
+            }
+
+            var exceptions = new ConcurrentQueue<Exception>();
+            
+            Parallel.ForEach(groupKeys, (groupKey) =>
+            {
+                try
+                {
+                    factory.GetBulkhead(groupKey);
+                }
+                catch (Exception e)
+                {
+                    exceptions.Enqueue(e);
+                }
+            });
+            
+            Assert.Empty(exceptions);
+        }
     }
 }