Bound LightEpoch waiter semaphore signals to prevent SemaphoreFullException

libs/server/StoreWrapper.cs

@@ -714,23 +714,34 @@ async Task CommitTaskAsync(int commitFrequencyMs, CancellationToken token = defa
             }
         }
 
-        async Task CompactionTaskAsync(int compactionFrequencySecs, CancellationToken token = default)
+        /// <summary>
+        /// Run <paramref name="runCycle"/> every <paramref name="interval"/> until <paramref name="token"/> is
+        /// cancelled. A cycle that throws is logged and retried on the next interval, so a transient fault
+        /// cannot disable a maintenance task for the lifetime of the process. Cancellation ends the loop
+        /// without logging.
+        /// </summary>
+        /// <param name="runCycle">The work to perform on each cycle.</param>
+        /// <param name="interval">Delay between cycles.</param>
+        /// <param name="taskDescription">Task name used in the failure log message.</param>
+        /// <param name="logger">Logger for failed cycles, may be null.</param>
+        /// <param name="token">Cancellation token that ends the loop.</param>
+        internal static async Task RunMaintenanceLoopAsync(Func<CancellationToken, ValueTask> runCycle,
+            TimeSpan interval, string taskDescription, ILogger logger, CancellationToken token)
         {
-            Debug.Assert(compactionFrequencySecs > 0);
             try
             {
-                while (true)
+                while (!token.IsCancellationRequested)
                 {
-                    if (token.IsCancellationRequested) return;
-
-                    await databaseManager.DoCompactionAsync(token, logger).ConfigureAwait(false);
-
-                    if (!runtimeConfig.GetBool(ServerConfigType.COMPACTION_FORCE_DELETE))
-                        logger?.LogInformation("NOTE: Take a checkpoint (SAVE/BGSAVE) in order to actually delete the older data segments (files) from disk");
-                    else
-                        logger?.LogInformation("NOTE: Compaction will delete files, make sure checkpoint/recovery is not being used");
+                    try
+                    {
+                        await runCycle(token).ConfigureAwait(false);
+                    }
+                    catch (Exception ex) when (!token.IsCancellationRequested)
+                    {
+                        logger?.LogError(ex, "Unknown exception received for {taskDescription}. Retrying on the next cycle.", taskDescription);
+                    }
 
-                    await Task.Delay(compactionFrequencySecs * 1000, token).ConfigureAwait(false);
+                    await Task.Delay(interval, token).ConfigureAwait(false);
                 }
             }
             catch (OperationCanceledException) when (token.IsCancellationRequested)
@@ -740,62 +751,51 @@ async Task CompactionTaskAsync(int compactionFrequencySecs, CancellationToken to
             }
             catch (Exception ex)
             {
-                logger?.LogError(ex, "CompactionTask exception received.");
+                logger?.LogCritical(ex, "Unknown exception received for {taskDescription}. The task won't be resumed.", taskDescription);
             }
         }
 
-        async Task ObjectCollectTaskAsync(int objectCollectFrequencySecs, CancellationToken token = default)
+        Task CompactionTaskAsync(int compactionFrequencySecs, CancellationToken token = default)
         {
-            Debug.Assert(objectCollectFrequencySecs > 0);
-            try
+            Debug.Assert(compactionFrequencySecs > 0);
+
+            return RunMaintenanceLoopAsync(async (cancellationToken) =>
             {
-                if (serverOptions.DisableObjects)
-                {
-                    logger?.LogWarning("ExpiredObjectCollectionFrequencySecs option is configured but Object store is disabled. Stopping the background hash collect task.");
-                    return;
-                }
+                await databaseManager.DoCompactionAsync(cancellationToken, logger).ConfigureAwait(false);
 
-                while (true)
-                {
-                    if (token.IsCancellationRequested) return;
+                if (!runtimeConfig.GetBool(ServerConfigType.COMPACTION_FORCE_DELETE))
+                    logger?.LogInformation("NOTE: Take a checkpoint (SAVE/BGSAVE) in order to actually delete the older data segments (files) from disk");
+                else
+                    logger?.LogInformation("NOTE: Compaction will delete files, make sure checkpoint/recovery is not being used");
+            }, TimeSpan.FromSeconds(compactionFrequencySecs), "background compaction task", logger, token);
+        }
 
-                    databaseManager.ExecuteObjectCollection();
+        Task ObjectCollectTaskAsync(int objectCollectFrequencySecs, CancellationToken token = default)
+        {
+            Debug.Assert(objectCollectFrequencySecs > 0);
 
-                    await Task.Delay(TimeSpan.FromSeconds(objectCollectFrequencySecs), token).ConfigureAwait(false);
-                }
-            }
-            catch (TaskCanceledException) when (token.IsCancellationRequested)
+            if (serverOptions.DisableObjects)
             {
-                // Suppress the exception if the task was cancelled because of store wrapper disposal
+                logger?.LogWarning("ExpiredObjectCollectionFrequencySecs option is configured but Object store is disabled. Stopping the background hash collect task.");
+                return Task.CompletedTask;
             }
