diff --git a/src/GameCult.Caching/Contracts/cultcache-store-composition.md b/src/GameCult.Caching/Contracts/cultcache-store-composition.md index 5287455e..83aaa61b 100644 --- a/src/GameCult.Caching/Contracts/cultcache-store-composition.md +++ b/src/GameCult.Caching/Contracts/cultcache-store-composition.md @@ -49,6 +49,16 @@ which route wrote it. published after the cache releases its gate and before the call that made it returns, on that caller's thread; no call publishes another call's changes. A direct `PullAll` on an attached store is such a call. +- Changes are delivered in save order. Across threads, a cache delivers changes + in the order it admitted them, so a later change to a key never reaches an + observer before an earlier one. A call whose changes were admitted after + another call's waits, outside the gate, until that delivery finishes. +- A write made by an observer during delivery is not delivered before that + write returns. It is queued behind the delivery in progress, in save order, + and delivered on the same thread before the outermost call returns; its + `OnUpdate` exceptions are rethrown to that outermost caller. +- An observer must not block waiting on another thread's write to a cache: + that thread's delivery may be waiting for this one to finish. - An `OnUpdate` handler exception is rethrown to that caller after all of that call's changes are delivered (an `AggregateException` if several threw). If the call itself failed, its own exception is rethrown and handler exceptions diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index b3f01f6c..ddc5c7a1 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -1180,8 +1180,13 @@ namespace GameCult.Caching private readonly ConditionalWeakTable _handles = new(); private readonly Subject _changes = new(); private readonly object _gate = new(); + private readonly object _deliveryGate = new(); + private long _admitted; + private long _delivered; // The changes admitted by this thread's outermost hold, published by that hold when it exits. - [ThreadStatic] private static List<(CultCache Cache, Change Change, bool Loaded)>? _held; + [ThreadStatic] private static List<(Change Change, bool Loaded)>? _held; + // Non-null while this thread delivers: tickets it has yet to deliver, in the order its holds closed. + [ThreadStatic] private static List<(CultCache Cache, long Ticket, List<(Change Change, bool Loaded)> Changes)>? _delivering; private bool _dirtyInMemory; public CultCache(CultDocumentRegistry? registry = null) @@ -1574,14 +1579,15 @@ namespace GameCult.Caching if (outcome != CultCommitOutcome.Committed) return outcome; foreach (var change in Apply(admitted, evicted, source)) - _held!.Add((this, change, source != null)); + _held.Add((change, source != null)); return CultCommitOutcome.Committed; }); - // Every admission runs in a hold. A nested hold adds to the outermost one, which publishes exactly its own changes - // after it leaves the gate, before it returns, on its caller's thread: observers never run under the gate. - // Every change is delivered even if an observer throws; then the first observer exception (an AggregateException - // for several) is rethrown. If the body itself threw, its exception wins and observer exceptions are dropped. + // Every admission runs in a hold. A nested hold on the same cache adds to the outermost one, which takes a ticket + // under the gate and, after leaving it, publishes exactly its own changes in ticket order, on its caller's thread: + // observers never run under the gate and see changes in save order. A hold closed while its thread is delivering + // queues its ticket behind that delivery. Every change is delivered even if an OnUpdate handler throws; then the + // first handler exception (an AggregateException for several) is rethrown. If the body threw, its exception wins. internal T Held(Func body) { if (Monitor.IsEntered(_gate)) @@ -1589,23 +1595,28 @@ namespace GameCult.Caching if (_held != null) throw new InvalidOperationException( "A cache hold was entered inside another cache's hold; a thread holds one cache's gate at a time."); - var outer = _held; - var mine = _held = new List<(CultCache Cache, Change Change, bool Loaded)>(); - T result; - try + var mine = _held = new List<(Change Change, bool Loaded)>(); + var ticket = -1L; + T result = default!; + ExceptionDispatchInfo? thrown = null; + lock (_gate) { - lock (_gate) + try + { result = body(); - } - catch - { - _held = outer; - Deliver(mine); - throw; + } + catch (Exception exception) + { + thrown = ExceptionDispatchInfo.Capture(exception); + } + + _held = null; + if (mine.Count > 0) + ticket = _admitted++; } - _held = outer; - var failures = Deliver(mine); + var failures = ticket < 0 ? new List() : Publish(ticket, mine); + thrown?.Throw(); if (failures.Count == 1) ExceptionDispatchInfo.Capture(failures[0]).Throw(); if (failures.Count > 1) @@ -1613,28 +1624,80 @@ namespace GameCult.Caching return result; } - private static List Deliver(List<(CultCache Cache, Change Change, bool Loaded)> changes) + private List Publish(long ticket, List<(Change Change, bool Loaded)> changes) { var failures = new List(); - foreach (var (cache, change, loaded) in changes) + if (_delivering != null) { - // R3 routes a throwing subscriber to its unhandled-exception handler; OnNext does not throw. - cache._changes.OnNext(change); - if (!loaded) - continue; - try + _delivering.Add((this, ticket, changes)); + return failures; + } + + var queue = _delivering = new List<(CultCache Cache, long Ticket, List<(Change Change, bool Loaded)> Changes)> + { + (this, ticket, changes) + }; + try + { + // A queued ticket that is never delivered would stall every later writer of its cache, so none is skipped. + for (var i = 0; i < queue.Count; i++) { - cache.OnUpdate?.Invoke(change.Previous, change.Document); - } - catch (Exception exception) - { - failures.Add(exception); + try + { + queue[i].Cache.Deliver(queue[i].Ticket, queue[i].Changes, failures); + } + catch (Exception exception) + { + failures.Add(exception); + } } } + finally + { + _delivering = null; + } return failures; } + // A thread waits for its turn holding no gate and no turn, and holds a turn only while delivering, which never + // waits: every wait is for a lower ticket, so deliveries cannot deadlock one another. + private void Deliver(long ticket, List<(Change Change, bool Loaded)> changes, List failures) + { + lock (_deliveryGate) + { + while (_delivered != ticket) + Monitor.Wait(_deliveryGate); + } + + try + { + foreach (var (change, loaded) in changes) + { + // R3 routes a throwing subscriber to its unhandled-exception handler; OnNext does not throw. + _changes.OnNext(change); + if (!loaded) + continue; + try + { + OnUpdate?.Invoke(change.Previous, change.Document); + } + catch (Exception exception) + { + failures.Add(exception); + } + } + } + finally + { + lock (_deliveryGate) + { + _delivered++; + Monitor.PulseAll(_deliveryGate); + } + } + } + private CacheBackingStore? Validate( IReadOnlyList admitted, IReadOnlyList evicted, diff --git a/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs b/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs index 373f9df0..0e668450 100644 --- a/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs +++ b/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs @@ -757,6 +757,67 @@ namespace GameCult.Caching.Tests batch.Upsert(new Counter { Name = "counter", Value = value }, new CultRecordHandle(Key)); }); + // Delivered in save order: each change replaces the document the previously delivered change installed. + [Test] + public void ConcurrentWritersDeliverInSaveOrder() + { + using var cache = new CultCache(Registry); + var key = new CultRecordKey("ordered"); + var observed = new System.Collections.Generic.List<(object? Document, object? Previous)>(); + using var subscription = cache.Watch().Subscribe(change => + { + lock (observed) + observed.Add((change.Document, change.PreviousDocument)); + }); + + var writers = Enumerable.Range(0, 8).Select(thread => Task.Run(() => + { + for (var i = 0; i < 2000; i++) + cache.UpsertAsync(new Counter { Name = "ordered", Value = i }, new CultRecordHandle(key)); + })).ToArray(); + + Assert.That(Task.WaitAll(writers, TimeSpan.FromSeconds(60)), Is.True); + Assert.That(observed, Has.Count.EqualTo(16000)); + Assert.That(FirstOutOfOrder(observed), Is.EqualTo(-1), "a change was delivered out of save order"); + } + + // The cache is not disposed on failure because a deadlocked delivery would hang the test thread too. + [Test] + public void ObserverWritingTheSameCacheDuringConcurrentDeliveryDoesNotDeadlock() + { + var cache = new CultCache(Registry); + var tally = new CultRecordKey("tally"); + var observed = new System.Collections.Generic.List<(object? Document, object? Previous)>(); + using var counters = cache.Watch().Subscribe(_ => + cache.UpsertAsync(new Tally { Name = "tally" }, new CultRecordHandle(tally))); + using var tallies = cache.Watch().Subscribe(change => + { + lock (observed) + observed.Add((change.Document, change.PreviousDocument)); + }); + + var writers = Enumerable.Range(0, 8).Select(thread => Task.Run(() => + { + for (var i = 0; i < 500; i++) + cache.UpsertAsync(new Counter { Name = $"c:{thread}", Value = i }, new CultRecordHandle(new CultRecordKey($"c:{thread}"))); + })).ToArray(); + + Assert.That(Task.WaitAll(writers, TimeSpan.FromSeconds(60)), Is.True, "an observer writing the cache during delivery deadlocked"); + Assert.That(observed, Has.Count.EqualTo(4000)); + Assert.That(FirstOutOfOrder(observed), Is.EqualTo(-1), "an observer's write was delivered out of save order"); + cache.Dispose(); + } + + private static int FirstOutOfOrder(System.Collections.Generic.List<(object? Document, object? Previous)> observed) + { + for (var i = 1; i < observed.Count; i++) + { + if (!ReferenceEquals(observed[i].Previous, observed[i - 1].Document)) + return i; + } + return -1; + } + private string PathOf(string fileName) => System.IO.Path.Combine(_directory, fileName); private static readonly CultDocumentRegistry Registry = CultDocumentRegistry.ForTypes(new[] { typeof(Counter), typeof(Tally), typeof(Slow) });