Implement native UDP transport reliability (#52)
Some checks failed
Native code generation / deterministic (push) Failing after 7m48s
Imaging and meshing gate / native (push) Failing after 17s
JPEG 2000 feature / linux (push) Failing after 58s
Skia feature / linux (push) Failing after 1m33s

This commit is contained in:
2026-08-09 11:39:59 +00:00
parent b413cd6f4d
commit c7777d42f5
13 changed files with 2593 additions and 214 deletions

53
Cargo.lock generated
View File

@@ -369,6 +369,7 @@ dependencies = [
"libremetaverse-structured-data",
"libremetaverse-types",
"roxmltree",
"tokio",
]
[[package]]
@@ -575,6 +576,17 @@ dependencies = [
"simd-adler32",
]
[[package]]
name = "mio"
version = "1.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427"
dependencies = [
"libc",
"wasi",
"windows-sys",
]
[[package]]
name = "nom"
version = "7.1.3"
@@ -845,6 +857,16 @@ version = "0.4.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
[[package]]
name = "socket2"
version = "0.6.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4"
dependencies = [
"libc",
"windows-sys",
]
[[package]]
name = "stats_alloc"
version = "0.1.10"
@@ -884,6 +906,31 @@ dependencies = [
"xattr",
]
[[package]]
name = "tokio"
version = "1.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed"
dependencies = [
"libc",
"mio",
"pin-project-lite",
"socket2",
"tokio-macros",
"windows-sys",
]
[[package]]
name = "tokio-macros"
version = "2.7.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e"
dependencies = [
"proc-macro2",
"quote",
"syn 3.0.3",
]
[[package]]
name = "toml"
version = "1.1.4+spec-1.1.0"
@@ -958,6 +1005,12 @@ version = "0.9.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
[[package]]
name = "wasi"
version = "0.11.1+wasi-snapshot-preview1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
[[package]]
name = "wasm-bindgen"
version = "0.2.127"

View File

@@ -262,3 +262,13 @@ token and are shut down idempotently in network, manager, HTTP, then rate-limite
order. Injected clocks support deterministic tests, while `Debug` output omits
endpoint values and service internals. The ownership and executor requirements
are documented in [`docs/client-core.md`](docs/client-core.md).
The native Tokio UDP layer now implements the C# packet buffers and throttle
encoding plus bounded socket receive, coordination, and single-writer tasks.
It assigns wrapping protocol sequences, aggregates and consumes ACKs, retries
reliable packets with duplicate suppression, preserves zerocoding and the
1,200-byte MTU contract, applies independent task/texture/asset token buckets,
and exposes payload-free transport statistics. All queues, peer state,
zerocode expansion, ACK state, and reliable windows have explicit limits;
linked cancellation and final drop release every socket task. The executor,
wire, backpressure, retry, and security contracts are documented in
[`docs/udp-transport.md`](docs/udp-transport.md).

View File

@@ -4,7 +4,7 @@ Generated by `python3 tools/generate_api_shims.py`; do not edit by hand.
| Assembly | Types | Members | Status |
|---|---:|---:|---|
| `LibreMetaverse` | 2,711 | 27,281 | native implementation: 28 types / 13,428 members; remaining surface is callable failure-only shims |
| `LibreMetaverse` | 2,711 | 27,281 | native implementation: 32 types / 13,459 members; remaining surface is callable failure-only shims |
| `LibreMetaverse.Imaging.Abstractions` | 3 | 20 | native implementation: 3 types / 20 members; no generated shims remain |
| `LibreMetaverse.Imaging.Skia` | 1 | 3 | native implementation: 1 type / 3 members; no generated shims remain |
| `LibreMetaverse.LslTools` | 164 | 768 | callable failure-only shim |

View File

@@ -18,6 +18,10 @@ libremetaverse-imaging = { path = "../libremetaverse-imaging" }
libremetaverse-structured-data = { path = "../libremetaverse-structured-data" }
libremetaverse-types = { path = "../libremetaverse-types" }
roxmltree = "0.21.1"
tokio = { version = "1.47.1", features = ["macros", "net", "rt", "sync", "time"] }
[dev-dependencies]
tokio = { version = "1.47.1", features = ["macros", "net", "rt-multi-thread", "sync", "test-util", "time"] }
[lints]
workspace = true

View File

@@ -113,6 +113,7 @@ struct ClientRuntime {
state: AtomicU8,
cancellation: CancellationTokenSource,
services: Mutex<Vec<ServiceRegistration>>,
agent_throttle_sender: Mutex<Option<Arc<dyn crate::udp_transport::AgentThrottleSender>>>,
shutdown_complete: Condvar,
shutdown_wait: Mutex<()>,
}
@@ -132,6 +133,7 @@ impl ClientRuntime {
})
.collect(),
),
agent_throttle_sender: Mutex::new(None),
shutdown_complete: Condvar::new(),
shutdown_wait: Mutex::new(()),
}
@@ -200,6 +202,10 @@ impl ClientRuntime {
}
}
services.clear();
self.agent_throttle_sender
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take();
let shutdown_wait = self
.shutdown_wait
.lock()
@@ -257,6 +263,41 @@ impl GridClient {
self.runtime.register(service)
}
/// Installs the network-owned callback used by mapped `AgentThrottle.Set`.
///
/// # Errors
///
/// Registration fails after client shutdown begins.
pub fn set_agent_throttle_sender(
&mut self,
sender: Arc<dyn crate::udp_transport::AgentThrottleSender>,
) -> Result<(), ClientCoreError> {
let mut registered = self
.runtime
.agent_throttle_sender
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let state = self.lifecycle_state();
if state != ClientLifecycleState::Active {
return Err(ClientCoreError::InvalidLifecycle {
operation: "register the agent throttle sender",
state,
});
}
*registered = Some(sender);
Ok(())
}
pub(crate) fn agent_throttle_sender(
&self,
) -> Option<Arc<dyn crate::udp_transport::AgentThrottleSender>> {
self.runtime
.agent_throttle_sender
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone()
}
pub(crate) fn shutdown(&self) -> Result<(), ClientCoreError> {
self.runtime.shutdown()
}

