From 27e1506c7b38db6dbc7dc8814e03a037a884d8df Mon Sep 17 00:00:00 2001 From: botoddly <250804054+botoddly@users.noreply.github.com> Date: Thu, 10 Sep 2026 10:06:56 +0200 Subject: [PATCH 1/5] Add Pixely.Observations, an append-only log of value entries that readers drain at their own pace --- AGENTS.md | 1 + Pixely.slnx | 2 + docs/observations.md | 109 +++++++++ packaging/Pixely/Pixely.Package.csproj | 1 + src/Pixely.Observations/IObservationLog.cs | 14 ++ src/Pixely.Observations/IObservationWriter.cs | 9 + src/Pixely.Observations/ObservationCursor.cs | 39 ++++ src/Pixely.Observations/ObservationLog.cs | 162 +++++++++++++ .../Pixely.Observations.csproj | 10 + .../ObservationLogTests.cs | 219 ++++++++++++++++++ .../Pixely.Observations.Tests.csproj | 27 +++ .../PackageIntegrationTests.cs | 1 + 12 files changed, 594 insertions(+) create mode 100644 docs/observations.md create mode 100644 src/Pixely.Observations/IObservationLog.cs create mode 100644 src/Pixely.Observations/IObservationWriter.cs create mode 100644 src/Pixely.Observations/ObservationCursor.cs create mode 100644 src/Pixely.Observations/ObservationLog.cs create mode 100644 src/Pixely.Observations/Pixely.Observations.csproj create mode 100644 tests/Pixely.Observations.Tests/ObservationLogTests.cs create mode 100644 tests/Pixely.Observations.Tests/Pixely.Observations.Tests.csproj diff --git a/AGENTS.md b/AGENTS.md index 45573a35..9d655a88 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -4,6 +4,7 @@ - `docs/class-registration.md` - ServiceCollection/ServiceProvider API: registration overloads, source generator requirements, lifecycle, aliases, multi-registration, parent/child provider callback merging and scoped lifecycles - `docs/events.md` - Pixely.Events EventBus, event handlers, publishing, and DI auto-subscription +- `docs/observations.md` - Pixely.Observations: appending value entries, per-reader cursors, trimming behind the slowest reader, capacity and stall detection, registration - `docs/static-factory-methods.md` - Static Create() method pattern - `docs/componentize.md` - Pixely.Componentize setup and usage - `docs/components.md` - GameWorld, GameObject, GameComponent lifecycle, Services, UpdateSystem diff --git a/Pixely.slnx b/Pixely.slnx index 7b214f78..b6cdb8b1 100644 --- a/Pixely.slnx +++ b/Pixely.slnx @@ -12,6 +12,7 @@ + @@ -27,6 +28,7 @@ + diff --git a/docs/observations.md b/docs/observations.md new file mode 100644 index 00000000..676d6a34 --- /dev/null +++ b/docs/observations.md @@ -0,0 +1,109 @@ +# Observations + +`ObservationLog` in `Pixely.Observations` is an append-only log of value entries that readers drain at their +own pace. Rules append; nothing is called back. A reader holds a cursor, reads when it suits its own point in +the frame, and the log drops an entry once every cursor has passed it. + +Use it for a transition that two reads of game state one frame apart cannot derive: a unit that walked a path +and arrived, one that appeared and was gone again, an action that resolved and was reversed inside the same +frame. State answers what is true now, and a rule can resolve many transitions between two reads, so any +"most recent transition" field is overwritten before a reader looks at it. + +## The entry type + +One log carries one entry type. To carry several kinds of entry in one order, make `TEntry` a tagged value +type; the log never looks inside it. + +```csharp +public readonly record struct UnitMovedEntry(UnitId Unit, TilePoint From, TilePoint To); +public readonly record struct UnitDiedEntry(UnitId Unit); + +public enum ObservationKind { UnitMoved, UnitDied } + +public readonly record struct Observation(ObservationKind Kind, ParticipantId Perceiver, UnitMovedEntry Moved, UnitDiedEntry Died); +``` + +Entries are past-tense records of ids and value types, never a live reference into game state. Nothing +allocates per entry: the log stores `TEntry` in an array, `Append` takes it by `in`, and `TryRead` copies it out. + +## Writing + +Inject `IObservationWriter` where rules append, so a writer cannot read: + +```csharp +internal sealed class MoveMechanic +{ + private readonly IObservationWriter _observations; + + internal MoveMechanic(IObservationWriter observations) => _observations = observations; + + internal void Move(UnitId unit, TilePoint destination) + { + // mutate state, then record what happened + _observations.Append(new Observation(ObservationKind.UnitMoved, perceiver, moved, default)); + } +} +``` + +## Reading + +Inject `IObservationLog`, create a cursor named after the reader, and drain it in the reader's own +update. Dispose the cursor with the reader: + +```csharp +internal sealed class UnitSpritePresenter : IUpdatable, IDisposable +{ + private readonly ObservationCursor _observations; + + internal UnitSpritePresenter(IObservationLog observations) + { + _observations = observations.CreateCursor(nameof(UnitSpritePresenter)); + } + + public void Update() + { + while (_observations.TryRead(out Observation observation)) + { + // skip what another participant perceived, then switch on the tag + } + } + + public void Dispose() => _observations.Dispose(); +} +``` + +A cursor starts positioned after the last appended entry, so a reader created part-way through a run sees +only what is appended from then on. Cursors read independently: entries appended this frame may be drained by +one reader now and by another several frames later. + +Addressing an entry to a subset of readers is the game's business, not the log's. Carry the perceiver in +`TEntry` and have the reader skip what it did not perceive. A reader bound to one participant should wrap its +cursor once rather than repeat the check in every consumer. + +## Bounds and stalls + +The buffer starts small and doubles towards the maximum capacity given to the constructor. It never shrinks, +so it settles at the high-water mark of its bursts and is reclaimed when the log's scope dies. + +Maximum capacity is a stall detector, not a working size. Pick a number far past any legitimate burst, knowing +a slot costs the size of `TEntry`. A reader that stops draining holds the trim point where it stopped; once the +log fills, `Append` throws and names the cursor that stopped and how far behind it is. Dropping the oldest +entry instead would hide exactly that failure. + +```text +Observation log reached its maximum capacity of 4096 entries. Cursor 'UnitSpritePresenter' stopped draining +4096 entries ago. +``` + +## Registration + +The log takes its capacity as a constructor argument, so register the instance and alias the two halves: + +```csharp +services.AddSingleton(new ObservationLog(4096)); +services.AddAlias, ObservationLog>(); +services.AddAlias, ObservationLog>(); +``` + +Register it in the scope it belongs to. A log registered in a stage's child provider dies with that stage, and +nothing carries into the next one. diff --git a/packaging/Pixely/Pixely.Package.csproj b/packaging/Pixely/Pixely.Package.csproj index 8891c044..254399bb 100644 --- a/packaging/Pixely/Pixely.Package.csproj +++ b/packaging/Pixely/Pixely.Package.csproj @@ -48,6 +48,7 @@ + diff --git a/src/Pixely.Observations/IObservationLog.cs b/src/Pixely.Observations/IObservationLog.cs new file mode 100644 index 00000000..a41ed95d --- /dev/null +++ b/src/Pixely.Observations/IObservationLog.cs @@ -0,0 +1,14 @@ +namespace Pixely.Observations; + +/// +/// The read half of an . Inject it where a reader creates its own cursor +/// and drains it on its own cadence. +/// +public interface IObservationLog where TEntry : struct +{ + /// + /// Creates a cursor positioned after the last appended entry, so it sees only what is appended from now on. + /// The name identifies the cursor when a stalled reader fills the log, so give it the reader's own name. + /// + ObservationCursor CreateCursor(string name = "unnamed"); +} diff --git a/src/Pixely.Observations/IObservationWriter.cs b/src/Pixely.Observations/IObservationWriter.cs new file mode 100644 index 00000000..815121ec --- /dev/null +++ b/src/Pixely.Observations/IObservationWriter.cs @@ -0,0 +1,9 @@ +namespace Pixely.Observations; + +/// +/// The write half of an . Inject it where rules append and nothing reads. +/// +public interface IObservationWriter where TEntry : struct +{ + void Append(in TEntry entry); +} diff --git a/src/Pixely.Observations/ObservationCursor.cs b/src/Pixely.Observations/ObservationCursor.cs new file mode 100644 index 00000000..367a6e88 --- /dev/null +++ b/src/Pixely.Observations/ObservationCursor.cs @@ -0,0 +1,39 @@ +namespace Pixely.Observations; + +/// +/// One reader's position in an . Dispose it when the reader goes away, +/// otherwise it holds the log's trim point where it stopped and the log fills. +/// +public sealed class ObservationCursor : IDisposable where TEntry : struct +{ + private readonly ObservationLog _log; + private bool _disposed; + + internal ObservationCursor(ObservationLog log, string name, long nextSequence) + { + _log = log; + Name = name; + NextSequence = nextSequence; + } + + public string Name { get; } + + internal long NextSequence { get; set; } + + public bool TryRead(out TEntry entry) + { + ObjectDisposedException.ThrowIf(_disposed, this); + return _log.TryRead(this, out entry); + } + + public void Dispose() + { + if (_disposed) + { + return; + } + + _disposed = true; + _log.RemoveCursor(this); + } +} diff --git a/src/Pixely.Observations/ObservationLog.cs b/src/Pixely.Observations/ObservationLog.cs new file mode 100644 index 00000000..d300c1ce --- /dev/null +++ b/src/Pixely.Observations/ObservationLog.cs @@ -0,0 +1,162 @@ +using System.Runtime.CompilerServices; + +namespace Pixely.Observations; + +/// +/// An append-only log of value entries that readers drain at their own pace through their own cursors. +/// Appending never calls a reader. An entry is dropped once every cursor has passed it, so the log is bounded +/// by the slowest reader rather than by how long the run lasts. +/// +/// +/// The single entry type of this log. Carry several kinds of entry in one log by making this a tagged value +/// type; the log never looks inside it. +/// +public sealed class ObservationLog : IObservationLog, IObservationWriter where TEntry : struct +{ + private const int InitialCapacity = 16; + + private readonly int _maximumCapacity; + private readonly List> _cursors = new(); + private TEntry[] _entries; + private int _head; + private int _count; + private long _firstSequence; + private long _nextSequence; + + /// + /// How many entries the log may retain before throws. The buffer starts small and grows + /// towards this bound, so it is a stall detector rather than a working size: pick a number far past any + /// legitimate burst, knowing a slot costs the size of . + /// + public ObservationLog(int maximumCapacity) + { + ArgumentOutOfRangeException.ThrowIfLessThan(maximumCapacity, 1); + _maximumCapacity = maximumCapacity; + _entries = new TEntry[Math.Min(InitialCapacity, maximumCapacity)]; + } + + public void Append(in TEntry entry) + { + if (_count == _maximumCapacity) + { + throw new InvalidOperationException(DescribeOverflow()); + } + + EnsureCapacity(_count + 1); + _entries[PhysicalIndex(_count)] = entry; + _count++; + _nextSequence++; + Trim(); + } + + public ObservationCursor CreateCursor(string name = "unnamed") + { + ObservationCursor cursor = new ObservationCursor(this, name, _nextSequence); + _cursors.Add(cursor); + return cursor; + } + + internal bool TryRead(ObservationCursor cursor, out TEntry entry) + { + int offset = checked((int)(cursor.NextSequence - _firstSequence)); + if (offset >= _count) + { + entry = default; + return false; + } + + entry = _entries[PhysicalIndex(offset)]; + cursor.NextSequence++; + Trim(); + return true; + } + + internal void RemoveCursor(ObservationCursor cursor) + { + _cursors.Remove(cursor); + Trim(); + } + + private void Trim() + { + if (_count == 0) + { + return; + } + + int removeCount = checked((int)(SlowestSequence() - _firstSequence)); + if (removeCount == 0) + { + return; + } + + // Only worth clearing when a slot can keep an object alive; the check folds away for the rest. + if (RuntimeHelpers.IsReferenceOrContainsReferences()) + { + for (int i = 0; i < removeCount; i++) + { + _entries[PhysicalIndex(i)] = default; + } + } + + _head = PhysicalIndex(removeCount); + _count -= removeCount; + _firstSequence += removeCount; + } + + private long SlowestSequence() + { + long slowest = _nextSequence; + foreach (ObservationCursor cursor in _cursors) + { + if (cursor.NextSequence < slowest) + { + slowest = cursor.NextSequence; + } + } + + return slowest; + } + + private void EnsureCapacity(int requiredCapacity) + { + if (requiredCapacity <= _entries.Length) + { + return; + } + + int newCapacity = _entries.Length * 2; + while (newCapacity < requiredCapacity) + { + newCapacity *= 2; + } + + TEntry[] newEntries = new TEntry[Math.Min(newCapacity, _maximumCapacity)]; + int untilWrap = Math.Min(_count, _entries.Length - _head); + Array.Copy(_entries, _head, newEntries, 0, untilWrap); + Array.Copy(_entries, 0, newEntries, untilWrap, _count - untilWrap); + _entries = newEntries; + _head = 0; + } + + private int PhysicalIndex(int offset) + { + return (_head + offset) % _entries.Length; + } + + private string DescribeOverflow() + { + long slowest = SlowestSequence(); + List stalled = new(); + foreach (ObservationCursor cursor in _cursors) + { + if (cursor.NextSequence == slowest) + { + stalled.Add(cursor.Name); + } + } + + return $"Observation log reached its maximum capacity of {_maximumCapacity} entries. " + + $"Cursor '{string.Join("', '", stalled)}' stopped draining {_nextSequence - slowest} entries ago."; + } +} diff --git a/src/Pixely.Observations/Pixely.Observations.csproj b/src/Pixely.Observations/Pixely.Observations.csproj new file mode 100644 index 00000000..a3cb2a51 --- /dev/null +++ b/src/Pixely.Observations/Pixely.Observations.csproj @@ -0,0 +1,10 @@ + + + + net10.0 + 14 + enable + enable + + + diff --git a/tests/Pixely.Observations.Tests/ObservationLogTests.cs b/tests/Pixely.Observations.Tests/ObservationLogTests.cs new file mode 100644 index 00000000..81bdc7a4 --- /dev/null +++ b/tests/Pixely.Observations.Tests/ObservationLogTests.cs @@ -0,0 +1,219 @@ +namespace Pixely.Observations.Tests; + +public readonly record struct TestEntry(int Value); + +[TestFixture] +public sealed class ObservationLogTests +{ + [Test] + public void Cursor_ReadsAppendedEntriesInOrder() + { + ObservationLog log = new ObservationLog(64); + ObservationCursor cursor = log.CreateCursor(); + + log.Append(new TestEntry(1)); + log.Append(new TestEntry(2)); + log.Append(new TestEntry(3)); + + Assert.That(Drain(cursor), Is.EqualTo(new[] { 1, 2, 3 })); + Assert.That(cursor.TryRead(out _), Is.False); + } + + [Test] + public void Cursor_OnlySeesEntriesAppendedAfterItsCreation() + { + ObservationLog log = new ObservationLog(64); + log.Append(new TestEntry(1)); + + ObservationCursor cursor = log.CreateCursor(); + log.Append(new TestEntry(2)); + + Assert.That(Drain(cursor), Is.EqualTo(new[] { 2 })); + } + + [Test] + public void Cursors_DrainIndependentlyAtTheirOwnPace() + { + ObservationLog log = new ObservationLog(64); + ObservationCursor fast = log.CreateCursor("fast"); + ObservationCursor slow = log.CreateCursor("slow"); + + log.Append(new TestEntry(1)); + log.Append(new TestEntry(2)); + + Assert.That(Drain(fast), Is.EqualTo(new[] { 1, 2 })); + + log.Append(new TestEntry(3)); + + // The slow cursor still sees everything from where it started. + Assert.That(Drain(slow), Is.EqualTo(new[] { 1, 2, 3 })); + Assert.That(Drain(fast), Is.EqualTo(new[] { 3 })); + } + + [Test] + public void Buffer_GrowsBeyondInitialCapacityPreservingOrder() + { + ObservationLog log = new ObservationLog(1024); + ObservationCursor cursor = log.CreateCursor(); + + // Far beyond the initial capacity of 16, without draining, forcing growth. + int[] expected = Enumerable.Range(0, 100).ToArray(); + foreach (int value in expected) + { + log.Append(new TestEntry(value)); + } + + Assert.That(Drain(cursor), Is.EqualTo(expected)); + } + + [Test] + public void Buffer_GrowsNoFurtherThanTheMaximumCapacity() + { + ObservationLog log = new ObservationLog(40); + ObservationCursor cursor = log.CreateCursor(); + + // Doubling from 16 would overshoot 40; the buffer has to stop at it and still keep order. + int[] expected = Enumerable.Range(0, 40).ToArray(); + foreach (int value in expected) + { + log.Append(new TestEntry(value)); + } + + Assert.That(Drain(cursor), Is.EqualTo(expected)); + } + + [Test] + public void MaximumCapacity_BelowTheInitialCapacityIsHonoured() + { + ObservationLog log = new ObservationLog(4); + ObservationCursor cursor = log.CreateCursor("stalled"); + + for (int i = 0; i < 4; i++) + { + log.Append(new TestEntry(i)); + } + + Assert.That(() => log.Append(new TestEntry(4)), Throws.InvalidOperationException); + Assert.That(Drain(cursor), Is.EqualTo(new[] { 0, 1, 2, 3 })); + } + + [Test] + public void Buffer_WrapsAroundWhenInterleavingAppendAndRead() + { + ObservationLog log = new ObservationLog(64); + ObservationCursor cursor = log.CreateCursor(); + + // Interleaving advances the head past the modulo boundary repeatedly. + List read = new(); + for (int i = 0; i < 100; i++) + { + log.Append(new TestEntry(i)); + Assert.That(cursor.TryRead(out TestEntry entry), Is.True); + read.Add(entry.Value); + } + + Assert.That(read, Is.EqualTo(Enumerable.Range(0, 100))); + } + + [Test] + public void Trimming_ReleasesEntriesOnceEveryCursorHasPassedThem() + { + ObservationLog log = new ObservationLog(32); + ObservationCursor cursor = log.CreateCursor(); + + // With a single cursor that keeps up, the log never fills no matter how many entries flow through it. + for (int i = 0; i < 32 * 100; i++) + { + log.Append(new TestEntry(i)); + cursor.TryRead(out _); + } + + Assert.Pass(); + } + + [Test] + public void Trimming_KeepsNothingWhenNoCursorExists() + { + ObservationLog log = new ObservationLog(4); + + // Nothing reads, so nothing is retained and the maximum capacity is never reached. + for (int i = 0; i < 100; i++) + { + log.Append(new TestEntry(i)); + } + + Assert.That(Drain(log.CreateCursor()), Is.Empty); + } + + [Test] + public void Append_ThrowsNamingTheCursorThatStoppedDraining() + { + ObservationLog log = new ObservationLog(8); + ObservationCursor keepingUp = log.CreateCursor("keeping-up"); + log.CreateCursor("presenter"); + + for (int i = 0; i < 8; i++) + { + log.Append(new TestEntry(i)); + keepingUp.TryRead(out _); + } + + Assert.That(() => log.Append(new TestEntry(8)), + Throws.InvalidOperationException.With.Message.Contains("presenter") + .And.Message.Contains("8 entries ago") + .And.Message.Not.Contains("keeping-up")); + } + + [Test] + public void DisposingStalledCursor_FreesTheLogForTrimming() + { + ObservationLog log = new ObservationLog(8); + ObservationCursor stalled = log.CreateCursor("stalled"); + + for (int i = 0; i < 8; i++) + { + log.Append(new TestEntry(i)); + } + + stalled.Dispose(); + + Assert.That(() => log.Append(new TestEntry(8)), Throws.Nothing); + } + + [Test] + public void DisposedCursor_ThrowsOnRead() + { + ObservationLog log = new ObservationLog(64); + ObservationCursor cursor = log.CreateCursor(); + cursor.Dispose(); + + Assert.That(() => cursor.TryRead(out _), Throws.TypeOf()); + } + + [Test] + public void Cursor_OnEmptyLogReturnsFalseAndTheDefaultEntry() + { + ObservationLog log = new ObservationLog(64); + ObservationCursor cursor = log.CreateCursor(); + + Assert.That(cursor.TryRead(out TestEntry entry), Is.False); + Assert.That(entry, Is.EqualTo(default(TestEntry))); + } + + [Test] + public void MaximumCapacity_MustBeAtLeastOne() + { + Assert.That(() => new ObservationLog(0), Throws.TypeOf()); + } + + private static int[] Drain(ObservationCursor cursor) + { + List values = new(); + while (cursor.TryRead(out TestEntry entry)) + { + values.Add(entry.Value); + } + + return values.ToArray(); + } +} diff --git a/tests/Pixely.Observations.Tests/Pixely.Observations.Tests.csproj b/tests/Pixely.Observations.Tests/Pixely.Observations.Tests.csproj new file mode 100644 index 00000000..90e7a459 --- /dev/null +++ b/tests/Pixely.Observations.Tests/Pixely.Observations.Tests.csproj @@ -0,0 +1,27 @@ + + + + net10.0 + 14 + enable + enable + false + + + + + + + + + + + + + + + + + + + diff --git a/tests/Pixely.Package.Tests/PackageIntegrationTests.cs b/tests/Pixely.Package.Tests/PackageIntegrationTests.cs index 9a4b164a..c784a16d 100644 --- a/tests/Pixely.Package.Tests/PackageIntegrationTests.cs +++ b/tests/Pixely.Package.Tests/PackageIntegrationTests.cs @@ -26,6 +26,7 @@ public class PackageIntegrationTests "Pixely.DependencyInjection", "Pixely.Events", "Pixely.Logging", + "Pixely.Observations", "Pixely.Pencuil", "Pixely.ShaderCommon", "Pixely.Utils", From 78a2604da328b0f160b096a80eb028e89a56ccca Mon Sep 17 00:00:00 2001 From: botoddly <250804054+botoddly@users.noreply.github.com> Date: Thu, 10 Sep 2026 10:14:03 +0200 Subject: [PATCH 2/5] Allow a reference entry type in the observation log --- docs/observations.md | 15 ++++++---- src/Pixely.Observations/IObservationLog.cs | 2 +- src/Pixely.Observations/IObservationWriter.cs | 2 +- src/Pixely.Observations/ObservationCursor.cs | 6 ++-- src/Pixely.Observations/ObservationLog.cs | 17 ++++++----- .../ObservationLogTests.cs | 29 +++++++++++++++++++ 6 files changed, 55 insertions(+), 16 deletions(-) diff --git a/docs/observations.md b/docs/observations.md index 676d6a34..679030a2 100644 --- a/docs/observations.md +++ b/docs/observations.md @@ -1,6 +1,6 @@ # Observations -`ObservationLog` in `Pixely.Observations` is an append-only log of value entries that readers drain at their +`ObservationLog` in `Pixely.Observations` is an append-only log of entries that readers drain at their own pace. Rules append; nothing is called back. A reader holds a cursor, reads when it suits its own point in the frame, and the log drops an entry once every cursor has passed it. @@ -11,8 +11,8 @@ frame. State answers what is true now, and a rule can resolve many transitions b ## The entry type -One log carries one entry type. To carry several kinds of entry in one order, make `TEntry` a tagged value -type; the log never looks inside it. +One log carries one entry type. To carry several kinds of entry in one order, make `TEntry` a tagged type; the +log never looks inside it. A value type is the one to reach for, because appending it costs no allocation. ```csharp public readonly record struct UnitMovedEntry(UnitId Unit, TilePoint From, TilePoint To); @@ -23,8 +23,13 @@ public enum ObservationKind { UnitMoved, UnitDied } public readonly record struct Observation(ObservationKind Kind, ParticipantId Perceiver, UnitMovedEntry Moved, UnitDiedEntry Died); ``` -Entries are past-tense records of ids and value types, never a live reference into game state. Nothing -allocates per entry: the log stores `TEntry` in an array, `Append` takes it by `in`, and `TryRead` copies it out. +Entries are past-tense records of ids and value types, never a live reference into game state. With a value +entry nothing allocates per entry: the log stores `TEntry` inline in an array, `Append` takes it by `in`, and +`TryRead` copies it out. + +`TEntry` may be a class where that suits the game better, and the log releases each slot as it trims so a +drained entry is not held alive. It costs an allocation per append, so it does not belong on a path that +appends every frame. ## Writing diff --git a/src/Pixely.Observations/IObservationLog.cs b/src/Pixely.Observations/IObservationLog.cs index a41ed95d..19438628 100644 --- a/src/Pixely.Observations/IObservationLog.cs +++ b/src/Pixely.Observations/IObservationLog.cs @@ -4,7 +4,7 @@ namespace Pixely.Observations; /// The read half of an . Inject it where a reader creates its own cursor /// and drains it on its own cadence. /// -public interface IObservationLog where TEntry : struct +public interface IObservationLog { /// /// Creates a cursor positioned after the last appended entry, so it sees only what is appended from now on. diff --git a/src/Pixely.Observations/IObservationWriter.cs b/src/Pixely.Observations/IObservationWriter.cs index 815121ec..1dd2c4a2 100644 --- a/src/Pixely.Observations/IObservationWriter.cs +++ b/src/Pixely.Observations/IObservationWriter.cs @@ -3,7 +3,7 @@ namespace Pixely.Observations; /// /// The write half of an . Inject it where rules append and nothing reads. /// -public interface IObservationWriter where TEntry : struct +public interface IObservationWriter { void Append(in TEntry entry); } diff --git a/src/Pixely.Observations/ObservationCursor.cs b/src/Pixely.Observations/ObservationCursor.cs index 367a6e88..18e30cf3 100644 --- a/src/Pixely.Observations/ObservationCursor.cs +++ b/src/Pixely.Observations/ObservationCursor.cs @@ -1,10 +1,12 @@ +using System.Diagnostics.CodeAnalysis; + namespace Pixely.Observations; /// /// One reader's position in an . Dispose it when the reader goes away, /// otherwise it holds the log's trim point where it stopped and the log fills. /// -public sealed class ObservationCursor : IDisposable where TEntry : struct +public sealed class ObservationCursor : IDisposable { private readonly ObservationLog _log; private bool _disposed; @@ -20,7 +22,7 @@ internal ObservationCursor(ObservationLog log, string name, long nextSeq internal long NextSequence { get; set; } - public bool TryRead(out TEntry entry) + public bool TryRead([MaybeNullWhen(false)] out TEntry entry) { ObjectDisposedException.ThrowIf(_disposed, this); return _log.TryRead(this, out entry); diff --git a/src/Pixely.Observations/ObservationLog.cs b/src/Pixely.Observations/ObservationLog.cs index d300c1ce..8cadd201 100644 --- a/src/Pixely.Observations/ObservationLog.cs +++ b/src/Pixely.Observations/ObservationLog.cs @@ -1,17 +1,19 @@ +using System.Diagnostics.CodeAnalysis; using System.Runtime.CompilerServices; namespace Pixely.Observations; /// -/// An append-only log of value entries that readers drain at their own pace through their own cursors. +/// An append-only log of entries that readers drain at their own pace through their own cursors. /// Appending never calls a reader. An entry is dropped once every cursor has passed it, so the log is bounded /// by the slowest reader rather than by how long the run lasts. /// /// -/// The single entry type of this log. Carry several kinds of entry in one log by making this a tagged value -/// type; the log never looks inside it. +/// The single entry type of this log; the log never looks inside it. Carry several kinds of entry in one log +/// by making it a tagged type. A value type keeps appending free of allocation, which is why the entries a +/// game appends every frame should be one. /// -public sealed class ObservationLog : IObservationLog, IObservationWriter where TEntry : struct +public sealed class ObservationLog : IObservationLog, IObservationWriter { private const int InitialCapacity = 16; @@ -56,7 +58,7 @@ public ObservationCursor CreateCursor(string name = "unnamed") return cursor; } - internal bool TryRead(ObservationCursor cursor, out TEntry entry) + internal bool TryRead(ObservationCursor cursor, [MaybeNullWhen(false)] out TEntry entry) { int offset = checked((int)(cursor.NextSequence - _firstSequence)); if (offset >= _count) @@ -90,12 +92,13 @@ private void Trim() return; } - // Only worth clearing when a slot can keep an object alive; the check folds away for the rest. + // Only worth clearing when a slot can keep an object alive; the check folds away for the rest. A cleared + // slot is past every cursor, so nothing reads it back before an append overwrites it. if (RuntimeHelpers.IsReferenceOrContainsReferences()) { for (int i = 0; i < removeCount; i++) { - _entries[PhysicalIndex(i)] = default; + _entries[PhysicalIndex(i)] = default!; } } diff --git a/tests/Pixely.Observations.Tests/ObservationLogTests.cs b/tests/Pixely.Observations.Tests/ObservationLogTests.cs index 81bdc7a4..f510f4ad 100644 --- a/tests/Pixely.Observations.Tests/ObservationLogTests.cs +++ b/tests/Pixely.Observations.Tests/ObservationLogTests.cs @@ -2,6 +2,8 @@ namespace Pixely.Observations.Tests; public readonly record struct TestEntry(int Value); +public sealed record TestReferenceEntry(int Value); + [TestFixture] public sealed class ObservationLogTests { @@ -206,6 +208,33 @@ public void MaximumCapacity_MustBeAtLeastOne() Assert.That(() => new ObservationLog(0), Throws.TypeOf()); } + [Test] + public void ReferenceEntry_ReadsBackTheAppendedInstancesInOrder() + { + ObservationLog log = new ObservationLog(64); + ObservationCursor cursor = log.CreateCursor(); + TestReferenceEntry first = new TestReferenceEntry(1); + TestReferenceEntry second = new TestReferenceEntry(2); + + log.Append(first); + log.Append(second); + + Assert.That(cursor.TryRead(out TestReferenceEntry? read), Is.True); + Assert.That(read, Is.SameAs(first)); + Assert.That(cursor.TryRead(out read), Is.True); + Assert.That(read, Is.SameAs(second)); + } + + [Test] + public void ReferenceEntry_OnEmptyLogReturnsFalseAndNull() + { + ObservationLog log = new ObservationLog(64); + ObservationCursor cursor = log.CreateCursor(); + + Assert.That(cursor.TryRead(out TestReferenceEntry? entry), Is.False); + Assert.That(entry, Is.Null); + } + private static int[] Drain(ObservationCursor cursor) { List values = new(); From 1acf53e2fba94e1099562174cccb399ff8e90189 Mon Sep 17 00:00:00 2001 From: botoddly <250804054+botoddly@users.noreply.github.com> Date: Thu, 10 Sep 2026 10:35:08 +0200 Subject: [PATCH 3/5] Stop the observation log doubling past int range and cover the wrapped growth copy --- src/Pixely.Observations/ObservationLog.cs | 10 +- .../ObservationLogTests.cs | 110 ++++++++++++++++++ 2 files changed, 113 insertions(+), 7 deletions(-) diff --git a/src/Pixely.Observations/ObservationLog.cs b/src/Pixely.Observations/ObservationLog.cs index 8cadd201..7cebb67d 100644 --- a/src/Pixely.Observations/ObservationLog.cs +++ b/src/Pixely.Observations/ObservationLog.cs @@ -128,13 +128,9 @@ private void EnsureCapacity(int requiredCapacity) return; } - int newCapacity = _entries.Length * 2; - while (newCapacity < requiredCapacity) - { - newCapacity *= 2; - } - - TEntry[] newEntries = new TEntry[Math.Min(newCapacity, _maximumCapacity)]; + // Doubling only while it stays under the bound, so the last step lands on the bound instead of overflowing. + int newCapacity = _entries.Length <= _maximumCapacity / 2 ? _entries.Length * 2 : _maximumCapacity; + TEntry[] newEntries = new TEntry[newCapacity]; int untilWrap = Math.Min(_count, _entries.Length - _head); Array.Copy(_entries, _head, newEntries, 0, untilWrap); Array.Copy(_entries, 0, newEntries, untilWrap, _count - untilWrap); diff --git a/tests/Pixely.Observations.Tests/ObservationLogTests.cs b/tests/Pixely.Observations.Tests/ObservationLogTests.cs index f510f4ad..4a8a596f 100644 --- a/tests/Pixely.Observations.Tests/ObservationLogTests.cs +++ b/tests/Pixely.Observations.Tests/ObservationLogTests.cs @@ -1,3 +1,5 @@ +using System.Runtime.CompilerServices; + namespace Pixely.Observations.Tests; public readonly record struct TestEntry(int Value); @@ -235,6 +237,114 @@ public void ReferenceEntry_OnEmptyLogReturnsFalseAndNull() Assert.That(entry, Is.Null); } + [Test] + public void Buffer_GrowsWhileWrappedPreservingOrder() + { + ObservationLog log = new ObservationLog(1024); + ObservationCursor cursor = log.CreateCursor(); + + // Drain ten entries first so the head sits mid-buffer and the retained entries wrap around the end, + // which is what makes growth copy them in two segments. + for (int i = 0; i < 10; i++) + { + log.Append(new TestEntry(i)); + cursor.TryRead(out _); + } + + int[] expected = Enumerable.Range(100, 20).ToArray(); + foreach (int value in expected) + { + log.Append(new TestEntry(value)); + } + + Assert.That(Drain(cursor), Is.EqualTo(expected)); + } + + [Test] + public void DisposingOneCursor_KeepsTheEntriesAnotherStillNeeds() + { + ObservationLog log = new ObservationLog(64); + ObservationCursor leaving = log.CreateCursor("leaving"); + ObservationCursor staying = log.CreateCursor("staying"); + + log.Append(new TestEntry(1)); + log.Append(new TestEntry(2)); + leaving.Dispose(); + + Assert.That(Drain(staying), Is.EqualTo(new[] { 1, 2 })); + } + + [Test] + public void DisposingACursorTwice_IsHarmless() + { + ObservationLog log = new ObservationLog(64); + ObservationCursor cursor = log.CreateCursor(); + cursor.Dispose(); + + Assert.That(() => cursor.Dispose(), Throws.Nothing); + } + + [Test] + public void Append_NamesEveryCursorTiedAtTheBack() + { + ObservationLog log = new ObservationLog(4); + log.CreateCursor("presenter"); + log.CreateCursor("audio"); + + for (int i = 0; i < 4; i++) + { + log.Append(new TestEntry(i)); + } + + Assert.That(() => log.Append(new TestEntry(4)), + Throws.InvalidOperationException.With.Message.Contains("presenter") + .And.Message.Contains("audio")); + } + + [Test] + public void Append_SucceedsAgainOnceTheStalledCursorDrains() + { + ObservationLog log = new ObservationLog(4); + ObservationCursor stalled = log.CreateCursor("stalled"); + + for (int i = 0; i < 4; i++) + { + log.Append(new TestEntry(i)); + } + + Assert.That(() => log.Append(new TestEntry(4)), Throws.InvalidOperationException); + Assert.That(Drain(stalled), Is.EqualTo(new[] { 0, 1, 2, 3 })); + + log.Append(new TestEntry(5)); + + Assert.That(Drain(stalled), Is.EqualTo(new[] { 5 })); + } + + [Test] + public void Trimming_ReleasesAReferenceEntryOnceEveryCursorHasPassedIt() + { + ObservationLog log = new ObservationLog(64); + ObservationCursor cursor = log.CreateCursor(); + + WeakReference reference = AppendAndDrainOne(log, cursor); + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + + Assert.That(reference.IsAlive, Is.False); + } + + // The only strong reference lives in this frame, so returning drops it and leaves the log's slot as the + // one thing that could still keep the entry alive. + [MethodImpl(MethodImplOptions.NoInlining)] + private static WeakReference AppendAndDrainOne(ObservationLog log, ObservationCursor cursor) + { + TestReferenceEntry entry = new TestReferenceEntry(1); + log.Append(entry); + cursor.TryRead(out TestReferenceEntry? _); + return new WeakReference(entry); + } + private static int[] Drain(ObservationCursor cursor) { List values = new(); From dc646a731ef977d2b7aac0ab5d2e18765d3c3a17 Mon Sep 17 00:00:00 2001 From: botoddly <250804054+botoddly@users.noreply.github.com> Date: Thu, 10 Sep 2026 10:48:03 +0200 Subject: [PATCH 4/5] Split the observation log into a log, a writer and a reader instead of interfaces over one type --- AGENTS.md | 2 +- docs/observations.md | 55 +-- src/Pixely.Observations/IObservationLog.cs | 14 - src/Pixely.Observations/IObservationWriter.cs | 9 - src/Pixely.Observations/ObservationCursor.cs | 41 --- src/Pixely.Observations/ObservationLog.cs | 62 ++-- src/Pixely.Observations/ObservationReader.cs | 45 +++ src/Pixely.Observations/ObservationWriter.cs | 20 ++ .../ObservationLogTests.cs | 316 +++++++++--------- 9 files changed, 299 insertions(+), 265 deletions(-) delete mode 100644 src/Pixely.Observations/IObservationLog.cs delete mode 100644 src/Pixely.Observations/IObservationWriter.cs delete mode 100644 src/Pixely.Observations/ObservationCursor.cs create mode 100644 src/Pixely.Observations/ObservationReader.cs create mode 100644 src/Pixely.Observations/ObservationWriter.cs diff --git a/AGENTS.md b/AGENTS.md index 9d655a88..2f7e98bf 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -4,7 +4,7 @@ - `docs/class-registration.md` - ServiceCollection/ServiceProvider API: registration overloads, source generator requirements, lifecycle, aliases, multi-registration, parent/child provider callback merging and scoped lifecycles - `docs/events.md` - Pixely.Events EventBus, event handlers, publishing, and DI auto-subscription -- `docs/observations.md` - Pixely.Observations: appending value entries, per-reader cursors, trimming behind the slowest reader, capacity and stall detection, registration +- `docs/observations.md` - Pixely.Observations: the log, its writer and its readers, trimming behind the slowest reader, capacity and stall detection, registration - `docs/static-factory-methods.md` - Static Create() method pattern - `docs/componentize.md` - Pixely.Componentize setup and usage - `docs/components.md` - GameWorld, GameObject, GameComponent lifecycle, Services, UpdateSystem diff --git a/docs/observations.md b/docs/observations.md index 679030a2..a25a4342 100644 --- a/docs/observations.md +++ b/docs/observations.md @@ -1,14 +1,24 @@ # Observations -`ObservationLog` in `Pixely.Observations` is an append-only log of entries that readers drain at their -own pace. Rules append; nothing is called back. A reader holds a cursor, reads when it suits its own point in -the frame, and the log drops an entry once every cursor has passed it. +`Pixely.Observations` is an append-only log of entries that readers drain at their own pace. Rules append; +nothing is called back. A reader holds its own position, reads when it suits its own point in the frame, and +the log drops an entry once every reader has passed it. Use it for a transition that two reads of game state one frame apart cannot derive: a unit that walked a path and arrived, one that appeared and was gone again, an action that resolved and was reversed inside the same frame. State answers what is true now, and a rule can resolve many transitions between two reads, so any "most recent transition" field is overwritten before a reader looks at it. +Three types, each with one job: + +- `ObservationLog` is the storage. It is constructed and then handed to the other two; it has no other + public members. +- `ObservationWriter` appends. It cannot read. +- `ObservationReader` is one reader's position in the log. It cannot append. + +So a rule holding a writer has no way to drain the log, and a reader has no way to record an observation no +rule produced. + ## The entry type One log carries one entry type. To carry several kinds of entry in one order, make `TEntry` a tagged type; the @@ -33,14 +43,14 @@ appends every frame. ## Writing -Inject `IObservationWriter` where rules append, so a writer cannot read: +Inject `ObservationWriter` where rules record what happened: ```csharp internal sealed class MoveMechanic { - private readonly IObservationWriter _observations; + private readonly ObservationWriter _observations; - internal MoveMechanic(IObservationWriter observations) => _observations = observations; + internal MoveMechanic(ObservationWriter observations) => _observations = observations; internal void Move(UnitId unit, TilePoint destination) { @@ -52,17 +62,17 @@ internal sealed class MoveMechanic ## Reading -Inject `IObservationLog`, create a cursor named after the reader, and drain it in the reader's own -update. Dispose the cursor with the reader: +Inject the log, construct a reader named after the consumer, and drain it in the consumer's own update. Dispose +the reader with the consumer: ```csharp internal sealed class UnitSpritePresenter : IUpdatable, IDisposable { - private readonly ObservationCursor _observations; + private readonly ObservationReader _observations; - internal UnitSpritePresenter(IObservationLog observations) + internal UnitSpritePresenter(ObservationLog log) { - _observations = observations.CreateCursor(nameof(UnitSpritePresenter)); + _observations = new ObservationReader(log, nameof(UnitSpritePresenter)); } public void Update() @@ -77,13 +87,16 @@ internal sealed class UnitSpritePresenter : IUpdatable, IDisposable } ``` -A cursor starts positioned after the last appended entry, so a reader created part-way through a run sees -only what is appended from then on. Cursors read independently: entries appended this frame may be drained by +Each consumer constructs its own reader rather than being handed one, because every reader of a given log is +the same closed type and the container resolves by type. Constructing it also lets the consumer name it. + +A reader starts positioned after the last appended entry, so a consumer created part-way through a run sees +only what is appended from then on. Readers drain independently: entries appended this frame may be drained by one reader now and by another several frames later. Addressing an entry to a subset of readers is the game's business, not the log's. Carry the perceiver in -`TEntry` and have the reader skip what it did not perceive. A reader bound to one participant should wrap its -cursor once rather than repeat the check in every consumer. +`TEntry` and have the reader skip what it did not perceive. A consumer bound to one participant should wrap its +reader once rather than repeat the check in every place that drains. ## Bounds and stalls @@ -92,22 +105,22 @@ so it settles at the high-water mark of its bursts and is reclaimed when the log Maximum capacity is a stall detector, not a working size. Pick a number far past any legitimate burst, knowing a slot costs the size of `TEntry`. A reader that stops draining holds the trim point where it stopped; once the -log fills, `Append` throws and names the cursor that stopped and how far behind it is. Dropping the oldest +log fills, appending throws and names the reader that stopped and how far behind it is. Dropping the oldest entry instead would hide exactly that failure. ```text -Observation log reached its maximum capacity of 4096 entries. Cursor 'UnitSpritePresenter' stopped draining +Observation log reached its maximum capacity of 4096 entries. Reader 'UnitSpritePresenter' stopped draining 4096 entries ago. ``` ## Registration -The log takes its capacity as a constructor argument, so register the instance and alias the two halves: +Construct the log, then register it alongside a writer over it: ```csharp -services.AddSingleton(new ObservationLog(4096)); -services.AddAlias, ObservationLog>(); -services.AddAlias, ObservationLog>(); +ObservationLog log = new ObservationLog(4096); +services.AddSingleton(log); +services.AddSingleton(new ObservationWriter(log)); ``` Register it in the scope it belongs to. A log registered in a stage's child provider dies with that stage, and diff --git a/src/Pixely.Observations/IObservationLog.cs b/src/Pixely.Observations/IObservationLog.cs deleted file mode 100644 index 19438628..00000000 --- a/src/Pixely.Observations/IObservationLog.cs +++ /dev/null @@ -1,14 +0,0 @@ -namespace Pixely.Observations; - -/// -/// The read half of an . Inject it where a reader creates its own cursor -/// and drains it on its own cadence. -/// -public interface IObservationLog -{ - /// - /// Creates a cursor positioned after the last appended entry, so it sees only what is appended from now on. - /// The name identifies the cursor when a stalled reader fills the log, so give it the reader's own name. - /// - ObservationCursor CreateCursor(string name = "unnamed"); -} diff --git a/src/Pixely.Observations/IObservationWriter.cs b/src/Pixely.Observations/IObservationWriter.cs deleted file mode 100644 index 1dd2c4a2..00000000 --- a/src/Pixely.Observations/IObservationWriter.cs +++ /dev/null @@ -1,9 +0,0 @@ -namespace Pixely.Observations; - -/// -/// The write half of an . Inject it where rules append and nothing reads. -/// -public interface IObservationWriter -{ - void Append(in TEntry entry); -} diff --git a/src/Pixely.Observations/ObservationCursor.cs b/src/Pixely.Observations/ObservationCursor.cs deleted file mode 100644 index 18e30cf3..00000000 --- a/src/Pixely.Observations/ObservationCursor.cs +++ /dev/null @@ -1,41 +0,0 @@ -using System.Diagnostics.CodeAnalysis; - -namespace Pixely.Observations; - -/// -/// One reader's position in an . Dispose it when the reader goes away, -/// otherwise it holds the log's trim point where it stopped and the log fills. -/// -public sealed class ObservationCursor : IDisposable -{ - private readonly ObservationLog _log; - private bool _disposed; - - internal ObservationCursor(ObservationLog log, string name, long nextSequence) - { - _log = log; - Name = name; - NextSequence = nextSequence; - } - - public string Name { get; } - - internal long NextSequence { get; set; } - - public bool TryRead([MaybeNullWhen(false)] out TEntry entry) - { - ObjectDisposedException.ThrowIf(_disposed, this); - return _log.TryRead(this, out entry); - } - - public void Dispose() - { - if (_disposed) - { - return; - } - - _disposed = true; - _log.RemoveCursor(this); - } -} diff --git a/src/Pixely.Observations/ObservationLog.cs b/src/Pixely.Observations/ObservationLog.cs index 7cebb67d..db5be36f 100644 --- a/src/Pixely.Observations/ObservationLog.cs +++ b/src/Pixely.Observations/ObservationLog.cs @@ -4,21 +4,25 @@ namespace Pixely.Observations; /// -/// An append-only log of entries that readers drain at their own pace through their own cursors. -/// Appending never calls a reader. An entry is dropped once every cursor has passed it, so the log is bounded -/// by the slowest reader rather than by how long the run lasts. +/// An append-only log of entries that readers drain at their own pace. Appending never calls a reader. An entry +/// is dropped once every reader has passed it, so the log is bounded by the slowest reader rather than by how +/// long the run lasts. /// +/// +/// The log is storage and nothing else: it is reached through an or an +/// , so neither role can do the other's job. +/// /// /// The single entry type of this log; the log never looks inside it. Carry several kinds of entry in one log /// by making it a tagged type. A value type keeps appending free of allocation, which is why the entries a /// game appends every frame should be one. /// -public sealed class ObservationLog : IObservationLog, IObservationWriter +public sealed class ObservationLog { private const int InitialCapacity = 16; private readonly int _maximumCapacity; - private readonly List> _cursors = new(); + private readonly List> _readers = new(); private TEntry[] _entries; private int _head; private int _count; @@ -26,9 +30,9 @@ public sealed class ObservationLog : IObservationLog, IObservati private long _nextSequence; /// - /// How many entries the log may retain before throws. The buffer starts small and grows - /// towards this bound, so it is a stall detector rather than a working size: pick a number far past any - /// legitimate burst, knowing a slot costs the size of . + /// How many entries the log may retain before appending throws. The buffer starts small and grows towards + /// this bound, so it is a stall detector rather than a working size: pick a number far past any legitimate + /// burst, knowing a slot costs the size of . /// public ObservationLog(int maximumCapacity) { @@ -37,7 +41,7 @@ public ObservationLog(int maximumCapacity) _entries = new TEntry[Math.Min(InitialCapacity, maximumCapacity)]; } - public void Append(in TEntry entry) + internal void Append(in TEntry entry) { if (_count == _maximumCapacity) { @@ -51,16 +55,9 @@ public void Append(in TEntry entry) Trim(); } - public ObservationCursor CreateCursor(string name = "unnamed") + internal bool TryRead(ObservationReader reader, [MaybeNullWhen(false)] out TEntry entry) { - ObservationCursor cursor = new ObservationCursor(this, name, _nextSequence); - _cursors.Add(cursor); - return cursor; - } - - internal bool TryRead(ObservationCursor cursor, [MaybeNullWhen(false)] out TEntry entry) - { - int offset = checked((int)(cursor.NextSequence - _firstSequence)); + int offset = checked((int)(reader.NextSequence - _firstSequence)); if (offset >= _count) { entry = default; @@ -68,14 +65,21 @@ internal bool TryRead(ObservationCursor cursor, [MaybeNullWhen(false)] o } entry = _entries[PhysicalIndex(offset)]; - cursor.NextSequence++; + reader.NextSequence++; Trim(); return true; } - internal void RemoveCursor(ObservationCursor cursor) + // A reader starts after the last appended entry, so it sees only what is appended from now on. + internal void AddReader(ObservationReader reader) + { + reader.NextSequence = _nextSequence; + _readers.Add(reader); + } + + internal void RemoveReader(ObservationReader reader) { - _cursors.Remove(cursor); + _readers.Remove(reader); Trim(); } @@ -93,7 +97,7 @@ private void Trim() } // Only worth clearing when a slot can keep an object alive; the check folds away for the rest. A cleared - // slot is past every cursor, so nothing reads it back before an append overwrites it. + // slot is past every reader, so nothing reads it back before an append overwrites it. if (RuntimeHelpers.IsReferenceOrContainsReferences()) { for (int i = 0; i < removeCount; i++) @@ -110,11 +114,11 @@ private void Trim() private long SlowestSequence() { long slowest = _nextSequence; - foreach (ObservationCursor cursor in _cursors) + foreach (ObservationReader reader in _readers) { - if (cursor.NextSequence < slowest) + if (reader.NextSequence < slowest) { - slowest = cursor.NextSequence; + slowest = reader.NextSequence; } } @@ -147,15 +151,15 @@ private string DescribeOverflow() { long slowest = SlowestSequence(); List stalled = new(); - foreach (ObservationCursor cursor in _cursors) + foreach (ObservationReader reader in _readers) { - if (cursor.NextSequence == slowest) + if (reader.NextSequence == slowest) { - stalled.Add(cursor.Name); + stalled.Add(reader.Name); } } return $"Observation log reached its maximum capacity of {_maximumCapacity} entries. " - + $"Cursor '{string.Join("', '", stalled)}' stopped draining {_nextSequence - slowest} entries ago."; + + $"Reader '{string.Join("', '", stalled)}' stopped draining {_nextSequence - slowest} entries ago."; } } diff --git a/src/Pixely.Observations/ObservationReader.cs b/src/Pixely.Observations/ObservationReader.cs new file mode 100644 index 00000000..43b401f1 --- /dev/null +++ b/src/Pixely.Observations/ObservationReader.cs @@ -0,0 +1,45 @@ +using System.Diagnostics.CodeAnalysis; + +namespace Pixely.Observations; + +/// +/// One reader's own position in an . Several readers drain the same log at +/// their own pace, each through its own instance. Dispose it when the reader goes away, otherwise it holds the +/// log's trim point where it stopped and the log fills. +/// +public sealed class ObservationReader : IDisposable +{ + private readonly ObservationLog _log; + private bool _disposed; + + /// + /// Identifies this reader when it stops draining and fills the log, so give it the reader's own name. + /// + public ObservationReader(ObservationLog log, string name = "unnamed") + { + _log = log; + Name = name; + log.AddReader(this); + } + + public string Name { get; } + + internal long NextSequence { get; set; } + + public bool TryRead([MaybeNullWhen(false)] out TEntry entry) + { + ObjectDisposedException.ThrowIf(_disposed, this); + return _log.TryRead(this, out entry); + } + + public void Dispose() + { + if (_disposed) + { + return; + } + + _disposed = true; + _log.RemoveReader(this); + } +} diff --git a/src/Pixely.Observations/ObservationWriter.cs b/src/Pixely.Observations/ObservationWriter.cs new file mode 100644 index 00000000..f638d1f3 --- /dev/null +++ b/src/Pixely.Observations/ObservationWriter.cs @@ -0,0 +1,20 @@ +namespace Pixely.Observations; + +/// +/// Appends to an . Inject it where rules record what happened, so nothing +/// that writes can also read. +/// +public sealed class ObservationWriter +{ + private readonly ObservationLog _log; + + public ObservationWriter(ObservationLog log) + { + _log = log; + } + + public void Append(in TEntry entry) + { + _log.Append(entry); + } +} diff --git a/tests/Pixely.Observations.Tests/ObservationLogTests.cs b/tests/Pixely.Observations.Tests/ObservationLogTests.cs index 4a8a596f..4ac4a003 100644 --- a/tests/Pixely.Observations.Tests/ObservationLogTests.cs +++ b/tests/Pixely.Observations.Tests/ObservationLogTests.cs @@ -10,46 +10,49 @@ public sealed record TestReferenceEntry(int Value); public sealed class ObservationLogTests { [Test] - public void Cursor_ReadsAppendedEntriesInOrder() + public void Reader_ReadsAppendedEntriesInOrder() { ObservationLog log = new ObservationLog(64); - ObservationCursor cursor = log.CreateCursor(); + ObservationWriter writer = new ObservationWriter(log); + ObservationReader reader = new ObservationReader(log); - log.Append(new TestEntry(1)); - log.Append(new TestEntry(2)); - log.Append(new TestEntry(3)); + writer.Append(new TestEntry(1)); + writer.Append(new TestEntry(2)); + writer.Append(new TestEntry(3)); - Assert.That(Drain(cursor), Is.EqualTo(new[] { 1, 2, 3 })); - Assert.That(cursor.TryRead(out _), Is.False); + Assert.That(Drain(reader), Is.EqualTo(new[] { 1, 2, 3 })); + Assert.That(reader.TryRead(out _), Is.False); } [Test] - public void Cursor_OnlySeesEntriesAppendedAfterItsCreation() + public void Reader_OnlySeesEntriesAppendedAfterItsCreation() { ObservationLog log = new ObservationLog(64); - log.Append(new TestEntry(1)); + ObservationWriter writer = new ObservationWriter(log); + writer.Append(new TestEntry(1)); - ObservationCursor cursor = log.CreateCursor(); - log.Append(new TestEntry(2)); + ObservationReader reader = new ObservationReader(log); + writer.Append(new TestEntry(2)); - Assert.That(Drain(cursor), Is.EqualTo(new[] { 2 })); + Assert.That(Drain(reader), Is.EqualTo(new[] { 2 })); } [Test] - public void Cursors_DrainIndependentlyAtTheirOwnPace() + public void Readers_DrainIndependentlyAtTheirOwnPace() { ObservationLog log = new ObservationLog(64); - ObservationCursor fast = log.CreateCursor("fast"); - ObservationCursor slow = log.CreateCursor("slow"); + ObservationWriter writer = new ObservationWriter(log); + ObservationReader fast = new ObservationReader(log, "fast"); + ObservationReader slow = new ObservationReader(log, "slow"); - log.Append(new TestEntry(1)); - log.Append(new TestEntry(2)); + writer.Append(new TestEntry(1)); + writer.Append(new TestEntry(2)); Assert.That(Drain(fast), Is.EqualTo(new[] { 1, 2 })); - log.Append(new TestEntry(3)); + writer.Append(new TestEntry(3)); - // The slow cursor still sees everything from where it started. + // The slow reader still sees everything from where it started. Assert.That(Drain(slow), Is.EqualTo(new[] { 1, 2, 3 })); Assert.That(Drain(fast), Is.EqualTo(new[] { 3 })); } @@ -58,61 +61,95 @@ public void Cursors_DrainIndependentlyAtTheirOwnPace() public void Buffer_GrowsBeyondInitialCapacityPreservingOrder() { ObservationLog log = new ObservationLog(1024); - ObservationCursor cursor = log.CreateCursor(); + ObservationWriter writer = new ObservationWriter(log); + ObservationReader reader = new ObservationReader(log); // Far beyond the initial capacity of 16, without draining, forcing growth. int[] expected = Enumerable.Range(0, 100).ToArray(); foreach (int value in expected) { - log.Append(new TestEntry(value)); + writer.Append(new TestEntry(value)); } - Assert.That(Drain(cursor), Is.EqualTo(expected)); + Assert.That(Drain(reader), Is.EqualTo(expected)); + } + + [Test] + public void Buffer_GrowsWhileWrappedPreservingOrder() + { + ObservationLog log = new ObservationLog(1024); + ObservationWriter writer = new ObservationWriter(log); + ObservationReader reader = new ObservationReader(log); + + // Drain ten entries first so the head sits mid-buffer and the retained entries wrap around the end, + // which is what makes growth copy them in two segments. + for (int i = 0; i < 10; i++) + { + writer.Append(new TestEntry(i)); + reader.TryRead(out _); + } + + int[] expected = Enumerable.Range(100, 20).ToArray(); + foreach (int value in expected) + { + writer.Append(new TestEntry(value)); + } + + Assert.That(Drain(reader), Is.EqualTo(expected)); } [Test] public void Buffer_GrowsNoFurtherThanTheMaximumCapacity() { ObservationLog log = new ObservationLog(40); - ObservationCursor cursor = log.CreateCursor(); + ObservationWriter writer = new ObservationWriter(log); + ObservationReader reader = new ObservationReader(log); // Doubling from 16 would overshoot 40; the buffer has to stop at it and still keep order. int[] expected = Enumerable.Range(0, 40).ToArray(); foreach (int value in expected) { - log.Append(new TestEntry(value)); + writer.Append(new TestEntry(value)); } - Assert.That(Drain(cursor), Is.EqualTo(expected)); + Assert.That(Drain(reader), Is.EqualTo(expected)); } [Test] public void MaximumCapacity_BelowTheInitialCapacityIsHonoured() { ObservationLog log = new ObservationLog(4); - ObservationCursor cursor = log.CreateCursor("stalled"); + ObservationWriter writer = new ObservationWriter(log); + ObservationReader reader = new ObservationReader(log, "stalled"); for (int i = 0; i < 4; i++) { - log.Append(new TestEntry(i)); + writer.Append(new TestEntry(i)); } - Assert.That(() => log.Append(new TestEntry(4)), Throws.InvalidOperationException); - Assert.That(Drain(cursor), Is.EqualTo(new[] { 0, 1, 2, 3 })); + Assert.That(() => writer.Append(new TestEntry(4)), Throws.InvalidOperationException); + Assert.That(Drain(reader), Is.EqualTo(new[] { 0, 1, 2, 3 })); + } + + [Test] + public void MaximumCapacity_MustBeAtLeastOne() + { + Assert.That(() => new ObservationLog(0), Throws.TypeOf()); } [Test] public void Buffer_WrapsAroundWhenInterleavingAppendAndRead() { ObservationLog log = new ObservationLog(64); - ObservationCursor cursor = log.CreateCursor(); + ObservationWriter writer = new ObservationWriter(log); + ObservationReader reader = new ObservationReader(log); // Interleaving advances the head past the modulo boundary repeatedly. List read = new(); for (int i = 0; i < 100; i++) { - log.Append(new TestEntry(i)); - Assert.That(cursor.TryRead(out TestEntry entry), Is.True); + writer.Append(new TestEntry(i)); + Assert.That(reader.TryRead(out TestEntry entry), Is.True); read.Add(entry.Value); } @@ -120,213 +157,192 @@ public void Buffer_WrapsAroundWhenInterleavingAppendAndRead() } [Test] - public void Trimming_ReleasesEntriesOnceEveryCursorHasPassedThem() + public void Trimming_ReleasesEntriesOnceEveryReaderHasPassedThem() { ObservationLog log = new ObservationLog(32); - ObservationCursor cursor = log.CreateCursor(); + ObservationWriter writer = new ObservationWriter(log); + ObservationReader reader = new ObservationReader(log); - // With a single cursor that keeps up, the log never fills no matter how many entries flow through it. + // With a single reader that keeps up, the log never fills no matter how many entries flow through it. for (int i = 0; i < 32 * 100; i++) { - log.Append(new TestEntry(i)); - cursor.TryRead(out _); + writer.Append(new TestEntry(i)); + reader.TryRead(out _); } Assert.Pass(); } [Test] - public void Trimming_KeepsNothingWhenNoCursorExists() + public void Trimming_KeepsNothingWhenNoReaderExists() { ObservationLog log = new ObservationLog(4); + ObservationWriter writer = new ObservationWriter(log); // Nothing reads, so nothing is retained and the maximum capacity is never reached. for (int i = 0; i < 100; i++) { - log.Append(new TestEntry(i)); + writer.Append(new TestEntry(i)); } - Assert.That(Drain(log.CreateCursor()), Is.Empty); + Assert.That(Drain(new ObservationReader(log)), Is.Empty); } [Test] - public void Append_ThrowsNamingTheCursorThatStoppedDraining() + public void Append_ThrowsNamingTheReaderThatStoppedDraining() { ObservationLog log = new ObservationLog(8); - ObservationCursor keepingUp = log.CreateCursor("keeping-up"); - log.CreateCursor("presenter"); + ObservationWriter writer = new ObservationWriter(log); + ObservationReader keepingUp = new ObservationReader(log, "keeping-up"); + _ = new ObservationReader(log, "presenter"); for (int i = 0; i < 8; i++) { - log.Append(new TestEntry(i)); + writer.Append(new TestEntry(i)); keepingUp.TryRead(out _); } - Assert.That(() => log.Append(new TestEntry(8)), + Assert.That(() => writer.Append(new TestEntry(8)), Throws.InvalidOperationException.With.Message.Contains("presenter") .And.Message.Contains("8 entries ago") .And.Message.Not.Contains("keeping-up")); } [Test] - public void DisposingStalledCursor_FreesTheLogForTrimming() + public void Append_NamesEveryReaderTiedAtTheBack() { - ObservationLog log = new ObservationLog(8); - ObservationCursor stalled = log.CreateCursor("stalled"); + ObservationLog log = new ObservationLog(4); + ObservationWriter writer = new ObservationWriter(log); + _ = new ObservationReader(log, "presenter"); + _ = new ObservationReader(log, "audio"); - for (int i = 0; i < 8; i++) + for (int i = 0; i < 4; i++) { - log.Append(new TestEntry(i)); + writer.Append(new TestEntry(i)); } - stalled.Dispose(); - - Assert.That(() => log.Append(new TestEntry(8)), Throws.Nothing); + Assert.That(() => writer.Append(new TestEntry(4)), + Throws.InvalidOperationException.With.Message.Contains("presenter") + .And.Message.Contains("audio")); } [Test] - public void DisposedCursor_ThrowsOnRead() + public void Append_SucceedsAgainOnceTheStalledReaderDrains() { - ObservationLog log = new ObservationLog(64); - ObservationCursor cursor = log.CreateCursor(); - cursor.Dispose(); + ObservationLog log = new ObservationLog(4); + ObservationWriter writer = new ObservationWriter(log); + ObservationReader stalled = new ObservationReader(log, "stalled"); - Assert.That(() => cursor.TryRead(out _), Throws.TypeOf()); - } + for (int i = 0; i < 4; i++) + { + writer.Append(new TestEntry(i)); + } - [Test] - public void Cursor_OnEmptyLogReturnsFalseAndTheDefaultEntry() - { - ObservationLog log = new ObservationLog(64); - ObservationCursor cursor = log.CreateCursor(); + Assert.That(() => writer.Append(new TestEntry(4)), Throws.InvalidOperationException); + Assert.That(Drain(stalled), Is.EqualTo(new[] { 0, 1, 2, 3 })); - Assert.That(cursor.TryRead(out TestEntry entry), Is.False); - Assert.That(entry, Is.EqualTo(default(TestEntry))); - } + writer.Append(new TestEntry(5)); - [Test] - public void MaximumCapacity_MustBeAtLeastOne() - { - Assert.That(() => new ObservationLog(0), Throws.TypeOf()); + Assert.That(Drain(stalled), Is.EqualTo(new[] { 5 })); } [Test] - public void ReferenceEntry_ReadsBackTheAppendedInstancesInOrder() + public void DisposingStalledReader_FreesTheLogForTrimming() { - ObservationLog log = new ObservationLog(64); - ObservationCursor cursor = log.CreateCursor(); - TestReferenceEntry first = new TestReferenceEntry(1); - TestReferenceEntry second = new TestReferenceEntry(2); + ObservationLog log = new ObservationLog(8); + ObservationWriter writer = new ObservationWriter(log); + ObservationReader stalled = new ObservationReader(log, "stalled"); + + for (int i = 0; i < 8; i++) + { + writer.Append(new TestEntry(i)); + } - log.Append(first); - log.Append(second); + stalled.Dispose(); - Assert.That(cursor.TryRead(out TestReferenceEntry? read), Is.True); - Assert.That(read, Is.SameAs(first)); - Assert.That(cursor.TryRead(out read), Is.True); - Assert.That(read, Is.SameAs(second)); + Assert.That(() => writer.Append(new TestEntry(8)), Throws.Nothing); } [Test] - public void ReferenceEntry_OnEmptyLogReturnsFalseAndNull() + public void DisposingOneReader_KeepsTheEntriesAnotherStillNeeds() { - ObservationLog log = new ObservationLog(64); - ObservationCursor cursor = log.CreateCursor(); + ObservationLog log = new ObservationLog(64); + ObservationWriter writer = new ObservationWriter(log); + ObservationReader leaving = new ObservationReader(log, "leaving"); + ObservationReader staying = new ObservationReader(log, "staying"); - Assert.That(cursor.TryRead(out TestReferenceEntry? entry), Is.False); - Assert.That(entry, Is.Null); + writer.Append(new TestEntry(1)); + writer.Append(new TestEntry(2)); + leaving.Dispose(); + + Assert.That(Drain(staying), Is.EqualTo(new[] { 1, 2 })); } [Test] - public void Buffer_GrowsWhileWrappedPreservingOrder() + public void DisposingAReaderTwice_IsHarmless() { - ObservationLog log = new ObservationLog(1024); - ObservationCursor cursor = log.CreateCursor(); - - // Drain ten entries first so the head sits mid-buffer and the retained entries wrap around the end, - // which is what makes growth copy them in two segments. - for (int i = 0; i < 10; i++) - { - log.Append(new TestEntry(i)); - cursor.TryRead(out _); - } - - int[] expected = Enumerable.Range(100, 20).ToArray(); - foreach (int value in expected) - { - log.Append(new TestEntry(value)); - } + ObservationLog log = new ObservationLog(64); + ObservationReader reader = new ObservationReader(log); + reader.Dispose(); - Assert.That(Drain(cursor), Is.EqualTo(expected)); + Assert.That(() => reader.Dispose(), Throws.Nothing); } [Test] - public void DisposingOneCursor_KeepsTheEntriesAnotherStillNeeds() + public void DisposedReader_ThrowsOnRead() { ObservationLog log = new ObservationLog(64); - ObservationCursor leaving = log.CreateCursor("leaving"); - ObservationCursor staying = log.CreateCursor("staying"); + ObservationReader reader = new ObservationReader(log); + reader.Dispose(); - log.Append(new TestEntry(1)); - log.Append(new TestEntry(2)); - leaving.Dispose(); - - Assert.That(Drain(staying), Is.EqualTo(new[] { 1, 2 })); + Assert.That(() => reader.TryRead(out _), Throws.TypeOf()); } [Test] - public void DisposingACursorTwice_IsHarmless() + public void Reader_OnEmptyLogReturnsFalseAndTheDefaultEntry() { ObservationLog log = new ObservationLog(64); - ObservationCursor cursor = log.CreateCursor(); - cursor.Dispose(); + ObservationReader reader = new ObservationReader(log); - Assert.That(() => cursor.Dispose(), Throws.Nothing); + Assert.That(reader.TryRead(out TestEntry entry), Is.False); + Assert.That(entry, Is.EqualTo(default(TestEntry))); } [Test] - public void Append_NamesEveryCursorTiedAtTheBack() + public void ReferenceEntry_ReadsBackTheAppendedInstancesInOrder() { - ObservationLog log = new ObservationLog(4); - log.CreateCursor("presenter"); - log.CreateCursor("audio"); + ObservationLog log = new ObservationLog(64); + ObservationWriter writer = new ObservationWriter(log); + ObservationReader reader = new ObservationReader(log); + TestReferenceEntry first = new TestReferenceEntry(1); + TestReferenceEntry second = new TestReferenceEntry(2); - for (int i = 0; i < 4; i++) - { - log.Append(new TestEntry(i)); - } + writer.Append(first); + writer.Append(second); - Assert.That(() => log.Append(new TestEntry(4)), - Throws.InvalidOperationException.With.Message.Contains("presenter") - .And.Message.Contains("audio")); + Assert.That(reader.TryRead(out TestReferenceEntry? read), Is.True); + Assert.That(read, Is.SameAs(first)); + Assert.That(reader.TryRead(out read), Is.True); + Assert.That(read, Is.SameAs(second)); } [Test] - public void Append_SucceedsAgainOnceTheStalledCursorDrains() + public void ReferenceEntry_OnEmptyLogReturnsFalseAndNull() { - ObservationLog log = new ObservationLog(4); - ObservationCursor stalled = log.CreateCursor("stalled"); - - for (int i = 0; i < 4; i++) - { - log.Append(new TestEntry(i)); - } - - Assert.That(() => log.Append(new TestEntry(4)), Throws.InvalidOperationException); - Assert.That(Drain(stalled), Is.EqualTo(new[] { 0, 1, 2, 3 })); - - log.Append(new TestEntry(5)); + ObservationLog log = new ObservationLog(64); + ObservationReader reader = new ObservationReader(log); - Assert.That(Drain(stalled), Is.EqualTo(new[] { 5 })); + Assert.That(reader.TryRead(out TestReferenceEntry? entry), Is.False); + Assert.That(entry, Is.Null); } [Test] - public void Trimming_ReleasesAReferenceEntryOnceEveryCursorHasPassedIt() + public void Trimming_ReleasesAReferenceEntryOnceEveryReaderHasPassedIt() { ObservationLog log = new ObservationLog(64); - ObservationCursor cursor = log.CreateCursor(); + ObservationReader reader = new ObservationReader(log); - WeakReference reference = AppendAndDrainOne(log, cursor); + WeakReference reference = AppendAndDrainOne(log, reader); GC.Collect(); GC.WaitForPendingFinalizers(); GC.Collect(); @@ -337,18 +353,18 @@ public void Trimming_ReleasesAReferenceEntryOnceEveryCursorHasPassedIt() // The only strong reference lives in this frame, so returning drops it and leaves the log's slot as the // one thing that could still keep the entry alive. [MethodImpl(MethodImplOptions.NoInlining)] - private static WeakReference AppendAndDrainOne(ObservationLog log, ObservationCursor cursor) + private static WeakReference AppendAndDrainOne(ObservationLog log, ObservationReader reader) { TestReferenceEntry entry = new TestReferenceEntry(1); - log.Append(entry); - cursor.TryRead(out TestReferenceEntry? _); + new ObservationWriter(log).Append(entry); + reader.TryRead(out TestReferenceEntry? _); return new WeakReference(entry); } - private static int[] Drain(ObservationCursor cursor) + private static int[] Drain(ObservationReader reader) { List values = new(); - while (cursor.TryRead(out TestEntry entry)) + while (reader.TryRead(out TestEntry entry)) { values.Add(entry.Value); } From f8699a6f46a6cc0320c5c097bb601b565a6c2852 Mon Sep 17 00:00:00 2001 From: botoddly <250804054+botoddly@users.noreply.github.com> Date: Thu, 10 Sep 2026 11:01:27 +0200 Subject: [PATCH 5/5] Wrap the observation log index by subtraction and state its single-threaded contract --- docs/observations.md | 3 +++ src/Pixely.Observations/ObservationLog.cs | 8 ++++++-- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/docs/observations.md b/docs/observations.md index a25a4342..7aa31b04 100644 --- a/docs/observations.md +++ b/docs/observations.md @@ -19,6 +19,9 @@ Three types, each with one job: So a rule holding a writer has no way to drain the log, and a reader has no way to record an observation no rule produced. +The log belongs to one frame loop and is not thread safe. Appending, reading and constructing a reader all +happen on the same thread. + ## The entry type One log carries one entry type. To carry several kinds of entry in one order, make `TEntry` a tagged type; the diff --git a/src/Pixely.Observations/ObservationLog.cs b/src/Pixely.Observations/ObservationLog.cs index db5be36f..bcf100f1 100644 --- a/src/Pixely.Observations/ObservationLog.cs +++ b/src/Pixely.Observations/ObservationLog.cs @@ -10,7 +10,8 @@ namespace Pixely.Observations; /// /// /// The log is storage and nothing else: it is reached through an or an -/// , so neither role can do the other's job. +/// , so neither role can do the other's job. It belongs to one frame +/// loop and is not thread safe: appending, reading and constructing a reader must all happen on the same thread. /// /// /// The single entry type of this log; the log never looks inside it. Carry several kinds of entry in one log @@ -142,9 +143,12 @@ private void EnsureCapacity(int requiredCapacity) _head = 0; } + // Wrapping by subtraction rather than by modulo, so a buffer whose head plus offset would pass int range + // still indexes correctly, and the hot path costs a compare instead of a division. private int PhysicalIndex(int offset) { - return (_head + offset) % _entries.Length; + int untilWrap = _entries.Length - _head; + return offset < untilWrap ? _head + offset : offset - untilWrap; } private string DescribeOverflow()