perf(smoltcp): unify device packet type on BytesMut for zero-copy TX

The smoltcp -> ZCPacket TX path still memcpy'd each outbound packet:
the device emitted Vec<u8>, and `Vec<u8> -> BytesMut` has no zero-copy
conversion in the bytes crate, so socks5/tcp_proxy paid a
`BytesMut::from(Bytes::from(data))` copy per packet.

Switch tokio_smoltcp's packet type to BytesMut end to end so smoltcp
writes into the headroom-reserved buffer that ZCPacket wraps directly.

- device: `Packet = BytesMut`; BufferTxToken allocates via
  BytesMut::with_capacity + resize; BufferRxToken derefs to &[u8].
- channel_device: Stream/Sink/channel carry BytesMut instead of Vec<u8>.
- socks5 / tcp_proxy: TX wraps the BytesMut via new_from_buf directly,
  removing the per-packet memcpy; inbound keeps an equivalent copy via
  BytesMut::from(payload).
- reactor: adapts through the Packet alias, no code change.

easytier/src/gateway/socks5.rs

@@ -28,7 +28,7 @@ use crate::{
         ip_reassembler::IpReassembler,
         tokio_smoltcp::{BufferSize, Net, NetConfig, channel_device},
     },
-    tunnel::packet_def::{PacketType, ZCPacket},
+    tunnel::packet_def::{PacketType, ZCPacket, ZCPacketType},
 };
 use anyhow::Context;
 use dashmap::DashMap;
@@ -363,7 +363,10 @@ impl Socks5ServerNet {
             let mut smoltcp_stack_receiver = packet_recv.lock().await;
             while let Some(packet) = smoltcp_stack_receiver.recv().await {
                 tracing::trace!(?packet, "receive from peer send to smoltcp packet");
-                if let Err(e) = stack_sink.send(Ok(packet.payload().to_vec())).await {
+                if let Err(e) = stack_sink
+                    .send(Ok(bytes::BytesMut::from(packet.payload())))
+                    .await
+                {
                     tracing::error!("send to smoltcp stack failed: {:?}", e);
                 }
             }
@@ -377,13 +380,16 @@ impl Socks5ServerNet {
                     "receive from smoltcp stack and send to peer mgr packet, len = {}",
                     data.len()
                 );
-                let Some(ipv4) = Ipv4Packet::new(&data) else {
-                    tracing::error!(?data, "smoltcp stack stream get non ipv4 packet");
+                let packet = ZCPacket::new_from_buf(data, ZCPacketType::NIC);
+                let Some(ipv4) = Ipv4Packet::new(packet.payload()) else {
+                    tracing::error!(
+                        payload_len = packet.payload_len(),
+                        "smoltcp stack stream get non ipv4 packet"
+                    );
                     continue;
                 };
 
                 let dst = ipv4.get_destination();
-                let packet = ZCPacket::new_with_payload(&data);
                 let Some(peer_manager) = peer_manager.upgrade() else {
                     tracing::warn!("peer manager is gone, smoltcp sender exited");
                     return;
@@ -412,7 +418,8 @@ impl Socks5ServerNet {
                     tcp_tx_size: 1024 * 128,
                     ..Default::default()
                 }),
-            ),
+            )
+            .with_packet_tx_headroom(ZCPacketType::NIC.get_packet_offsets().payload_offset),
         );
 
         let forward_tasks = Arc::new(std::sync::Mutex::new(forward_tasks));

easytier/src/gateway/tcp_proxy.rs

@@ -39,6 +39,8 @@ use super::CidrSet;
 
 #[cfg(feature = "smoltcp")]
 use super::tokio_smoltcp::{self, Net, NetConfig, channel_device};
+#[cfg(feature = "smoltcp")]
+use crate::tunnel::packet_def::ZCPacketType;
 
 #[async_trait::async_trait]
 pub(crate) trait NatDstConnector: Send + Sync + Clone + 'static {
