Mitigate lock convoys when AtomicFactory throws (#477)

Author: bitfaster

BitFaster.Caching.UnitTests/Atomic/AtomicFactoryTests.cs

@@ -1,4 +1,5 @@
 
+using System;
 using System.Threading;
 using System.Threading.Tasks;
 using BitFaster.Caching.Atomic;
@@ -118,7 +119,9 @@ public void WhenArgObjectValuesAreSameEqualsTrue()
         [Fact]
         public async Task WhenCallersRunConcurrentlyResultIsFromWinner()
         {
-            var enter = new ManualResetEvent(false);
+            var enter1 = new ManualResetEvent(false);
+            var enter2 = new ManualResetEvent(false);
+            var factory = new ManualResetEvent(false);
             var resume = new ManualResetEvent(false);
 
             var atomicFactory = new AtomicFactory<int, int>();
@@ -127,9 +130,10 @@ public async Task WhenCallersRunConcurrentlyResultIsFromWinner()
 
             Task<int> first = Task.Run(() =>
             {
+                enter1.Set();
                 return atomicFactory.GetValue(1, k =>
                 {
-                    enter.Set();
+                    factory.Set();
                     resume.WaitOne();
 
                     result = 1;
@@ -140,9 +144,10 @@ public async Task WhenCallersRunConcurrentlyResultIsFromWinner()
 
             Task<int> second = Task.Run(() =>
             {
+                enter2.Set();
                 return atomicFactory.GetValue(1, k =>
                 {
-                    enter.Set();
+                    factory.Set();
                     resume.WaitOne();
 
                     result = 2;
@@ -151,13 +156,114 @@ public async Task WhenCallersRunConcurrentlyResultIsFromWinner()
                 });
             });
 
-            enter.WaitOne();
+            enter1.WaitOne();
+            enter2.WaitOne();
+            factory.WaitOne();
             resume.Set();
 
             (await first).Should().Be(result);
             (await second).Should().Be(result);
 
             winnerCount.Should().Be(1);
         }
+
+        [Fact]
+        public async Task WhenCallersRunConcurrentlyAndFailExceptionIsPropogated()
+        {
+            var enter1 = new ManualResetEvent(false);
+            var enter2 = new ManualResetEvent(false);
+            var factory = new ManualResetEvent(false);
+            var resume = new ManualResetEvent(false);
+
+            var atomicFactory = new AtomicFactory<int, int>();
+            var throwCount = 0;
+
+            Task<int> first = Task.Run(() =>
+            {
+                enter1.Set();
+                return atomicFactory.GetValue(1, k =>
+                {
+                    factory.Set();
+                    resume.WaitOne();
+
+                    Interlocked.Increment(ref throwCount);
+                    throw new Exception();
+                });
+            });
+
+            Task<int> second = Task.Run(() =>
+            {
+                enter2.Set();
+                return atomicFactory.GetValue(1, k =>
+                {
+                    factory.Set();
+                    resume.WaitOne();
+
+                    Interlocked.Increment(ref throwCount);
+                    throw new Exception();
+                });
+            });
+
+            enter1.WaitOne();
+            enter2.WaitOne();
+            factory.WaitOne();
+            resume.Set();
+
+            Func<Task> act1 = () => first;
+            Func<Task> act2 = () => second;
+
+            await act1.Should().ThrowAsync<Exception>();
+            await act2.Should().ThrowAsync<Exception>();
+
+            // verify only one exception was thrown
+            throwCount.Should().Be(1);
+        }
+
+        [Fact]
+        public async Task WhenCallersRunConcurrentlyAndFailNewCallerStartsClean()
+        {
+            var enter1 = new ManualResetEvent(false);
+            var enter2 = new ManualResetEvent(false);
+            var factory = new ManualResetEvent(false);
+            var resume = new ManualResetEvent(false);
+
+            var atomicFactory = new AtomicFactory<int, int>();
+
+            Task<int> first = Task.Run(() =>
+            {
+                enter1.Set();
+                return atomicFactory.GetValue(1, k =>
+                {
+                    factory.Set();
+                    resume.WaitOne();
+                    throw new Exception();
+                });
+            });
+
+            Task<int> second = Task.Run(() =>
+            {
+                enter2.Set();
+                return atomicFactory.GetValue(1, k =>
+                {
+                    factory.Set();
+                    resume.WaitOne();
+                    throw new Exception();
+                });
+            });
+
+            enter1.WaitOne();
+            enter2.WaitOne();
+            factory.WaitOne();
+            resume.Set();
+
+            Func<Task> act1 = () => first;
+            Func<Task> act2 = () => second;
+
+            await act1.Should().ThrowAsync<Exception>();
+            await act2.Should().ThrowAsync<Exception>();
+
+            // verify exception is no longer cached
+            atomicFactory.GetValue(1, k => k).Should().Be(1);
+        }
     }
 }

BitFaster.Caching/Atomic/AtomicFactory.cs

@@ -1,13 +1,14 @@
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.Runtime.ExceptionServices;
 using System.Threading;
 
 namespace BitFaster.Caching.Atomic
 {
     /// <summary>
-    /// A class that provides simple, lightweight exactly once initialization for values
-    /// stored in a cache.
+    /// A class that provides simple, lightweight exactly once initialization for values stored
+    /// in a cache. Exceptions are propogated to the caller.
     /// </summary>
     /// <typeparam name="K">The type of the key.</typeparam>
     /// <typeparam name="V">The type of the value.</typeparam>
@@ -92,14 +93,35 @@ public V ValueIfCreated
             }
         }
 
+        /// <summary>
+        /// Note the failure case works like this:
+        /// 1. Thread A enters AtomicFactory.CreateValue then Initializer.CreateValue and holds the lock.
+        /// 2. Thread B enters AtomicFactory.CreateValue then Initializer.CreateValue and queues on the lock.
+        /// 3. Thread A calls value factory, and after 1 second throws an exception. The exception is 
+        /// captured in exceptionDispatch, lock is released, and an exeption is thrown.
+        /// 4. AtomicFactory.CreateValue catches the exception and creates a fresh initializer.
+        /// 5. Thread B enters the lock, finds exceptionDispatch is populated and immediately throws.
+        /// 6. Thread C can now start from a clean state.
+        /// This mitigates lock convoys where many queued threads will fail slowly one by one, introducing delays
+        /// and multiplying the number of calls to the failing resource.
+        /// </summary>
         private V CreateValue<TFactory>(K key, TFactory valueFactory) where TFactory : struct, IValueFactory<K, V>
         {
             var init = Volatile.Read(ref initializer);
 
             if (init != null)
             {
-                value = init.CreateValue(key, valueFactory);
-                Volatile.Write(ref initializer, null); // volatile write must occur after setting value
+                try
+                {
+                    value = init.CreateValue(key, valueFactory);
+                    Volatile.Write(ref initializer, null); // volatile write must occur after setting value
+                }
+                catch
+                {
+                    // Overwrite the initializer with a fresh copy. New threads will start from a clean state.
+                    Volatile.Write(ref initializer, new Initializer());
+                    throw;
+                }
             }
 
             return value;
@@ -138,6 +160,7 @@ private class Initializer
         {
             private bool isInitialized;
             private V value;
+            private ExceptionDispatchInfo exceptionDispatch;
 
             public V CreateValue<TFactory>(K key, TFactory valueFactory) where TFactory : struct, IValueFactory<K, V>
             {
@@ -148,9 +171,24 @@ public V CreateValue<TFactory>(K key, TFactory valueFactory) where TFactory : st
                         return value;
                     }
 
-                    value = valueFactory.Create(key);
-                    isInitialized = true;
-                    return value;
+                    // If a previous thread called the factory and failed, throw the same error instead
+                    // of calling the factory again.
+                    if (exceptionDispatch != null)
+                    {
+                        exceptionDispatch.Throw();
+                    }
+
+                    try
+                    {
+                        value = valueFactory.Create(key);
+                        isInitialized = true;
+                        return value;
+                    }
+                    catch (Exception ex)
+                    {
+                        exceptionDispatch = ExceptionDispatchInfo.Capture(ex);
+                        throw;
+                    }
                 }
             }
         }