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.
This commit is contained in:
fanyang
2026-06-27 12:30:35 +08:00
parent b092e78523
commit b8d1d6b32c
4 changed files with 25 additions and 24 deletions
+5 -5
View File
@@ -363,7 +363,10 @@ impl Socks5ServerNet {
let mut smoltcp_stack_receiver = packet_recv.lock().await; let mut smoltcp_stack_receiver = packet_recv.lock().await;
while let Some(packet) = smoltcp_stack_receiver.recv().await { while let Some(packet) = smoltcp_stack_receiver.recv().await {
tracing::trace!(?packet, "receive from peer send to smoltcp packet"); 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); tracing::error!("send to smoltcp stack failed: {:?}", e);
} }
} }
@@ -377,10 +380,7 @@ impl Socks5ServerNet {
"receive from smoltcp stack and send to peer mgr packet, len = {}", "receive from smoltcp stack and send to peer mgr packet, len = {}",
data.len() data.len()
); );
let packet = ZCPacket::new_from_buf( let packet = ZCPacket::new_from_buf(data, ZCPacketType::NIC);
bytes::BytesMut::from(bytes::Bytes::from(data)),
ZCPacketType::NIC,
);
let Some(ipv4) = Ipv4Packet::new(packet.payload()) else { let Some(ipv4) = Ipv4Packet::new(packet.payload()) else {
tracing::error!( tracing::error!(
payload_len = packet.payload_len(), payload_len = packet.payload_len(),
+5 -5
View File
@@ -563,7 +563,10 @@ impl<C: NatDstConnector> TcpProxy<C> {
self.tasks.lock().unwrap().spawn(async move { self.tasks.lock().unwrap().spawn(async move {
while let Some(packet) = smoltcp_stack_receiver.recv().await { while let Some(packet) = smoltcp_stack_receiver.recv().await {
tracing::trace!(?packet, "receive from peer send to smoltcp packet"); 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); tracing::error!("send to smoltcp stack failed: {:?}", e);
} }
} }
@@ -577,10 +580,7 @@ impl<C: NatDstConnector> TcpProxy<C> {
?data, ?data,
"receive from smoltcp stack and send to peer mgr packet" "receive from smoltcp stack and send to peer mgr packet"
); );
let packet = ZCPacket::new_from_buf( let packet = ZCPacket::new_from_buf(data, ZCPacketType::NIC);
bytes::BytesMut::from(bytes::Bytes::from(data)),
ZCPacketType::NIC,
);
let Some(ipv4) = Ipv4Packet::new(packet.payload()) else { let Some(ipv4) = Ipv4Packet::new(packet.payload()) else {
tracing::error!( tracing::error!(
payload_len = packet.payload_len(), payload_len = packet.payload_len(),
@@ -1,3 +1,4 @@
use bytes::BytesMut;
use futures::{Sink, Stream}; use futures::{Sink, Stream};
use smoltcp::phy::DeviceCapabilities; use smoltcp::phy::DeviceCapabilities;
use std::{ use std::{
@@ -12,15 +13,15 @@ use super::device::AsyncDevice;
/// A device that send and receive packets using a channel. /// A device that send and receive packets using a channel.
pub struct ChannelDevice { pub struct ChannelDevice {
recv: Receiver<io::Result<Vec<u8>>>, recv: Receiver<io::Result<BytesMut>>,
send: PollSender<Vec<u8>>, send: PollSender<BytesMut>,
caps: DeviceCapabilities, caps: DeviceCapabilities,
} }
pub type ChannelDeviceNewRet = ( pub type ChannelDeviceNewRet = (
ChannelDevice, ChannelDevice,
Sender<io::Result<Vec<u8>>>, Sender<io::Result<BytesMut>>,
Receiver<Vec<u8>>, Receiver<BytesMut>,
); );
impl ChannelDevice { impl ChannelDevice {
@@ -43,25 +44,25 @@ impl ChannelDevice {
} }
impl Stream for 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>> { fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
self.recv.poll_recv(cx) 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) io::Error::other(e)
} }
impl Sink<Vec<u8>> for ChannelDevice { impl Sink<BytesMut> for ChannelDevice {
type Error = io::Error; type Error = io::Error;
fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::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) 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) self.send.send_item(item).map_err(map_err)
} }
+6 -6
View File
@@ -1,3 +1,4 @@
use bytes::BytesMut;
use futures::{Sink, Stream}; use futures::{Sink, Stream};
pub use smoltcp::phy::DeviceCapabilities; pub use smoltcp::phy::DeviceCapabilities;
use smoltcp::{ use smoltcp::{
@@ -10,7 +11,7 @@ use std::{collections::VecDeque, io};
pub const DEFAULT_MAX_BURST_SIZE: usize = 100; pub const DEFAULT_MAX_BURST_SIZE: usize = 100;
/// A packet used in `AsyncDevice`. /// A packet used in `AsyncDevice`.
pub type Packet = Vec<u8>; pub type Packet = BytesMut;
/// A device that send and receive packets asynchronously. /// A device that send and receive packets asynchronously.
pub trait AsyncDevice: pub trait AsyncDevice:
@@ -42,13 +43,11 @@ pub struct BufferDevice {
pub struct BufferRxToken(Packet); pub struct BufferRxToken(Packet);
impl RxToken for BufferRxToken { impl RxToken for BufferRxToken {
fn consume<R, F>(mut self, f: F) -> R fn consume<R, F>(self, f: F) -> R
where where
F: FnOnce(&[u8]) -> R, F: FnOnce(&[u8]) -> R,
{ {
let p = &mut self.0; f(&self.0[..])
f(p)
} }
} }
@@ -61,7 +60,8 @@ impl<'d> TxToken for BufferTxToken<'d> {
F: FnOnce(&mut [u8]) -> R, F: FnOnce(&mut [u8]) -> R,
{ {
let tx_headroom = self.0.tx_headroom; let tx_headroom = self.0.tx_headroom;
let mut buffer = vec![0u8; tx_headroom + len]; let mut buffer = BytesMut::with_capacity(tx_headroom + len);
buffer.resize(tx_headroom + len, 0);
let result = f(&mut buffer[tx_headroom..]); let result = f(&mut buffer[tx_headroom..]);
self.0.send_queue.push_back(buffer); self.0.send_queue.push_back(buffer);