diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 69367eb5..786f72bb 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -180,16 +180,18 @@ Current progress: dropped data packet in both directions between TypeScript and C#/Kotlin/Python/Rust. The drops happen below the transport connection through a UDP bridge, so the recovery is RUDP resend/ack behavior, not application-level retry. -- TypeScript and Python now have the first RUDP fragmentation implementation: - the session can split oversized reliable ordered payloads into fragment - packets, reassemble them before delivery, and the socket transport can carry - fragmented `schema` frames when `maxFragmentBytes` is configured. C#, Rust, - and Kotlin still need the same implementation before fragmentation can claim - cross-runtime parity. +- TypeScript, C#, Rust, Python, and Kotlin now share the first RUDP + fragmentation implementation: the session can split oversized reliable + ordered payloads into fragment packets, reassemble them before delivery, and + the socket transport can carry fragmented `schema` frames when + `maxFragmentBytes` is configured. +- The TypeScript interop harness now proves fragmented reliable ordered + `schema` frames in both request and response direction between TypeScript and + C#/Kotlin/Python/Rust one-shot UDP peers. - The remaining parity work is to deepen Python's server-side use of its port and move each runtime's existing TCP/LiteNetLib/WebSocket bodies behind those ports, then broaden RUDP interop into the full loss, reordering, - fragmentation, reconnect, and schema-message matrix. + reconnect, backpressure, and schema-message matrix. ## RUDP Packet Contract V0 diff --git a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt index e482afdd..de7e8e64 100644 --- a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt +++ b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt @@ -171,18 +171,28 @@ data class CultNetRudpSendOptions( ) private data class PendingReliablePacket(var packet: CultNetRudpPacket, var lastSentAtMs: Long) +private data class PendingOrderedFrame(val frame: CultNetRudpDeliveredFrame, val nextSequence: Long) +private data class FragmentBuffer( + val channelId: String, + val ordered: Boolean, + val fragmentCount: Int, + val payloads: MutableMap = linkedMapOf(), + val sequences: MutableMap = linkedMapOf(), +) class CultNetRudpSession(options: CultNetRudpSessionOptions) { val connectionId: Long = uint32(options.connectionId, "connectionId") val resendDelayMs: Long = options.resendDelayMs private var nextSequence = uint32(options.initialSequence, "initialSequence") + private var nextFragmentId = 1 var connected: Boolean = false private set private var highestReceivedSequence: Long? = null private val receivedSequences = linkedSetOf() private val pendingReliable = linkedMapOf() private val orderedNextSequenceByChannel = linkedMapOf() - private val orderedBuffers = linkedMapOf>() + private val orderedBuffers = linkedMapOf>() + private val fragmentBuffers = linkedMapOf, FragmentBuffer>() val pendingReliableSequences: List get() = pendingReliable.keys.sorted() @@ -204,10 +214,41 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { } fun send(channelId: String, payload: ByteArray, options: CultNetRudpSendOptions = CultNetRudpSendOptions()): CultNetRudpPacket { + return sendMany(channelId, payload, options).first() + } + + fun sendMany( + channelId: String, + payload: ByteArray, + options: CultNetRudpSendOptions = CultNetRudpSendOptions(), + maxFragmentBytes: Int? = null, + ): List { if (!connected) throw IOException("Cannot send RUDP data before the session is connected") + if (maxFragmentBytes != null && maxFragmentBytes <= 0) throw IOException("RUDP maxFragmentBytes must be greater than zero") + if (maxFragmentBytes != null && payload.size > maxFragmentBytes) { + val fragmentCount = (payload.size + maxFragmentBytes - 1) / maxFragmentBytes + if (fragmentCount > 0xffff) throw IOException("RUDP payload requires more than 65535 fragments") + val fragmentId = allocateFragmentId() + return (0 until fragmentCount).map { index -> + val start = index * maxFragmentBytes + val packet = createPacket( + channelId = channelId, + packetType = CultNetRudpPacketType.Data, + payload = payload.copyOfRange(start, minOf(start + maxFragmentBytes, payload.size)), + reliable = options.reliable, + ordered = options.ordered, + sequenced = options.sequenced, + fragmentId = fragmentId, + fragmentIndex = index, + fragmentCount = fragmentCount, + ) + if (packet.reliable) trackReliable(packet, options.nowMs) + packet + } + } val packet = createPacket(channelId = channelId, packetType = CultNetRudpPacketType.Data, payload = payload, reliable = options.reliable, ordered = options.ordered, sequenced = options.sequenced) if (packet.reliable) trackReliable(packet, options.nowMs) - return packet + return listOf(packet) } fun receive(packet: CultNetRudpPacket, nowMs: Long = 0): CultNetRudpReceiveResult { @@ -236,8 +277,8 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { val duplicate = receivedSequences.contains(packet.sequence) rememberReceived(packet.sequence) if (duplicate) return CultNetRudpReceiveResult() - val frame = CultNetRudpDeliveredFrame(packet.channelId, packet.payload.copyOf(), packet.sequence) - return CultNetRudpReceiveResult(delivered = if (packet.ordered) deliverOrdered(frame) else listOf(frame)) + val reassembled = reassemble(packet) ?: return CultNetRudpReceiveResult() + return CultNetRudpReceiveResult(delivered = if (reassembled.ordered) deliverOrdered(reassembled.frame, reassembled.nextSequence) else listOf(reassembled.frame)) } fun createAck(): CultNetRudpPacket = createPacket(CultNetRudpPacketType.Ack, "control", ByteArray(0)) @@ -256,11 +297,14 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { reliable: Boolean = false, ordered: Boolean = false, sequenced: Boolean = false, + fragmentId: Int = 0, + fragmentIndex: Int = 0, + fragmentCount: Int = 0, ): CultNetRudpPacket { val sequence = nextSequence nextSequence = uint32(nextSequence + 1, "sequence") val (ack, ackMask) = ackState() - return CultNetRudpPacket(packetType, connectionId, sequence, ack, ackMask, channelId, reliable, ordered, sequenced, payload = payload.copyOf()) + return CultNetRudpPacket(packetType, connectionId, sequence, ack, ackMask, channelId, reliable, ordered, sequenced, fragmentId, fragmentIndex, fragmentCount, payload.copyOf()) } private fun trackReliable(packet: CultNetRudpPacket, nowMs: Long) { @@ -290,18 +334,48 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { return ack to ackMask } - private fun deliverOrdered(frame: CultNetRudpDeliveredFrame): List { + private data class ReassembledFrame(val frame: CultNetRudpDeliveredFrame, val ordered: Boolean, val nextSequence: Long) + + private fun reassemble(packet: CultNetRudpPacket): ReassembledFrame? { + if (packet.fragmentCount == 0) { + return ReassembledFrame(CultNetRudpDeliveredFrame(packet.channelId, packet.payload.copyOf(), packet.sequence), packet.ordered, packet.sequence + 1) + } + if (packet.fragmentId == 0) throw IOException("RUDP fragmented packet must have a non-zero fragment id") + if (packet.fragmentIndex >= packet.fragmentCount) throw IOException("RUDP fragment index must be lower than fragment count") + val key = packet.channelId to packet.fragmentId + val buffer = fragmentBuffers.getOrPut(key) { + FragmentBuffer(packet.channelId, packet.ordered, packet.fragmentCount) + } + if (buffer.fragmentCount != packet.fragmentCount || buffer.ordered != packet.ordered) { + throw IOException("RUDP fragment metadata changed within a fragment set") + } + buffer.payloads[packet.fragmentIndex] = packet.payload.copyOf() + buffer.sequences[packet.fragmentIndex] = packet.sequence + if (buffer.payloads.size < packet.fragmentCount) return null + val payload = ByteArray(buffer.payloads.values.sumOf { it.size }) + var offset = 0 + for (index in 0 until packet.fragmentCount) { + val chunk = buffer.payloads[index] ?: return null + chunk.copyInto(payload, offset) + offset += chunk.size + } + val sequences = buffer.sequences.values + fragmentBuffers.remove(key) + return ReassembledFrame(CultNetRudpDeliveredFrame(buffer.channelId, payload, sequences.minOrNull() ?: packet.sequence), buffer.ordered, (sequences.maxOrNull() ?: packet.sequence) + 1) + } + + private fun deliverOrdered(frame: CultNetRudpDeliveredFrame, nextSequenceAfterFrame: Long): List { val next = orderedNextSequenceByChannel[frame.channelId] if (next == null) { - orderedNextSequenceByChannel[frame.channelId] = frame.sequence + 1 + orderedNextSequenceByChannel[frame.channelId] = nextSequenceAfterFrame return listOf(frame) + drainOrdered(frame.channelId) } if (frame.sequence < next) return emptyList() if (frame.sequence > next) { - orderedBuffers.getOrPut(frame.channelId) { TreeMap() }[frame.sequence] = frame + orderedBuffers.getOrPut(frame.channelId) { TreeMap() }[frame.sequence] = PendingOrderedFrame(frame, nextSequenceAfterFrame) return emptyList() } - orderedNextSequenceByChannel[frame.channelId] = next + 1 + orderedNextSequenceByChannel[frame.channelId] = nextSequenceAfterFrame return listOf(frame) + drainOrdered(frame.channelId) } @@ -310,13 +384,20 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { val buffer = orderedBuffers[channelId] ?: return delivered while (true) { val next = orderedNextSequenceByChannel[channelId] ?: break - val frame = buffer.remove(next) ?: break - delivered.add(frame) - orderedNextSequenceByChannel[channelId] = next + 1 + val pending = buffer.remove(next) ?: break + delivered.add(pending.frame) + orderedNextSequenceByChannel[channelId] = pending.nextSequence } return delivered } + private fun allocateFragmentId(): Int { + val fragmentId = nextFragmentId + nextFragmentId += 1 + if (nextFragmentId > 0xffff) nextFragmentId = 1 + return fragmentId + } + private fun requireConnection(packet: CultNetRudpPacket) { if (packet.connectionId != connectionId) throw IOException("RUDP packet connection id ${packet.connectionId} does not match $connectionId") } @@ -332,6 +413,7 @@ class CultNetRudpSocketTransportConnection( remoteAddress: InetSocketAddress? = null, initialSequence: Long = 1, resendDelayMs: Long = 250, + private val maxFragmentBytes: Int? = null, ) : AutoCloseable { private val session = CultNetRudpSession(CultNetRudpSessionOptions(connectionId, initialSequence, resendDelayMs)) private var remote = remoteAddress @@ -345,6 +427,7 @@ class CultNetRudpSocketTransportConnection( runtimeId = runtimeId, host = socket.localAddress.hostAddress, port = socket.localPort, + maxFragmentBytes = maxFragmentBytes, ) val connected: Boolean get() = session.connected val stats: CultNetTransportStats get() = CultNetTransportStats(bytesReceived, bytesSent, framesReceived, framesSent) @@ -355,7 +438,7 @@ class CultNetRudpSocketTransportConnection( } fun send(channelId: String, payload: ByteArray) { - sendPacket(session.send(channelId, payload, channelSendOptions(channelId, nowMs()))) + session.sendMany(channelId, payload, channelSendOptions(channelId, nowMs()), maxFragmentBytes).forEach { sendPacket(it) } framesSent += 1 } @@ -494,7 +577,9 @@ private fun nowMs(): Long = Instant.now().toEpochMilli() fun main(args: Array) { if (args.isEmpty()) { rudpPacketCodecUsesDeterministicReliableOrderedFixture() + rudpSessionFragmentsAndReassemblesReliableOrderedPayloads() rudpSocketTransportHandshakesAndCarriesReliableOrderedSchemaFrames() + rudpSocketTransportCarriesFragmentedReliableOrderedSchemaFrames() return } @@ -509,6 +594,9 @@ fun main(args: Array) { private fun rudpServeOnce(options: Map) { val bindHost = options["bind-host"] ?: "127.0.0.1" val bindPort = options["bind-port"]?.toInt() ?: 0 + val expectedClientPayload = (options["client-payload"] ?: "ts-kotlin-client-state").toByteArray(StandardCharsets.UTF_8) + val serverPayload = (options["server-payload"] ?: "kotlin-server-state").toByteArray(StandardCharsets.UTF_8) + val maxFragmentBytes = options["max-fragment-bytes"]?.toInt() val socket = DatagramSocket(bindPort, InetAddress.getByName(bindHost)).also { it.soTimeout = 20 } CultNetRudpSocketTransportConnection( socket = socket, @@ -517,14 +605,15 @@ private fun rudpServeOnce(options: Map) { connectionId = 0x446688aaL, initialSequence = 100, resendDelayMs = 25, + maxFragmentBytes = maxFragmentBytes, ).use { transport -> println("""{"status":"ready","port":${socket.localPort}}""") val deadline = System.nanoTime() + 5_000_000_000L while (System.nanoTime() < deadline) { val frame = transport.receiveOnce() if (frame != null) { - requireRudpFrame(frame, "schema", "ts-kotlin-client-state") - transport.send("schema", "kotlin-server-state".toByteArray(StandardCharsets.UTF_8)) + requireRudpFrame(frame, "schema", expectedClientPayload) + transport.send("schema", serverPayload) pollRudpAfterSend(transport, 250) println("""{"status":"ok"}""") return @@ -581,8 +670,11 @@ private fun pollRudpAfterSend(transport: CultNetRudpSocketTransportConnection, d } private fun requireRudpFrame(frame: CultNetTransportFrame, expectedChannelId: String, expectedPayload: String) { - val expectedBytes = expectedPayload.toByteArray(StandardCharsets.UTF_8) - if (frame.channelId != expectedChannelId || !frame.payload.contentEquals(expectedBytes)) { + requireRudpFrame(frame, expectedChannelId, expectedPayload.toByteArray(StandardCharsets.UTF_8)) +} + +private fun requireRudpFrame(frame: CultNetTransportFrame, expectedChannelId: String, expectedPayload: ByteArray) { + if (frame.channelId != expectedChannelId || !frame.payload.contentEquals(expectedPayload)) { error("Unexpected RUDP frame: channel=${frame.channelId}, payload=${String(frame.payload, StandardCharsets.UTF_8)}") } } @@ -637,6 +729,31 @@ private fun rudpPacketCodecUsesDeterministicReliableOrderedFixture() { check(String(decoded.payload, StandardCharsets.UTF_8) == "hello") } +private fun rudpSessionFragmentsAndReassemblesReliableOrderedPayloads() { + val sender = CultNetRudpSession(CultNetRudpSessionOptions(connectionId = 456, initialSequence = 1)) + val receiver = CultNetRudpSession(CultNetRudpSessionOptions(connectionId = 456, initialSequence = 100)) + sender.receive(CultNetRudpPacket(CultNetRudpPacketType.Accept, 456, 90, 0, 0, "control")) + receiver.receive(CultNetRudpPacket(CultNetRudpPacketType.Accept, 456, 91, 0, 0, "control")) + + val packets = sender.sendMany( + "schema", + "fragment-me-please".toByteArray(StandardCharsets.UTF_8), + CultNetRudpSendOptions(reliable = true, ordered = true, nowMs = 10), + maxFragmentBytes = 5, + ) + check(packets.size == 4) + check(packets.map { it.fragmentCount } == listOf(4, 4, 4, 4)) + check(packets.map { it.fragmentIndex } == listOf(0, 1, 2, 3)) + check(packets.all { it.fragmentId == packets.first().fragmentId }) + check(receiver.receive(packets[0]).delivered.isEmpty()) + check(receiver.receive(packets[1]).delivered.isEmpty()) + check(receiver.receive(packets[2]).delivered.isEmpty()) + val delivered = receiver.receive(packets[3]).delivered + check(delivered.size == 1) + check(String(delivered.first().payload, StandardCharsets.UTF_8) == "fragment-me-please") + check(delivered.first().sequence == packets.first().sequence) +} + private fun rudpSocketTransportHandshakesAndCarriesReliableOrderedSchemaFrames() { val loopback = InetAddress.getByName("127.0.0.1") val serverSocket = DatagramSocket(0, loopback).also { it.soTimeout = 20 } @@ -680,6 +797,43 @@ private fun rudpSocketTransportHandshakesAndCarriesReliableOrderedSchemaFrames() } } +private fun rudpSocketTransportCarriesFragmentedReliableOrderedSchemaFrames() { + val loopback = InetAddress.getByName("127.0.0.1") + val serverSocket = DatagramSocket(0, loopback).also { it.soTimeout = 20 } + val clientSocket = DatagramSocket(0, loopback).also { it.soTimeout = 20 } + val connectionId = 0x10203041L + CultNetRudpSocketTransportConnection( + socket = serverSocket, + mode = CultNetRudpSocketMode.Server, + runtimeId = "kotlin-rudp-fragment-server", + connectionId = connectionId, + initialSequence = 100, + resendDelayMs = 25, + maxFragmentBytes = 8, + ).use { server -> + CultNetRudpSocketTransportConnection( + socket = clientSocket, + mode = CultNetRudpSocketMode.Client, + runtimeId = "kotlin-rudp-fragment-client", + connectionId = connectionId, + remoteAddress = InetSocketAddress(loopback, serverSocket.localPort), + initialSequence = 1, + resendDelayMs = 25, + maxFragmentBytes = 8, + ).use { client -> + val payload = "this-schema-frame-is-larger-than-one-rudp-fragment".toByteArray(StandardCharsets.UTF_8) + client.connect("join".toByteArray(StandardCharsets.UTF_8)) + pumpRudpHandshake(client, server) + client.send("schema", payload) + val serverFrame = receiveRudpFrame(server) + check(serverFrame.channelId == "schema") + check(serverFrame.payload.contentEquals(payload)) + check(client.stats.framesSent == 1L) + check(server.stats.framesReceived == 1L) + } + } +} + private fun pumpRudpHandshake( client: CultNetRudpSocketTransportConnection, server: CultNetRudpSocketTransportConnection, diff --git a/packages/cultnet-rs/examples/cultnet_interop_peer.rs b/packages/cultnet-rs/examples/cultnet_interop_peer.rs index 56609de9..6bc5dbe8 100644 --- a/packages/cultnet-rs/examples/cultnet_interop_peer.rs +++ b/packages/cultnet-rs/examples/cultnet_interop_peer.rs @@ -458,6 +458,19 @@ fn rudp_serve_once(options: &BTreeMap) -> Result<()> { .transpose() .with_context(|| "argument --bind-port must be a u16")? .unwrap_or(0); + let expected_client_payload = options + .get("client-payload") + .map(|value| value.as_bytes().to_vec()) + .unwrap_or_else(|| b"ts-rust-client-state".to_vec()); + let server_payload = options + .get("server-payload") + .map(|value| value.as_bytes().to_vec()) + .unwrap_or_else(|| b"rust-server-state".to_vec()); + let max_fragment_bytes = options + .get("max-fragment-bytes") + .map(|value| value.parse::()) + .transpose() + .with_context(|| "argument --max-fragment-bytes must be a u32")?; let socket = UdpSocket::bind((bind_host.as_str(), bind_port)) .with_context(|| format!("failed to bind RUDP server on {bind_host}:{bind_port}"))?; socket.set_read_timeout(Some(Duration::from_millis(20)))?; @@ -466,6 +479,7 @@ fn rudp_serve_once(options: &BTreeMap) -> Result<()> { CultNetRudpSocketTransportOptions::server("rust-rudp-interop", socket, 0x22446688); transport_options.initial_sequence = 100; transport_options.resend_delay_ms = 25; + transport_options.max_fragment_bytes = max_fragment_bytes; let mut transport = CultNetRudpSocketTransportConnection::new(transport_options)?; print_json(&serde_json::json!({ @@ -476,14 +490,14 @@ fn rudp_serve_once(options: &BTreeMap) -> Result<()> { let deadline = Instant::now() + Duration::from_secs(5); while Instant::now() < deadline { if let Some(frame) = transport.receive_once()? { - if frame.channel_id != "schema" || frame.payload != b"ts-rust-client-state" { + if frame.channel_id != "schema" || frame.payload != expected_client_payload { return Err(anyhow!( "unexpected RUDP frame: channel={} payload={:?}", frame.channel_id, frame.payload )); } - transport.send("schema", b"rust-server-state".to_vec())?; + transport.send("schema", server_payload.clone())?; poll_rudp_after_send(&mut transport, Duration::from_millis(250))?; print_json(&serde_json::json!({ "status": "ok" }))?; return Ok(()); diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index b5aead78..22c5e8c4 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -94,16 +94,33 @@ struct PendingReliablePacket { last_sent_at_ms: u64, } +#[derive(Clone, Debug, PartialEq, Eq)] +struct PendingOrderedFrame { + frame: CultNetRudpDeliveredFrame, + next_sequence: u32, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct FragmentBuffer { + channel_id: String, + ordered: bool, + fragment_count: u16, + payloads: BTreeMap>, + sequences: BTreeMap, +} + pub struct CultNetRudpSession { connection_id: u32, resend_delay_ms: u64, next_sequence: u32, + next_fragment_id: u16, connected: bool, highest_received_sequence: Option, received_sequences: BTreeSet, pending_reliable: BTreeMap, ordered_next_sequence_by_channel: BTreeMap, - ordered_buffers: BTreeMap>, + ordered_buffers: BTreeMap>, + fragment_buffers: BTreeMap<(String, u16), FragmentBuffer>, } impl CultNetRudpSession { @@ -112,12 +129,14 @@ impl CultNetRudpSession { connection_id: options.connection_id, resend_delay_ms: options.resend_delay_ms, next_sequence: options.initial_sequence, + next_fragment_id: 1, connected: false, highest_received_sequence: None, received_sequences: BTreeSet::new(), pending_reliable: BTreeMap::new(), ordered_next_sequence_by_channel: BTreeMap::new(), ordered_buffers: BTreeMap::new(), + fragment_buffers: BTreeMap::new(), } } @@ -184,12 +203,59 @@ impl CultNetRudpSession { payload: Vec, options: CultNetRudpSendOptions, ) -> Result { + self.send_many(channel_id, payload, options, None)? + .into_iter() + .next() + .ok_or_else(|| anyhow!("RUDP send produced no packets")) + } + + pub fn send_many( + &mut self, + channel_id: &str, + payload: Vec, + options: CultNetRudpSendOptions, + max_fragment_bytes: Option, + ) -> Result> { if !self.connected { return Err(anyhow!( "Cannot send RUDP data before the session is connected" )); } + if let Some(max_fragment_bytes) = max_fragment_bytes { + if max_fragment_bytes == 0 { + return Err(anyhow!("RUDP max_fragment_bytes must be greater than zero")); + } + if payload.len() > max_fragment_bytes { + let fragment_count = payload.len().div_ceil(max_fragment_bytes); + if fragment_count > u16::MAX as usize { + return Err(anyhow!("RUDP payload requires more than 65535 fragments")); + } + let fragment_id = self.allocate_fragment_id(); + let mut packets = Vec::new(); + for index in 0..fragment_count { + let start = index * max_fragment_bytes; + let end = (start + max_fragment_bytes).min(payload.len()); + let packet = self.create_packet_with_fragments( + CultNetRudpPacketType::Data, + channel_id, + payload[start..end].to_vec(), + options.reliable, + options.ordered, + options.sequenced, + fragment_id, + index as u16, + fragment_count as u16, + ); + if packet.reliable { + self.track_reliable(packet.clone(), options.now_ms); + } + packets.push(packet); + } + return Ok(packets); + } + } + let packet = self.create_packet( CultNetRudpPacketType::Data, channel_id, @@ -201,7 +267,7 @@ impl CultNetRudpSession { if packet.reliable { self.track_reliable(packet.clone(), options.now_ms); } - Ok(packet) + Ok(vec![packet]) } pub fn receive( @@ -262,13 +328,14 @@ impl CultNetRudpSession { }); } - let frame = CultNetRudpDeliveredFrame { - channel_id: packet.channel_id.clone(), - payload: packet.payload.clone(), - sequence: packet.sequence, + let Some((frame, ordered, next_sequence)) = self.reassemble(packet)? else { + return Ok(CultNetRudpReceiveResult { + delivered: Vec::new(), + reply: None, + }); }; - let delivered = if packet.ordered { - self.deliver_ordered(frame) + let delivered = if ordered { + self.deliver_ordered(frame, next_sequence) } else { vec![frame] }; @@ -309,6 +376,32 @@ impl CultNetRudpSession { reliable: bool, ordered: bool, sequenced: bool, + ) -> CultNetRudpPacket { + self.create_packet_with_fragments( + packet_type, + channel_id, + payload, + reliable, + ordered, + sequenced, + 0, + 0, + 0, + ) + } + + #[allow(clippy::too_many_arguments)] + fn create_packet_with_fragments( + &mut self, + packet_type: CultNetRudpPacketType, + channel_id: &str, + payload: Vec, + reliable: bool, + ordered: bool, + sequenced: bool, + fragment_id: u16, + fragment_index: u16, + fragment_count: u16, ) -> CultNetRudpPacket { let sequence = self.next_sequence; self.next_sequence = self @@ -326,9 +419,9 @@ impl CultNetRudpSession { reliable, ordered, sequenced, - fragment_id: 0, - fragment_index: 0, - fragment_count: 0, + fragment_id, + fragment_index, + fragment_count, payload, } } @@ -373,9 +466,88 @@ impl CultNetRudpSession { (ack, ack_mask) } + fn reassemble( + &mut self, + packet: &CultNetRudpPacket, + ) -> Result> { + if packet.fragment_count == 0 { + return Ok(Some(( + CultNetRudpDeliveredFrame { + channel_id: packet.channel_id.clone(), + payload: packet.payload.clone(), + sequence: packet.sequence, + }, + packet.ordered, + packet.sequence + 1, + ))); + } + if packet.fragment_id == 0 { + return Err(anyhow!( + "RUDP fragmented packet must have a non-zero fragment id" + )); + } + if packet.fragment_index >= packet.fragment_count { + return Err(anyhow!( + "RUDP fragment index must be lower than fragment count" + )); + } + + let key = (packet.channel_id.clone(), packet.fragment_id); + let buffer = self + .fragment_buffers + .entry(key.clone()) + .or_insert_with(|| FragmentBuffer { + channel_id: packet.channel_id.clone(), + ordered: packet.ordered, + fragment_count: packet.fragment_count, + payloads: BTreeMap::new(), + sequences: BTreeMap::new(), + }); + if buffer.fragment_count != packet.fragment_count || buffer.ordered != packet.ordered { + return Err(anyhow!( + "RUDP fragment metadata changed within a fragment set" + )); + } + buffer + .payloads + .insert(packet.fragment_index, packet.payload.clone()); + buffer + .sequences + .insert(packet.fragment_index, packet.sequence); + if buffer.payloads.len() < packet.fragment_count as usize { + return Ok(None); + } + + let mut payload = Vec::new(); + let mut sequences = Vec::new(); + for index in 0..packet.fragment_count { + let Some(chunk) = buffer.payloads.get(&index) else { + return Ok(None); + }; + let Some(sequence) = buffer.sequences.get(&index) else { + return Ok(None); + }; + payload.extend_from_slice(chunk); + sequences.push(*sequence); + } + let channel_id = buffer.channel_id.clone(); + let ordered = buffer.ordered; + self.fragment_buffers.remove(&key); + Ok(Some(( + CultNetRudpDeliveredFrame { + channel_id, + payload, + sequence: *sequences.iter().min().unwrap(), + }, + ordered, + sequences.iter().max().unwrap() + 1, + ))) + } + fn deliver_ordered( &mut self, frame: CultNetRudpDeliveredFrame, + next_sequence_after_frame: u32, ) -> Vec { let channel_id = frame.channel_id.clone(); let Some(next) = self @@ -384,7 +556,7 @@ impl CultNetRudpSession { .copied() else { self.ordered_next_sequence_by_channel - .insert(channel_id.clone(), frame.sequence + 1); + .insert(channel_id.clone(), next_sequence_after_frame); let mut delivered = vec![frame]; delivered.extend(self.drain_ordered(&channel_id)); return delivered; @@ -395,15 +567,18 @@ impl CultNetRudpSession { } if frame.sequence > next { - self.ordered_buffers - .entry(channel_id) - .or_default() - .insert(frame.sequence, frame); + self.ordered_buffers.entry(channel_id).or_default().insert( + frame.sequence, + PendingOrderedFrame { + frame, + next_sequence: next_sequence_after_frame, + }, + ); return Vec::new(); } self.ordered_next_sequence_by_channel - .insert(channel_id.clone(), next + 1); + .insert(channel_id.clone(), next_sequence_after_frame); let mut delivered = vec![frame]; delivered.extend(self.drain_ordered(&channel_id)); delivered @@ -422,16 +597,25 @@ impl CultNetRudpSession { let Some(buffer) = self.ordered_buffers.get_mut(channel_id) else { break; }; - let Some(frame) = buffer.remove(&next) else { + let Some(pending) = buffer.remove(&next) else { break; }; - delivered.push(frame); + delivered.push(pending.frame); self.ordered_next_sequence_by_channel - .insert(channel_id.to_string(), next + 1); + .insert(channel_id.to_string(), pending.next_sequence); } delivered } + fn allocate_fragment_id(&mut self) -> u16 { + let fragment_id = self.next_fragment_id; + self.next_fragment_id = self.next_fragment_id.saturating_add(1); + if self.next_fragment_id == 0 { + self.next_fragment_id = 1; + } + fragment_id + } + fn require_connection(&self, packet: &CultNetRudpPacket) -> Result<()> { if packet.connection_id != self.connection_id { return Err(anyhow!( @@ -508,6 +692,7 @@ pub struct CultNetRudpSocketTransportConnection { pub profile: CultNetTransportProfile, stats: CultNetTransportStats, delivered_frames: VecDeque, + max_fragment_bytes: Option, } impl CultNetRudpSocketTransportConnection { @@ -535,6 +720,7 @@ impl CultNetRudpSocketTransportConnection { profile, stats: CultNetTransportStats::default(), delivered_frames: VecDeque::new(), + max_fragment_bytes: options.max_fragment_bytes.map(|value| value as usize), }) } @@ -558,8 +744,12 @@ impl CultNetRudpSocketTransportConnection { pub fn send(&mut self, channel_id: &str, payload: Vec) -> Result<()> { let options = channel_send_options(channel_id, now_ms()); - let packet = self.session.send(channel_id, payload, options)?; - self.send_packet(&packet)?; + let packets = + self.session + .send_many(channel_id, payload, options, self.max_fragment_bytes)?; + for packet in packets { + self.send_packet(&packet)?; + } self.stats.frames_sent += 1; Ok(()) } diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index 7ecc0b5c..333685d8 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -571,6 +571,97 @@ fn rudp_session_suppresses_duplicates_and_delivers_reliable_ordered_payloads_in_ Ok(()) } +#[test] +fn rudp_session_fragments_and_reassembles_reliable_ordered_payloads() -> Result<()> { + let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 456, + initial_sequence: 1, + ..CultNetRudpSessionOptions::default() + }); + let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 456, + initial_sequence: 100, + ..CultNetRudpSessionOptions::default() + }); + sender.receive( + &CultNetRudpPacket { + packet_type: CultNetRudpPacketType::Accept, + connection_id: 456, + sequence: 90, + ack: 0, + ack_mask: 0, + channel_id: "control".to_string(), + reliable: false, + ordered: false, + sequenced: false, + fragment_id: 0, + fragment_index: 0, + fragment_count: 0, + payload: Vec::new(), + }, + 0, + )?; + receiver.receive( + &CultNetRudpPacket { + packet_type: CultNetRudpPacketType::Accept, + connection_id: 456, + sequence: 91, + ack: 0, + ack_mask: 0, + channel_id: "control".to_string(), + reliable: false, + ordered: false, + sequenced: false, + fragment_id: 0, + fragment_index: 0, + fragment_count: 0, + payload: Vec::new(), + }, + 0, + )?; + + let packets = sender.send_many( + "schema", + b"fragment-me-please".to_vec(), + CultNetRudpSendOptions { + reliable: true, + ordered: true, + now_ms: 10, + ..CultNetRudpSendOptions::default() + }, + Some(5), + )?; + assert_eq!(packets.len(), 4); + assert_eq!( + packets + .iter() + .map(|packet| packet.fragment_count) + .collect::>(), + vec![4, 4, 4, 4] + ); + assert_eq!( + packets + .iter() + .map(|packet| packet.fragment_index) + .collect::>(), + vec![0, 1, 2, 3] + ); + assert!( + packets + .iter() + .all(|packet| packet.fragment_id == packets[0].fragment_id) + ); + + assert!(receiver.receive(&packets[0], 0)?.delivered.is_empty()); + assert!(receiver.receive(&packets[1], 0)?.delivered.is_empty()); + assert!(receiver.receive(&packets[2], 0)?.delivered.is_empty()); + let delivered = receiver.receive(&packets[3], 0)?.delivered; + assert_eq!(delivered.len(), 1); + assert_eq!(delivered[0].payload, b"fragment-me-please"); + assert_eq!(delivered[0].sequence, packets[0].sequence); + Ok(()) +} + #[test] fn rudp_socket_transport_handshakes_and_carries_reliable_ordered_schema_frames() -> Result<()> { let server_socket = bind_udp_socket()?; @@ -628,6 +719,52 @@ fn rudp_socket_transport_handshakes_and_carries_reliable_ordered_schema_frames() Ok(()) } +#[test] +fn rudp_socket_transport_carries_fragmented_reliable_ordered_schema_frames() -> Result<()> { + let server_socket = bind_udp_socket()?; + let client_socket = bind_udp_socket()?; + let server_addr = server_socket.local_addr()?; + let connection_id = 0x1020_3041; + let mut server = + CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + runtime_id: "rust-rudp-fragment-server".to_string(), + socket: server_socket, + mode: CultNetRudpSocketMode::Server, + remote_addr: None, + connection_id, + initial_sequence: 100, + resend_delay_ms: 25, + transport_id: None, + max_payload_bytes: None, + max_fragment_bytes: Some(8), + })?; + let mut client = + CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + runtime_id: "rust-rudp-fragment-client".to_string(), + socket: client_socket, + mode: CultNetRudpSocketMode::Client, + remote_addr: Some(server_addr), + connection_id, + initial_sequence: 1, + resend_delay_ms: 25, + transport_id: None, + max_payload_bytes: None, + max_fragment_bytes: Some(8), + })?; + + client.connect(b"join".to_vec())?; + pump_rudp_handshake(&mut client, &mut server)?; + + let payload = b"this-schema-frame-is-larger-than-one-rudp-fragment".to_vec(); + client.send("schema", payload.clone())?; + let server_frame = receive_rudp_frame(&mut server)?; + assert_eq!(server_frame.channel_id, "schema"); + assert_eq!(server_frame.payload, payload); + assert_eq!(client.stats().frames_sent, 1); + assert_eq!(server.stats().frames_received, 1); + Ok(()) +} + #[test] fn legacy_gamecult_networking_contract_round_trips_login_union() -> Result<()> { let message = CultNetMessage::Login { diff --git a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts index bbade2d5..a8117a87 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts @@ -72,12 +72,18 @@ interface RudpServerInteropPeerSpec { name: string; connectionId: number; build: () => Promise; - spawn: () => RunningPythonRudpPeer; + spawn: (options?: RudpServerSpawnOptions) => RunningPythonRudpPeer; joinPayload: string; clientPayload: string; expectedPayload: string; } +interface RudpServerSpawnOptions { + clientPayload?: string; + serverPayload?: string; + maxFragmentBytes?: number; +} + function rudpServerInteropPeers(): RudpServerInteropPeerSpec[] { return [ { @@ -120,6 +126,7 @@ function rudpServerInteropPeers(): RudpServerInteropPeerSpec[] { } const pythonRudpPeerScript = String.raw` +import os import socket import time @@ -129,6 +136,11 @@ from cultnet_py import ( CultNetRudpSocketTransportOptions, ) +expected_client_payload = os.environ.get("RUDP_CLIENT_PAYLOAD", "client-state").encode() +server_payload = os.environ.get("RUDP_SERVER_PAYLOAD", "python-state").encode() +max_fragment_bytes_raw = os.environ.get("RUDP_MAX_FRAGMENT_BYTES") +max_fragment_bytes = int(max_fragment_bytes_raw) if max_fragment_bytes_raw else None + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.bind(("127.0.0.1", 0)) sock.settimeout(0.02) @@ -139,6 +151,7 @@ transport = CultNetRudpSocketTransportConnection(CultNetRudpSocketTransportOptio connection_id=0x55667788, initial_sequence=100, resend_delay_ms=25, + max_fragment_bytes=max_fragment_bytes, )) print(f"PORT {sock.getsockname()[1]}", flush=True) @@ -149,10 +162,10 @@ while time.time() < deadline: if frame is None: time.sleep(0.005) continue - if frame.channel_id != "schema" or frame.payload != b"client-state": + if frame.channel_id != "schema" or frame.payload != expected_client_payload: transport.close() raise SystemExit(f"unexpected RUDP frame: {frame.channel_id!r} {frame.payload!r}") - transport.send("schema", b"python-state") + transport.send("schema", server_payload) print("OK", flush=True) ack_deadline = time.time() + 0.25 while time.time() < ack_deadline: @@ -1153,6 +1166,58 @@ test("CultNet RUDP reliable schema response survives one dropped server data pac } }); +test("CultNet RUDP fragmented schema frames cross runtime boundaries", async () => { + for (const peer of rudpServerInteropPeers()) { + await peer.build(); + const clientPayload = `${peer.clientPayload}:fragmented:${"x".repeat(96)}`; + const serverPayload = `${peer.expectedPayload}:fragmented:${"y".repeat(96)}`; + const serverPeer = peer.spawn({ + clientPayload, + serverPayload, + maxFragmentBytes: 16, + }); + const clientSocket = await bindUdpSocket(); + let client: CultNetRudpSocketTransportConnection | undefined; + + try { + client = new CultNetRudpSocketTransportConnection({ + runtimeId: `ts-fragmented-${peer.name.replace(/\W/g, "").toLowerCase()}-rudp-interop`, + socket: clientSocket, + mode: "client", + remoteHost: "127.0.0.1", + remotePort: await serverPeer.ready, + connectionId: peer.connectionId, + initialSequence: 1, + resendDelayMs: 25, + resendPollMs: 5, + maxFragmentBytes: 16, + }); + + const receivedFrame = once(client, "frame"); + client.connect(Buffer.from(peer.joinPayload)); + await waitFor(() => client?.connected === true, `TypeScript RUDP client to complete ${peer.name} fragmented handshake`); + client.send("schema", Buffer.from(clientPayload)); + + const [frame] = await withTimeout(receivedFrame, 2_000, `${peer.name} fragmented RUDP response frame`); + assert.equal(frame.channelId, "schema"); + assert.deepEqual(Buffer.from(frame.payload), Buffer.from(serverPayload)); + + const [exitCode] = await withTimeout(once(serverPeer.child, "exit"), 2_000, `${peer.name} RUDP peer exit after fragmented exchange`); + assert.equal(exitCode, 0, serverPeer.stderr.join("")); + } finally { + if (client) { + client.close(); + } else { + clientSocket.close(); + } + if (serverPeer.child.exitCode === null && !serverPeer.child.killed) { + serverPeer.child.kill("SIGTERM"); + await once(serverPeer.child, "exit").catch(() => undefined); + } + } + } +}); + async function buildInteropPeers(): Promise { await buildRustInteropPeer(); await buildCSharpInteropPeer(); @@ -1305,10 +1370,16 @@ interface RunningPythonRudpPeer { stderr: string[]; } -function spawnPythonRudpPeer(): RunningPythonRudpPeer { +function spawnPythonRudpPeer(options: RudpServerSpawnOptions = {}): RunningPythonRudpPeer { const child = spawn(pythonCommand, ["-c", pythonRudpPeerScript], { cwd: cultcachePyRoot, - env: { ...process.env, PYTHONPATH: cultcachePySrc }, + env: { + ...process.env, + PYTHONPATH: cultcachePySrc, + ...(options.clientPayload ? { RUDP_CLIENT_PAYLOAD: options.clientPayload } : {}), + ...(options.serverPayload ? { RUDP_SERVER_PAYLOAD: options.serverPayload } : {}), + ...(options.maxFragmentBytes ? { RUDP_MAX_FRAGMENT_BYTES: String(options.maxFragmentBytes) } : {}), + }, stdio: ["ignore", "pipe", "pipe"], }); const stderr: string[] = []; @@ -1353,8 +1424,14 @@ function spawnPythonRudpPeer(): RunningPythonRudpPeer { return { child, ready, stderr }; } -function spawnRustRudpServer(): RunningPythonRudpPeer { - const child = spawn(rustBinaryPath, ["rudp-serve-once", "--bind-host", "127.0.0.1"], { +function spawnRustRudpServer(options: RudpServerSpawnOptions = {}): RunningPythonRudpPeer { + const child = spawn(rustBinaryPath, [ + "rudp-serve-once", + "--bind-host", "127.0.0.1", + ...(options.clientPayload ? ["--client-payload", options.clientPayload] : []), + ...(options.serverPayload ? ["--server-payload", options.serverPayload] : []), + ...(options.maxFragmentBytes ? ["--max-fragment-bytes", String(options.maxFragmentBytes)] : []), + ], { cwd: cultnetRsRoot, env: process.env, stdio: ["ignore", "pipe", "pipe"], @@ -1403,8 +1480,15 @@ function spawnRustRudpServer(): RunningPythonRudpPeer { return { child, ready, stderr }; } -function spawnCSharpRudpServer(): RunningPythonRudpPeer { - const child = spawn(dotnetCommand, [csharpDllPath, "rudp-serve-once", "--bind-host", "127.0.0.1"], { +function spawnCSharpRudpServer(options: RudpServerSpawnOptions = {}): RunningPythonRudpPeer { + const child = spawn(dotnetCommand, [ + csharpDllPath, + "rudp-serve-once", + "--bind-host", "127.0.0.1", + ...(options.clientPayload ? ["--client-payload", options.clientPayload] : []), + ...(options.serverPayload ? ["--server-payload", options.serverPayload] : []), + ...(options.maxFragmentBytes ? ["--max-fragment-bytes", String(options.maxFragmentBytes)] : []), + ], { cwd: cultLibRoot, env: process.env, stdio: ["ignore", "pipe", "pipe"], @@ -1453,12 +1537,15 @@ function spawnCSharpRudpServer(): RunningPythonRudpPeer { return { child, ready, stderr }; } -function spawnKotlinRudpServer(): RunningPythonRudpPeer { +function spawnKotlinRudpServer(options: RudpServerSpawnOptions = {}): RunningPythonRudpPeer { const child = spawn(kotlinJavaCommand, [ "-cp", kotlinClasspath, "org.gamecult.cultmesh.CultMeshKt", "rudp-serve-once", "--bind-host", "127.0.0.1", + ...(options.clientPayload ? ["--client-payload", options.clientPayload] : []), + ...(options.serverPayload ? ["--server-payload", options.serverPayload] : []), + ...(options.maxFragmentBytes ? ["--max-fragment-bytes", String(options.maxFragmentBytes)] : []), ], { cwd: cultLibRoot, env: process.env, diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index e9b08871..5a8b9ec3 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -561,14 +561,31 @@ namespace GameCult.Networking public long LastSentAtMs { get; set; } } + private sealed class PendingOrderedFrame + { + public CultNetRudpDeliveredFrame Frame { get; set; } = new CultNetRudpDeliveredFrame(); + public uint NextSequence { get; set; } + } + + private sealed class FragmentBuffer + { + public string ChannelId { get; set; } = string.Empty; + public bool Ordered { get; set; } + public ushort FragmentCount { get; set; } + public Dictionary Payloads { get; } = new Dictionary(); + public Dictionary Sequences { get; } = new Dictionary(); + } + private uint _nextSequence; + private ushort _nextFragmentId = 1; private bool _connected; private uint? _highestReceivedSequence; private readonly HashSet _receivedSequences = new HashSet(); private readonly Dictionary _pendingReliable = new Dictionary(); private readonly Dictionary _orderedNextSequenceByChannel = new Dictionary(StringComparer.Ordinal); - private readonly Dictionary> _orderedBuffers = - new Dictionary>(StringComparer.Ordinal); + private readonly Dictionary> _orderedBuffers = + new Dictionary>(StringComparer.Ordinal); + private readonly Dictionary _fragmentBuffers = new Dictionary(StringComparer.Ordinal); /// /// Initializes a RUDP reliability session. @@ -630,6 +647,14 @@ namespace GameCult.Networking /// Creates a data packet. /// public CultNetRudpPacket Send(string channelId, byte[] payload, CultNetRudpSendOptions? options = null) + { + return SendMany(channelId, payload, options).First(); + } + + /// + /// Creates one or more data packets, fragmenting when requested. + /// + public IReadOnlyList SendMany(string channelId, byte[] payload, CultNetRudpSendOptions? options = null, int? maxFragmentBytes = null) { if (!_connected) { @@ -637,6 +662,47 @@ namespace GameCult.Networking } options ??= new CultNetRudpSendOptions(); + payload ??= Array.Empty(); + if (maxFragmentBytes.HasValue && maxFragmentBytes.Value <= 0) + { + throw new ArgumentOutOfRangeException(nameof(maxFragmentBytes), "RUDP maxFragmentBytes must be greater than zero."); + } + + if (maxFragmentBytes.HasValue && payload.Length > maxFragmentBytes.Value) + { + var fragmentCount = (payload.Length + maxFragmentBytes.Value - 1) / maxFragmentBytes.Value; + if (fragmentCount > ushort.MaxValue) + { + throw new InvalidOperationException("RUDP payload requires more than 65535 fragments."); + } + + var fragmentId = AllocateFragmentId(); + var packets = new List(); + for (var index = 0; index < fragmentCount; index++) + { + var start = index * maxFragmentBytes.Value; + var length = Math.Min(maxFragmentBytes.Value, payload.Length - start); + var chunk = new byte[length]; + Array.Copy(payload, start, chunk, 0, length); + var fragmentPacket = CreatePacket( + CultNetRudpPacketType.Data, + channelId, + chunk, + options.Reliable, + options.Ordered, + options.Sequenced, + fragmentId, + (ushort)index, + (ushort)fragmentCount); + if (fragmentPacket.Reliable) + { + TrackReliable(fragmentPacket, options.NowMs); + } + packets.Add(fragmentPacket); + } + return packets; + } + var packet = CreatePacket( CultNetRudpPacketType.Data, channelId, @@ -648,7 +714,7 @@ namespace GameCult.Networking { TrackReliable(packet, options.NowMs); } - return packet; + return new[] { packet }; } /// @@ -699,16 +765,17 @@ namespace GameCult.Networking return new CultNetRudpReceiveResult(); } - var frame = new CultNetRudpDeliveredFrame + var reassembled = Reassemble(packet); + if (reassembled == null) { - ChannelId = packet.ChannelId, - Payload = packet.Payload ?? Array.Empty(), - Sequence = packet.Sequence - }; + return new CultNetRudpReceiveResult(); + } return new CultNetRudpReceiveResult { - Delivered = packet.Ordered ? DeliverOrdered(frame) : new[] { frame } + Delivered = reassembled.Ordered + ? DeliverOrdered(reassembled.Frame, reassembled.NextSequence) + : new[] { reassembled.Frame } }; } @@ -745,6 +812,20 @@ namespace GameCult.Networking bool reliable, bool ordered, bool sequenced) + { + return CreatePacket(packetType, channelId, payload, reliable, ordered, sequenced, 0, 0, 0); + } + + private CultNetRudpPacket CreatePacket( + CultNetRudpPacketType packetType, + string channelId, + byte[] payload, + bool reliable, + bool ordered, + bool sequenced, + ushort fragmentId, + ushort fragmentIndex, + ushort fragmentCount) { var sequence = _nextSequence; _nextSequence = checked(_nextSequence + 1); @@ -760,6 +841,9 @@ namespace GameCult.Networking Reliable = reliable, Ordered = ordered, Sequenced = sequenced, + FragmentId = fragmentId, + FragmentIndex = fragmentIndex, + FragmentCount = fragmentCount, Payload = payload ?? Array.Empty() }; } @@ -809,11 +893,90 @@ namespace GameCult.Networking return (ack, ackMask); } - private IReadOnlyList DeliverOrdered(CultNetRudpDeliveredFrame frame) + private sealed class ReassembledFrame + { + public CultNetRudpDeliveredFrame Frame { get; set; } = new CultNetRudpDeliveredFrame(); + public bool Ordered { get; set; } + public uint NextSequence { get; set; } + } + + private ReassembledFrame? Reassemble(CultNetRudpPacket packet) + { + if (packet.FragmentCount == 0) + { + return new ReassembledFrame + { + Frame = new CultNetRudpDeliveredFrame + { + ChannelId = packet.ChannelId, + Payload = packet.Payload ?? Array.Empty(), + Sequence = packet.Sequence + }, + Ordered = packet.Ordered, + NextSequence = packet.Sequence + 1 + }; + } + if (packet.FragmentId == 0) + { + throw new InvalidOperationException("RUDP fragmented packet must have a non-zero fragment id."); + } + if (packet.FragmentIndex >= packet.FragmentCount) + { + throw new InvalidOperationException("RUDP fragment index must be lower than fragment count."); + } + + var key = $"{packet.ChannelId}\0{packet.FragmentId}"; + if (!_fragmentBuffers.TryGetValue(key, out var buffer)) + { + buffer = new FragmentBuffer + { + ChannelId = packet.ChannelId, + Ordered = packet.Ordered, + FragmentCount = packet.FragmentCount + }; + _fragmentBuffers[key] = buffer; + } + if (buffer.FragmentCount != packet.FragmentCount || buffer.Ordered != packet.Ordered) + { + throw new InvalidOperationException("RUDP fragment metadata changed within a fragment set."); + } + + buffer.Payloads[packet.FragmentIndex] = packet.Payload?.ToArray() ?? Array.Empty(); + buffer.Sequences[packet.FragmentIndex] = packet.Sequence; + if (buffer.Payloads.Count < packet.FragmentCount) + { + return null; + } + + var payloadLength = buffer.Payloads.Values.Sum(chunk => chunk.Length); + var payload = new byte[payloadLength]; + var offset = 0; + for (ushort index = 0; index < packet.FragmentCount; index++) + { + var chunk = buffer.Payloads[index]; + Array.Copy(chunk, 0, payload, offset, chunk.Length); + offset += chunk.Length; + } + var sequences = buffer.Sequences.Values.ToArray(); + _fragmentBuffers.Remove(key); + return new ReassembledFrame + { + Frame = new CultNetRudpDeliveredFrame + { + ChannelId = buffer.ChannelId, + Payload = payload, + Sequence = sequences.Min() + }, + Ordered = buffer.Ordered, + NextSequence = sequences.Max() + 1 + }; + } + + private IReadOnlyList DeliverOrdered(CultNetRudpDeliveredFrame frame, uint nextSequenceAfterFrame) { if (!_orderedNextSequenceByChannel.TryGetValue(frame.ChannelId, out var next)) { - _orderedNextSequenceByChannel[frame.ChannelId] = frame.Sequence + 1; + _orderedNextSequenceByChannel[frame.ChannelId] = nextSequenceAfterFrame; return new[] { frame }.Concat(DrainOrdered(frame.ChannelId)).ToArray(); } @@ -826,15 +989,15 @@ namespace GameCult.Networking { if (!_orderedBuffers.TryGetValue(frame.ChannelId, out var buffer)) { - buffer = new SortedDictionary(); + buffer = new SortedDictionary(); _orderedBuffers[frame.ChannelId] = buffer; } - buffer[frame.Sequence] = frame; + buffer[frame.Sequence] = new PendingOrderedFrame { Frame = frame, NextSequence = nextSequenceAfterFrame }; return Array.Empty(); } - _orderedNextSequenceByChannel[frame.ChannelId] = next + 1; + _orderedNextSequenceByChannel[frame.ChannelId] = nextSequenceAfterFrame; return new[] { frame }.Concat(DrainOrdered(frame.ChannelId)).ToArray(); } @@ -846,16 +1009,27 @@ namespace GameCult.Networking return delivered; } - while (_orderedNextSequenceByChannel.TryGetValue(channelId, out var next) && buffer.TryGetValue(next, out var frame)) + while (_orderedNextSequenceByChannel.TryGetValue(channelId, out var next) && buffer.TryGetValue(next, out var pending)) { buffer.Remove(next); - delivered.Add(frame); - _orderedNextSequenceByChannel[channelId] = next + 1; + delivered.Add(pending.Frame); + _orderedNextSequenceByChannel[channelId] = pending.NextSequence; } return delivered; } + private ushort AllocateFragmentId() + { + var fragmentId = _nextFragmentId; + _nextFragmentId++; + if (_nextFragmentId == 0) + { + _nextFragmentId = 1; + } + return fragmentId; + } + private void RequireConnection(CultNetRudpPacket packet) { if (packet.ConnectionId != ConnectionId) @@ -893,6 +1067,7 @@ namespace GameCult.Networking private readonly Socket _socket; private readonly CultNetRudpSession _session; private readonly CultNetRudpSocketMode _mode; + private readonly int? _maxFragmentBytes; private readonly CultNetTransportStats _stats = new CultNetTransportStats(); private readonly Queue _deliveredFrames = new Queue(); private EndPoint? _remoteEndPoint; @@ -908,6 +1083,7 @@ namespace GameCult.Networking _socket = options.Socket ?? throw new ArgumentNullException(nameof(options.Socket)); _mode = options.Mode; _remoteEndPoint = options.RemoteEndPoint; + _maxFragmentBytes = options.MaxFragmentBytes; _session = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = options.ConnectionId, @@ -961,8 +1137,10 @@ namespace GameCult.Networking /// public void Send(string channelId, byte[] payload) { - var packet = _session.Send(channelId, payload, ChannelSendOptions(channelId)); - SendPacket(packet); + foreach (var packet in _session.SendMany(channelId, payload, ChannelSendOptions(channelId), _maxFragmentBytes)) + { + SendPacket(packet); + } _stats.FramesSent++; } diff --git a/tests/GameCult.Networking.InteropPeer/Program.cs b/tests/GameCult.Networking.InteropPeer/Program.cs index 1f6eb374..88e30097 100644 --- a/tests/GameCult.Networking.InteropPeer/Program.cs +++ b/tests/GameCult.Networking.InteropPeer/Program.cs @@ -583,6 +583,15 @@ static void RudpServeOnce(Dictionary options) var bindPort = options.TryGetValue("bind-port", out var configuredBindPort) ? int.Parse(configuredBindPort, CultureInfo.InvariantCulture) : 0; + var expectedClientPayload = Ascii(options.TryGetValue("client-payload", out var configuredClientPayload) + ? configuredClientPayload + : "ts-csharp-client-state"); + var serverPayload = Ascii(options.TryGetValue("server-payload", out var configuredServerPayload) + ? configuredServerPayload + : "csharp-server-state"); + var maxFragmentBytes = options.TryGetValue("max-fragment-bytes", out var configuredMaxFragmentBytes) + ? int.Parse(configuredMaxFragmentBytes, CultureInfo.InvariantCulture) + : (int?)null; using var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp); socket.Bind(new IPEndPoint(IPAddress.Parse(bindHost), bindPort)); @@ -595,7 +604,8 @@ static void RudpServeOnce(Dictionary options) Mode = CultNetRudpSocketMode.Server, ConnectionId = 0x33557799, InitialSequence = 100, - ResendDelayMs = 25 + ResendDelayMs = 25, + MaxFragmentBytes = maxFragmentBytes }); WriteJsonLine(new @@ -610,8 +620,8 @@ static void RudpServeOnce(Dictionary options) var frame = transport.ReceiveOnce(); if (frame != null) { - RequireRudpFrame(frame, "schema", "ts-csharp-client-state"); - transport.Send("schema", Ascii("csharp-server-state")); + RequireRudpFrameBytes(frame, "schema", expectedClientPayload); + transport.Send("schema", serverPayload); PollRudpAfterSend(transport, TimeSpan.FromMilliseconds(250)); WriteJsonLine(new { status = "ok" }); return; @@ -680,9 +690,14 @@ static void PollRudpAfterSend(CultNetRudpSocketTransportConnection transport, Ti } static void RequireRudpFrame(CultNetTransportFrame frame, string expectedChannelId, string expectedPayload) +{ + RequireRudpFrameBytes(frame, expectedChannelId, Ascii(expectedPayload)); +} + +static void RequireRudpFrameBytes(CultNetTransportFrame frame, string expectedChannelId, byte[] expectedPayload) { if (!string.Equals(frame.ChannelId, expectedChannelId, StringComparison.Ordinal) - || !frame.Payload.SequenceEqual(Ascii(expectedPayload))) + || !frame.Payload.SequenceEqual(expectedPayload)) { throw new InvalidOperationException( $"Unexpected RUDP frame: channel={frame.ChannelId}, payload={Convert.ToHexString(frame.Payload)}."); diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index e613f61d..7258310d 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -494,6 +494,42 @@ namespace GameCult.Networking.Tests Is.EqualTo(new[] { "second", "third" })); } + [Test] + public void RudpSession_FragmentsAndReassemblesReliableOrderedPayloads() + { + var sender = new CultNetRudpSession(new CultNetRudpSessionOptions + { + ConnectionId = 456, + InitialSequence = 1 + }); + var receiver = new CultNetRudpSession(new CultNetRudpSessionOptions + { + ConnectionId = 456, + InitialSequence = 100 + }); + sender.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 456, Sequence = 90, ChannelId = "control" }); + receiver.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 456, Sequence = 91, ChannelId = "control" }); + + var packets = sender.SendMany( + "schema", + Encoding.UTF8.GetBytes("fragment-me-please"), + new CultNetRudpSendOptions { Reliable = true, Ordered = true, NowMs = 10 }, + maxFragmentBytes: 5).ToArray(); + + Assert.That(packets, Has.Length.EqualTo(4)); + Assert.That(packets.Select(packet => packet.FragmentCount).ToArray(), Is.EqualTo(new ushort[] { 4, 4, 4, 4 })); + Assert.That(packets.Select(packet => packet.FragmentIndex).ToArray(), Is.EqualTo(new ushort[] { 0, 1, 2, 3 })); + Assert.That(packets.All(packet => packet.FragmentId == packets[0].FragmentId), Is.True); + + Assert.That(receiver.Receive(packets[0]).Delivered, Is.Empty); + Assert.That(receiver.Receive(packets[1]).Delivered, Is.Empty); + Assert.That(receiver.Receive(packets[2]).Delivered, Is.Empty); + var delivered = receiver.Receive(packets[3]).Delivered.ToArray(); + Assert.That(delivered, Has.Length.EqualTo(1)); + Assert.That(Encoding.UTF8.GetString(delivered[0].Payload), Is.EqualTo("fragment-me-please")); + Assert.That(delivered[0].Sequence, Is.EqualTo(packets[0].Sequence)); + } + [Test] public void RudpSocketTransport_HandshakesAndCarriesReliableOrderedSchemaFrames() { @@ -540,6 +576,46 @@ namespace GameCult.Networking.Tests Assert.That(server.Stats.FramesReceived, Is.EqualTo(1)); } + [Test] + public void RudpSocketTransport_CarriesFragmentedReliableOrderedSchemaFrames() + { + using var serverSocket = BindUdpSocket(); + using var clientSocket = BindUdpSocket(); + var serverEndPoint = serverSocket.LocalEndPoint!; + const uint connectionId = 0x10203041; + using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions + { + RuntimeId = "csharp-rudp-fragment-server", + Socket = serverSocket, + Mode = CultNetRudpSocketMode.Server, + ConnectionId = connectionId, + InitialSequence = 100, + ResendDelayMs = 25, + MaxFragmentBytes = 8 + }); + using var client = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions + { + RuntimeId = "csharp-rudp-fragment-client", + Socket = clientSocket, + Mode = CultNetRudpSocketMode.Client, + RemoteEndPoint = serverEndPoint, + ConnectionId = connectionId, + InitialSequence = 1, + ResendDelayMs = 25, + MaxFragmentBytes = 8 + }); + + var payload = Encoding.UTF8.GetBytes("this-schema-frame-is-larger-than-one-rudp-fragment"); + client.Connect(Encoding.UTF8.GetBytes("join")); + PumpRudpHandshake(client, server); + client.Send("schema", payload); + var serverFrame = ReceiveRudpFrame(server); + Assert.That(serverFrame.ChannelId, Is.EqualTo("schema")); + Assert.That(serverFrame.Payload, Is.EqualTo(payload)); + Assert.That(client.Stats.FramesSent, Is.EqualTo(1)); + Assert.That(server.Stats.FramesReceived, Is.EqualTo(1)); + } + [Test] public void CultNetSchemaMessageSerialization_RoundTrips_RawSnapshotResponse() {