diff --git a/Directory.Build.props b/Directory.Build.props index e8bad6c0..4d0f86c5 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -14,7 +14,7 @@ MIT © Meta Vens and contributors true - 1.0.25 + 1.0.26 true diff --git a/src/GameCult.Mesh/docs/reliability-migration-plan.md b/src/GameCult.Mesh/docs/reliability-migration-plan.md index 650ab7fd..d0594d44 100644 --- a/src/GameCult.Mesh/docs/reliability-migration-plan.md +++ b/src/GameCult.Mesh/docs/reliability-migration-plan.md @@ -964,6 +964,37 @@ body path when remote demand exists; Phase 6 must replace that expiry-sensitive traffic with a stream/datagram body plane rather than enlarging snapshot or CDN timeouts. +### RUDP session mutation body map + +**Owner:** each socket transport binding serializes access to its own RUDP +reliability session. The server owns one serialization gate per peer; a client +connection owns one gate for its session. + +**Inputs:** concurrent typed publication, command and receipt sends, inbound +ACK/data processing, handshake state, timeout checks, and resend polling. + +**Outputs:** one coherent packet sequence and fragment stream per peer, with +reliable-packet bookkeeping updated in the same critical section as packet +creation and transmission. + +**Derived state:** schema messages, database subscriptions, and Eve reactive +variables are consumers of the reliable control plane. They do not own packet +ordering and must not serialize their own publication loops. + +**Forbidden writers:** daemon schedulers, renderers, and provider sessions +cannot mutate `CultNetRudpSession` directly or add retries, replacement +snapshots, and artificial publication delays to compensate for concurrent +transport calls. + +**Shared paths:** application send, receive/ACK, resend, connect, disconnect, +ping, and timeout inspection all enter through the same per-session gate. +Fragments belonging to one logical send are emitted before another thread may +allocate sequence numbers for that peer. + +**Cut line:** all socket-backed `CultNetRudpSession` mutation is behind the +transport-owned gate. Product code no longer has to choose between concurrent +simulation publication and reliable delivery. + Reliable control messages larger than the RUDP receive mask explicitly acknowledge the packet just received. A retransmit older than the rolling 32-packet summary can therefore always leave the sender queue; the rolling mask diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index de4b15e2..f50c4985 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -1542,6 +1542,7 @@ namespace GameCult.Networking private const int WireBurstPackets = 32; private readonly Socket _socket; private readonly CultNetRudpSession _session; + private readonly object _sessionGate = new object(); private readonly CultNetRudpSocketMode _mode; private readonly int? _maxFragmentBytes; private readonly CultNetTransportStats _stats = new CultNetTransportStats(); @@ -1594,7 +1595,7 @@ namespace GameCult.Networking /// /// Gets whether the RUDP handshake has completed. /// - public bool Connected => _session.Connected; + public bool Connected { get { lock (_sessionGate) return _session.Connected; } } /// /// Gets a snapshot of the current transfer counters. @@ -1630,7 +1631,8 @@ namespace GameCult.Networking throw new InvalidOperationException("Only a client RUDP socket transport can initiate connect."); } - SendPacket(_session.CreateConnect(NowMs(), payload ?? Array.Empty())); + lock (_sessionGate) + SendPacket(_session.CreateConnect(NowMs(), payload ?? Array.Empty())); } /// @@ -1691,7 +1693,8 @@ namespace GameCult.Networking /// public void Send(string channelId, byte[] payload) { - SendPackets(_session.SendMany(channelId, payload, ChannelSendOptions(channelId), _maxFragmentBytes)); + lock (_sessionGate) + SendPackets(_session.SendMany(channelId, payload, ChannelSendOptions(channelId), _maxFragmentBytes)); _stats.FramesSent++; } @@ -1844,7 +1847,8 @@ namespace GameCult.Networking /// public void Disconnect(byte[]? reason = null) { - SendPacket(_session.CreateDisconnect(reason ?? Array.Empty())); + lock (_sessionGate) + SendPacket(_session.CreateDisconnect(reason ?? Array.Empty())); } /// @@ -1852,7 +1856,8 @@ namespace GameCult.Networking /// public void Ping(byte[]? payload = null) { - SendPacket(_session.CreatePing(payload ?? Array.Empty())); + lock (_sessionGate) + SendPacket(_session.CreatePing(payload ?? Array.Empty())); } /// @@ -1860,7 +1865,8 @@ namespace GameCult.Networking /// public bool CheckTimeout(long timeoutMs) { - return _session.CheckTimeout(NowMs(), timeoutMs); + lock (_sessionGate) + return _session.CheckTimeout(NowMs(), timeoutMs); } /// @@ -1929,14 +1935,21 @@ namespace GameCult.Networking TracePacket("rx", packet, received, remote); if (_mode == CultNetRudpSocketMode.Server && packet.PacketType == CultNetRudpPacketType.Connect) { - SendPacket(_session.AcceptConnect(packet, NowMs())); + lock (_sessionGate) + SendPacket(_session.AcceptConnect(packet, NowMs())); return true; } - var result = _session.Receive(packet, NowMs()); - if (result.Reply != null) + CultNetRudpReceiveResult result; + CultNetRudpPacket? acknowledgement = null; + lock (_sessionGate) { - SendPacket(result.Reply); + result = _session.Receive(packet, NowMs()); + if (result.Reply != null) + SendPacket(result.Reply); + if (packet.PacketType == CultNetRudpPacketType.Accept || + packet.PacketType == CultNetRudpPacketType.Data) + acknowledgement = _session.CreateAck(packet.Sequence); } if (result.Pong) { @@ -1959,11 +1972,8 @@ namespace GameCult.Networking } delivered = _deliveredFrames.Count > 0 ? _deliveredFrames.Dequeue() : null; - if (packet.PacketType == CultNetRudpPacketType.Accept || - packet.PacketType == CultNetRudpPacketType.Data) - { - SendPacket(_session.CreateAck(packet.Sequence)); - } + if (acknowledgement != null) + SendPacket(acknowledgement); return true; } @@ -1975,7 +1985,8 @@ namespace GameCult.Networking { if (_socket.Poll(0, SelectMode.SelectRead)) return; - SendPackets(_session.DueResends(NowMs())); + lock (_sessionGate) + SendPackets(_session.DueResends(NowMs())); } private void SendPackets(IReadOnlyList packets) @@ -2102,6 +2113,7 @@ namespace GameCult.Networking } internal CultNetRudpSession Session { get; } + internal object SessionGate { get; } = new object(); /// /// Gets the UDP endpoint for this peer. @@ -2110,11 +2122,11 @@ namespace GameCult.Networking /// /// Gets whether the peer RUDP handshake has completed. /// - public bool Connected => Session.Connected; + public bool Connected { get { lock (SessionGate) return Session.Connected; } } /// /// Gets the number of reliable packets still awaiting acknowledgement. /// - public int PendingReliablePacketCount => Session.PendingReliableSequences.Count; + public int PendingReliablePacketCount { get { lock (SessionGate) return Session.PendingReliableSequences.Count; } } /// /// Gets the last transport-level remote disconnect reason, if one was received. /// @@ -2215,7 +2227,8 @@ namespace GameCult.Networking public void Send(CultNetRudpSocketServerPeer peer, string channelId, byte[] payload) { if (peer == null) throw new ArgumentNullException(nameof(peer)); - SendPackets(peer, peer.Session.SendMany(channelId, payload, ChannelSendOptions(channelId), _maxFragmentBytes)); + lock (peer.SessionGate) + SendPackets(peer, peer.Session.SendMany(channelId, payload, ChannelSendOptions(channelId), _maxFragmentBytes)); _stats.FramesSent++; } @@ -2326,7 +2339,8 @@ namespace GameCult.Networking MaxPendingReliablePackets = _maxPendingReliablePackets })); _peers[peerKey] = peer; - SendPacket(peer.RemoteEndPoint, peer.Session.AcceptConnect(packet, NowMs(), _acceptPayload)); + lock (peer.SessionGate) + SendPacket(peer.RemoteEndPoint, peer.Session.AcceptConnect(packet, NowMs(), _acceptPayload)); return true; } @@ -2335,10 +2349,15 @@ namespace GameCult.Networking return true; } - var result = existingPeer.Session.Receive(packet, NowMs()); - if (result.Reply != null) + CultNetRudpReceiveResult result; + CultNetRudpPacket? acknowledgement = null; + lock (existingPeer.SessionGate) { - SendPacket(existingPeer.RemoteEndPoint, result.Reply); + result = existingPeer.Session.Receive(packet, NowMs()); + if (result.Reply != null) + SendPacket(existingPeer.RemoteEndPoint, result.Reply); + if (packet.PacketType == CultNetRudpPacketType.Data) + acknowledgement = existingPeer.Session.CreateAck(packet.Sequence); } if (result.Disconnected) { @@ -2363,10 +2382,8 @@ namespace GameCult.Networking } delivered = _deliveredFrames.Count > 0 ? _deliveredFrames.Dequeue() : null; - if (packet.PacketType == CultNetRudpPacketType.Data) - { - SendPacket(existingPeer.RemoteEndPoint, existingPeer.Session.CreateAck(packet.Sequence)); - } + if (acknowledgement != null) + SendPacket(existingPeer.RemoteEndPoint, acknowledgement); return true; } @@ -2380,7 +2397,8 @@ namespace GameCult.Networking return; foreach (var peer in _peers.Values.ToArray()) { - SendPackets(peer, peer.Session.DueResends(NowMs())); + lock (peer.SessionGate) + SendPackets(peer, peer.Session.DueResends(NowMs())); } } diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 98e8438c..ed03fc57 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1864,6 +1864,84 @@ namespace GameCult.Networking.Tests "A lossless large response must not remain trapped in the reliable resend queue."); } + [Test] + public async Task RudpCultNetSchemaServer_SerializesConcurrentFragmentedSendsPerPeer() + { + const int publisherCount = 4; + const int messagesPerPublisher = 4; + const int expectedMessages = publisherCount * messagesPerPublisher; + using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions + { + RuntimeId = "csharp-rudp-concurrent-publish-host", + Socket = BindUdpSocket(), + MaxFragmentBytes = 512, + MaxPendingReliablePackets = 1024 + }); + var receivedIds = new ConcurrentDictionary(); + var receivedAll = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var serverFailure = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + using var cancellation = new CancellationTokenSource(); + var serverThread = new Thread(() => + { + try + { + while (!cancellation.IsCancellationRequested) + { + _ = server.PollOnceAsync().GetAwaiter().GetResult(); + Thread.Sleep(1); + } + } + catch (Exception error) { serverFailure.TrySetResult(error); } + }) { IsBackground = true }; + serverThread.Start(); + + using var client = CultNetSchemaClients.CreateRudp( + runtimeId: "csharp-rudp-concurrent-publish-client", + maxFragmentBytes: 512); + client.OnCultNet(response => + { + if (receivedIds.TryAdd(response.MessageId, 0) && receivedIds.Count == expectedMessages) + receivedAll.TrySetResult(true); + }); + client.Connect("127.0.0.1", server.LocalEndPoint.Port); + await WaitUntilAsync(() => client.Connected && server.Peers.Count == 1, TimeSpan.FromSeconds(2)); + var peer = server.Peers.Single(); + using var sendBarrier = new Barrier(publisherCount); + + await Task.WhenAll(Enumerable.Range(0, publisherCount).Select(publisher => Task.Run(() => + { + sendBarrier.SignalAndWait(); + for (var message = 0; message < messagesPerPublisher; message++) + { + var messageId = $"publisher-{publisher}-message-{message}"; + server.SendCultNet(peer, new CultNetSnapshotResponseRawMessage + { + MessageId = messageId, + Documents = + [ + new CultNetRawDocumentRecord + { + SchemaId = "test.concurrent.publish.v1", + RecordKey = $"test:concurrent:{messageId}", + PayloadEncoding = "messagepack", + Payload = Enumerable.Range(0, 4 * 1024) + .Select(index => (byte)((index + publisher + message) % 251)).ToArray() + } + ] + }); + } + }))); + + var completed = await Task.WhenAny(receivedAll.Task, serverFailure.Task, Task.Delay(5000)); + if (completed == serverFailure.Task) throw await serverFailure.Task; + if (completed != receivedAll.Task) + Assert.Fail($"Concurrent publish timed out after receiving {receivedIds.Count} of {expectedMessages} logical messages."); + cancellation.Cancel(); + + Assert.That(receivedIds.Count, Is.EqualTo(expectedMessages)); + } + [Test] public void RudpSocketTransport_ErgonomicHelpersCarryNamedChannels() { diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll index 9a808a27..2120ac47 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb index e26bd528..58f5f6b1 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll index ae6f1f1b..99e54017 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb index 225e7e75..d28288f9 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll index 32b160b4..3d048eab 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb index eeed669e..522637a5 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll index 2af790f0..e5b713f6 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb index 76c376bf..e2b26038 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll index 54615bd3..74644fa7 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb index 108cad92..3f6b774f 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb differ diff --git a/unity/org.gamecult.cultlib/package.json b/unity/org.gamecult.cultlib/package.json index 8b570e95..3edc87bc 100644 --- a/unity/org.gamecult.cultlib/package.json +++ b/unity/org.gamecult.cultlib/package.json @@ -1,7 +1,7 @@ { "name": "org.gamecult.cultlib", "displayName": "CultLib", - "version": "1.0.25", + "version": "1.0.26", "unity": "2021.3", "description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.", "license": "MIT",