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

Acknowledge reliable RUDP windows cumulatively

Browse source
This commit is contained in:
Meta Vens 2026-08-23 19:00:33 +02:00
commit fda51fad50
10 changed files with 59 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

2
packages/cultcache-py/src/cultmesh_py/server.py
View file

@ -268,7 +268,7 @@ class CultMeshLocalServer:
msgpack.packb(response, use_bin_type=True),
)
if packet.reliable or packet.packet_type == CultNetRudpPacketType.DATA or result.delivered:
self._send_rudp_packet(rudp_socket, remote_addr, peer.session.create_ack_for(packet.sequence))
self._send_rudp_packet(rudp_socket, remote_addr, peer.session.create_ack())
def _poll_rudp_resends(
self,

2
packages/cultcache-py/src/cultnet_py/interop_peer.py
View file

@ -320,7 +320,7 @@ def rudp_loop(
for response in handle_server_message(state, message, peer.subscriptions):
send_rudp_schema_frame(sock, remote, peer.session, response)
if packet.reliable or packet.packet_type == CultNetRudpPacketType.DATA or result.delivered:
send_rudp_packet(sock, remote, peer.session.create_ack_for(packet.sequence))
send_rudp_packet(sock, remote, peer.session.create_ack())
except Exception as error:
sys.stderr.write(json.dumps({
"event": "rudpMessageError",

13
packages/cultcache-py/src/cultnet_py/transport.py
View file

@ -599,7 +599,16 @@ class CultNetRudpSession:
)
def create_ack(self) -> CultNetRudpPacket:
return self._create_packet(CultNetRudpPacketType.ACK, "control", b"")
ack, ack_mask = self._ack_state()
return CultNetRudpPacket(
packet_type=CultNetRudpPacketType.ACK,
connection_id=self.connection_id,
sequence=0,
ack=ack,
ack_mask=ack_mask,
channel_id="control",
payload=b"",
)
def create_ack_for(self, sequence: int) -> CultNetRudpPacket:
return CultNetRudpPacket(
@ -955,7 +964,7 @@ class CultNetRudpSocketTransportConnection:
frame = self._delivered_frames.popleft() if self._delivered_frames else None
if packet.reliable or packet.packet_type == CultNetRudpPacketType.ACCEPT or frame is not None:
self._send_packet(self.session.create_ack_for(packet.sequence))
self._send_packet(self.session.create_ack())
return frame
def receive(self, timeout_seconds: float | None = None) -> CultNetTransportFrame:

8
packages/cultcache-py/tests/test_cultcache.py
View file

@ -1178,9 +1178,11 @@ class CultCacheTests(unittest.TestCase):
delivered = []
while wire:
packet = wire.pop(0)
delivered.extend(receiver.receive(packet, 2).delivered)
acknowledged = sender.receive(receiver.create_ack_for(packet.sequence), 3)
admitted = len(wire)
for _ in range(admitted):
packet = wire.pop(0)
delivered.extend(receiver.receive(packet, 2).delivered)
acknowledged = sender.receive(receiver.create_ack(), 3)
wire.extend(acknowledged.ready_to_send)
self.assertEqual(sender.outstanding_reliable_packet_count, 0)

24
packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt
View file

@ -2764,7 +2764,17 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) {
)
}
fun createAck(): CultNetRudpPacket = createPacket(CultNetRudpPacketType.Ack, "control", ByteArray(0))
fun createAck(): CultNetRudpPacket {
val (ack, ackMask) = ackState()
return CultNetRudpPacket(
packetType = CultNetRudpPacketType.Ack,
connectionId = connectionId,
sequence = 0,
ack = ack,
ackMask = ackMask,
channelId = "control",
)
}
fun createAckFor(sequence: Long): CultNetRudpPacket = CultNetRudpPacket(
packetType = CultNetRudpPacketType.Ack,
@ -3205,7 +3215,7 @@ class CultNetRudpSocketTransportConnection(
framesReceived += 1
}
val frame = if (delivered.isEmpty()) null else delivered.removeFirst()
if (packet.reliable || packet.packetType == CultNetRudpPacketType.Accept || frame != null) sendPacket(session.createAckFor(packet.sequence))
if (packet.reliable || packet.packetType == CultNetRudpPacketType.Accept || frame != null) sendPacket(session.createAck())
return frame
}
@ -3972,7 +3982,7 @@ private fun startInteropRudpServer(
val result = session.receive(packet, nowMs())
result.reply?.let { sendInteropRudpPacket(socket, remote, it) }
result.readyToSend.forEach { sendInteropRudpPacket(socket, remote, it) }
if (packet.reliable || packet.packetType == CultNetRudpPacketType.Data) sendInteropRudpPacket(socket, remote, session.createAckFor(packet.sequence))
if (packet.reliable || packet.packetType == CultNetRudpPacketType.Data) sendInteropRudpPacket(socket, remote, session.createAck())
for (frame in result.delivered) {
if (frame.channelId != "schema") continue
val sender = InteropRudpSessionSender(socket, remote, session)
@ -5323,9 +5333,11 @@ private fun rudpSessionAdvancesLargeFragmentSetsThroughBoundedReliableWindow() {
val delivered = mutableListOf<CultNetRudpDeliveredFrame>()
while (!wire.isEmpty()) {
val packet = wire.removeFirst()
delivered.addAll(receiver.receive(packet, 2).delivered)
val acknowledged = sender.receive(receiver.createAckFor(packet.sequence), 3)
val admitted = wire.size
repeat(admitted) {
delivered.addAll(receiver.receive(wire.removeFirst(), 2).delivered)
}
val acknowledged = sender.receive(receiver.createAck(), 3)
acknowledged.readyToSend.forEach { wire.addLast(it) }
}
check(sender.outstandingReliablePacketCount == 0)

6
packages/cultnet-rs/src/rudp.rs
View file

@ -1288,7 +1288,7 @@ impl CultNetRudpServerHub {
let reply = peer
.session
.pending_accept_for_resend(now_ms())
.unwrap_or_else(|| peer.session.create_ack_for(packet.sequence));
.unwrap_or_else(|| peer.session.create_ack());
self.send_packet(remote_addr, &reply)?;
return Ok(None);
}
@ -1340,7 +1340,7 @@ impl CultNetRudpServerHub {
let result = peer.session.receive(&packet, now_ms())?;
let context = peer.context.clone();
let ack = if packet.reliable {
Some(peer.session.create_ack_for(packet.sequence))
Some(peer.session.create_ack())
} else {
None
};
@ -1657,7 +1657,7 @@ impl CultNetRudpSocketTransportConnection {
let frame = self.delivered_frames.pop_front();
if packet.reliable || packet.packet_type == CultNetRudpPacketType::Accept || frame.is_some()
{
let ack = self.session.create_ack_for(packet.sequence);
let ack = self.session.create_ack();
self.send_packet(&ack)?;
}
Ok(frame)

12
packages/cultnet-rs/tests/cultnet.rs
View file

@ -1308,10 +1308,14 @@ fn rudp_session_advances_large_fragment_sets_through_a_bounded_reliable_window()
let mut wire = VecDeque::from(initial);
let mut delivered = Vec::new();
while let Some(packet) = wire.pop_front() {
let received = receiver.receive(&packet, 2)?;
delivered.extend(received.delivered);
let acknowledged = sender.receive(&receiver.create_ack_for(packet.sequence), 3)?;
while !wire.is_empty() {
let admitted = wire.len();
for _ in 0..admitted {
let packet = wire.pop_front().expect("admitted packet remains queued");
let received = receiver.receive(&packet, 2)?;
delivered.extend(received.delivered);
}
let acknowledged = sender.receive(&receiver.create_ack(), 3)?;
wire.extend(acknowledged.ready_to_send);
}

9
packages/cultnet-ts/test/cultnet.test.ts
View file

@ -1322,9 +1322,12 @@ test("rudp session advances large fragment sets through a bounded reliable windo
const delivered = [];
while (wire.length > 0) {
const packet = wire.shift()!;
delivered.push(...receiver.receive(packet, 2).delivered);
const acknowledged = sender.receive(receiver.createAckFor(packet.sequence), 3);
const admitted = wire.length;
for (let index = 0; index < admitted; index += 1) {
const packet = wire.shift()!;
delivered.push(...receiver.receive(packet, 2).delivered);
}
const acknowledged = sender.receive(receiver.createAck(), 3);
wire.push(...(acknowledged.readyToSend ?? []));
}

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

@ -2077,7 +2077,7 @@ namespace GameCult.Networking
SendPackets(result.ReadyToSend);
if (packet.PacketType == CultNetRudpPacketType.Accept ||
packet.PacketType == CultNetRudpPacketType.Data)
acknowledgement = _session.CreateAck(packet.Sequence);
acknowledgement = _session.CreateAck();
}
if (result.Pong)
{
@ -2486,7 +2486,7 @@ namespace GameCult.Networking
SendPacket(existingPeer.RemoteEndPoint, result.Reply);
SendPackets(existingPeer, result.ReadyToSend);
if (packet.PacketType == CultNetRudpPacketType.Data)
acknowledgement = existingPeer.Session.CreateAck(packet.Sequence);
acknowledgement = existingPeer.Session.CreateAck();
}
if (result.Disconnected)
{

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

@ -1096,9 +1096,10 @@ namespace GameCult.Networking.Tests
var delivered = new List<CultNetRudpDeliveredFrame>();
while (wire.Count > 0)
{
var packet = wire.Dequeue();
delivered.AddRange(receiver.Receive(packet, 2).Delivered);
var acknowledged = sender.Receive(receiver.CreateAck(packet.Sequence), 3);
var admitted = wire.Count;
for (var index = 0; index < admitted; index++)
delivered.AddRange(receiver.Receive(wire.Dequeue(), 2).Delivered);
var acknowledged = sender.Receive(receiver.CreateAck(), 3);
foreach (var ready in acknowledged.ReadyToSend)
wire.Enqueue(ready);
}