diff --git a/libs/server/StoreWrapper.cs b/libs/server/StoreWrapper.cs index f85d63a0859..82bb28e2ab8 100644 --- a/libs/server/StoreWrapper.cs +++ b/libs/server/StoreWrapper.cs @@ -714,23 +714,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 (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); } /// 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..05ab0bce7c8 100644 --- a/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs +++ b/libs/storage/Tsavorite/cs/src/core/Epochs/LightEpoch.cs @@ -126,6 +126,28 @@ private class Metadata /// volatile int waiterCount = 0; + /// + /// 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. + /// + /// 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; + /// /// Flag (MSB) used to mark the epoch as disposed in . /// @@ -550,7 +572,46 @@ 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 , 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 + /// 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 +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) 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/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 4e81519e29d..1b945a1142a 100644 --- a/website/docs/dev/tsavorite/epochprotection.md +++ b/website/docs/dev/tsavorite/epochprotection.md @@ -29,7 +29,25 @@ 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 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. + +- 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`. ## Relevant Public Methods and How to use them