-            catch (Exception ex)
+
+            return RunMaintenanceLoopAsync((_) =>
             {
-                logger?.LogCritical(ex, "Unknown exception received for background hash collect task. Object collect task won't be resumed.");
-            }
+                databaseManager.ExecuteObjectCollection();
+                return ValueTask.CompletedTask;
+            }, TimeSpan.FromSeconds(objectCollectFrequencySecs), "background hash collect task", logger, token);
         }
 
-        async Task ExpiredKeyDeletionScanTaskAsync(int expiredKeyDeletionScanFrequencySecs, CancellationToken token = default)
+        Task ExpiredKeyDeletionScanTaskAsync(int expiredKeyDeletionScanFrequencySecs, CancellationToken token = default)
         {
             Debug.Assert(expiredKeyDeletionScanFrequencySecs > 0);
-            try
-            {
-                while (true)
-                {
-                    if (token.IsCancellationRequested) return;
-
-                    databaseManager.ExpiredKeyDeletionScan();
 
-                    await Task.Delay(TimeSpan.FromSeconds(expiredKeyDeletionScanFrequencySecs), token).ConfigureAwait(false);
-                }
-            }
-            catch (TaskCanceledException) when (token.IsCancellationRequested)
+            return RunMaintenanceLoopAsync((_) =>
             {
-                // Suppress the exception if the task was cancelled because of store wrapper disposal
-            }
-            catch (Exception ex)
-            {
-                logger?.LogCritical(ex, "Unknown exception received for background expired key deletion scan task. The task won't be resumed.");
-            }
+                databaseManager.ExpiredKeyDeletionScan();
+                return ValueTask.CompletedTask;
+            }, TimeSpan.FromSeconds(expiredKeyDeletionScanFrequencySecs), "background expired key deletion scan task", logger, token);
         }
 
         /// <summary>

libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.TestHooks.cs

@@ -37,5 +37,27 @@ internal long TestHookThisThreadAnnouncedEpoch()
         /// Capacity of the drain list.
         /// </summary>
         internal static int TestHookDrainListCapacity => kDrainListSize;
+
+        /// <summary>
+        /// Number of entries in the epoch table.
+        /// </summary>
+        internal static int TestHookTableSize => kTableSize;
+
+        /// <summary>
+        /// Number of threads currently waiting for an epoch table entry.
+        /// </summary>
+        internal int TestHookWaiterCount => waiterCount & ~kDisposedFlag;
+
+        /// <summary>
+        /// Signals issued to waiters but not yet consumed. This is the count that overflows
+        /// <see cref="System.Threading.SemaphoreSlim"/> if <see cref="Release"/> signals unconditionally.
+        /// </summary>
+        internal int TestHookOutstandingWaiterSignals => waiterSemaphore.CurrentCount;
+
+        /// <summary>
+        /// Signal reservations taken by <see cref="SignalWaiter"/>, an upper bound on
+        /// <see cref="TestHookOutstandingWaiterSignals"/>.
+        /// </summary>
+        internal int TestHookPendingWaiterSignals => pendingWaiterSignals;
     }
 }

libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs

@@ -126,6 +126,28 @@ private class Metadata
         /// </summary>
         volatile int waiterCount = 0;
 