View File

@@ -3333,102 +3333,20 @@ impl AgentState {
}
/// C# type: `T:LibreMetaverse.AgentThrottle`.
pub struct AgentThrottle;
impl AgentThrottle {
/// C# member: `M:LibreMetaverse.AgentThrottle.#ctor(LibreMetaverse.GridClient)`.
pub fn new_with_grid_client(client: libremetaverse::GridClient) -> Result<Self, crate::Error> {
libremetaverse_types::not_implemented(
"M:LibreMetaverse.AgentThrottle.#ctor(LibreMetaverse.GridClient)",
)
}
/// C# member: `M:LibreMetaverse.AgentThrottle.#ctor(System.Byte[],System.Int32)`.
pub fn new_with_bytes_int32(data: Vec<u8>, pos: i32) -> Result<Self, crate::Error> {
libremetaverse_types::not_implemented(
"M:LibreMetaverse.AgentThrottle.#ctor(System.Byte[],System.Int32)",
)
}
/// C# member: `M:LibreMetaverse.AgentThrottle.Set`.
pub fn set_with_method(&self) -> Result<(), crate::Error> {
libremetaverse_types::not_implemented("M:LibreMetaverse.AgentThrottle.Set")
}
/// C# member: `M:LibreMetaverse.AgentThrottle.Set(LibreMetaverse.Simulator)`.
pub fn set_with_simulator(
&self,
simulator: Option<libremetaverse::Simulator>,
) -> Result<(), crate::Error> {
libremetaverse_types::not_implemented(
"M:LibreMetaverse.AgentThrottle.Set(LibreMetaverse.Simulator)",
)
}
/// C# member: `M:LibreMetaverse.AgentThrottle.ToBytes`.
pub fn to_bytes(&self) -> Result<Vec<u8>, crate::Error> {
libremetaverse_types::not_implemented("M:LibreMetaverse.AgentThrottle.ToBytes")
}
/// C# member: `P:LibreMetaverse.AgentThrottle.Asset`.
pub fn asset(&self) -> f32 {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Asset")
}
/// Setter for C# member: `P:LibreMetaverse.AgentThrottle.Asset`.
pub fn set_asset(&mut self, value: f32) {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Asset")
}
/// C# member: `P:LibreMetaverse.AgentThrottle.Cloud`.
pub fn cloud(&self) -> f32 {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Cloud")
}
/// Setter for C# member: `P:LibreMetaverse.AgentThrottle.Cloud`.
pub fn set_cloud(&mut self, value: f32) {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Cloud")
}
/// C# member: `P:LibreMetaverse.AgentThrottle.Land`.
pub fn land(&self) -> f32 {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Land")
}
/// Setter for C# member: `P:LibreMetaverse.AgentThrottle.Land`.
pub fn set_land(&mut self, value: f32) {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Land")
}
/// C# member: `P:LibreMetaverse.AgentThrottle.Resend`.
pub fn resend(&self) -> f32 {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Resend")
}
/// Setter for C# member: `P:LibreMetaverse.AgentThrottle.Resend`.
pub fn set_resend(&mut self, value: f32) {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Resend")
}
/// C# member: `P:LibreMetaverse.AgentThrottle.Task`.
pub fn task(&self) -> f32 {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Task")
}
/// Setter for C# member: `P:LibreMetaverse.AgentThrottle.Task`.
pub fn set_task(&mut self, value: f32) {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Task")
}
/// C# member: `P:LibreMetaverse.AgentThrottle.Texture`.
pub fn texture(&self) -> f32 {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Texture")
}
/// Setter for C# member: `P:LibreMetaverse.AgentThrottle.Texture`.
pub fn set_texture(&mut self, value: f32) {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Texture")
}
/// C# member: `P:LibreMetaverse.AgentThrottle.Total`.
pub fn total(&self) -> f32 {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Total")
}
/// Setter for C# member: `P:LibreMetaverse.AgentThrottle.Total`.
pub fn set_total(&mut self, value: f32) {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Total")
}
/// C# member: `P:LibreMetaverse.AgentThrottle.Wind`.
pub fn wind(&self) -> f32 {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Wind")
}
/// Setter for C# member: `P:LibreMetaverse.AgentThrottle.Wind`.
pub fn set_wind(&mut self, value: f32) {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.AgentThrottle.Wind")
}
}
pub use crate::udp_transport::AgentThrottle;
/// C# type: `T:LibreMetaverse.AgentWearablesReplyEventArgs`.
pub struct AgentWearablesReplyEventArgs;
@@ -13666,21 +13584,9 @@ pub enum ImageType {
}
/// C# type: `T:LibreMetaverse.IncomingPacketIDCollection`.
pub struct IncomingPacketIDCollection;
impl IncomingPacketIDCollection {
/// C# member: `M:LibreMetaverse.IncomingPacketIDCollection.#ctor(System.Int32)`.
pub fn new(capacity: i32) -> Result<Self, crate::Error> {
libremetaverse_types::not_implemented(
"M:LibreMetaverse.IncomingPacketIDCollection.#ctor(System.Int32)",
)
}
/// C# member: `M:LibreMetaverse.IncomingPacketIDCollection.TryEnqueue(System.UInt32)`.
pub fn try_enqueue(&self, ack: u32) -> bool {
libremetaverse_types::unimplemented_api!(
"M:LibreMetaverse.IncomingPacketIDCollection.TryEnqueue(System.UInt32)"
)
}
}
pub use crate::udp_transport::IncomingPacketIDCollection;
/// C# type: `T:LibreMetaverse.InitiateDownloadEventArgs`.
pub struct InitiateDownloadEventArgs;
@@ -27995,106 +27901,26 @@ pub use crate::foliage_catalog::TreeDefinition;
pub use crate::foliage_catalog::TreeDefinitions;
/// C# type: `T:LibreMetaverse.UDPBase`.
pub struct UDPBase;
impl UDPBase {
/// C# member: `M:LibreMetaverse.UDPBase.AsyncBeginSend(LibreMetaverse.UDPPacketBuffer)`.
pub fn async_begin_send(
&self,
buf: libremetaverse::UDPPacketBuffer,
) -> Result<(), crate::Error> {
libremetaverse_types::not_implemented(
"M:LibreMetaverse.UDPBase.AsyncBeginSend(LibreMetaverse.UDPPacketBuffer)",
)
}
/// C# member: `M:LibreMetaverse.UDPBase.Start`.
pub fn start(&self) -> Result<(), crate::Error> {
libremetaverse_types::not_implemented("M:LibreMetaverse.UDPBase.Start")
}
/// C# member: `M:LibreMetaverse.UDPBase.Stop`.
pub fn stop(&self) -> Result<(), crate::Error> {
libremetaverse_types::not_implemented("M:LibreMetaverse.UDPBase.Stop")
}
/// C# member: `P:LibreMetaverse.UDPBase.IsRunning`.
pub fn is_running(&self) -> bool {
libremetaverse_types::unimplemented_api!("P:LibreMetaverse.UDPBase.IsRunning")
}
}
pub use crate::udp_transport::UDPBase;
/// C# type: `T:LibreMetaverse.UDPPacketBuffer`.
pub struct UDPPacketBuffer {
/// C# member: `F:LibreMetaverse.UDPPacketBuffer.Data`.
pub data: Vec<u8>,
/// C# member: `F:LibreMetaverse.UDPPacketBuffer.DataLength`.
pub data_length: i32,
/// C# member: `F:LibreMetaverse.UDPPacketBuffer.RemoteEndPoint`.
pub remote_end_point: std::net::SocketAddr,
}
impl UDPPacketBuffer {
/// C# member: `F:LibreMetaverse.UDPPacketBuffer.DEFAULT_BUFFER_SIZE`.
pub const DEFAULT_BUFFER_SIZE: i32 = 4096;
/// C# member: `F:LibreMetaverse.UDPPacketBuffer.Data`.
/// C# member: `F:LibreMetaverse.UDPPacketBuffer.DataLength`.
/// C# member: `F:LibreMetaverse.UDPPacketBuffer.RemoteEndPoint`.
/// C# member: `M:LibreMetaverse.UDPPacketBuffer.#ctor`.
pub fn new_with_constructor() -> Result<Self, crate::Error> {
libremetaverse_types::not_implemented("M:LibreMetaverse.UDPPacketBuffer.#ctor")
}
/// C# member: `M:LibreMetaverse.UDPPacketBuffer.#ctor(System.Byte[],System.Int32,System.Net.IPEndPoint,System.Int32)`.
pub fn new_with_bytes_int32_ip_end_point_int32(
buffer: Vec<u8>,
buffer_size: i32,
destination: std::net::SocketAddr,
category: i32,
) -> Result<Self, crate::Error> {
libremetaverse_types::not_implemented(
"M:LibreMetaverse.UDPPacketBuffer.#ctor(System.Byte[],System.Int32,System.Net.IPEndPoint,System.Int32)",
)
}
/// C# member: `M:LibreMetaverse.UDPPacketBuffer.#ctor(System.Net.IPEndPoint)`.
pub fn new_with_ip_end_point(end_point: std::net::SocketAddr) -> Result<Self, crate::Error> {
libremetaverse_types::not_implemented(
"M:LibreMetaverse.UDPPacketBuffer.#ctor(System.Net.IPEndPoint)",
)
}
/// C# member: `M:LibreMetaverse.UDPPacketBuffer.#ctor(System.Net.IPEndPoint,System.Byte[])`.
pub fn new_with_ip_end_point_bytes(
end_point: std::net::SocketAddr,
data: Vec<u8>,
) -> Result<Self, crate::Error> {
libremetaverse_types::not_implemented(
"M:LibreMetaverse.UDPPacketBuffer.#ctor(System.Net.IPEndPoint,System.Byte[])",
)
}
/// C# member: `M:LibreMetaverse.UDPPacketBuffer.#ctor(System.Net.IPEndPoint,System.Int32)`.
pub fn new_with_ip_end_point_int32(
end_point: std::net::SocketAddr,
buffer_size: i32,
) -> Result<Self, crate::Error> {
libremetaverse_types::not_implemented(
"M:LibreMetaverse.UDPPacketBuffer.#ctor(System.Net.IPEndPoint,System.Int32)",
)
}
/// C# member: `M:LibreMetaverse.UDPPacketBuffer.CopyFrom(System.Array)`.
pub fn copy_from_with_array(
&self,
src: libremetaverse_types::compat::Array,
) -> Result<(), crate::Error> {
libremetaverse_types::not_implemented(
"M:LibreMetaverse.UDPPacketBuffer.CopyFrom(System.Array)",
)
}
/// C# member: `M:LibreMetaverse.UDPPacketBuffer.CopyFrom(System.Array,System.Int32)`.
pub fn copy_from_with_array_int32(
&self,
src: libremetaverse_types::compat::Array,
length: i32,
) -> Result<(), crate::Error> {
libremetaverse_types::not_implemented(
"M:LibreMetaverse.UDPPacketBuffer.CopyFrom(System.Array,System.Int32)",
)
}
/// C# member: `M:LibreMetaverse.UDPPacketBuffer.ResetEndpoint`.
pub fn reset_endpoint(&self) -> Result<(), crate::Error> {
libremetaverse_types::not_implemented("M:LibreMetaverse.UDPPacketBuffer.ResetEndpoint")
}
}
pub use crate::udp_transport::UDPPacketBuffer;
/// C# type: `T:LibreMetaverse.UUIDNameReplyEventArgs`.
pub struct UUIDNameReplyEventArgs;

