From e99e3dccf58605fa61451d62591ac3b42708b8d0 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Mon, 11 Sep 2023 18:14:08 -0400 Subject: [PATCH] Use buf in test. --- Cargo.lock | 10 +++++----- src/lib.rs | 55 +++++++++++++++++++++--------------------------------- 2 files changed, 26 insertions(+), 39 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4244d81..481b83c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10,9 +10,9 @@ checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa" [[package]] name = "base64" -version = "0.21.3" +version = "0.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "414dcefbc63d77c526a76b3afcf6fbb9b5e2791c19c3aa2297733208750c6e53" +checksum = "9ba43ea6f343b788c8764558649e08df62f86c6ef251fdaeb1ffd010a9ae50a2" [[package]] name = "bitflags" @@ -173,9 +173,9 @@ dependencies = [ [[package]] name = "syn" -version = "2.0.29" +version = "2.0.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c324c494eba9d92503e6f1ef2e6df781e78f6a7705a0202d9801b198807d518a" +checksum = "239814284fd6f1a4ffe4ca893952cdd93c224b6a1571c9a9eadd670295c0c9e2" dependencies = [ "proc-macro2", "quote", @@ -310,7 +310,7 @@ dependencies = [ [[package]] name = "zerotier-common-utils" version = "0.1.0" -source = "git+https://github.com/zerotier/common-utils.git?branch=main#c24a5b1dbf50f3c25cbba5a43c0f927026144f08" +source = "git+https://github.com/zerotier/common-utils.git?branch=main#517d12fd366046a7e76aa79a86dbbd72bb72ff45" dependencies = [ "base64", "libc", diff --git a/src/lib.rs b/src/lib.rs index 4370fe1..9922560 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -45,9 +45,11 @@ pub trait PacketHandler: Send + Sync + Sized + 'static { /// Obtain a buffer to receive a new packet. /// - /// The buffer's contents do not need to be zeroed prior to being used. Its as_mut() function - /// must return a slice large enough to receive a UDP packet or packets will be lost. - fn get_buffer(&self) -> Self::Buffer; + /// Receive buffers must be sized to accept the largest possible UDP packet or packets may + /// be lost. In other words if the buffer is e.g. a Vec its size must be set with + /// resize() prior to returning from this so that as_mut() will return a destination buffer + /// for reading. The buffer does not need to be cleared as any data will be overwritten. + fn get_receive_buffer(&self) -> Self::Buffer; /// Called by the engine to return a buffer after a send completes (or fails). /// @@ -55,9 +57,12 @@ pub trait PacketHandler: Send + Sync + Sized + 'static { /// If you are doing pooling you could instead call this internally inside on_udp_packet when /// processing is complete. /// - /// The default implementation is a no-op. Override if needed to e.g. return buffers to a pool. - #[allow(unused_variables)] - fn return_buffer(&self, buffer: Self::Buffer) {} + /// The default implementation just drops. Override if anything needs to be done with used + /// buffers other than dropping them. + #[inline(always)] + fn return_buffer(&self, buffer: Self::Buffer) { + drop(buffer); + } /// Called whan a UDP packet is received. /// @@ -292,7 +297,7 @@ impl EngineThread { let d: &UdpSocket = unsafe { &*(*key as *const UdpSocket) }; let handler = d.handler.as_ref(); loop { - let mut buf = handler.get_buffer(); + let mut buf = handler.get_receive_buffer(); let buf_inner = buf.as_mut(); let mut addrlen = size_of::() as libc::socklen_t; let packet_size = unsafe { @@ -390,33 +395,23 @@ impl UdpSocket { mod tests { use super::*; use std::sync::atomic::AtomicU64; + use zerotier_common_utils::buf::Buf; #[derive(Default)] struct TestHandler { - pool: Mutex>>, received: AtomicU64, echo: bool, } impl PacketHandler for TestHandler { - type Buffer = Box<[u8]>; + type Buffer = Buf; type Ref = Arc; type SocketApplicationData = (); - fn get_buffer(&self) -> Self::Buffer { - let mut pool = self.pool.lock().unwrap(); - if pool.is_empty() { - drop(pool); - let mut v = Vec::new(); - v.resize(2048, 0); - v.into_boxed_slice() - } else { - pool.pop().unwrap() - } - } - - fn return_buffer(&self, buffer: Self::Buffer) { - self.pool.lock().unwrap().push(buffer); + fn get_receive_buffer(&self) -> Self::Buffer { + let mut b = Buf::new(1504); + unsafe { b.set_size(1500) }; + b } fn on_udp_packet( @@ -440,11 +435,7 @@ mod tests { fn loopback() { let eng = Engine::::new(); - let sender_handler = Arc::new(TestHandler { - pool: Mutex::new(Vec::new()), - received: AtomicU64::new(0), - echo: false, - }); + let sender_handler = Arc::new(TestHandler { received: AtomicU64::new(0), echo: false }); let sender = eng .bind( &InetAddress::from_ip_port(&[127, 0, 0, 1], 11111), @@ -455,11 +446,7 @@ mod tests { ) .unwrap(); - let receiver_handler = Arc::new(TestHandler { - pool: Mutex::new(Vec::new()), - received: AtomicU64::new(0), - echo: true, - }); + let receiver_handler = Arc::new(TestHandler { received: AtomicU64::new(0), echo: true }); let sendto_addr = InetAddress::from_ip_port(&[127, 0, 0, 1], 11112); let _receiver = eng .bind(&sendto_addr, None, true, receiver_handler.clone(), ()) @@ -467,7 +454,7 @@ mod tests { const PACKET_COUNT: usize = 1024; for _ in 0..PACKET_COUNT { - let mut b = sender_handler.get_buffer(); + let mut b = sender_handler.get_receive_buffer(); b.as_mut().fill(1); sender.send(&sendto_addr, b, 1024); }