diff --git a/src/GameCult.Mesh/docs/transport-planes.md b/src/GameCult.Mesh/docs/transport-planes.md index 0a86ccab..f4155dc6 100644 --- a/src/GameCult.Mesh/docs/transport-planes.md +++ b/src/GameCult.Mesh/docs/transport-planes.md @@ -84,3 +84,22 @@ reached 10.7 MiB/s while emitting 18,112,867 wire bytes. CultMesh TCP includes chunk hashing, durable checkpoints, final hashing, and atomic promotion, so this is not a claim of raw protocol parity. It is a repeatable workload comparison and leaves the remaining framework overhead visible. + +The same harness now has a state-only workload. On 2026-07-19, 1,000 latest-only +frames with 16 KiB payloads produced the following loopback result: + +| Plane | Offered rate | Delivered | Highest generation | Settled time | +| --- | ---: | ---: | ---: | ---: | +| CultMesh QUIC latest-only | 23.8 MiB/s | 989 / 1,000 | 999 | 655.4 ms | +| CultNet RUDP latest | 45.6 MiB/s | 373 / 1,000 | 516 | 602.7 ms | + +Offered rate measures publisher-side enqueue/write completion, not successful +delivery. RUDP accepted bytes faster but failed to deliver the terminal +generation after a 250 ms drain; that is a delivery failure, not useful +throughput. QUIC's reliable independent streams reached the terminal generation +while the receiver coalesced eleven superseded frames. Reproduce the workload +with: + +```text +dotnet run --project tools/GameCult.TransportParity -- --state-only --state-bytes 16384 --state-frames 1000 +``` diff --git a/tools/GameCult.TransportParity/GameCult.TransportParity.csproj b/tools/GameCult.TransportParity/GameCult.TransportParity.csproj index 4a073fe7..e4b704da 100644 --- a/tools/GameCult.TransportParity/GameCult.TransportParity.csproj +++ b/tools/GameCult.TransportParity/GameCult.TransportParity.csproj @@ -7,5 +7,6 @@ + diff --git a/tools/GameCult.TransportParity/Program.cs b/tools/GameCult.TransportParity/Program.cs index 2661e9fa..1d724631 100644 --- a/tools/GameCult.TransportParity/Program.cs +++ b/tools/GameCult.TransportParity/Program.cs @@ -1,4 +1,5 @@ using System.Diagnostics; +using System.Buffers.Binary; using System.IO.MemoryMappedFiles; using System.Net; using System.Net.Quic; @@ -9,6 +10,7 @@ using System.Security.Cryptography; using System.Security.Cryptography.X509Certificates; using GameCult.Caching; using GameCult.Mesh; +using GameCult.Mesh.Quic; using GameCult.Networking; var payloadBytes = ReadPayloadBytes(args); @@ -17,17 +19,29 @@ new Random(0xAE7E).NextBytes(payload); var expectedHash = Convert.ToHexString(SHA256.HashData(payload)); Console.WriteLine($"payloadBytes={payloadBytes} quicSupported={QuicListener.IsSupported}"); -Print(MeasureMappedFile(payload)); -var tcp = await MeasureTcpAsync(payload, expectedHash); -Print(tcp); -Print(await MeasureCultMeshTcpAsync(payload)); +if (!args.Contains("--state-only", StringComparer.Ordinal)) +{ + Print(MeasureMappedFile(payload)); + var tcp = await MeasureTcpAsync(payload, expectedHash); + Print(tcp); + Print(await MeasureCultMeshTcpAsync(payload)); + if (QuicListener.IsSupported) + Print(await MeasureQuicAsync(payload, expectedHash)); + else + Console.WriteLine("plane=quic status=unsupported"); + + Print(await MeasureCultMeshRudpAsync(payload)); +} + +var statePayloadBytes = ReadPositiveOption(args, "--state-bytes", 16 * 1024); +var stateFrameCount = ReadPositiveOption(args, "--state-frames", 1_000); +Console.WriteLine($"statePayloadBytes={statePayloadBytes} stateFrames={stateFrameCount}"); if (QuicListener.IsSupported) - Print(await MeasureQuicAsync(payload, expectedHash)); + PrintState(await MeasureCultMeshQuicStateAsync(statePayloadBytes, stateFrameCount)); else - Console.WriteLine("plane=quic status=unsupported"); - -Print(await MeasureCultMeshRudpAsync(payload)); + Console.WriteLine("plane=cultmesh-quic-latest status=unsupported"); +PrintState(await MeasureCultNetRudpLatestStateAsync(statePayloadBytes, stateFrameCount)); static int ReadPayloadBytes(string[] args) { @@ -39,12 +53,29 @@ static int ReadPayloadBytes(string[] args) return bytes; } +static int ReadPositiveOption(string[] args, string option, int defaultValue) +{ + var index = Array.IndexOf(args, option); + if (index < 0) return defaultValue; + if (index + 1 >= args.Length || !int.TryParse(args[index + 1], out var value) || value <= 0) + throw new ArgumentException($"{option} requires a positive 32-bit integer."); + return value; +} + static void Print(Result result) => Console.WriteLine( $"plane={result.Plane} elapsedMs={result.Elapsed.TotalMilliseconds:F1} " + (result.GoodputMiBps.HasValue ? $"goodputMiBps={result.GoodputMiBps.Value:F1}" : "") + (result.WireBytes.HasValue ? $" wireBytes={result.WireBytes.Value}" : string.Empty)); +static void PrintState(StateResult result) => + Console.WriteLine( + $"plane={result.Plane} sendElapsedMs={result.SendElapsed.TotalMilliseconds:F1} " + + $"settledElapsedMs={result.SettledElapsed.TotalMilliseconds:F1} " + + $"offeredMiBps={result.OfferedMiBps:F1} deliveredFrames={result.DeliveredFrames}/{result.SentFrames} " + + $"highestSequence={result.HighestSequence}" + + (result.WireBytes.HasValue ? $" wireBytes={result.WireBytes.Value}" : string.Empty)); + static Result MeasureMappedFile(byte[] payload) { var source = TempPath("mapped-source"); @@ -152,6 +183,143 @@ static async Task MeasureQuicAsync(byte[] payload, string expectedHash) return Result.Create("quic-msquic", payload.LongLength, elapsed.Elapsed); } +static async Task MeasureCultMeshQuicStateAsync(int payloadBytes, int frameCount) +{ + using var certificate = CreateCertificate(); + await using var server = await CultMeshQuicRealtimeServer.ListenAsync( + new CultMeshQuicRealtimeServerOptions + { + ListenEndPoint = new IPEndPoint(IPAddress.Loopback, 0), + ServerCertificate = certificate + }); + var endpoint = $"{CultMeshQuicRealtimeTransportConnector.Scheme}://127.0.0.1:{server.LocalEndPoint.Port}"; + using var discovery = new CultMeshDiscoveryService([new RouteSource(endpoint)]); + using var sessions = new CultMeshSessionManager( + discovery, + Array.Empty(), + Array.Empty(), + [new CultMeshQuicRealtimeTransportConnector(new CultMeshQuicRealtimeConnectorOptions + { + ValidateProviderCertificate = (_, _, _, _) => true + })]); + var session = await sessions.ConnectRealtimeAsync(CultMeshEndpointId.Parse("service:parity.provider")); + var deadline = DateTime.UtcNow.AddSeconds(5); + while (server.ConnectionCount == 0) + { + if (DateTime.UtcNow >= deadline) throw new TimeoutException("QUIC state client did not connect."); + await Task.Delay(1); + } + + var receivedFrames = 0; + var highestSequence = -1L; + using var receiveDeadline = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + var receive = Task.Run(async () => + { + while (true) + { + var frame = await session.ReceiveAsync(receiveDeadline.Token); + receivedFrames++; + highestSequence = Math.Max(highestSequence, frame.Sequence); + if (frame.ProducerEpoch == 1 && frame.Sequence == frameCount - 1) return; + } + }); + var payload = new byte[payloadBytes]; + var elapsed = Stopwatch.StartNew(); + for (var sequence = 0; sequence < frameCount; sequence++) + { + await server.BroadcastAsync(new CultMeshRealtimeFrame + { + ChannelId = "parity.entities", + SchemaId = "parity.entity_soa.v1", + BodyId = "body:parity:entities", + ProducerEpoch = 1, + Sequence = sequence, + Delivery = CultMeshRealtimeDelivery.LatestOnly, + Payload = payload + }); + } + var sendElapsed = elapsed.Elapsed; + await receive; + elapsed.Stop(); + return StateResult.Create( + "cultmesh-quic-latest", + payloadBytes, + frameCount, + receivedFrames, + highestSequence, + sendElapsed, + elapsed.Elapsed); +} + +static async Task MeasureCultNetRudpLatestStateAsync(int payloadBytes, int frameCount) +{ + using var serverSocket = BindUdpSocket(); + using var clientSocket = BindUdpSocket(); + const uint connectionId = 0x50415249; + using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions + { + RuntimeId = "parity-rudp-server", + Socket = serverSocket, + Mode = CultNetRudpSocketMode.Server, + ConnectionId = connectionId, + InitialSequence = 100, + MaxFragmentBytes = 1024 + }); + using var client = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions + { + RuntimeId = "parity-rudp-client", + Socket = clientSocket, + Mode = CultNetRudpSocketMode.Client, + RemoteEndPoint = serverSocket.LocalEndPoint, + ConnectionId = connectionId, + InitialSequence = 1, + MaxFragmentBytes = 1024 + }); + client.Connect("parity"); + var handshakeDeadline = DateTime.UtcNow.AddSeconds(5); + while (!client.Connected || !server.Connected) + { + server.ReceiveOnce(); + client.ReceiveOnce(); + if (DateTime.UtcNow >= handshakeDeadline) throw new TimeoutException("RUDP state peers did not connect."); + } + + var receivedFrames = 0; + var highestSequence = -1L; + using var receiveDeadline = new CancellationTokenSource(); + var receive = Task.Run(() => + { + while (!receiveDeadline.IsCancellationRequested) + { + var frame = server.ReceiveOnce(); + if (frame == null) continue; + receivedFrames++; + highestSequence = Math.Max(highestSequence, BinaryPrimitives.ReadInt64LittleEndian(frame.Payload)); + } + }); + var payload = new byte[Math.Max(payloadBytes, sizeof(long))]; + var elapsed = Stopwatch.StartNew(); + for (var sequence = 0; sequence < frameCount; sequence++) + { + BinaryPrimitives.WriteInt64LittleEndian(payload, sequence); + client.SendLatest(payload); + } + var sendElapsed = elapsed.Elapsed; + await Task.Delay(250); + receiveDeadline.Cancel(); + await receive; + elapsed.Stop(); + return StateResult.Create( + "cultnet-rudp-latest", + payload.Length, + frameCount, + receivedFrames, + highestSequence, + sendElapsed, + elapsed.Elapsed, + client.Stats.BytesSent); +} + static async Task MeasureCultMeshRudpAsync(byte[] payload) { var artifact = CultMesh.PackCdnArtifact( @@ -274,6 +442,14 @@ static X509Certificate2 CreateCertificate() static string TempPath(string plane) => Path.Combine(Path.GetTempPath(), $"cultmesh-parity-{plane}-{Guid.NewGuid():N}.body"); +static Socket BindUdpSocket() +{ + var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp); + socket.Bind(new IPEndPoint(IPAddress.Loopback, 0)); + socket.ReceiveTimeout = 20; + return socket; +} + static void VerifyAndDelete(string path, string expectedHash) { try @@ -317,3 +493,33 @@ internal sealed record Result(string Plane, TimeSpan Elapsed, double? GoodputMiB public static Result CreateOpen(string plane, TimeSpan elapsed) => new(plane, elapsed, null, null); } + +internal sealed record StateResult( + string Plane, + int SentFrames, + int DeliveredFrames, + long HighestSequence, + TimeSpan SendElapsed, + TimeSpan SettledElapsed, + double OfferedMiBps, + long? WireBytes) +{ + public static StateResult Create( + string plane, + int payloadBytes, + int sentFrames, + int deliveredFrames, + long highestSequence, + TimeSpan sendElapsed, + TimeSpan settledElapsed, + long? wireBytes = null) => + new( + plane, + sentFrames, + deliveredFrames, + highestSequence, + sendElapsed, + settledElapsed, + (long)payloadBytes * sentFrames / 1024d / 1024d / sendElapsed.TotalSeconds, + wireBytes); +}