From fbf93f3570008c7c3d9aa2010c9de487523a4129 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 18 Aug 2026 02:37:13 +0200 Subject: [PATCH] Measure and bound reactive document cost --- packages/cultcache-py/src/cultmesh_py/node.py | 14 +- .../measure-reactive-document-performance.ps1 | 28 +++ src/GameCult.Mesh/CultMeshPrimitives.cs | 21 +- .../GameCult.Mesh.PerformanceProbe.csproj | 11 + .../GameCult.Mesh.PerformanceProbe/Program.cs | 211 ++++++++++++++++++ 5 files changed, 273 insertions(+), 12 deletions(-) create mode 100644 scripts/measure-reactive-document-performance.ps1 create mode 100644 tools/GameCult.Mesh.PerformanceProbe/GameCult.Mesh.PerformanceProbe.csproj create mode 100644 tools/GameCult.Mesh.PerformanceProbe/Program.cs diff --git a/packages/cultcache-py/src/cultmesh_py/node.py b/packages/cultcache-py/src/cultmesh_py/node.py index fb302c8c..09ac2012 100644 --- a/packages/cultcache-py/src/cultmesh_py/node.py +++ b/packages/cultcache-py/src/cultmesh_py/node.py @@ -1566,11 +1566,13 @@ class CultMeshReactiveDocument: with self._lock: if self._disposed: return - canonical = self._clone(change.value) + replacement = None if self._dirty or self._flushing: - predicted = self._clone(self._current) - delta = self._create_delta(predicted, canonical) + delta = self._create_delta(self._current, change.value) if delta: + canonical = self._clone(change.value) + predicted = self._clone(self._current) + replacement = canonical self._reconciliation_version += 1 self.reconciliation = CultMeshReactiveDocumentReconciliation( canonical=canonical, @@ -1583,7 +1585,11 @@ class CultMeshReactiveDocument: self.reconciliation = None if not self._options.replace_dirty_current_on_canonical_snapshot: return - self._current = canonical + self._current = ( + replacement + if replacement is not None + else self._clone(change.value) + ) self.reconciliation = None def _schedule_flush_locked(self) -> None: diff --git a/scripts/measure-reactive-document-performance.ps1 b/scripts/measure-reactive-document-performance.ps1 new file mode 100644 index 00000000..45598ed0 --- /dev/null +++ b/scripts/measure-reactive-document-performance.ps1 @@ -0,0 +1,28 @@ +param( + [switch] $Quick +) + +$ErrorActionPreference = "Stop" +$root = Split-Path -Parent $PSScriptRoot +$project = Join-Path $root "tools\GameCult.Mesh.PerformanceProbe\GameCult.Mesh.PerformanceProbe.csproj" +$buildArguments = @( + "build", + $project, + "--configuration", "Release", + "--verbosity", "quiet", + "/p:NoWarn=1591%3BCS8632" +) +& dotnet @buildArguments +if ($LASTEXITCODE -ne 0) { + throw "CultMesh reactive document performance probe build failed." +} + +$arguments = @("run", "--project", $project, "--configuration", "Release", "--no-build") +if ($Quick) { + $arguments += @("--", "--quick") +} + +& dotnet @arguments +if ($LASTEXITCODE -ne 0) { + throw "CultMesh reactive document performance probe failed." +} diff --git a/src/GameCult.Mesh/CultMeshPrimitives.cs b/src/GameCult.Mesh/CultMeshPrimitives.cs index d31f8397..d136595f 100644 --- a/src/GameCult.Mesh/CultMeshPrimitives.cs +++ b/src/GameCult.Mesh/CultMeshPrimitives.cs @@ -1426,6 +1426,8 @@ namespace GameCult.Mesh public sealed class CultMeshReactiveDocument : IDisposable where TDocument : class { + private static readonly CultMeshComparableMember[] ComparableMembers = + GetComparableMembers(typeof(TDocument)).ToArray(); private readonly CultMeshDocumentWriter _writer; private readonly CultMeshDocumentHandle _document; private readonly CultMeshReactiveDocumentOptions _options; @@ -1604,30 +1606,33 @@ namespace GameCult.Mesh if (_disposed) return; - var nextCanonical = CultMeshDocumentHandle.CloneDocument(canonical); + TDocument? replacement = null; if (_dirty || _flushing) { - var predicted = CultMeshDocumentHandle.CloneDocument(_current); - var delta = CreateReconciliationDelta(predicted, nextCanonical); + var delta = CreateReconciliationDelta(_current, canonical); if (delta.Count == 0) { Reconciliation = null; + if (!_options.ReplaceDirtyCurrentOnCanonicalSnapshot) + return; } else { + var nextCanonical = CultMeshDocumentHandle.CloneDocument(canonical); + var predicted = CultMeshDocumentHandle.CloneDocument(_current); + replacement = nextCanonical; Reconciliation = new CultMeshReactiveDocumentReconciliation( nextCanonical, predicted, delta, ++_reconciliationVersion, DateTimeOffset.UtcNow); + if (!_options.ReplaceDirtyCurrentOnCanonicalSnapshot) + return; } - - if (!_options.ReplaceDirtyCurrentOnCanonicalSnapshot) - return; } - _current = nextCanonical; + _current = replacement ?? CultMeshDocumentHandle.CloneDocument(canonical); Reconciliation = null; } } @@ -1689,7 +1694,7 @@ namespace GameCult.Mesh TDocument canonical) { var delta = new Dictionary(StringComparer.Ordinal); - foreach (var member in GetComparableMembers(typeof(TDocument))) + foreach (var member in ComparableMembers) { var predictedValue = member.GetValue(predicted); var canonicalValue = member.GetValue(canonical); diff --git a/tools/GameCult.Mesh.PerformanceProbe/GameCult.Mesh.PerformanceProbe.csproj b/tools/GameCult.Mesh.PerformanceProbe/GameCult.Mesh.PerformanceProbe.csproj new file mode 100644 index 00000000..4a073fe7 --- /dev/null +++ b/tools/GameCult.Mesh.PerformanceProbe/GameCult.Mesh.PerformanceProbe.csproj @@ -0,0 +1,11 @@ + + + Exe + net10.0 + enable + enable + + + + + diff --git a/tools/GameCult.Mesh.PerformanceProbe/Program.cs b/tools/GameCult.Mesh.PerformanceProbe/Program.cs new file mode 100644 index 00000000..948d8e90 --- /dev/null +++ b/tools/GameCult.Mesh.PerformanceProbe/Program.cs @@ -0,0 +1,211 @@ +using System.Diagnostics; +using System.Text.Json; +using GameCult.Caching; +using GameCult.Caching.MessagePack; +using GameCult.Mesh; +using MessagePack; +using R3; + +var quick = args.Contains("--quick", StringComparer.OrdinalIgnoreCase); +var idleDuration = quick ? TimeSpan.FromMilliseconds(250) : TimeSpan.FromSeconds(10); +var activeDuration = quick ? TimeSpan.FromMilliseconds(500) : TimeSpan.FromSeconds(10); +var documentCounts = quick ? new[] { 1, 100, 1_000 } : new[] { 1, 100, 1_000 }; +var results = new List(); + +foreach (var documentCount in documentCounts) + results.Add(await MeasureAsync(documentCount, idleDuration, activeDuration)); + +Console.WriteLine(JsonSerializer.Serialize(new +{ + workload = new + { + documentCounts, + payloadBytes = 16 * 1024, + idleSeconds = idleDuration.TotalSeconds, + activeSeconds = activeDuration.TotalSeconds, + updateRateHz = 60, + changedFraction = 0.01 + }, + results +}, new JsonSerializerOptions { WriteIndented = true })); + +return results.All(result => + result.ActualPublishes == result.ExpectedPublishes && + result.PayloadBytesPublished > 0 && + result.AllocationToPayloadRatio < 10 && + result.P99PublishLatencyMilliseconds < 250) + ? 0 + : 1; + +static async Task MeasureAsync( + int documentCount, + TimeSpan idleDuration, + TimeSpan activeDuration) +{ + var payload = new string('x', 16 * 1024); + var changedDocumentCount = Math.Max(1, documentCount / 100); + var subjects = new Subject[documentCount]; + var reactive = new CultMeshReactiveDocument[documentCount]; + var publishSignals = new TaskCompletionSource?[documentCount]; + var updateStartedAt = new long[documentCount]; + var latencies = new List(); + var measurementClock = Stopwatch.StartNew(); + long payloadBytesPublished = 0; + var publishes = 0; + + for (var index = 0; index < documentCount; index++) + { + var documentIndex = index; + var current = new PerformanceDocument + { + Id = "performance:" + index, + Payload = payload, + Revision = 1 + }; + var subject = new Subject(); + subjects[index] = subject; + var handle = CultMesh.Document( + "performance:" + index, + CultMesh.Verse("performance", "probe"), + _ => Task.FromResult(current), + _ => subject, + value => + { + Interlocked.Add( + ref payloadBytesPublished, + CultDocumentMessagePackSerialization.SerializeUntyped(value, typeof(PerformanceDocument)).Length); + lock (latencies) + { + latencies.Add(Stopwatch.GetElapsedTime(updateStartedAt[documentIndex]).TotalMilliseconds); + } + Interlocked.Increment(ref publishes); + current = value; + subject.OnNext(value); + publishSignals[documentIndex]?.TrySetResult(true); + return Task.CompletedTask; + }); + reactive[index] = await handle.AuthoritativeWriter().ReactiveAsync( + new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.Zero }); + } + + try + { + ForceCollection(); + var process = Process.GetCurrentProcess(); + var idleAllocatedStart = GC.GetTotalAllocatedBytes(precise: true); + var idleCpuStart = process.TotalProcessorTime; + var idleThreadsStart = process.Threads.Count; + await Task.Delay(idleDuration); + process.Refresh(); + var idleAllocatedBytes = GC.GetTotalAllocatedBytes(precise: true) - idleAllocatedStart; + var idleCpu = process.TotalProcessorTime - idleCpuStart; + var idleThreadsEnd = process.Threads.Count; + + ForceCollection(); + var activeAllocatedStart = GC.GetTotalAllocatedBytes(precise: true); + var activeCpuStart = process.TotalProcessorTime; + var activeThreadsStart = process.Threads.Count; + var activeClock = Stopwatch.StartNew(); + var frameInterval = TimeSpan.FromSeconds(1.0 / 60.0); + var frames = 0; + while (activeClock.Elapsed < activeDuration) + { + var completions = new Task[changedDocumentCount]; + for (var index = 0; index < changedDocumentCount; index++) + { + var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + publishSignals[index] = completion; + updateStartedAt[index] = Stopwatch.GetTimestamp(); + reactive[index].Update(document => document.Revision++); + completions[index] = completion.Task; + } + await Task.WhenAll(completions); + frames++; + var nextFrameAt = frameInterval * frames; + var remaining = nextFrameAt - activeClock.Elapsed; + if (remaining > TimeSpan.Zero) await Task.Delay(remaining); + } + + process.Refresh(); + var activeAllocatedBytes = GC.GetTotalAllocatedBytes(precise: true) - activeAllocatedStart; + var activeCpu = process.TotalProcessorTime - activeCpuStart; + var measuredActiveSeconds = activeClock.Elapsed.TotalSeconds; + var activeThreadsEnd = process.Threads.Count; + double[] orderedLatencies; + lock (latencies) orderedLatencies = latencies.OrderBy(value => value).ToArray(); + + return new ProbeResult( + documentCount, + changedDocumentCount, + frames, + frames * changedDocumentCount, + publishes, + payloadBytesPublished, + idleAllocatedBytes, + idleCpu.TotalMilliseconds, + idleThreadsStart, + idleThreadsEnd, + activeAllocatedBytes, + activeCpu.TotalMilliseconds, + activeThreadsStart, + activeThreadsEnd, + activeAllocatedBytes / measuredActiveSeconds, + payloadBytesPublished / measuredActiveSeconds, + (double)activeAllocatedBytes / payloadBytesPublished, + Percentile(orderedLatencies, 0.50), + Percentile(orderedLatencies, 0.95), + Percentile(orderedLatencies, 0.99), + measurementClock.Elapsed.TotalSeconds); + } + finally + { + foreach (var document in reactive) document.Dispose(); + foreach (var subject in subjects) subject.Dispose(); + } +} + +static double Percentile(double[] ordered, double percentile) +{ + if (ordered.Length == 0) return 0; + var index = (int)Math.Ceiling(percentile * ordered.Length) - 1; + return ordered[Math.Clamp(index, 0, ordered.Length - 1)]; +} + +static void ForceCollection() +{ + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); +} + +internal sealed record ProbeResult( + int DocumentCount, + int ChangedDocumentCount, + int Frames, + int ExpectedPublishes, + int ActualPublishes, + long PayloadBytesPublished, + long IdleAllocatedBytes, + double IdleCpuMilliseconds, + int IdleThreadsStart, + int IdleThreadsEnd, + long ActiveAllocatedBytes, + double ActiveCpuMilliseconds, + int ActiveThreadsStart, + int ActiveThreadsEnd, + double ActiveAllocatedBytesPerSecond, + double PublishedPayloadBytesPerSecond, + double AllocationToPayloadRatio, + double P50PublishLatencyMilliseconds, + double P95PublishLatencyMilliseconds, + double P99PublishLatencyMilliseconds, + double ScenarioWallSeconds); + +[CultDocument("gamecult.mesh.performance_probe", "gamecult.mesh.performance_probe.v1")] +[MessagePackObject] +public sealed class PerformanceDocument +{ + [Key(0), CultName] public string Id { get; set; } = string.Empty; + [Key(1)] public string Payload { get; set; } = string.Empty; + [Key(2)] public int Revision { get; set; } +}