diff --git a/crates/shadowsocks-service/src/local/tun/mod.rs b/crates/shadowsocks-service/src/local/tun/mod.rs index 814112c0..da4fc003 100644 --- a/crates/shadowsocks-service/src/local/tun/mod.rs +++ b/crates/shadowsocks-service/src/local/tun/mod.rs @@ -227,7 +227,7 @@ impl Tun { ); } - self.tcp.drive_interface_state(frame); + self.tcp.drive_interface_state(frame).await; } IpProtocol::Udp => { if !self.mode.enable_udp() { diff --git a/crates/shadowsocks-service/src/local/tun/tcp.rs b/crates/shadowsocks-service/src/local/tun/tcp.rs index 042b1f29..c17a0af6 100644 --- a/crates/shadowsocks-service/src/local/tun/tcp.rs +++ b/crates/shadowsocks-service/src/local/tun/tcp.rs @@ -150,6 +150,7 @@ pub struct TcpTun { manager_notify: Arc, balancer: PingBalancer, iface_rx: mpsc::UnboundedReceiver>, + iface_tx: mpsc::Sender>, } impl Drop for TcpTun { @@ -164,8 +165,7 @@ impl TcpTun { capabilities.medium = Medium::Ip; capabilities.max_transmission_unit = mtu as usize; - let (iface_tx, iface_rx) = mpsc::unbounded_channel::>(); - let virt = VirtTunDevice::new(capabilities, iface_tx); + let (virt, iface_rx, iface_tx) = VirtTunDevice::new(capabilities); let iface_builder = InterfaceBuilder::new(virt, vec![]); let iface_ipaddrs = [ @@ -226,6 +226,7 @@ impl TcpTun { manager_notify, balancer, iface_rx, + iface_tx, } } @@ -267,10 +268,9 @@ impl TcpTun { Ok(()) } - pub fn drive_interface_state(&mut self, frame: &[u8]) { - { - let mut manager = self.manager.lock().unwrap(); - manager.iface.device_mut().inject_packet(frame.to_vec()); + pub async fn drive_interface_state(&mut self, frame: &[u8]) { + if let Err(..) = self.iface_tx.send(frame.to_vec()).await { + panic!("interface send channel closed unexpectly"); } // Wake up and poll the interface. diff --git a/crates/shadowsocks-service/src/local/tun/virt_device.rs b/crates/shadowsocks-service/src/local/tun/virt_device.rs index a6a9e66c..e907671e 100644 --- a/crates/shadowsocks-service/src/local/tun/virt_device.rs +++ b/crates/shadowsocks-service/src/local/tun/virt_device.rs @@ -1,7 +1,5 @@ //! Virtual Device for receiving packets from tun -use std::collections::VecDeque; - use smoltcp::{ phy::{self, Device, DeviceCapabilities}, time::Instant, @@ -10,21 +8,24 @@ use tokio::sync::mpsc; pub struct VirtTunDevice { capabilities: DeviceCapabilities, - in_buf: VecDeque>, + in_buf: mpsc::Receiver>, out_buf: mpsc::UnboundedSender>, } impl VirtTunDevice { - pub fn new(capabilities: DeviceCapabilities, iface_tx: mpsc::UnboundedSender>) -> Self { - Self { - capabilities, - in_buf: VecDeque::new(), - out_buf: iface_tx, - } - } + pub fn new(capabilities: DeviceCapabilities) -> (Self, mpsc::UnboundedReceiver>, mpsc::Sender>) { + let (iface_tx, iface_output) = mpsc::unbounded_channel(); + let (iface_input, iface_rx) = mpsc::channel(512); - pub fn inject_packet(&mut self, buffer: Vec) { - self.in_buf.push_back(buffer); + ( + Self { + capabilities, + in_buf: iface_rx, + out_buf: iface_tx, + }, + iface_output, + iface_input, + ) } } @@ -33,7 +34,7 @@ impl<'a> Device<'a> for VirtTunDevice { type TxToken = VirtTxToken<'a>; fn receive(&'a mut self) -> Option<(Self::RxToken, Self::TxToken)> { - if let Some(buffer) = self.in_buf.pop_front() { + if let Ok(buffer) = self.in_buf.try_recv() { let rx = Self::RxToken { buffer }; let tx = VirtTxToken(self); return Some((rx, tx));