@@ -561,7 +563,10 @@ impl<C: NatDstConnector> TcpProxy<C> {
             self.tasks.lock().unwrap().spawn(async move {
                 while let Some(packet) = smoltcp_stack_receiver.recv().await {
                     tracing::trace!(?packet, "receive from peer send to smoltcp packet");
-                    if let Err(e) = stack_sink.send(Ok(packet.payload().to_vec())).await {
+                    if let Err(e) = stack_sink
+                        .send(Ok(bytes::BytesMut::from(packet.payload())))
+                        .await
+                    {
                         tracing::error!("send to smoltcp stack failed: {:?}", e);
                     }
                 }
@@ -575,13 +580,16 @@ impl<C: NatDstConnector> TcpProxy<C> {
                         ?data,
                         "receive from smoltcp stack and send to peer mgr packet"
                     );
-                    let Some(ipv4) = Ipv4Packet::new(&data) else {
-                        tracing::error!(?data, "smoltcp stack stream get non ipv4 packet");
+                    let packet = ZCPacket::new_from_buf(data, ZCPacketType::NIC);
+                    let Some(ipv4) = Ipv4Packet::new(packet.payload()) else {
+                        tracing::error!(
+                            payload_len = packet.payload_len(),
+                            "smoltcp stack stream get non ipv4 packet"
+                        );
                         continue;
                     };
 
                     let dst = ipv4.get_destination();
-                    let packet = ZCPacket::new_with_payload(&data);
                     let Some(peer_mgr) = peer_mgr.upgrade() else {
                         tracing::warn!("peer manager is gone, smoltcp sender exited");
                         return;
@@ -610,7 +618,8 @@ impl<C: NatDstConnector> TcpProxy<C> {
                         tcp_tx_size: 1024 * 16,
                         ..Default::default()
                     }),
-                ),
+                )
+                .with_packet_tx_headroom(ZCPacketType::NIC.get_packet_offsets().payload_offset),
             );
             net.set_any_ip(true);
             self.smoltcp_net.lock().await.replace(net);

easytier/src/gateway/tokio_smoltcp/channel_device.rs

@@ -1,3 +1,4 @@
+use bytes::BytesMut;
 use futures::{Sink, Stream};
 use smoltcp::phy::DeviceCapabilities;
 use std::{
@@ -12,15 +13,15 @@ use super::device::AsyncDevice;
 
 /// A device that send and receive packets using a channel.
 pub struct ChannelDevice {
-    recv: Receiver<io::Result<Vec<u8>>>,
-    send: PollSender<Vec<u8>>,
+    recv: Receiver<io::Result<BytesMut>>,
+    send: PollSender<BytesMut>,
     caps: DeviceCapabilities,
 }
 
 pub type ChannelDeviceNewRet = (
     ChannelDevice,
-    Sender<io::Result<Vec<u8>>>,
-    Receiver<Vec<u8>>,
+    Sender<io::Result<BytesMut>>,
+    Receiver<BytesMut>,
 );
 
 impl ChannelDevice {
@@ -43,25 +44,25 @@ impl ChannelDevice {
 }
 
 impl Stream for ChannelDevice {
-    type Item = io::Result<Vec<u8>>;
+    type Item = io::Result<BytesMut>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         self.recv.poll_recv(cx)
     }
 }
 
-fn map_err(e: PollSendError<Vec<u8>>) -> io::Error {
+fn map_err(e: PollSendError<BytesMut>) -> io::Error {
     io::Error::other(e)
 }
 
-impl Sink<Vec<u8>> for ChannelDevice {
+impl Sink<BytesMut> for ChannelDevice {
     type Error = io::Error;
 
     fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         self.send.poll_reserve(cx).map_err(map_err)
     }
 
-    fn start_send(mut self: Pin<&mut Self>, item: Vec<u8>) -> Result<(), Self::Error> {
+    fn start_send(mut self: Pin<&mut Self>, item: BytesMut) -> Result<(), Self::Error> {
         self.send.send_item(item).map_err(map_err)
     }
 

easytier/src/gateway/tokio_smoltcp/device.rs

@@ -1,3 +1,4 @@
+use bytes::BytesMut;
 use futures::{Sink, Stream};
 pub use smoltcp::phy::DeviceCapabilities;
 use smoltcp::{
@@ -10,7 +11,7 @@ use std::{collections::VecDeque, io};
 pub const DEFAULT_MAX_BURST_SIZE: usize = 100;
 
 /// A packet used in `AsyncDevice`.
-pub type Packet = Vec<u8>;
+pub type Packet = BytesMut;
 
 /// A device that send and receive packets asynchronously.
 pub trait AsyncDevice:
@@ -33,6 +34,7 @@ where
 pub struct BufferDevice {
     caps: DeviceCapabilities,
     max_burst_size: usize,
+    tx_headroom: usize,
     recv_queue: VecDeque<Packet>,
     send_queue: VecDeque<Packet>,
 }
@@ -41,13 +43,11 @@ pub struct BufferDevice {
 pub struct BufferRxToken(Packet);
 
 impl RxToken for BufferRxToken {
-    fn consume<R, F>(mut self, f: F) -> R
+    fn consume<R, F>(self, f: F) -> R
     where
         F: FnOnce(&[u8]) -> R,
     {
-        let p = &mut self.0;
-
-        f(p)
+        f(&self.0[..])
     }
 }
 
@@ -59,8 +59,10 @@ impl<'d> TxToken for BufferTxToken<'d> {
     where
         F: FnOnce(&mut [u8]) -> R,
     {
-        let mut buffer = vec![0u8; len];
-        let result = f(&mut buffer);
+        let tx_headroom = self.0.tx_headroom;
+        let mut buffer = BytesMut::with_capacity(tx_headroom + len);
+        buffer.resize(tx_headroom + len, 0);
+        let result = f(&mut buffer[tx_headroom..]);
 
         self.0.send_queue.push_back(buffer);
 
@@ -98,11 +100,12 @@ impl Device for BufferDevice {
 }
 
 impl BufferDevice {
-    pub(crate) fn new(caps: DeviceCapabilities) -> BufferDevice {
+    pub(crate) fn new(caps: DeviceCapabilities, tx_headroom: usize) -> BufferDevice {
         let max_burst_size = caps.max_burst_size.unwrap_or(DEFAULT_MAX_BURST_SIZE);
         BufferDevice {
             caps,
             max_burst_size,
+            tx_headroom,
             recv_queue: VecDeque::with_capacity(max_burst_size),
             send_queue: VecDeque::with_capacity(max_burst_size),
         }
@@ -123,3 +126,27 @@ impl BufferDevice {
         self.recv_queue.is_empty()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn buffer_device_reserves_tx_headroom() {
+        let mut caps = DeviceCapabilities::default();
+        caps.max_burst_size = Some(1);
+        let mut device = BufferDevice::new(caps, 16);
+
+        let token = device.transmit(Instant::now()).unwrap();
+        token.consume(4, |buf| {
+            assert_eq!(buf.len(), 4);
+            buf.copy_from_slice(&[1, 2, 3, 4]);
+        });
+
+        let mut queue = device.take_send_queue();
+        let packet = queue.pop_front().unwrap();
+        assert_eq!(packet.len(), 20);
+        assert_eq!(&packet[..16], &[0; 16]);
+        assert_eq!(&packet[16..], &[1, 2, 3, 4]);
+    }
+}

easytier/src/gateway/tokio_smoltcp/mod.rs

@@ -51,6 +51,7 @@ pub struct NetConfig {
     pub ip_addr: IpCidr,
     pub gateway: Vec<IpAddress>,
     pub buffer_size: BufferSize,
+    pub(crate) packet_tx_headroom: usize,
 }
 
 impl NetConfig {
@@ -65,8 +66,14 @@ impl NetConfig {
             ip_addr,
             gateway,
             buffer_size: buffer_size.unwrap_or_default(),
+            packet_tx_headroom: 0,
         }
     }
+
+    pub fn with_packet_tx_headroom(mut self, packet_tx_headroom: usize) -> Self {
+        self.packet_tx_headroom = packet_tx_headroom;
+        self
+    }
 }
 
 /// `Net` is the main interface to the network stack.
@@ -97,7 +104,8 @@ impl Net {
     }
 
     fn new2<D: device::AsyncDevice + 'static>(device: D, config: NetConfig) -> Net {
-        let mut buffer_device = BufferDevice::new(device.capabilities().clone());
+        let mut buffer_device =
+            BufferDevice::new(device.capabilities().clone(), config.packet_tx_headroom);
         let mut iface = Interface::new(config.interface_config, &mut buffer_device, Instant::now());
         let ip_addr = config.ip_addr;
         iface.update_ip_addrs(|ip_addrs| {

easytier/src/tunnel/packet_def.rs

@@ -764,6 +764,7 @@ impl ZCPacket {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::{hint::black_box, time::Instant};
 
     #[test]
     fn test_zc_packet() {
@@ -809,4 +810,68 @@ mod tests {
 
         assert!(packet.mut_wg_tunnel_header().is_none());
     }
+
+    fn bench_smoltcp_zcpacket_construct(payload_len: usize, iterations: usize) {
+        let nic_offset = ZCPacketType::NIC.get_packet_offsets().payload_offset;
+
+        // Correctness check (outside the timed section): both construction paths
+        // must yield equivalent payloads for the perf comparison to be meaningful.
+        {
+            let data = vec![7u8; payload_len];
+            let p_copy = ZCPacket::new_with_payload(&data);
+
+            let mut buf = BytesMut::with_capacity(nic_offset + payload_len);
+            buf.resize(nic_offset + payload_len, 0);
+            buf[nic_offset..].fill(7);
+            let p_zero = ZCPacket::new_from_buf(buf, ZCPacketType::NIC);
+
+            assert_eq!(p_copy.payload(), p_zero.payload());
+        }
+
+        // copy path: smoltcp emits a bare payload buf; socks5/tcp_proxy copy it
+        // via ZCPacket::new_with_payload (pre-f5ce0848 behavior).
+        let now = Instant::now();
+        let mut checksum = 0usize;
+        for _ in 0..iterations {
+            let data = vec![7u8; payload_len];
+            let p = ZCPacket::new_with_payload(black_box(&data));
+            // black_box forces the side-effect-free construction to be emitted;
+            // payload_len is stable per run so it cannot skew the numbers.
+            checksum = checksum.wrapping_add(black_box(&p).payload_len());
+        }
+        let copy_elapsed = now.elapsed().as_secs_f64();
+
+        // zerocopy path: device reserves NIC headroom in the buf; socks5/tcp_proxy
+        // wrap it zero-copy via ZCPacket::new_from_buf (f5ce0848 behavior).
+        let now = Instant::now();
+        let mut checksum2 = 0usize;
+        for _ in 0..iterations {
+            let mut buf = BytesMut::with_capacity(nic_offset + payload_len);
+            buf.resize(nic_offset + payload_len, 0);
+            buf[nic_offset..].fill(7);
+            let p = ZCPacket::new_from_buf(black_box(buf), ZCPacketType::NIC);
+            checksum2 = checksum2.wrapping_add(black_box(&p).payload_len());
+        }
+        let zerocopy_elapsed = now.elapsed().as_secs_f64();
+
+        println!(
+            "smoltcp_zcpacket payload_len={} iterations={} copy_pps={:.0} copy_bytes_per_sec={:.0} zerocopy_pps={:.0} zerocopy_bytes_per_sec={:.0} speedup={:.2}x checksums={}/{}",
+            payload_len,
+            iterations,
+            iterations as f64 / copy_elapsed,
+            (payload_len * iterations) as f64 / copy_elapsed,
+            iterations as f64 / zerocopy_elapsed,
+            (payload_len * iterations) as f64 / zerocopy_elapsed,
+            copy_elapsed / zerocopy_elapsed,
+            checksum,
+            checksum2
+        );
+    }
+
+    #[test]
+    #[ignore = "benchmark helper; run with --ignored --nocapture"]
+    fn smoltcp_zcpacket_construct_bench() {
+        bench_smoltcp_zcpacket_construct(1280, 1_000_000);
+        bench_smoltcp_zcpacket_construct(4096, 500_000);
+    }
 }