diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index 1b3e49ac..06c6cf1a 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -893,6 +893,7 @@ namespace GameCult.Networking private long? _lastReceivedAtMs; private uint? _highestReceivedSequence; private readonly HashSet _receivedSequences = new HashSet(); + private readonly object _pendingReliableGate = new object(); private readonly Dictionary _pendingReliable = new Dictionary(); private readonly Dictionary _orderedNextSequenceByChannel = new Dictionary(StringComparer.Ordinal); private readonly Dictionary> _orderedBuffers = @@ -935,7 +936,16 @@ namespace GameCult.Networking /// /// Gets reliable packet sequences awaiting acknowledgement. /// - public IReadOnlyList PendingReliableSequences => _pendingReliable.Keys.OrderBy(value => value).ToArray(); + public IReadOnlyList PendingReliableSequences + { + get + { + lock (_pendingReliableGate) + { + return _pendingReliable.Keys.OrderBy(value => value).ToArray(); + } + } + } /// /// Creates a reliable ordered connect packet. @@ -1186,12 +1196,15 @@ namespace GameCult.Networking public IReadOnlyList DueResends(long nowMs) { var due = new List(); - foreach (var pending in _pendingReliable.Values) + lock (_pendingReliableGate) { - if (nowMs - pending.LastSentAtMs >= ResendDelayMs) + foreach (var pending in _pendingReliable.Values) { - pending.LastSentAtMs = nowMs; - due.Add(ClonePacket(pending.Packet)); + if (nowMs - pending.LastSentAtMs >= ResendDelayMs) + { + pending.LastSentAtMs = nowMs; + due.Add(ClonePacket(pending.Packet)); + } } } @@ -1243,11 +1256,14 @@ namespace GameCult.Networking private void TrackReliable(CultNetRudpPacket packet, long nowMs) { - _pendingReliable[packet.Sequence] = new PendingReliablePacket + lock (_pendingReliableGate) { - Packet = ClonePacket(packet), - LastSentAtMs = nowMs - }; + _pendingReliable[packet.Sequence] = new PendingReliablePacket + { + Packet = ClonePacket(packet), + LastSentAtMs = nowMs + }; + } } private void EnsureReliableCapacity(int packetCount) @@ -1257,20 +1273,26 @@ namespace GameCult.Networking return; } - if (_pendingReliable.Count + packetCount > _maxPendingReliablePackets.Value) + lock (_pendingReliableGate) { - throw new InvalidOperationException("RUDP reliable send queue is full."); + if (_pendingReliable.Count + packetCount > _maxPendingReliablePackets.Value) + { + throw new InvalidOperationException("RUDP reliable send queue is full."); + } } } private void ApplyAcknowledgements(CultNetRudpPacket packet) { - _pendingReliable.Remove(packet.Ack); - for (var bit = 0; bit < 32; bit++) + lock (_pendingReliableGate) { - if ((packet.AckMask & (1u << bit)) != 0 && packet.Ack > bit) + _pendingReliable.Remove(packet.Ack); + for (var bit = 0; bit < 32; bit++) { - _pendingReliable.Remove(packet.Ack - (uint)bit - 1); + if ((packet.AckMask & (1u << bit)) != 0 && packet.Ack > bit) + { + _pendingReliable.Remove(packet.Ack - (uint)bit - 1); + } } } } diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 07848be5..5c79ab50 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -771,6 +771,51 @@ namespace GameCult.Networking.Tests Assert.That(session.DueResends(250), Is.Empty); } + [Test] + public async Task RudpSession_ResendsAndAcknowledgementsCanRunConcurrently() + { + var session = new CultNetRudpSession(new CultNetRudpSessionOptions + { + ConnectionId = 100, + InitialSequence = 1, + ResendDelayMs = 1 + }); + session.Receive(new CultNetRudpPacket + { + PacketType = CultNetRudpPacketType.Accept, + ConnectionId = 100, + Sequence = 5000, + ChannelId = "control" + }); + var packets = Enumerable.Range(0, 1000) + .Select(index => session.Send("schema", Encoding.UTF8.GetBytes(index.ToString()), + new CultNetRudpSendOptions { Reliable = true, Ordered = true, NowMs = 0 })) + .ToArray(); + + await Task.WhenAll( + Task.Run(() => + { + for (var now = 1; now <= 1000; now++) + _ = session.DueResends(now); + }), + Task.Run(() => + { + foreach (var packet in packets) + { + session.Receive(new CultNetRudpPacket + { + PacketType = CultNetRudpPacketType.Ack, + ConnectionId = 100, + Sequence = 5001, + Ack = packet.Sequence, + ChannelId = "control" + }); + } + })); + + Assert.That(session.PendingReliableSequences, Is.Empty); + } + [Test] public void RudpSession_PingsAndDetectsReceiveTimeout() {