+        /// <summary>
+        /// Reservations taken by <see cref="SignalWaiter"/> for <see cref="waiterSemaphore"/> signals that
+        /// have been issued but not yet consumed by a waiter. A reservation is only taken while this is below
+        /// <see cref="waiterCount"/>, so it never exceeds the peak number of concurrent waiters. The bound is
+        /// against the peak rather than the instantaneous count, because a waiter can claim a slot and leave
+        /// while a reservation taken for it is still in flight, briefly leaving a signal outstanding with no
+        /// waiter present. A leftover signal is consumed by the next waiter, which decrements this and
+        /// re-probes the table, so leftovers drain instead of accumulating. Any such bound is enough: without
+        /// one the count grows by one per epoch release for as long as any waiter exists, and eventually
+        /// exceeds <see cref="SemaphoreSlim"/>'s maximum.
+        ///
+        /// The counter is explicit rather than a comparison of <see cref="SemaphoreSlim.CurrentCount"/> against
+        /// <see cref="waiterCount"/> at signal time. Such a comparison is not atomic with the
+        /// <see cref="SemaphoreSlim.Release()"/> that would follow it, so concurrent releasers observe the same
+        /// below-threshold count and each signal on it, which is the common case under the contention that makes
+        /// the bound matter. The resulting surplus also persists, since the count only falls when a waiter
+        /// consumes a signal, so it carries from one contention burst into the next. Taking the reservation by
+        /// CAS instead makes the increment itself the permission to signal, and the reservation is returned when
+        /// the signal is consumed.
+        /// </summary>
+        volatile int pendingWaiterSignals = 0;
+
         /// <summary>
         /// Flag (MSB) used to mark the epoch as disposed in <see cref="waiterCount"/>.
         /// </summary>
@@ -550,7 +572,46 @@ void Release()
 
             entry = kInvalidIndex;
             if (waiterCount > 0)
-                waiterSemaphore.Release();
+                SignalWaiter();
+        }
+
+        /// <summary>
+        /// Wake one thread waiting in <see cref="ReserveEntryWait"/> for the slot just freed, unless every
+        /// current waiter already has a wake pending.
+        ///
+        /// The increment is what reserves the right to signal, so it always precedes its
+        /// <see cref="SemaphoreSlim.Release()"/> and the matching decrement always follows a successful wait.
+        /// The semaphore's count is therefore never above <see cref="pendingWaiterSignals"/>, and a
+        /// reservation is only taken while that is below <see cref="waiterCount"/>, so both are capped by the
+        /// peak number of concurrent waiters. The cap is against the peak rather than the instantaneous count:
+        /// a waiter can claim the freed slot on its re-probe and decrement <see cref="waiterCount"/> while a
+        /// reservation taken for it is still in flight, leaving a signal outstanding with no waiter present.
+        /// The next waiter consumes that signal, decrements, and re-probes, so leftovers drain.
+        ///
+        /// Suppressing a signal cannot lose a wakeup. A waiter only blocks when the semaphore's count is zero,
+        /// and a signal is only suppressed when a wake is already queued for every waiter, so at least one is
+        /// inbound. Waiters are released in FIFO order and re-probe the whole table before blocking again, so
+        /// whichever one wakes finds this slot (or a later one). The read of <see cref="pendingWaiterSignals"/>
+        /// here follows the release-store that published the slot as free, so a waiter woken by an
+        /// already-outstanding signal sees it.
+        ///
+        /// During <see cref="Dispose"/> the MSB of <see cref="waiterCount"/> makes it negative, so no signal is
+        /// issued; waiters unwind through <see cref="cts"/> instead.
+        /// </summary>
+        [MethodImpl(MethodImplOptions.NoInlining)]
+        void SignalWaiter()
+        {
+            var pending = pendingWaiterSignals;
+            while (pending < waiterCount)
+            {
+                var prev = Interlocked.CompareExchange(ref pendingWaiterSignals, pending + 1, pending);
+                if (prev == pending)
+                {
+                    waiterSemaphore.Release();
+                    return;
+                }
+                pending = prev;
+            }
         }
 
         /// <summary>
@@ -644,6 +705,10 @@ void ReserveEntryWait(ref int entry)
 
                     // No slot available, wait for a signal from Release()
                     waiterSemaphore.Wait(cts.Token);
+
+                    // A signal was consumed, so release the reservation SignalWaiter() took for it. Only
+                    // reached when the wait succeeded; a cancelled wait consumes nothing.
+                    _ = Interlocked.Decrement(ref pendingWaiterSignals);
                 }
             }
             catch (OperationCanceledException)