GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Deliver each cache's changes in save order

Browse source 
The outermost hold takes a ticket under the gate and, after releasing it,
waits on a per-cache delivery gate for its turn, delivers, then advances
the turn. Delivery order across threads equals admission order. A hold
closed while its thread is delivering queues its ticket behind that
delivery on the same thread instead of waiting, so an observer writing
the cache cannot deadlock against a writer: a waiting thread holds no gate
and no turn, and a turn holder never waits.

Hand-over-hand from the gate to a delivery lock was rejected: a writer
blocked on the delivery lock while holding the gate deadlocks with an
observer that writes the same cache.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-13 20:09:23 +02:00
commit bcae4839d0
3 changed files with 165 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

10
src/GameCult.Caching/Contracts/cultcache-store-composition.md
View file

@ -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

125
src/GameCult.Caching/CultCache.cs
View file

@ -1180,8 +1180,13 @@ namespace GameCult.Caching
private readonly ConditionalWeakTable<object, KeyBox> _handles = new();
private readonly Subject<Change> _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<T>(Func<T> 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<Exception>() : 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<Exception> Deliver(List<(CultCache Cache, Change Change, bool Loaded)> changes)
private List<Exception> Publish(long ticket, List<(Change Change, bool Loaded)> changes)
{
var failures = new List<Exception>();
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<Exception> 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<CultStoredDocument> admitted,
IReadOnlyList<CultStoredDocument> evicted,

61
tests/GameCult.Caching.Tests/ConditionalCommitTests.cs
View file

@ -757,6 +757,67 @@ namespace GameCult.Caching.Tests
batch.Upsert(new Counter { Name = "counter", Value = value }, new CultRecordHandle<Counter>(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<Counter>().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<Counter>(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<Counter>().Subscribe(_ =>
cache.UpsertAsync(new Tally { Name = "tally" }, new CultRecordHandle<Tally>(tally)));
using var tallies = cache.Watch<Tally>().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<Counter>(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) });