View File

@@ -20,6 +20,7 @@ mod skeleton;
#[rustfmt::skip] // Deterministic machine output is formatted by the pinned generator.
mod skeleton_catalog;
mod targa;
mod udp_transport;
#[rustfmt::skip] // Deterministic machine output is formatted by the pinned generator.
mod visual_catalog;
@@ -204,6 +205,10 @@ pub use libremetaverse_imaging as imaging_abstractions;
pub use libremetaverse_structured_data as structured_data;
pub use libremetaverse_types as types;
pub use libremetaverse_types::Error;
pub use udp_transport::{
AgentThrottleSender, UdpPacketHandler, UdpThrottleCategory, UdpTransportConfig,
UdpTransportError, UdpTransportStats,
};
/// Rust object-safe view of the C# `InventoryBase` inheritance hierarchy.
pub trait InventoryObjectClass: std::any::Any {

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,511 @@
use libremetaverse::packets::{
AgentThrottlePacket, CompletePingCheckPacket, Packet, PacketAckPacket,
PacketAckPacketPacketsBlock, PacketType, UseCircuitCodePacket,
};
use libremetaverse::{
AgentThrottle, AgentThrottleSender, GridClient, Helpers, IncomingPacketIDCollection, Simulator,
UDPBase, UDPPacketBuffer, UdpPacketHandler, UdpTransportConfig, UdpTransportError,
};
use libremetaverse_types::Error;
use libremetaverse_types::compat::{Array, CancellationToken, Object};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::net::UdpSocket;
#[derive(Default)]
struct RecordingHandler {
received: Mutex<Vec<Vec<u8>>>,
sent: Mutex<Vec<Vec<u8>>>,
drops: Mutex<usize>,
}
impl UdpPacketHandler for RecordingHandler {
fn packet_received(&self, buffer: UDPPacketBuffer) {
self.received
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.push(buffer.data[..usize::try_from(buffer.data_length).unwrap()].to_vec());
}
fn packet_sent(&self, buffer: UDPPacketBuffer, bytes_sent: usize) {
self.sent
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.push(buffer.data[..bytes_sent].to_vec());
}
fn packet_dropped(&self) {
*self
.drops
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) += 1;
}
}
fn loopback() -> SocketAddr {
SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)
}
fn reliable_packet() -> Vec<u8> {
let mut packet = UseCircuitCodePacket::new_with_constructor().expect("packet constructor");
packet.circuit_code.code = 0x1122_3344;
packet.to_bytes_with_method().expect("packet bytes")
}
fn packet_ack(sequence: u32) -> Vec<u8> {
let mut packet = PacketAckPacket::new_with_constructor().expect("ACK constructor");
packet.packets = vec![PacketAckPacketPacketsBlock { id: sequence }];
let mut bytes = packet.to_bytes_with_method().expect("ACK bytes");
bytes[0] &= !Helpers::MSG_RELIABLE;
bytes
}
fn complete_ping_packet() -> Vec<u8> {
let mut packet = CompletePingCheckPacket::new_with_constructor().expect("ping constructor");
packet.ping_id.ping_id = 7;
packet.to_bytes_with_method().expect("ping bytes")
}
async fn wait_until(mut condition: impl FnMut() -> bool) {
tokio::time::timeout(Duration::from_secs(2), async {
while !condition() {
tokio::task::yield_now().await;
tokio::time::sleep(Duration::from_millis(1)).await;
}
})
.await
.expect("condition timed out");
}
#[test]
fn packet_buffer_constructors_and_copy_match_the_reference() {
let endpoint = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 13000);
let mut packet = UDPPacketBuffer::new_with_ip_end_point_int32(endpoint, 4).unwrap();
assert_eq!(packet.data, vec![0; 4]);
assert_eq!(packet.data_length, 0);
packet
.copy_from_with_array(Array(vec![Object::Bytes(vec![1, 2, 3, 4])]))
.unwrap();
assert_eq!(packet.data, vec![1, 2, 3, 4]);
assert_eq!(packet.data_length, 0);
assert_eq!(
packet.copy_from_slice(&[1, 2, 3, 4, 5]),
Err(Error::IndexOutOfRange)
);
packet.reset_endpoint().unwrap();
assert_eq!(packet.remote_end_point, "0.0.0.0:0".parse().unwrap());
let borrowed = vec![9, 8, 7];
let packet = UDPPacketBuffer::new_with_ip_end_point_bytes(endpoint, borrowed).unwrap();
assert_eq!(packet.data, vec![9, 8, 7]);
assert_eq!(packet.data_length, 0);
}
#[test]
fn incoming_packet_archive_is_bounded_and_rejects_duplicates() {
let archive = IncomingPacketIDCollection::new(3).unwrap();
assert!(archive.try_enqueue(1));
assert!(!archive.try_enqueue(1));
assert!(archive.try_enqueue(2));
// The C# ring keeps one slot empty, so adding the third item evicts 1.
assert!(archive.try_enqueue(3));
assert!(archive.try_enqueue(1));
assert_eq!(
IncomingPacketIDCollection::new(0).unwrap_err(),
Error::Argument
);
}
#[test]
fn agent_throttle_clamps_and_round_trips_golden_little_endian_values() {
let throttle = AgentThrottle::default();
assert_eq!(throttle.resend(), 150_000.0);
assert_eq!(throttle.land(), 170_000.0);
assert_eq!(throttle.wind(), 34_000.0);
assert_eq!(throttle.cloud(), 34_000.0);
assert_eq!(throttle.task(), 360_448.0);
assert_eq!(throttle.texture(), 360_448.0);
assert_eq!(throttle.asset(), 220_000.0);
assert_eq!(throttle.total(), 1_328_896.0);
let bytes = throttle.to_bytes().unwrap();
assert_eq!(bytes.len(), 28);
let decoded = AgentThrottle::new_with_bytes_int32(bytes, 0).unwrap();
assert_eq!(decoded.to_bytes().unwrap(), throttle.to_bytes().unwrap());
let mut limits = AgentThrottle::default();
limits.set_total(-1.0);
assert_eq!(limits.resend(), 10_000.0);
assert_eq!(limits.task(), 4_000.0);
assert_eq!(limits.land(), 0.0);
}
#[derive(Default)]
struct RecordingThrottleSender(Mutex<Vec<Vec<u8>>>);
impl AgentThrottleSender for RecordingThrottleSender {
fn send_throttle(
&self,
throttle_bytes: &[u8],
_simulator: Option<&Simulator>,
) -> Result<(), Error> {
self.0.lock().unwrap().push(throttle_bytes.to_vec());
Ok(())
}
}
#[test]
fn mapped_agent_throttle_set_uses_the_client_network_binding() {
let sender = Arc::new(RecordingThrottleSender::default());
let mut client = GridClient::new().unwrap();
client.set_agent_throttle_sender(sender.clone()).unwrap();
let throttle = AgentThrottle::new_with_grid_client(client).unwrap();
throttle.set_with_method().unwrap();
assert_eq!(
sender.0.lock().unwrap().as_slice(),
&[throttle.to_bytes().unwrap()]
);
}
#[test]
fn construction_is_runtime_neutral_and_start_requires_an_injected_runtime() {
let transport = UDPBase::client_with_defaults("127.0.0.1:13000".parse().unwrap()).unwrap();
assert!(!transport.is_running());
assert_eq!(transport.start(), Err(Error::InvalidOperation));
assert!(!transport.is_running());
}
#[tokio::test(flavor = "current_thread")]
async fn bounded_command_queue_reports_backpressure_without_unbounded_buffering() {
let server = UdpSocket::bind(loopback()).await.unwrap();
let config = UdpTransportConfig {
command_queue_capacity: 1,
..UdpTransportConfig::default()
};
let transport = UDPBase::client(
server.local_addr().unwrap(),
config,
Arc::new(RecordingHandler::default()),
CancellationToken::default(),
)
.unwrap();
transport.start().unwrap();
// A current-thread runtime does not poll spawned tasks until this test
// yields, making the capacity-one boundary deterministic.
transport
.try_send_packet(
complete_ping_packet(),
server.local_addr().unwrap(),
PacketType::CompletePingCheck,
false,
)
.unwrap();
assert!(matches!(
transport.try_send_packet(
complete_ping_packet(),
server.local_addr().unwrap(),
PacketType::CompletePingCheck,
false,
),
Err(UdpTransportError::Backpressure)
));
assert_eq!(transport.stats().dropped_send_queue, 1);
transport.stop_async().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn local_socket_send_assigns_sequence_and_ack_stops_resends() {
let server = UdpSocket::bind(loopback()).await.unwrap();
let server_address = server.local_addr().unwrap();
let handler = Arc::new(RecordingHandler::default());
let config = UdpTransportConfig {
resend_timeout: Duration::from_millis(60),
network_tick_interval: Duration::from_millis(10),
..UdpTransportConfig::default()
};
let transport = UDPBase::client(
server_address,
config,
handler.clone(),
CancellationToken::default(),
)
.unwrap();
transport.start().unwrap();
let sequence = transport
.send_packet(
reliable_packet(),
PacketType::UseCircuitCode,
false,
CancellationToken::default(),
)
.await
.unwrap();
assert_eq!(sequence, 1);
let mut receive = [0_u8; 4096];
let (length, client_address) =
tokio::time::timeout(Duration::from_secs(2), server.recv_from(&mut receive))
.await
.unwrap()
.unwrap();
assert_eq!(u32::from_be_bytes(receive[1..5].try_into().unwrap()), 1);
assert_ne!(receive[0] & Helpers::MSG_RELIABLE, 0);
server
.send_to(&packet_ack(sequence), client_address)
.await
.unwrap();
wait_until(|| transport.stats().acknowledgements_received == 1).await;
tokio::time::sleep(Duration::from_millis(100)).await;
assert_eq!(transport.stats().resent_datagrams, 0);
assert_eq!(handler.sent.lock().unwrap().len(), 1);
assert_eq!(length, handler.sent.lock().unwrap()[0].len());
transport.stop_async().await;
assert!(!transport.is_running());
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reliable_packets_resend_with_same_sequence_then_expire() {
let server = UdpSocket::bind(loopback()).await.unwrap();
let config = UdpTransportConfig {
resend_timeout: Duration::from_millis(20),
network_tick_interval: Duration::from_millis(5),
max_resend_count: 2,
..UdpTransportConfig::default()
};
let transport = UDPBase::client(
server.local_addr().unwrap(),
config,
Arc::new(RecordingHandler::default()),
CancellationToken::default(),
)
.unwrap();
transport.start().unwrap();
transport
.send_packet(
reliable_packet(),
PacketType::UseCircuitCode,
false,
CancellationToken::default(),
)
.await
.unwrap();
let mut datagrams = Vec::new();
let mut receive = [0_u8; 4096];
for _ in 0..3 {
let (length, _) =
tokio::time::timeout(Duration::from_secs(1), server.recv_from(&mut receive))
.await
.unwrap()
.unwrap();
datagrams.push(receive[..length].to_vec());
}
assert_eq!(
datagrams
.iter()
.map(|bytes| u32::from_be_bytes(bytes[1..5].try_into().unwrap()))
.collect::<Vec<_>>(),
vec![1, 1, 1]
);
assert_eq!(datagrams[0][0] & Helpers::MSG_RESENT, 0);
assert_ne!(datagrams[1][0] & Helpers::MSG_RESENT, 0);
wait_until(|| transport.stats().failed_resends == 1).await;
assert_eq!(transport.stats().resent_datagrams, 2);
transport.stop_async().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn incoming_reliable_packets_are_acked_and_duplicates_are_not_dispatched() {
let server = UdpSocket::bind(loopback()).await.unwrap();
let handler = Arc::new(RecordingHandler::default());
let config = UdpTransportConfig {
max_pending_acks: 2,
network_tick_interval: Duration::from_secs(1),
..UdpTransportConfig::default()
};
let transport = UDPBase::client(
server.local_addr().unwrap(),
config,
handler.clone(),
CancellationToken::default(),
)
.unwrap();
transport.start().unwrap();
let client_address = transport.local_address().unwrap();
let mut first = reliable_packet();
first[1..5].copy_from_slice(&10_u32.to_be_bytes());
let mut second = reliable_packet();
second[1..5].copy_from_slice(&12_u32.to_be_bytes());
let mut late = reliable_packet();
late[1..5].copy_from_slice(&11_u32.to_be_bytes());
server.send_to(&first, client_address).await.unwrap();
server.send_to(&first, client_address).await.unwrap();
server.send_to(&second, client_address).await.unwrap();
server.send_to(&late, client_address).await.unwrap();
let mut receive = [0_u8; 4096];
let mut ack_groups = Vec::new();
for _ in 0..2 {
let (length, _) =
tokio::time::timeout(Duration::from_secs(2), server.recv_from(&mut receive))
.await
.unwrap()
.unwrap();
let mut position = 0;
let ack = PacketAckPacket::new_with_bytes_int32(receive[..length].to_vec(), &mut position)
.unwrap();
ack_groups.push(ack.packets.iter().map(|block| block.id).collect::<Vec<_>>());
}
assert_eq!(ack_groups, vec![vec![10, 10], vec![12, 11]]);
wait_until(|| handler.received.lock().unwrap().len() == 3).await;
assert_eq!(transport.stats().duplicate_datagrams, 1);
assert_eq!(transport.stats().sequence_gaps, 1);
assert_eq!(transport.stats().out_of_order_datagrams, 1);
assert_eq!(transport.stats().acknowledgements_sent, 4);
transport.stop_async().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn pending_ack_is_piggybacked_before_the_mtu_boundary() {
let server = UdpSocket::bind(loopback()).await.unwrap();
let handler = Arc::new(RecordingHandler::default());
let config = UdpTransportConfig {
network_tick_interval: Duration::from_secs(10),
..UdpTransportConfig::default()
};
let transport = UDPBase::client(
server.local_addr().unwrap(),
config,
handler.clone(),
CancellationToken::default(),
)
.unwrap();
transport.start().unwrap();
let mut reliable = reliable_packet();
reliable[1..5].copy_from_slice(&20_u32.to_be_bytes());
server
.send_to(&reliable, transport.local_address().unwrap())
.await
.unwrap();
wait_until(|| handler.received.lock().unwrap().len() == 1).await;
transport
.send_packet(
complete_ping_packet(),
PacketType::CompletePingCheck,
false,
CancellationToken::default(),
)
.await
.unwrap();
let mut receive = [0_u8; 4096];
let (length, _) = server.recv_from(&mut receive).await.unwrap();
assert_ne!(receive[0] & Helpers::MSG_APPENDED_ACKS, 0);
let mut end = i32::try_from(length).unwrap() - 1;
let decoded = Packet::build_packet_with_bytes_int32_bytes(
receive[..length].to_vec(),
&mut end,
vec![0; 8192],
)
.unwrap();
assert_eq!(decoded.type_, PacketType::CompletePingCheck);
assert_eq!(decoded.header.ack_list, Some(vec![20]));
transport.stop_async().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn zerocoding_mtu_malformed_input_and_cancellation_are_bounded() {
let server = UdpSocket::bind(loopback()).await.unwrap();
let handler = Arc::new(RecordingHandler::default());
let transport = UDPBase::client(
server.local_addr().unwrap(),
UdpTransportConfig::default(),
handler.clone(),
CancellationToken::default(),
)
.unwrap();
transport.start().unwrap();
let client_address = transport.local_address().unwrap();
let mut packet = AgentThrottlePacket::new_with_constructor().unwrap();
packet.throttle.throttles = vec![0; 28];
let raw = packet.to_bytes_with_method().unwrap();
transport
.send_packet(
raw.clone(),
PacketType::AgentThrottle,
true,
CancellationToken::default(),
)
.await
.unwrap();
let mut receive = [0_u8; 4096];
let (length, _) = server.recv_from(&mut receive).await.unwrap();
assert_ne!(receive[0] & Helpers::MSG_ZEROCODED, 0);
assert!(length < raw.len());
let mut end = i32::try_from(length).unwrap() - 1;
Packet::build_packet_with_bytes_int32_bytes(
receive[..length].to_vec(),
&mut end,
vec![0; 8192],
)
.unwrap();
assert_eq!(
transport
.send_packet(
vec![0; 1201],
PacketType::AgentThrottle,
false,
CancellationToken::default(),
)
.await,
Err(UdpTransportError::MtuExceeded)
);
for malformed in [vec![], vec![0x40], vec![0x40, 0, 0, 0, 1, 0, 0xfe]] {
server.send_to(&malformed, client_address).await.unwrap();
}
wait_until(|| transport.stats().malformed_datagrams >= 3).await;
assert!(handler.received.lock().unwrap().is_empty());
transport.stop_async().await;
assert!(!transport.is_running());
assert_eq!(
transport
.send_packet(
reliable_packet(),
PacketType::UseCircuitCode,
false,
CancellationToken::default(),
)
.await,
Err(UdpTransportError::NotRunning)
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn dropping_the_last_transport_handle_releases_its_socket_tasks() {
let server = UdpSocket::bind(loopback()).await.unwrap();
let transport = UDPBase::client_with_defaults(server.local_addr().unwrap()).unwrap();
transport.start().unwrap();
let local_address = transport.local_address().unwrap();
drop(transport);
tokio::time::timeout(Duration::from_secs(2), async {
loop {
match UdpSocket::bind(local_address).await {
Ok(rebound) => break rebound,
Err(_) => tokio::time::sleep(Duration::from_millis(1)).await,
}
}
})
.await
.expect("transport tasks retained the UDP socket after drop");
}

77
docs/udp-transport.md Normal file
View File

@@ -0,0 +1,77 @@
# Native UDP transport
`libremetaverse::UDPBase` is the native Tokio implementation of the
LibreMetaverse UDP transport slice. Construction is runtime-neutral: it does
not bind a socket, start a thread, or create an executor. Call `start` from an
entered Tokio runtime after constructing either a client transport with
`UDPBase::client` or a server transport with `UDPBase::server`.
The implementation uses three bounded queues:
- one socket receive task writes complete datagrams to the receive queue with
drop-on-full backpressure;
- one coordinator validates packet framing, owns per-peer sequence, duplicate,
pending-ACK, and reliable resend state, and writes prepared datagrams to a
bounded writer queue;
- one writer task is the only socket send owner and applies the per-category
token buckets before calling `send_to`.
The default limits mirror the golden C# settings: 512 receive and command
entries, a 1,000-entry packet archive, ten ACKs before an immediate standalone
ACK, a 500 ms network tick, a 4 second resend timeout, and three retries.
Additional explicit limits bound the send queue, reliable window, ACK queue,
peer table, decoded zerocode buffer, and all allocations derived from incoming
datagrams. Client-mode source validation matches the reference by accepting
only the configured simulator IP address; replies continue to use the
configured endpoint.
## Wire behavior
Every newly prepared packet receives the next four-byte big-endian sequence,
using the full `u32` header width and wrapping through zero like the C#
`Interlocked.Increment(Int32)` plus `uint` cast. Reliable packets remain in the
bounded per-peer window until either an appended ACK or `PacketAck` block
removes them. Timed-out entries are retransmitted with `MSG_RESENT` and the
same sequence, then removed after the configured retry count. Reliable incoming
packets are ACKed even when duplicate; duplicate callbacks are suppressed.
Out-of-order packets are delivered immediately, as in the C# client, while
gaps and late arrivals are counted separately.
Pending ACKs are appended in big-endian form while the strict reference
`data_length + 5 < MTU` condition holds. Remaining ACKs are emitted as an
unreliable `PacketAck`. Zerocoding occurs before ACK appending. If zerocoding
would expand a packet beyond the 1,200-byte protocol MTU, the zerocode flag is
cleared and the original packet is sent. The transport does not invent an
application fragmentation format: packet codecs must use their existing
`ToBytesMultiple` behavior, and an oversized unsplit outbound payload returns
`UdpTransportError::MtuExceeded`.
Incoming storage is 4 KiB, matching `UDPPacketBuffer.DEFAULT_BUFFER_SIZE`, and
the receiver allocates one extra detection byte so an oversized datagram is
rejected rather than silently treated as complete. Packet parsing and
zerodecoding use fixed configured ceilings. Malformed or unknown packets are
counted and discarded without invoking application callbacks.
## Throttling, cancellation, and diagnostics
`AgentThrottle` reproduces all seven C# clamps and its 28-byte little-endian
wire layout. Outgoing control/handshake packets bypass throttling; task,
texture, and asset packets use independent 100 ms token buckets with the same
four-period burst and 200-byte minimum replenishment policy as the reference.
`update_throttle` swaps all three buckets in writer order.
All tasks share a linked cancellation source. `stop_async` cancels and joins
the receive, coordinator, and writer tasks; `stop` and final drop cancel and
abort them for synchronous C# compatibility. A stopped transport can be
started again with fresh task state unless its parent token is cancelled.
`UdpTransportStats` contains counts and byte totals only. The transport does
not log datagram bodies, credentials, capability URLs, or socket error text;
its `Debug` implementation also omits endpoint values. The implementation uses
only portable Rust and Tokio socket APIs on Linux, Windows, and macOS.
The isolated loopback suite is:
```sh
cargo test -p libremetaverse --test udp_transport --no-default-features
```

View File

@@ -194,6 +194,7 @@ dependencies = [
"libremetaverse-structured-data",
"libremetaverse-types",
"roxmltree",
"tokio",
]
[[package]]
@@ -337,6 +338,17 @@ dependencies = [
"libremetaverse-voice-webrtc",
]
[[package]]
name = "mio"
version = "1.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427"
dependencies = [
"libc",
"wasi",
"windows-sys",
]
[[package]]
name = "once_cell"
version = "1.21.4"
@@ -458,6 +470,16 @@ version = "0.4.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
[[package]]
name = "socket2"
version = "0.6.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4"
dependencies = [
"libc",
"windows-sys",
]
[[package]]
name = "syn"
version = "2.0.119"
@@ -480,6 +502,31 @@ dependencies = [
"unicode-ident",
]
[[package]]
name = "tokio"
version = "1.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed"
dependencies = [
"libc",
"mio",
"pin-project-lite",
"socket2",
"tokio-macros",
"windows-sys",
]
[[package]]
name = "tokio-macros"
version = "2.7.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e"
dependencies = [
"proc-macro2",
"quote",
"syn 3.0.3",
]
[[package]]
name = "typenum"
version = "1.20.1"
@@ -509,6 +556,12 @@ version = "0.9.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
[[package]]
name = "wasi"
version = "0.11.1+wasi-snapshot-preview1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
[[package]]
name = "wasm-bindgen"
version = "0.2.127"
@@ -554,6 +607,21 @@ dependencies = [
"unicode-ident",
]
[[package]]
name = "windows-link"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
[[package]]
name = "windows-sys"
version = "0.61.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
dependencies = [
"windows-link",
]
[[package]]
name = "zmij"
version = "1.0.23"

View File

@@ -106,8 +106,8 @@ fn client_and_settings_construct_without_starting_runtime_work() {
member_id(CapsRateLimiter::new_with_constructor()),
"M:LibreMetaverse.CapsRateLimiter.#ctor"
);
assert_eq!(
member_id(UDPPacketBuffer::new_with_constructor()),
"M:LibreMetaverse.UDPPacketBuffer.#ctor"
);
let udp_buffer = UDPPacketBuffer::new_with_constructor().expect("UDP packet buffer");
assert_eq!(udp_buffer.data.len(), 4096);
assert_eq!(udp_buffer.data_length, 0);
assert_eq!(udp_buffer.remote_end_point, "0.0.0.0:0".parse().unwrap());
}

View File

@@ -112,6 +112,10 @@ NATIVE_TYPES = {
"T:LibreMetaverse.SkeletalBoneInfo": "crate::visual_catalog::SkeletalBoneInfo",
"T:LibreMetaverse.TreeDefinition": "crate::foliage_catalog::TreeDefinition",
"T:LibreMetaverse.TreeDefinitions": "crate::foliage_catalog::TreeDefinitions",
"T:LibreMetaverse.AgentThrottle": "crate::udp_transport::AgentThrottle",
"T:LibreMetaverse.IncomingPacketIDCollection": "crate::udp_transport::IncomingPacketIDCollection",
"T:LibreMetaverse.UDPBase": "crate::udp_transport::UDPBase",
"T:LibreMetaverse.UDPPacketBuffer": "crate::udp_transport::UDPPacketBuffer",
"T:LibreMetaverse.VisualAlphaParam": "crate::visual_catalog::VisualAlphaParam",
"T:LibreMetaverse.VisualColorParam": "crate::visual_catalog::VisualColorParam",
"T:LibreMetaverse.VisualParam": "crate::visual_catalog::VisualParam",