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

Measure and bound reactive document cost

Browse source
This commit is contained in:
Meta Vens 2026-08-18 02:37:13 +02:00
commit fbf93f3570
5 changed files with 273 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

14
packages/cultcache-py/src/cultmesh_py/node.py
View file

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

28
scripts/measure-reactive-document-performance.ps1 Normal file
View file

@ -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."
}

21
src/GameCult.Mesh/CultMeshPrimitives.cs
View file

@ -1426,6 +1426,8 @@ namespace GameCult.Mesh
public sealed class CultMeshReactiveDocument<TDocument> : IDisposable
where TDocument : class
{
private static readonly CultMeshComparableMember[] ComparableMembers =
GetComparableMembers(typeof(TDocument)).ToArray();
private readonly CultMeshDocumentWriter<TDocument> _writer;
private readonly CultMeshDocumentHandle<TDocument> _document;
private readonly CultMeshReactiveDocumentOptions _options;
@ -1604,30 +1606,33 @@ namespace GameCult.Mesh
if (_disposed)
return;
var nextCanonical = CultMeshDocumentHandle<TDocument>.CloneDocument(canonical);
TDocument? replacement = null;
if (_dirty || _flushing)
{
var predicted = CultMeshDocumentHandle<TDocument>.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<TDocument>.CloneDocument(canonical);
var predicted = CultMeshDocumentHandle<TDocument>.CloneDocument(_current);
replacement = nextCanonical;
Reconciliation = new CultMeshReactiveDocumentReconciliation<TDocument>(
nextCanonical,
predicted,
delta,
++_reconciliationVersion,
DateTimeOffset.UtcNow);
if (!_options.ReplaceDirtyCurrentOnCanonicalSnapshot)
return;
}
if (!_options.ReplaceDirtyCurrentOnCanonicalSnapshot)
return;
}
_current = nextCanonical;
_current = replacement ?? CultMeshDocumentHandle<TDocument>.CloneDocument(canonical);
Reconciliation = null;
}
}
@ -1689,7 +1694,7 @@ namespace GameCult.Mesh
TDocument canonical)
{
var delta = new Dictionary<string, object?>(StringComparer.Ordinal);
foreach (var member in GetComparableMembers(typeof(TDocument)))
foreach (var member in ComparableMembers)
{
var predictedValue = member.GetValue(predicted);
var canonicalValue = member.GetValue(canonical);

11
tools/GameCult.Mesh.PerformanceProbe/GameCult.Mesh.PerformanceProbe.csproj Normal file
View file

@ -0,0 +1,11 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net10.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\src\GameCult.Mesh\GameCult.Mesh.csproj" />
</ItemGroup>
</Project>

211
tools/GameCult.Mesh.PerformanceProbe/Program.cs Normal file
View file

@ -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<ProbeResult>();
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<ProbeResult> 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<PerformanceDocument>[documentCount];
var reactive = new CultMeshReactiveDocument<PerformanceDocument>[documentCount];
var publishSignals = new TaskCompletionSource<bool>?[documentCount];
var updateStartedAt = new long[documentCount];
var latencies = new List<double>();
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<PerformanceDocument>();
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<bool>(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; }
}