optimization: local-tun virtual device lock-free input/output

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() {

crates/shadowsocks-service/src/local/tun/tcp.rs

@@ -150,6 +150,7 @@ pub struct TcpTun {
     manager_notify: Arc<Notify>,
     balancer: PingBalancer,
     iface_rx: mpsc::UnboundedReceiver<Vec<u8>>,
+    iface_tx: mpsc::Sender<Vec<u8>>,
 }
 
 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::<Vec<u8>>();
-        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.

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<Vec<u8>>,
+    in_buf: mpsc::Receiver<Vec<u8>>,
     out_buf: mpsc::UnboundedSender<Vec<u8>>,
 }
 
 impl VirtTunDevice {
-    pub fn new(capabilities: DeviceCapabilities, iface_tx: mpsc::UnboundedSender<Vec<u8>>) -> Self {
-        Self {
-            capabilities,
-            in_buf: VecDeque::new(),
-            out_buf: iface_tx,
-        }
-    }
+    pub fn new(capabilities: DeviceCapabilities) -> (Self, mpsc::UnboundedReceiver<Vec<u8>>, mpsc::Sender<Vec<u8>>) {
+        let (iface_tx, iface_output) = mpsc::unbounded_channel();
+        let (iface_input, iface_rx) = mpsc::channel(512);
 
-    pub fn inject_packet(&mut self, buffer: Vec<u8>) {
-        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));