From 2c9b91770cce8777b6edc423b8b1b191a6ea7a87 Mon Sep 17 00:00:00 2001
From: TedHartMS <15467143+TedHartMS@users.noreply.github.com>
Date: Tue, 15 Sep 2026 18:22:18 -0700
Subject: [PATCH 1/4] Bound LightEpoch waiter semaphore signals (#2135)
LightEpoch.Release() signalled waiterSemaphore once per epoch release for as
long as any thread was waiting for a table entry, but each waiter consumes
exactly one signal. ProtectAndDrain() hops through SuspendResume() on every
refresh while a waiter exists, so releases vastly outnumber waiters under
contention and the semaphore's count climbed monotonically. At int.MaxValue
SemaphoreSlim.Release() began throwing SemaphoreFullException out of every
epoch release, killing client sessions and permanently stopping the object
collect and expired key deletion tasks until the process was restarted.
Track outstanding signals in pendingWaiterSignals and only signal while it is
below waiterCount. The reservation is taken before the semaphore release and
returned after a successful wait, so the semaphore's count can never exceed the
number of waiters. Suppressing a signal cannot strand a waiter: a signal is
only withheld when every current waiter already has a wake pending, and a woken
waiter re-probes the whole table before blocking again.
Also stop the two background tasks in StoreWrapper from terminating for the
lifetime of the process on an unexpected exception. The per-cycle work is now
guarded so a failed cycle is logged and retried on the next interval.
---
libs/server/StoreWrapper.cs | 47 ++--
.../src/core/Epochs/LightEpoch.TestHooks.cs | 22 ++
.../cs/src/core/Epochs/LightEpoch.cs | 50 +++-
.../cs/test/test.epoch/WaiterSignalTests.cs | 240 ++++++++++++++++++
website/docs/dev/tsavorite/epochprotection.md | 16 +-
5 files changed, 357 insertions(+), 18 deletions(-)
create mode 100644 libs/storage/Tsavorite/cs/test/test.epoch/WaiterSignalTests.cs
diff --git a/libs/server/StoreWrapper.cs b/libs/server/StoreWrapper.cs
index 5e81d8be9fc..a3b5aa5dac5 100644
--- a/libs/server/StoreWrapper.cs
+++ b/libs/server/StoreWrapper.cs
@@ -722,24 +722,32 @@ async Task CompactionTaskAsync(int compactionFrequencySecs, CancellationToken to
async Task ObjectCollectTaskAsync(int objectCollectFrequencySecs, CancellationToken token = default)
{
Debug.Assert(objectCollectFrequencySecs > 0);
- try
+
+ if (serverOptions.DisableObjects)
{
- if (serverOptions.DisableObjects)
- {
- logger?.LogWarning("ExpiredObjectCollectionFrequencySecs option is configured but Object store is disabled. Stopping the background hash collect task.");
- return;
- }
+ logger?.LogWarning("ExpiredObjectCollectionFrequencySecs option is configured but Object store is disabled. Stopping the background hash collect task.");
+ return;
+ }
- while (true)
+ try
+ {
+ while (!token.IsCancellationRequested)
{
- if (token.IsCancellationRequested) return;
-
- databaseManager.ExecuteObjectCollection();
+ // A failed cycle must not take the task down with it; retry on the next interval so a
+ // transient fault cannot disable object collection for the lifetime of the process.
+ try
+ {
+ databaseManager.ExecuteObjectCollection();
+ }
+ catch (Exception ex) when (!token.IsCancellationRequested)
+ {
+ logger?.LogError(ex, "Unknown exception received for background hash collect task. Retrying on the next cycle.");
+ }
await Task.Delay(TimeSpan.FromSeconds(objectCollectFrequencySecs), token).ConfigureAwait(false);
}
}
- catch (TaskCanceledException) when (token.IsCancellationRequested)
+ catch (OperationCanceledException) when (token.IsCancellationRequested)
{
// Suppress the exception if the task was cancelled because of store wrapper disposal
}
@@ -754,16 +762,23 @@ async Task ExpiredKeyDeletionScanTaskAsync(int expiredKeyDeletionScanFrequencySe
Debug.Assert(expiredKeyDeletionScanFrequencySecs > 0);
try
{
- while (true)
+ while (!token.IsCancellationRequested)
{
- if (token.IsCancellationRequested) return;
-
- databaseManager.ExpiredKeyDeletionScan();
+ // A failed cycle must not take the task down with it; retry on the next interval so a
+ // transient fault cannot disable expired key deletion for the lifetime of the process.
+ try
+ {
+ databaseManager.ExpiredKeyDeletionScan();
+ }
+ catch (Exception ex) when (!token.IsCancellationRequested)
+ {
+ logger?.LogError(ex, "Unknown exception received for background expired key deletion scan task. Retrying on the next cycle.");
+ }
await Task.Delay(TimeSpan.FromSeconds(expiredKeyDeletionScanFrequencySecs), token).ConfigureAwait(false);
}
}
- catch (TaskCanceledException) when (token.IsCancellationRequested)
+ catch (OperationCanceledException) when (token.IsCancellationRequested)
{
// Suppress the exception if the task was cancelled because of store wrapper disposal
}
diff --git a/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.TestHooks.cs b/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.TestHooks.cs
index 105bfbd5c32..bfeb110fa4e 100644
--- a/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.TestHooks.cs
+++ b/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.TestHooks.cs
@@ -37,5 +37,27 @@ internal long TestHookThisThreadAnnouncedEpoch()
/// Capacity of the drain list.
///
internal static int TestHookDrainListCapacity => kDrainListSize;
+
+ ///
+ /// Number of entries in the epoch table.
+ ///
+ internal static int TestHookTableSize => kTableSize;
+
+ ///
+ /// Number of threads currently waiting for an epoch table entry.
+ ///
+ internal int TestHookWaiterCount => waiterCount & ~kDisposedFlag;
+
+ ///
+ /// Signals issued to waiters but not yet consumed. This is the count that overflows
+ /// if signals unconditionally.
+ ///
+ internal int TestHookOutstandingWaiterSignals => waiterSemaphore.CurrentCount;
+
+ ///
+ /// Signal reservations taken by , an upper bound on
+ /// .
+ ///
+ internal int TestHookPendingWaiterSignals => pendingWaiterSignals;
}
}
\ No newline at end of file
diff --git a/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs b/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs
index ca8e0929304..8de8d1c4020 100644
--- a/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs
+++ b/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs
@@ -126,6 +126,15 @@ private class Metadata
///
volatile int waiterCount = 0;
+ ///
+ /// Number of signals that have been issued but not yet consumed by a
+ /// waiter. only issues a signal while this is below ,
+ /// which bounds the semaphore's count by the number of waiters. Without that bound the count grows by
+ /// one per epoch release for as long as any waiter exists, and eventually overflows
+ /// 's maximum.
+ ///
+ volatile int pendingWaiterSignals = 0;
+
///
/// Flag (MSB) used to mark the epoch as disposed in .
///
@@ -550,7 +559,42 @@ void Release()
entry = kInvalidIndex;
if (waiterCount > 0)
- waiterSemaphore.Release();
+ SignalWaiter();
+ }
+
+ ///
+ /// Wake one thread waiting in 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
+ /// and the matching decrement always follows a successful wait.
+ /// The semaphore's count is therefore never above , which this loop
+ /// holds at or below .
+ ///
+ /// 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
+ /// here follows the release-store that published the slot as free, so a waiter woken by an
+ /// already-outstanding signal sees it.
+ ///
+ /// During the MSB of makes it negative, so no signal is
+ /// issued; waiters unwind through instead.
+ ///
+ [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;
+ }
}
///
@@ -644,6 +688,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)
diff --git a/libs/storage/Tsavorite/cs/test/test.epoch/WaiterSignalTests.cs b/libs/storage/Tsavorite/cs/test/test.epoch/WaiterSignalTests.cs
new file mode 100644
index 00000000000..fb7c718b5c6
--- /dev/null
+++ b/libs/storage/Tsavorite/cs/test/test.epoch/WaiterSignalTests.cs
@@ -0,0 +1,240 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+using System;
+using System.Threading;
+using NUnit.Framework;
+using Tsavorite.core;
+
+namespace Tsavorite.test.epoch
+{
+ ///
+ /// The waiter semaphore that parks threads when the epoch table has no free slot. A thread that
+ /// releases a slot wakes a waiter, and the signals it issues must stay bounded by the number of
+ /// waiters: signalling once per release for as long as any waiter exists grows the semaphore's
+ /// count without bound until throws
+ /// , which then escapes every epoch release process-wide.
+ ///
+ [TestFixture]
+ public class WaiterSignalTests : EpochTestBase
+ {
+ /// Longest any of these runs should ever take; exceeded only by a lost wakeup.
+ static readonly TimeSpan JoinTimeout = TimeSpan.FromSeconds(120);
+
+ /// What a single oversubscribed run observed.
+ readonly struct RunResult
+ {
+ /// Total refresh cycles completed across all threads.
+ internal long Cycles { get; init; }
+
+ /// Highest unconsumed signal count seen while the threads ran.
+ internal int MaxOutstandingSignals { get; init; }
+
+ /// Highest signal reservation count seen while the threads ran.
+ internal int MaxPendingSignals { get; init; }
+
+ /// Highest number of threads seen blocked waiting for a slot.
+ internal int MaxWaiters { get; init; }
+ }
+
+ [Test]
+ public void SignalCountStaysBoundedWhenTheTableIsOversubscribed()
+ {
+ var threadCount = LightEpoch.TestHookTableSize + 8;
+ var result = RunOversubscribed(threadCount, cyclesPerThread: 2_000);
+
+ Assert.Multiple(() =>
+ {
+ Assert.That(result.MaxWaiters, Is.GreaterThan(0),
+ "the table was never actually oversubscribed, so nothing was exercised");
+ Assert.That(result.Cycles, Is.GreaterThan(10_000),
+ "too few epoch releases for the bound to mean anything");
+ Assert.That(result.MaxOutstandingSignals, Is.LessThanOrEqualTo(threadCount),
+ "unconsumed signals outgrew the waiters that can consume them, so the count is unbounded");
+ Assert.That(result.MaxPendingSignals, Is.LessThanOrEqualTo(threadCount),
+ "signal reservations outgrew the waiters they were taken for");
+ Assert.That(result.MaxOutstandingSignals, Is.LessThanOrEqualTo(result.MaxPendingSignals),
+ "every signal must be covered by a reservation taken before it was issued");
+ });
+ }
+
+ ///
+ /// Suppressing a signal must never strand a waiter: with far more threads than slots, most of
+ /// them are parked at any moment and nearly every release is a candidate for suppression.
+ ///
+ [Test]
+ public void EveryThreadCompletesWhenTheTableIsOversubscribed()
+ {
+ var threadCount = LightEpoch.TestHookTableSize + 32;
+ var result = RunOversubscribed(threadCount, cyclesPerThread: 500);
+
+ Assert.Multiple(() =>
+ {
+ Assert.That(result.MaxWaiters, Is.GreaterThan(0),
+ "the table was never actually oversubscribed, so nothing was exercised");
+ Assert.That(result.Cycles, Is.EqualTo((long)threadCount * 500),
+ "every thread must complete all of its cycles");
+ Assert.That(epoch.TestHookWaiterCount, Is.Zero, "a thread was left waiting for a slot");
+ });
+ }
+
+ ///
+ /// Signals left over from one contention burst must not carry into the next. Accumulating across
+ /// bursts is what takes a long-lived server to the overflow.
+ ///
+ [Test]
+ public void SignalsDrainAfterContentionEnds()
+ {
+ var threadCount = LightEpoch.TestHookTableSize + 8;
+
+ for (var burst = 1; burst <= 3; burst++)
+ {
+ var result = RunOversubscribed(threadCount, cyclesPerThread: 500);
+
+ Assert.That(result.MaxWaiters, Is.GreaterThan(0),
+ $"burst {burst} never oversubscribed the table, so nothing was exercised");
+ Assert.That(epoch.TestHookPendingWaiterSignals, Is.LessThanOrEqualTo(threadCount),
+ $"signal reservations accumulated across bursts (after burst {burst})");
+ Assert.That(epoch.TestHookOutstandingWaiterSignals, Is.LessThanOrEqualTo(threadCount),
+ $"unconsumed signals accumulated across bursts (after burst {burst})");
+ }
+ }
+
+ ///
+ /// Run threads, more than the table has slots, each completing
+ /// refreshes. Slots are filled first and the surplus threads
+ /// are confirmed parked before any of them is allowed to churn, so the run always starts from a
+ /// genuinely oversubscribed table. then hops through
+ /// SuspendResume for as long as a waiter exists, so every cycle releases a slot while the
+ /// threads that lost the race for one are parked on the semaphore. A sampler thread watches the
+ /// signal counters throughout.
+ ///
+ RunResult RunOversubscribed(int threadCount, int cyclesPerThread)
+ {
+ var slotCount = LightEpoch.TestHookTableSize;
+ Assert.That(threadCount, Is.GreaterThan(slotCount), "the run must ask for more threads than the table has slots");
+
+ var contenderCount = threadCount - slotCount;
+ var slotsFilled = new CountdownEvent(slotCount);
+ var go = new ManualResetEventSlim();
+ var running = true;
+ long cycles = 0;
+ var maxOutstandingSignals = 0;
+ var maxPendingSignals = 0;
+ var maxWaiters = 0;
+
+ void Churn()
+ {
+ try
+ {
+ for (var cycle = 0; cycle < cyclesPerThread; cycle++)
+ epoch.ProtectAndDrain();
+ }
+ finally
+ {
+ epoch.Suspend();
+ }
+
+ _ = Interlocked.Add(ref cycles, cyclesPerThread);
+ }
+
+ var sampler = new Thread(() =>
+ {
+ while (Volatile.Read(ref running))
+ {
+ RecordMax(ref maxOutstandingSignals, epoch.TestHookOutstandingWaiterSignals);
+ RecordMax(ref maxPendingSignals, epoch.TestHookPendingWaiterSignals);
+ RecordMax(ref maxWaiters, epoch.TestHookWaiterCount);
+ Thread.Yield();
+ }
+ })
+ { IsBackground = true, Name = $"{nameof(RunOversubscribed)}-sampler" };
+
+ var workers = new Thread[threadCount];
+
+ // Holders take every slot and park, so the table is provably full before anyone else arrives.
+ for (var i = 0; i < slotCount; i++)
+ {
+ workers[i] = new Thread(() =>
+ {
+ epoch.Resume();
+ slotsFilled.Signal();
+ go.Wait();
+ Churn();
+ })
+ { IsBackground = true, Name = $"{nameof(RunOversubscribed)}-holder-{i}" };
+ }
+
+ // Contenders find no slot and block in ReserveEntryWait until a holder releases one.
+ for (var i = slotCount; i < threadCount; i++)
+ {
+ workers[i] = new Thread(() =>
+ {
+ epoch.Resume();
+ Churn();
+ })
+ { IsBackground = true, Name = $"{nameof(RunOversubscribed)}-contender-{i}" };
+ }
+
+ try
+ {
+ sampler.Start();
+
+ for (var i = 0; i < slotCount; i++)
+ workers[i].Start();
+ Assert.That(slotsFilled.Wait(JoinTimeout), Is.True, "the holders never filled the epoch table");
+
+ for (var i = slotCount; i < threadCount; i++)
+ workers[i].Start();
+ WaitForWaiters(contenderCount);
+
+ go.Set();
+
+ foreach (var worker in workers)
+ Assert.That(worker.Join(JoinTimeout), Is.True, $"{worker.Name} never finished; a waiter was likely stranded");
+ }
+ finally
+ {
+ // Unblock the holders even if an assertion above fired, so no thread is left parked.
+ go.Set();
+ Volatile.Write(ref running, false);
+ _ = sampler.Join(JoinTimeout);
+
+ slotsFilled.Dispose();
+ go.Dispose();
+ }
+
+ return new RunResult
+ {
+ Cycles = Volatile.Read(ref cycles),
+ MaxOutstandingSignals = Volatile.Read(ref maxOutstandingSignals),
+ MaxPendingSignals = Volatile.Read(ref maxPendingSignals),
+ MaxWaiters = Volatile.Read(ref maxWaiters)
+ };
+ }
+
+ /// Block until threads are parked waiting for a slot.
+ void WaitForWaiters(int expected)
+ {
+ var deadline = DateTime.UtcNow + JoinTimeout;
+ while (epoch.TestHookWaiterCount < expected)
+ {
+ Assert.That(DateTime.UtcNow, Is.LessThan(deadline),
+ $"only {epoch.TestHookWaiterCount} of {expected} threads parked waiting for a slot");
+ Thread.Yield();
+ }
+ }
+
+ static void RecordMax(ref int target, int observed)
+ {
+ var max = Volatile.Read(ref target);
+ while (observed > max)
+ {
+ var prev = Interlocked.CompareExchange(ref target, observed, max);
+ if (prev == max)
+ return;
+ max = prev;
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/website/docs/dev/tsavorite/epochprotection.md b/website/docs/dev/tsavorite/epochprotection.md
index 4e81519e29d..37bee47f82f 100644
--- a/website/docs/dev/tsavorite/epochprotection.md
+++ b/website/docs/dev/tsavorite/epochprotection.md
@@ -29,7 +29,21 @@ We need to ensure shared variables are not being read and mutated simultaneously
- We can add trigger actions that are executed when all threads are past a safe epoch ``` (For every thread T: SafeEpoch <= thread local Epoch <= Global Epoch) ```. Since the system holds all thread local epochs in a system accesible epoch table we can scan and find a safe epoch.
This gives us the ability to have an exactly-once invoked function that depends on all threads logically coordinating and not having any code being executed at the time of Epoch ending for the trigger.
-- If you look closely, what `epoch.Resume()` essentially does is have a thread find a free entry in the epoch table, put its id there, and "claim" the current epoch (the next thread will increment the epoch by 1 and claim the next epoch). Inside `epoch.Resume()`, there is a loop; if the current target epoch entry is already occupied by another thread, the current thread will yield. When it wakes up, it will try to occupy the next entry. If the epoch table is full, the rest of the threads will keep yielding.
+- If you look closely, what `epoch.Resume()` essentially does is have a thread find a free entry in the epoch table, put its id there, and "claim" the current epoch (the next thread will increment the epoch by 1 and claim the next epoch). Inside `epoch.Resume()`, the thread probes two hashed starting offsets and then circles the whole table twice looking for a free entry. If every entry is taken, the thread blocks rather than spinning; see *Waiting for an epoch table entry* below.
+
+### Waiting for an epoch table entry
+
+The table has a fixed number of slots, so more threads can want protection than there are entries. A thread that cannot claim one blocks on a semaphore until another thread frees a slot.
+
+- A thread that fails to claim a slot increments `waiterCount`, re-probes the table, and only then blocks. The re-probe is what closes the race against a releasing thread that read `waiterCount` as zero a moment earlier and therefore did not signal; without it, that thread could block on a slot that is already free.
+
+- `Release` (reached through `Suspend` and `SuspendResume`) frees the slot and then wakes one waiter. It publishes the slot as free *before* deciding whether to signal, so a waiter that wakes always sees the freed slot.
+
+- The number of signals issued but not yet consumed is bounded by the number of waiters, tracked in `pendingWaiterSignals`. The bound matters because `ProtectAndDrain` hops through `SuspendResume` on *every* refresh while any waiter exists, to give waiters a fair chance at a slot. Releases therefore vastly outnumber waiters under contention, and signalling on each one would grow the semaphore's count without bound until it overflowed. Suppressing a signal cannot strand a waiter: a signal is only withheld when every current waiter already has a wake pending, and a woken waiter re-probes the entire table before blocking again.
+
+- `Dispose` cancels the shared `CancellationTokenSource` so parked threads unwind as `ObjectDisposedException`, and sets the MSB of `waiterCount` to keep new waiters out.
+
+- Oversubscribing the table degrades into waiting, not failure. Sustained concurrency well above `N` shows up as added latency on `Resume`, so `N = max(128, ProcessorCount * 2)` is a throughput consideration rather than a hard limit on the number of threads that may use a `LightEpoch`.
## Relevant Public Methods and How to use them
From d873c5f09535349f10a664299e5a6ab2e9a15268 Mon Sep 17 00:00:00 2001
From: TedHartMS <15467143+TedHartMS@users.noreply.github.com>
Date: Tue, 15 Sep 2026 18:49:45 -0700
Subject: [PATCH 2/4] Make the compaction background task survive a failed
cycle
CompactionTaskAsync wrapped its whole loop in a single catch, so any exception
from a compaction cycle ended the task for the lifetime of the process. Guard
the per-cycle work instead, so a failed cycle is logged and retried on the next
interval, matching the object collect and expired key deletion tasks.
Cancellation is now caught as OperationCanceledException and suppressed rather
than logged as an error, so shutdown no longer reports a spurious failure.
---
libs/server/StoreWrapper.cs | 27 +++++++++++++++++++--------
1 file changed, 19 insertions(+), 8 deletions(-)
diff --git a/libs/server/StoreWrapper.cs b/libs/server/StoreWrapper.cs
index a3b5aa5dac5..164b3ebf6bd 100644
--- a/libs/server/StoreWrapper.cs
+++ b/libs/server/StoreWrapper.cs
@@ -699,20 +699,31 @@ async Task CompactionTaskAsync(int compactionFrequencySecs, CancellationToken to
Debug.Assert(compactionFrequencySecs > 0);
try
{
- while (true)
+ while (!token.IsCancellationRequested)
{
- if (token.IsCancellationRequested) return;
-
- await databaseManager.DoCompactionAsync(token, logger).ConfigureAwait(false);
+ // A failed cycle must not take the task down with it; retry on the next interval so a
+ // transient fault cannot disable compaction for the lifetime of the process.
+ try
+ {
+ 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");
+ 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");
+ }
+ catch (Exception ex) when (!token.IsCancellationRequested)
+ {
+ logger?.LogError(ex, "CompactionTask exception received. Retrying on the next cycle.");
+ }
await Task.Delay(compactionFrequencySecs * 1000, token).ConfigureAwait(false);
}
}
+ catch (OperationCanceledException) when (token.IsCancellationRequested)
+ {
+ // Suppress the exception if the task was cancelled because of store wrapper disposal
+ }
catch (Exception ex)
{
logger?.LogError(ex, "CompactionTask exception received.");
From 254b5edec781b740bd51d66c3cd9f23fecb44602 Mon Sep 17 00:00:00 2001
From: TedHartMS <15467143+TedHartMS@users.noreply.github.com>
Date: Tue, 15 Sep 2026 19:04:47 -0700
Subject: [PATCH 3/4] Address PR review: precise signal bound, tested
maintenance loop
Correct the documented bound on pendingWaiterSignals. A waiter can claim the
freed slot on its re-probe and decrement waiterCount while a reservation taken
for it is still in flight, so the cap is against the peak number of concurrent
waiters, not the instantaneous count. A leftover signal is consumed by the next
waiter, which decrements and re-probes, so leftovers drain rather than
accumulate. Any such bound prevents the overflow; the stronger invariant the
comments and docs asserted does not hold.
Extract the shared retry loop used by the compaction, object collect and
expired key deletion tasks into RunMaintenanceLoopAsync. The three loops were
identical after the resilience change, and a single seam makes the contract
testable so it cannot silently regress to a terminal catch.
Add MaintenanceLoopTests covering that contract with fault injection: a failed
cycle is logged as an error and the loop runs again, repeated failures keep
retrying without a terminal log, cancellation ends the loop without faulting or
logging, and a cycle cancelled during shutdown is not logged as a failure.
---
libs/server/StoreWrapper.cs | 108 +++++-------
.../cs/src/core/Epochs/LightEpoch.cs | 22 ++-
.../Garnet.test/MaintenanceLoopTests.cs | 163 ++++++++++++++++++
website/docs/dev/tsavorite/epochprotection.md | 4 +-
4 files changed, 224 insertions(+), 73 deletions(-)
create mode 100644 test/standalone/Garnet.test/MaintenanceLoopTests.cs
diff --git a/libs/server/StoreWrapper.cs b/libs/server/StoreWrapper.cs
index 164b3ebf6bd..3b333a12516 100644
--- a/libs/server/StoreWrapper.cs
+++ b/libs/server/StoreWrapper.cs
@@ -694,30 +694,34 @@ async Task CommitTaskAsync(int commitFrequencyMs, CancellationToken token = defa
}
}
- async Task CompactionTaskAsync(int compactionFrequencySecs, CancellationToken token = default)
+ ///
+ /// Run every until 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.
+ ///
+ /// The work to perform on each cycle.
+ /// Delay between cycles.
+ /// Task name used in the failure log message.
+ /// Logger for failed cycles, may be null.
+ /// Cancellation token that ends the loop.
+ internal static async Task RunMaintenanceLoopAsync(Func runCycle,
+ TimeSpan interval, string taskDescription, ILogger logger, CancellationToken token)
{
- Debug.Assert(compactionFrequencySecs > 0);
try
{
while (!token.IsCancellationRequested)
{
- // A failed cycle must not take the task down with it; retry on the next interval so a
- // transient fault cannot disable compaction for the lifetime of the process.
try
{
- 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");
+ await runCycle(token).ConfigureAwait(false);
}
catch (Exception ex) when (!token.IsCancellationRequested)
{
- logger?.LogError(ex, "CompactionTask exception received. Retrying on the next cycle.");
+ 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)
@@ -726,77 +730,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(compactionFrequencySecs > 0);
+
+ return RunMaintenanceLoopAsync(async (cancellationToken) =>
+ {
+ await databaseManager.DoCompactionAsync(cancellationToken, 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");
+ }, TimeSpan.FromSeconds(compactionFrequencySecs), "background compaction task", logger, token);
+ }
+
+ Task ObjectCollectTaskAsync(int objectCollectFrequencySecs, CancellationToken token = default)
{
Debug.Assert(objectCollectFrequencySecs > 0);
if (serverOptions.DisableObjects)
{
logger?.LogWarning("ExpiredObjectCollectionFrequencySecs option is configured but Object store is disabled. Stopping the background hash collect task.");
- return;
+ return Task.CompletedTask;
}
- try
+ return RunMaintenanceLoopAsync((_) =>
{
- while (!token.IsCancellationRequested)
- {
- // A failed cycle must not take the task down with it; retry on the next interval so a
- // transient fault cannot disable object collection for the lifetime of the process.
- try
- {
- databaseManager.ExecuteObjectCollection();
- }
- catch (Exception ex) when (!token.IsCancellationRequested)
- {
- logger?.LogError(ex, "Unknown exception received for background hash collect task. Retrying on the next cycle.");
- }
-
- await Task.Delay(TimeSpan.FromSeconds(objectCollectFrequencySecs), token).ConfigureAwait(false);
- }
- }
- catch (OperationCanceledException) when (token.IsCancellationRequested)
- {
- // Suppress the exception if the task was cancelled because of store wrapper disposal
- }
- catch (Exception ex)
- {
- 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 (!token.IsCancellationRequested)
- {
- // A failed cycle must not take the task down with it; retry on the next interval so a
- // transient fault cannot disable expired key deletion for the lifetime of the process.
- try
- {
- databaseManager.ExpiredKeyDeletionScan();
- }
- catch (Exception ex) when (!token.IsCancellationRequested)
- {
- logger?.LogError(ex, "Unknown exception received for background expired key deletion scan task. Retrying on the next cycle.");
- }
- await Task.Delay(TimeSpan.FromSeconds(expiredKeyDeletionScanFrequencySecs), token).ConfigureAwait(false);
- }
- }
- catch (OperationCanceledException) when (token.IsCancellationRequested)
- {
- // Suppress the exception if the task was cancelled because of store wrapper disposal
- }
- catch (Exception ex)
+ return RunMaintenanceLoopAsync((_) =>
{
- 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);
}
///
diff --git a/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs b/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs
index 8de8d1c4020..ac3f1c6ab9c 100644
--- a/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs
+++ b/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs
@@ -127,11 +127,15 @@ private class Metadata
volatile int waiterCount = 0;
///
- /// Number of signals that have been issued but not yet consumed by a
- /// waiter. only issues a signal while this is below ,
- /// which bounds the semaphore's count by the number of waiters. Without that bound the count grows by
- /// one per epoch release for as long as any waiter exists, and eventually overflows
- /// 's maximum.
+ /// Reservations taken by for signals that
+ /// have been issued but not yet consumed by a waiter. A reservation is only taken while this is below
+ /// , 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 's maximum.
///
volatile int pendingWaiterSignals = 0;
@@ -568,8 +572,12 @@ void Release()
///
/// The increment is what reserves the right to signal, so it always precedes its
/// and the matching decrement always follows a successful wait.
- /// The semaphore's count is therefore never above , which this loop
- /// holds at or below .
+ /// The semaphore's count is therefore never above , and a
+ /// reservation is only taken while that is below , 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 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
diff --git a/test/standalone/Garnet.test/MaintenanceLoopTests.cs b/test/standalone/Garnet.test/MaintenanceLoopTests.cs
new file mode 100644
index 00000000000..ead59906cbf
--- /dev/null
+++ b/test/standalone/Garnet.test/MaintenanceLoopTests.cs
@@ -0,0 +1,163 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Garnet.server;
+using Microsoft.Extensions.Logging;
+using NUnit.Framework;
+
+namespace Garnet.test
+{
+ ///
+ /// The retry contract shared by the background maintenance loops (compaction, object collect and
+ /// expired key deletion). A cycle that throws must be logged and retried on the next interval rather
+ /// than ending the task, because a terminated task stays dead for the lifetime of the process.
+ ///
+ [TestFixture]
+ public class MaintenanceLoopTests : TestBase
+ {
+ /// Longest any of these loops should need; exceeded only if the loop stopped running.
+ static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(30);
+
+ static readonly TimeSpan Interval = TimeSpan.FromMilliseconds(10);
+
+ [Test]
+ public async Task AFailedCycleIsLoggedAndTheLoopKeepsRunning()
+ {
+ using var cts = new CancellationTokenSource();
+ var logger = new RecordingLogger();
+ var cycles = 0;
+ var reachedSecondCycle = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ var loop = StoreWrapper.RunMaintenanceLoopAsync((_) =>
+ {
+ // Fail the first cycle only; the loop must come back for a second.
+ if (Interlocked.Increment(ref cycles) == 1)
+ throw new InvalidOperationException("injected first-cycle failure");
+
+ reachedSecondCycle.TrySetResult();
+ return ValueTask.CompletedTask;
+ }, Interval, "test task", logger, cts.Token);
+
+ Assert.That(await Task.WhenAny(reachedSecondCycle.Task, Task.Delay(WaitTimeout)), Is.SameAs(reachedSecondCycle.Task),
+ "the loop did not run another cycle after one threw");
+
+ await cts.CancelAsync();
+ await loop;
+
+ Assert.Multiple(() =>
+ {
+ Assert.That(logger.Entries, Has.Exactly(1).Matches(e =>
+ e.Level == LogLevel.Error && e.Exception is InvalidOperationException),
+ "the failed cycle should be logged exactly once, as an error");
+ Assert.That(logger.Entries, Has.None.Matches(e => e.Level == LogLevel.Critical),
+ "a recoverable cycle failure must not be logged as terminal");
+ });
+ }
+
+ [Test]
+ public async Task TheLoopKeepsRunningWhenEveryCycleFails()
+ {
+ using var cts = new CancellationTokenSource();
+ var logger = new RecordingLogger();
+ var cycles = 0;
+ var reachedThirdCycle = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ var loop = StoreWrapper.RunMaintenanceLoopAsync((_) =>
+ {
+ if (Interlocked.Increment(ref cycles) >= 3)
+ reachedThirdCycle.TrySetResult();
+ throw new InvalidOperationException("injected failure");
+ }, Interval, "test task", logger, cts.Token);
+
+ Assert.That(await Task.WhenAny(reachedThirdCycle.Task, Task.Delay(WaitTimeout)), Is.SameAs(reachedThirdCycle.Task),
+ "repeated failures ended the loop instead of being retried");
+
+ await cts.CancelAsync();
+ await loop;
+
+ Assert.That(logger.Entries, Has.None.Matches(e => e.Level == LogLevel.Critical),
+ "repeated recoverable failures must not be logged as terminal");
+ }
+
+ [Test]
+ public async Task CancellationEndsTheLoopWithoutFaultingOrLogging()
+ {
+ using var cts = new CancellationTokenSource();
+ var logger = new RecordingLogger();
+ var ranOnce = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ var loop = StoreWrapper.RunMaintenanceLoopAsync((_) =>
+ {
+ ranOnce.TrySetResult();
+ return ValueTask.CompletedTask;
+ }, Interval, "test task", logger, cts.Token);
+
+ Assert.That(await Task.WhenAny(ranOnce.Task, Task.Delay(WaitTimeout)), Is.SameAs(ranOnce.Task),
+ "the loop never ran a cycle");
+
+ await cts.CancelAsync();
+
+ Assert.That(await Task.WhenAny(loop, Task.Delay(WaitTimeout)), Is.SameAs(loop), "cancellation did not end the loop");
+ await loop;
+
+ Assert.Multiple(() =>
+ {
+ Assert.That(loop.IsCompletedSuccessfully, Is.True, "cancellation must not fault the task");
+ Assert.That(logger.Entries, Is.Empty, "cancellation is not a failure and must not be logged");
+ });
+ }
+
+ ///
+ /// A cycle that observes cancellation and throws for that reason is shutdown, not a failure, so it
+ /// must not be logged as a retryable error.
+ ///
+ [Test]
+ public async Task ACycleCancelledDuringShutdownIsNotLoggedAsAFailure()
+ {
+ using var cts = new CancellationTokenSource();
+ var logger = new RecordingLogger();
+
+ var loop = StoreWrapper.RunMaintenanceLoopAsync(async (cancellationToken) =>
+ {
+ await cts.CancelAsync();
+ cancellationToken.ThrowIfCancellationRequested();
+ }, Interval, "test task", logger, cts.Token);
+
+ Assert.That(await Task.WhenAny(loop, Task.Delay(WaitTimeout)), Is.SameAs(loop), "the loop did not end after cancellation");
+ await loop;
+
+ Assert.That(logger.Entries, Is.Empty, "a cycle cancelled by shutdown must not be logged as a failure");
+ }
+
+ readonly record struct LogEntry(LogLevel Level, Exception Exception);
+
+ sealed class RecordingLogger : ILogger
+ {
+ readonly List entries = [];
+
+ internal LogEntry[] Entries
+ {
+ get
+ {
+ lock (entries)
+ return [.. entries];
+ }
+ }
+
+ public IDisposable BeginScope(TState state) where TState : notnull => null;
+
+ public bool IsEnabled(LogLevel logLevel) => true;
+
+ public void Log(LogLevel logLevel, EventId eventId, TState state, Exception exception,
+ Func formatter)
+ {
+ lock (entries)
+ entries.Add(new LogEntry(logLevel, exception));
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/website/docs/dev/tsavorite/epochprotection.md b/website/docs/dev/tsavorite/epochprotection.md
index 37bee47f82f..50e63743fed 100644
--- a/website/docs/dev/tsavorite/epochprotection.md
+++ b/website/docs/dev/tsavorite/epochprotection.md
@@ -39,7 +39,9 @@ The table has a fixed number of slots, so more threads can want protection than
- `Release` (reached through `Suspend` and `SuspendResume`) frees the slot and then wakes one waiter. It publishes the slot as free *before* deciding whether to signal, so a waiter that wakes always sees the freed slot.
-- The number of signals issued but not yet consumed is bounded by the number of waiters, tracked in `pendingWaiterSignals`. The bound matters because `ProtectAndDrain` hops through `SuspendResume` on *every* refresh while any waiter exists, to give waiters a fair chance at a slot. Releases therefore vastly outnumber waiters under contention, and signalling on each one would grow the semaphore's count without bound until it overflowed. Suppressing a signal cannot strand a waiter: a signal is only withheld when every current waiter already has a wake pending, and a woken waiter re-probes the entire table before blocking again.
+- The number of signals issued but not yet consumed is tracked in `pendingWaiterSignals`, and a signal is only issued while that is below the current waiter count. This caps outstanding signals at the *peak* number of concurrent waiters rather than the instantaneous count: a waiter can claim the freed slot and leave while a signal reserved for it is still in flight, briefly leaving a signal outstanding with no waiter present. Such a leftover is consumed by the next waiter, which re-probes the table after waking, so leftovers drain instead of accumulating. Any such bound is what matters — the count is `O(threads)` rather than unbounded. The bound is needed because `ProtectAndDrain` hops through `SuspendResume` on *every* refresh while any waiter exists, to give waiters a fair chance at a slot. Releases therefore vastly outnumber waiters under contention, and signalling on each one would grow the semaphore's count until it overflowed.
+
+- Suppressing a signal cannot strand a waiter: a signal is only withheld when every current waiter already has a wake pending, and a woken waiter re-probes the entire table before blocking again.
- `Dispose` cancels the shared `CancellationTokenSource` so parked threads unwind as `ObjectDisposedException`, and sets the MSB of `waiterCount` to keep new waiters out.
From c7c7d3cc1b99e8aaa776113e57727f9b8bd6eb69 Mon Sep 17 00:00:00 2001
From: TedHartMS <15467143+TedHartMS@users.noreply.github.com>
Date: Wed, 16 Sep 2026 14:31:27 -0700
Subject: [PATCH 4/4] Document why the signal bound uses a counter instead of
CurrentCount
Records that comparing the semaphore's CurrentCount against waiterCount is a read-then-act with no atomicity against the Release that follows, and that the resulting surplus is never reclaimed, so the reservation counter is the bound that holds.
---
libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs | 9 +++++++++
website/docs/dev/tsavorite/epochprotection.md | 2 ++
2 files changed, 11 insertions(+)
diff --git a/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs b/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs
index ac3f1c6ab9c..05ab0bce7c8 100644
--- a/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs
+++ b/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs
@@ -136,6 +136,15 @@ private class Metadata
/// 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 's maximum.
+ ///
+ /// The counter is explicit rather than a comparison of against
+ /// at signal time. Such a comparison is not atomic with the
+ /// 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.
///
volatile int pendingWaiterSignals = 0;
diff --git a/website/docs/dev/tsavorite/epochprotection.md b/website/docs/dev/tsavorite/epochprotection.md
index 50e63743fed..1b945a1142a 100644
--- a/website/docs/dev/tsavorite/epochprotection.md
+++ b/website/docs/dev/tsavorite/epochprotection.md
@@ -43,6 +43,8 @@ The table has a fixed number of slots, so more threads can want protection than
- Suppressing a signal cannot strand a waiter: a signal is only withheld when every current waiter already has a wake pending, and a woken waiter re-probes the entire table before blocking again.
+- The counter is explicit rather than a check of the semaphore's `CurrentCount` against `waiterCount` at signal time. That check is a read-then-act: it is not atomic with the `Release` that follows it, so concurrent releasers all observe the same below-threshold count and all signal — the common case under exactly the contention that makes the bound matter. The surplus then persists, because `CurrentCount` only falls when a waiter consumes a signal, so it carries from one contention burst into the next. Reserving by CAS makes the increment itself the permission to signal, which is a hard bound, and returning the reservation on consumption keeps residue from accumulating across bursts.
+
- `Dispose` cancels the shared `CancellationTokenSource` so parked threads unwind as `ObjectDisposedException`, and sets the MSB of `waiterCount` to keep new waiters out.
- Oversubscribing the table degrades into waiting, not failure. Sustained concurrency well above `N` shows up as added latency on `Resume`, so `N = max(128, ProcessorCount * 2)` is a throughput consideration rather than a hard limit on the number of threads that may use a `LightEpoch`.