[mio] 增加模拟弱网参数

vnt/src/channel/context.rs

@@ -8,6 +8,7 @@ use std::time::{Duration, Instant};
 
 use crossbeam_utils::atomic::AtomicCell;
 use parking_lot::RwLock;
+use rand::Rng;
 
 use crate::channel::punch::NatType;
 use crate::channel::sender::{AcceptSocketSender, ChannelSender, PacketSender};
@@ -26,9 +27,21 @@ impl Context {
         use_channel_type: UseChannelType,
         first_latency: bool,
         is_tcp: bool,
+        packet_loss_rate: Option<f64>,
+        packet_delay: u32,
     ) -> Self {
         let channel_num = main_udp_socket.len();
         assert_ne!(channel_num, 0, "not channel");
+        let packet_loss_rate = packet_loss_rate
+            .map(|v| {
+                let v = (v * PACKET_LOSS_RATE_DENOMINATOR as f64) as u32;
+                if v > PACKET_LOSS_RATE_DENOMINATOR {
+                    PACKET_LOSS_RATE_DENOMINATOR
+                } else {
+                    v
+                }
+            })
+            .unwrap_or(0);
         let inner = ContextInner {
             main_udp_socket,
             sub_udp_socket: RwLock::new(Vec::with_capacity(64)),
@@ -36,6 +49,8 @@ impl Context {
             route_table: RouteTable::new(use_channel_type, first_latency, channel_num),
             is_tcp,
             state: AtomicBool::new(true),
+            packet_loss_rate,
+            packet_delay,
         };
         Self {
             inner: Arc::new(inner),
@@ -56,7 +71,7 @@ impl Deref for Context {
 
 /// 对称网络增加的udp socket数目，有助于增加打洞成功率
 pub const SYMMETRIC_CHANNEL_NUM: usize = 64;
-
+const PACKET_LOSS_RATE_DENOMINATOR: u32 = 100_0000;
 pub struct ContextInner {
     // 核心udp socket
     pub(crate) main_udp_socket: Vec<UdpSocket>,
@@ -70,6 +85,10 @@ pub struct ContextInner {
     is_tcp: bool,
     //状态
     state: AtomicBool,
+    //控制丢包率，取值v=[0,100_0000] 丢包率r=v/100_0000
+    packet_loss_rate: u32,
+    //控制延迟
+    packet_delay: u32,
 }
 
 impl ContextInner {
@@ -230,6 +249,27 @@ impl ContextInner {
             }
         }
     }
+    /// 发送网络数据
+    pub fn send_ipv4_by_id(
+        &self,
+        buf: &[u8],
+        id: &Ipv4Addr,
+        server_addr: SocketAddr,
+    ) -> io::Result<()> {
+        if self.packet_loss_rate > 0 {
+            if rand::thread_rng().gen_ratio(self.packet_loss_rate, PACKET_LOSS_RATE_DENOMINATOR) {
+                return Ok(());
+            }
+        }
+        if self.packet_delay > 0 {
+            std::thread::sleep(Duration::from_millis(self.packet_delay as _));
+        }
+        if self.send_by_id(buf, id).is_err() && !self.route_table.use_channel_type.is_only_p2p() {
+            self.send_default(buf, server_addr)
+        } else {
+            Ok(())
+        }
+    }
     /// 将数据发到指定id
     pub fn send_by_id(&self, buf: &[u8], id: &Ipv4Addr) -> io::Result<()> {
         let route = self.route_table.get_route_by_id(id)?;

vnt/src/channel/mod.rs

@@ -141,6 +141,8 @@ pub fn init_context(
     use_channel_type: UseChannelType,
     first_latency: bool,
     is_tcp: bool,
+    packet_loss_rate: Option<f64>,
+    packet_delay: u32,
 ) -> io::Result<(Context, mio::net::TcpListener)> {
     assert!(!ports.is_empty(), "not channel");
     let mut udps = Vec::with_capacity(ports.len());
@@ -158,7 +160,14 @@ pub fn init_context(
         main_channel.set_write_timeout(Some(Duration::from_secs(5)))?;
         udps.push(main_channel);
     }
-    let context = Context::new(udps, use_channel_type, first_latency, is_tcp);
+    let context = Context::new(
+        udps,
+        use_channel_type,
+        first_latency,
+        is_tcp,
+        packet_loss_rate,
+        packet_delay,
+    );
 
     let port = context.main_local_udp_port()?[0];
     //监听v6+v4双栈，tcp通道使用异步io

vnt/src/core/conn.rs

@@ -93,6 +93,8 @@ impl Vnt {
             config.use_channel_type,
             config.first_latency,
             config.tcp,
+            config.packet_loss_rate,
+            config.packet_delay,
         )?;
         let local_ipv4 = nat::local_ipv4();
         let local_ipv6 = nat::local_ipv6();

vnt/src/core/mod.rs

@@ -39,6 +39,9 @@ pub struct Config {
     #[cfg(target_os = "android")]
     pub device_fd: i32,
     pub use_channel_type: UseChannelType,
+    //控制丢包率
+    pub packet_loss_rate: Option<f64>,
+    pub packet_delay: u32,
 }
 
 impl Config {
@@ -67,6 +70,8 @@ impl Config {
         #[cfg(not(target_os = "android"))] device_name: Option<String>,
         #[cfg(target_os = "android")] device_fd: i32,
         use_channel_type: UseChannelType,
+        packet_loss_rate: Option<f64>,
+        packet_delay: u32,
     ) -> io::Result<Self> {
         for x in stun_server.iter_mut() {
             if !x.contains(":") {
@@ -111,6 +116,8 @@ impl Config {
             #[cfg(target_os = "android")]
             device_fd,
             use_channel_type,
+            packet_loss_rate,
+            packet_delay,
         })
     }
 }

vnt/src/handle/tun_tap/mod.rs

@@ -150,10 +150,5 @@ pub fn base_handle(
     }
     client_cipher.encrypt_ipv4(&mut net_packet)?;
     //优先发到直连到地址
-    if context.send_by_id(net_packet.buffer(), &dest_ip).is_err() {
-        if !context.use_channel_type().is_only_p2p() {
-            context.send_default(net_packet.buffer(), current_device.connect_server)?;
-        }
-    }
-    return Ok(());
+    context.send_ipv4_by_id(net_packet.buffer(), &dest_ip, current_device.connect_server)
 }

vnt/src/handle/tun_tap/tun_handler.rs

@@ -47,11 +47,11 @@ fn handle(
     client_cipher: &Cipher,
     server_cipher: &Cipher,
 ) -> io::Result<()> {
-    if len > 12 && data[12] >> 4 != 4 {
-        //忽略非ipv4包
-        return Ok(());
-    }
-    let ipv4_packet = IpV4Packet::new(&mut data[12..len])?;
+    //忽略掉结构不对的情况（ipv6数据、win tap会读到空数据），不然日志打印太多了
+    let ipv4_packet = match IpV4Packet::new(&mut data[12..len]) {
+        Ok(packet) => packet,
+        Err(_) => return Ok(()),
+    };
     let src_ip = ipv4_packet.source_ip();
     let dest_ip = ipv4_packet.destination_ip();
     if src_ip == dest_ip {