[C#] BumpCurrentEpoch calling threads must be protected (#871)

* - Ensure threads calling BumpCurrentEpoch are protected
- Back out change for shared bucket lock to wait for writer

* Move try/finally block for better inlining

Author: TedHartMS

cs/src/core/Epochs/EpochProtectedVersionScheme.cs

@@ -274,8 +274,8 @@ public EpochProtectedVersionScheme(LightEpoch epoch)
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private bool MakeTransition(VersionSchemeState expectedState, VersionSchemeState nextState)
         {
-            if (Interlocked.CompareExchange(ref state.Word, nextState.Word, expectedState.Word) !=
-                expectedState.Word) return false;
+            if (Interlocked.CompareExchange(ref state.Word, nextState.Word, expectedState.Word) != expectedState.Word) 
+                return false;
             Debug.WriteLine("Moved to {0}, {1}", nextState.Phase, nextState.Version);
             return true;
         }
@@ -358,28 +358,35 @@ internal void TryStepStateMachine(VersionSchemeStateMachine expectedMachine = nu
 
             var intermediate = VersionSchemeState.MakeIntermediate(oldState);
             if (!MakeTransition(oldState, intermediate)) return;
+
             // Avoid upfront memory allocation by going to a function
             StepMachineHeavy(machineLocal, oldState, nextState);
-
-            // Ensure that state machine is able to make progress if this thread is the only active thread
-            if (!epoch.ThisInstanceProtected())
-            {
-                epoch.Resume();
-                epoch.Suspend();
-            }
         }
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private void StepMachineHeavy(VersionSchemeStateMachine machineLocal, VersionSchemeState old, VersionSchemeState next)
         {
-            epoch.BumpCurrentEpoch(() =>
+            // // Resume epoch to ensure that state machine is able to make progress
+            // if this thread is the only active thread. Also, StepMachineHeavy calls BumpCurrentEpoch, which requires a protected thread.
+            bool isProtected = epoch.ThisInstanceProtected();
+            if (!isProtected)
+                epoch.Resume();
+            try
             {
-                machineLocal.OnEnteringState(old, next);
-                var success = MakeTransition(VersionSchemeState.MakeIntermediate(old), next);
-                machineLocal.AfterEnteringState(next);
-                Debug.Assert(success);
-                TryStepStateMachine(machineLocal);
-            });
+                epoch.BumpCurrentEpoch(() =>
+                {
+                    machineLocal.OnEnteringState(old, next);
+                    var success = MakeTransition(VersionSchemeState.MakeIntermediate(old), next);
+                    machineLocal.AfterEnteringState(next);
+                    Debug.Assert(success);
+                    TryStepStateMachine(machineLocal);
+                });
+            }
+            finally
+            {
+                if (!isProtected)
+                    epoch.Suspend();
+            }
         }
 
         /// <summary>

cs/src/core/Epochs/LightEpoch.cs

@@ -225,6 +225,7 @@ public void Resume()
         /// <returns></returns>
         long BumpCurrentEpoch()
         {
+            Debug.Assert(this.ThisInstanceProtected(), "BumpCurrentEpoch must be called on a protected thread");
             long nextEpoch = Interlocked.Increment(ref CurrentEpoch);
 
             if (drainCount > 0)

cs/src/core/FasterLog/FasterLog.cs

@@ -322,18 +322,22 @@ public void Dispose()
         /// <param name="spinWait"> whether to spin until log completion becomes committed </param>
         public void CompleteLog(bool spinWait = false)
         {
-
-            // Ensure all currently started entries will enqueue before we declare log closed
-            epoch.BumpCurrentEpoch(() =>
+            // Ensure progress even if there is no thread in epoch table. Also, BumpCurrentEpoch must be done on a protected thread.
+            bool isProtected = epoch.ThisInstanceProtected();
+            if (!isProtected)
+                epoch.Resume();
+            try
             {
-                CommitInternal(out _, out _, false, Array.Empty<byte>(), long.MaxValue, null);
-            });
-
-            // Ensure progress even if there is no thread in epoch table
-            if (!epoch.ThisInstanceProtected())
+                // Ensure all currently started entries will enqueue before we declare log closed
+                epoch.BumpCurrentEpoch(() =>
+                {
+                    CommitInternal(out _, out _, false, Array.Empty<byte>(), long.MaxValue, null);
+                });
+            }
+            finally
             {
-                epoch.Resume();
-                epoch.Suspend();
+                if (!isProtected)
+                    epoch.Suspend();
             }
 
             if (spinWait)

cs/src/core/Index/FASTER/FASTERBase.cs

@@ -116,27 +116,19 @@ public static bool TryAcquireSharedLatch(HashBucket* bucket)
 
             for (int spinCount = Constants.kMaxLockSpins; ; Thread.Yield())
             {
+                // Note: If reader starvation is encountered, consider rotating priority between reader and writer locks.
                 long expected_word = entry_word;
-                if ((expected_word & kSharedLatchBitMask) != kSharedLatchBitMask) // shared lock is not full
+                if ((expected_word & kExclusiveLatchBitMask) == 0) // not exclusively locked
                 {
-                    if (expected_word == Interlocked.CompareExchange(ref entry_word, expected_word + kSharedLatchIncrement, expected_word))
-                        break;
+                    if ((expected_word & kSharedLatchBitMask) != kSharedLatchBitMask) // shared lock is not full
+                    {
+                        if (expected_word == Interlocked.CompareExchange(ref entry_word, expected_word + kSharedLatchIncrement, expected_word))
+                            return true;
+                    }
                 }
                 if (--spinCount <= 0)
                     return false;
             }
-
-            // Wait for any writer to drain. Another session may hold the XLock on this bucket and need an epoch refresh to unlock, so limit this to avoid deadlock.
-            for (var ii = 0; ii < Constants.kMaxWriterLockDrainSpins; ++ii)
-            {
-                if ((entry_word & kExclusiveLatchBitMask) == 0)
-                    return true;
-                Thread.Yield();
-            }
-
-            // Release the shared-latch increment we added and return false so the caller will retry the operation.
-            Interlocked.Add(ref entry_word, -kSharedLatchIncrement);
-            return false;
         }
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]

cs/src/core/Index/Synchronization/IndexResizeStateMachine.cs

@@ -53,7 +53,18 @@ public void GlobalAfterEnteringState<Key, Value>(
             switch (next.Phase)
             {
                 case Phase.PREPARE_GROW:
-                    faster.epoch.BumpCurrentEpoch(() => allThreadsInPrepareGrow = true);
+                    bool isProtected = faster.epoch.ThisInstanceProtected();
+                    if (!isProtected)
+                        faster.epoch.Resume();
+                    try
+                    {
+                        faster.epoch.BumpCurrentEpoch(() => allThreadsInPrepareGrow = true);
+                    }
+                    finally
+                    {
+                        if (!isProtected)
+                            faster.epoch.Suspend();
+                    }
                     break;
                 case Phase.IN_PROGRESS_GROW:
                 case Phase.REST: