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

Synchronize RUDP resend acknowledgements

Browse source
This commit is contained in:
Meta Vens 2026-07-17 13:16:26 +01:00
commit bb7e06ba1e
2 changed files with 82 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

52
src/GameCult.Networking/CultNetTransport.cs
View file

@ -893,6 +893,7 @@ namespace GameCult.Networking
private long? _lastReceivedAtMs;
private uint? _highestReceivedSequence;
private readonly HashSet<uint> _receivedSequences = new HashSet<uint>();
private readonly object _pendingReliableGate = new object();
private readonly Dictionary<uint, PendingReliablePacket> _pendingReliable = new Dictionary<uint, PendingReliablePacket>();
private readonly Dictionary<string, uint> _orderedNextSequenceByChannel = new Dictionary<string, uint>(StringComparer.Ordinal);
private readonly Dictionary<string, SortedDictionary<uint, PendingOrderedFrame>> _orderedBuffers =
@ -935,7 +936,16 @@ namespace GameCult.Networking
/// <summary>
/// Gets reliable packet sequences awaiting acknowledgement.
/// </summary>
public IReadOnlyList<uint> PendingReliableSequences => _pendingReliable.Keys.OrderBy(value => value).ToArray();
public IReadOnlyList<uint> PendingReliableSequences
{
get
{
lock (_pendingReliableGate)
{
return _pendingReliable.Keys.OrderBy(value => value).ToArray();
}
}
}
/// <summary>
/// Creates a reliable ordered connect packet.
@ -1186,12 +1196,15 @@ namespace GameCult.Networking
public IReadOnlyList<CultNetRudpPacket> DueResends(long nowMs)
{
var due = new List<CultNetRudpPacket>();
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);
}
}
}
}

45
tests/GameCult.Networking.Tests/NetworkingTests.cs
View file

@ -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()
{