去除udp通道的arc包装

vnt-cli/src/config/mod.rs

@@ -158,7 +158,8 @@ pub fn read_config(file_path: &str) -> io::Result<(Config, bool)> {
         punch_model,
         file_conf.port,
         file_conf.first_latency,
-    );
+    )
+    .unwrap();
     Ok((config, file_conf.cmd))
 }
 

vnt-cli/src/main.rs

@@ -288,7 +288,8 @@ fn main() {
             punch_model,
             port,
             first_latency,
-        );
+        )
+        .unwrap();
         (config, cmd)
     };
     println!("version {}", vnt::VNT_VERSION);

vnt-jni/src/vnt_util.rs

@@ -130,7 +130,7 @@ fn new_sync(env: &mut JNIEnv, config: JObject) -> Result<VntUtilSync, Error> {
     for addr in stun_server_str.split(",") {
         stun_server.push(addr.trim().to_string());
     }
-    let config = Config::new(
+    let config = match Config::new(
         false,
         token,
         device_id,
@@ -154,7 +154,17 @@ fn new_sync(env: &mut JNIEnv, config: JObject) -> Result<VntUtilSync, Error> {
         PunchModel::All,
         port,
         first_latency,
-    );
+    ) {
+        Ok(config) => config,
+        Err(e) => {
+            env.throw_new(
+                "java/lang/RuntimeException",
+                format!("vnt start error {}", e),
+            )
+            .expect("throw");
+            return Err(Error::JavaException);
+        }
+    };
     match VntUtilSync::new(config) {
         Ok(vnt_util) => Ok(vnt_util),
         Err(e) => {

vnt/src/channel/channel.rs

@@ -8,7 +8,7 @@ use std::time::{Duration, Instant};
 use std::{io, thread};
 
 use crossbeam_utils::atomic::AtomicCell;
-use parking_lot::RwLock;
+use parking_lot::{Mutex, RwLock};
 use tokio::net::UdpSocket;
 use tokio::sync::watch::{channel, Receiver, Sender};
 
@@ -20,9 +20,9 @@ use crate::handle::CurrentDeviceInfo;
 
 pub struct ContextInner {
     //udp用于打洞、服务端通信(可选)
-    pub(crate) main_channel: Arc<StdUdpSocket>,
+    pub(crate) main_channel: StdUdpSocket,
     //在udp的基础上，可以选择使用tcp和服务端通信
-    pub(crate) main_tcp_channel: Option<std::sync::mpsc::SyncSender<Vec<u8>>>,
+    pub(crate) main_tcp_channel: Option<Mutex<TcpStream>>,
     pub(crate) route_table: RwLock<HashMap<Ipv4Addr, Vec<(Route, AtomicCell<Instant>)>>>,
     pub(crate) status_receiver: Receiver<Status>,
     pub(crate) status_sender: Sender<Status>,
@@ -39,8 +39,8 @@ pub struct Context {
 
 impl Context {
     pub fn new(
-        main_channel: Arc<StdUdpSocket>,
-        main_tcp_channel: Option<std::sync::mpsc::SyncSender<Vec<u8>>>,
+        main_channel: StdUdpSocket,
+        main_tcp_channel: Option<TcpStream>,
         current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
         _channel_num: usize,
         first_latency: bool,
@@ -48,6 +48,7 @@ impl Context {
         //当前版本只支持一个通道
         let channel_num = 1;
         let (status_sender, status_receiver) = channel(Status::Cone);
+        let main_tcp_channel = main_tcp_channel.map(|e| Mutex::new(e));
         let inner = Arc::new(ContextInner {
             main_channel,
             main_tcp_channel,
@@ -79,13 +80,16 @@ impl Context {
             );
         }
         if let Some(tcp) = &self.inner.main_tcp_channel {
-            let _ = tcp.send(vec![]);
+            tcp.lock().shutdown(Shutdown::Both)?;
         }
         Ok(())
     }
     pub fn is_main_tcp(&self) -> bool {
         self.inner.main_tcp_channel.is_some()
     }
+    pub fn is_first_latency(&self) -> bool {
+        self.inner.first_latency
+    }
     pub fn switch(&self, nat_type: NatType) {
         match nat_type {
             NatType::Symmetric => {
@@ -123,14 +127,32 @@ impl Context {
         }
         self.inner.main_channel.send_to(buf, addr)
     }
+    #[inline]
+    pub fn send_main_tcp(&self, buf: &[u8]) -> io::Result<usize> {
+        if let Some(sender) = &self.inner.main_tcp_channel {
+            let mut stream = sender.lock();
+            let mut head = [0; 4];
+            let len = buf.len();
+            head[2] = (len >> 8) as u8;
+            head[3] = (len & 0xFF) as u8;
+            stream.write_all(&head)?;
+            stream.write_all(buf)?;
+            Ok(len)
+        } else {
+            return Err(io::Error::new(io::ErrorKind::NotFound, "tcp not found"));
+        }
+    }
 
     pub fn send_main(&self, buf: &[u8], addr: SocketAddr) -> io::Result<usize> {
         if let Some(sender) = &self.inner.main_tcp_channel {
-            if sender.try_send(buf.to_vec()).is_ok() {
-                Ok(buf.len())
-            } else {
-                Err(io::Error::new(io::ErrorKind::Other, "send_main err"))
-            }
+            let mut stream = sender.lock();
+            let mut head = [0; 4];
+            let len = buf.len();
+            head[2] = (len >> 8) as u8;
+            head[3] = (len & 0xFF) as u8;
+            stream.write_all(&head)?;
+            stream.write_all(buf)?;
+            Ok(len)
         } else {
             self.send_main_udp(buf, addr)
         }
@@ -182,17 +204,7 @@ impl Context {
 
     pub async fn send_by_key(&self, buf: &[u8], route_key: &RouteKey) -> io::Result<usize> {
         match route_key.index {
-            TCP_ID => {
-                if let Some(sender) = &self.inner.main_tcp_channel {
-                    if sender.send(buf.to_vec()).is_ok() {
-                        Ok(buf.len())
-                    } else {
-                        Err(io::Error::new(io::ErrorKind::Other, "send_by_key err"))
-                    }
-                } else {
-                    Err(io::Error::new(io::ErrorKind::Other, "send_by_key err"))
-                }
-            }
+            TCP_ID => self.send_main_tcp(buf),
             UDP_ID => self.send_main_udp(buf, route_key.addr),
             _ => {
                 if let Some(udp) = self.get_udp_by_route(route_key) {
@@ -204,17 +216,7 @@ impl Context {
     }
     pub fn try_send_by_key(&self, buf: &[u8], route_key: &RouteKey) -> io::Result<usize> {
         match route_key.index {
-            TCP_ID => {
-                if let Some(sender) = &self.inner.main_tcp_channel {
-                    if sender.try_send(buf.to_vec()).is_ok() {
-                        Ok(buf.len())
-                    } else {
-                        Err(io::Error::new(io::ErrorKind::Other, "send_by_key err"))
-                    }
-                } else {
-                    Err(io::Error::new(io::ErrorKind::Other, "send_by_key err"))
-                }
-            }
+            TCP_ID => self.send_main_tcp(buf),
             UDP_ID => self.send_main_udp(buf, route_key.addr),
             _ => {
                 if let Some(udp) = self.get_udp_by_route(route_key) {
@@ -242,8 +244,8 @@ impl Context {
             .or_insert_with(|| Vec::with_capacity(4));
         let mut exist = false;
         for (x, time) in list.iter_mut() {
-            if x.metric < route.metric {
-                //不能比当前的路径更长
+            if x.metric < route.metric && !self.inner.first_latency {
+                //非优先延迟的情况下 不能比当前的路径更长
                 return;
             }
             if x.route_key() == key {
@@ -260,12 +262,16 @@ impl Context {
         if exist {
             list.sort_by_key(|(k, _)| k.rt);
         } else {
-            if route.metric == 1 && !self.inner.first_latency {
-                //非优先延迟的情况下 添加了直连的则排除非直连的
-                list.retain(|(k, _)| k.metric == 1);
-            }
+            let max_len = if self.inner.first_latency {
+                self.inner.channel_num + 1
+            } else {
+                if route.metric == 1 {
+                    //非优先延迟的情况下 添加了直连的则排除非直连的
+                    list.retain(|(k, _)| k.metric == 1);
+                }
+                self.inner.channel_num
+            };
             list.sort_by_key(|(k, _)| k.rt);
-            let max_len = self.inner.channel_num;
             if list.len() > max_len {
                 list.truncate(max_len);
             }
@@ -399,14 +405,17 @@ fn buf_channel_group(size: usize) -> (BufSenderGroup, BufReceiverGroup) {
 impl Channel {
     fn tcp_handle(
         tcp_r: &mut TcpStream,
-        context: Context,
-        handler: ChannelDataHandler,
+        context: &Context,
+        handler: &ChannelDataHandler,
         head_reserve: usize,
     ) -> io::Result<()> {
         let mut head = [0; 4];
         let addr = tcp_r.peer_addr()?;
         let key = RouteKey::new(TCP_ID, addr);
         loop {
+            if context.is_close() {
+                return Ok(());
+            }
             let mut buf = [0; 4096];
             tcp_r.read_exact(&mut head)?;
             let len = (((head[2] as u16) << 8) | head[3] as u16) as usize;
@@ -417,97 +426,57 @@ impl Channel {
                 ));
             }
             tcp_r.read_exact(&mut buf[head_reserve..head_reserve + len])?;
-            handler.handle(&mut buf, head_reserve, head_reserve + len, key, &context);
+            handler.handle(&mut buf, head_reserve, head_reserve + len, key, context);
         }
     }
     fn start_tcp(
-        worker: VntWorker,
         mut tcp_stream: TcpStream,
-        receiver: std::sync::mpsc::Receiver<Vec<u8>>,
         context: Context,
         handler: ChannelDataHandler,
         head_reserve: usize,
     ) {
         let current_device = context.inner.current_device.clone();
-        {
-            let mut tcp_r = tcp_stream.try_clone().unwrap();
-            let context = context.clone();
-            let handler = handler.clone();
-            thread::Builder::new()
-                .name("tcp_reader".into())
-                .spawn(move || {
-                    if let Err(e) = Self::tcp_handle(&mut tcp_r, context, handler, head_reserve) {
-                        log::info!("tcp链接断开:{:?}", e);
-                    }
-                    if let Err(e) = tcp_r.shutdown(Shutdown::Both) {
-                        log::info!("tcp链接关闭异常:{:?}", e);
-                    }
-                })
-                .unwrap();
-        }
-        let mut head = [0; 4];
         loop {
-            let data = match receiver.recv() {
-                Ok(data) => data,
-                Err(_) => {
-                    break;
-                }
-            };
-            let len = data.len();
-            if len == 0 {
-                break;
+            if let Err(e) = tcp_stream.set_nodelay(true) {
+                log::info!("set_nodelay:{:?}", e);
             }
-            head[2] = (len >> 8) as u8;
-            head[3] = (len & 0xFF) as u8;
-            let mut err = false;
-            if let Err(e) = tcp_stream.write_all(&head) {
-                err = true;
-                log::info!("发送失败,需要重连:{:?}", e);
-            } else if let Err(e) = tcp_stream.write_all(&data) {
-                err = true;
-                log::info!("发送失败,需要重连:{:?}", e);
+            if let Err(e) = tcp_stream.set_write_timeout(Some(Duration::from_secs(3))) {
+                log::info!("set_write_timeout:{:?}", e);
             }
-            if err {
-                if let Err(e) = tcp_stream.shutdown(Shutdown::Both) {
-                    log::info!("tcp链接关闭异常:{:?}", e);
+            if let Err(e) = tcp_stream.set_read_timeout(Some(Duration::from_secs(10))) {
+                log::info!("set_read_timeout:{:?}", e);
+            }
+            if let Err(e) = Self::tcp_handle(&mut tcp_stream, &context, &handler, head_reserve) {
+                log::info!("tcp链接断开:{:?}", e);
+            }
+            if let Err(e) = tcp_stream.shutdown(Shutdown::Both) {
+                log::info!("tcp链接关闭异常:{:?}", e);
+            }
+            loop {
+                if context.is_close() {
+                    return;
                 }
-                match TcpStream::connect(current_device.load().connect_server) {
+                let device_info = current_device.load();
+                match TcpStream::connect(device_info.connect_server) {
                     Ok(tcp) => {
-                        tcp.set_read_timeout(Some(Duration::from_secs(10))).unwrap();
-                        tcp_stream = tcp;
-                        let mut tcp_r = tcp_stream.try_clone().unwrap();
-                        let context = context.clone();
-                        let handler = handler.clone();
-                        thread::Builder::new()
-                            .name("tcp_reader".into())
-                            .spawn(move || {
-                                if let Err(e) =
-                                    Self::tcp_handle(&mut tcp_r, context, handler, head_reserve)
-                                {
-                                    log::info!("重连后 tcp链接断开:{:?}", e);
-                                }
-                                if let Err(e) = tcp_r.shutdown(Shutdown::Both) {
-                                    log::info!("重连后 tcp链接关闭异常:{:?}", e);
-                                }
-                            })
-                            .unwrap();
+                        tcp_stream = tcp.try_clone().unwrap();
+                        let mut guard = context.inner.main_tcp_channel.as_ref().unwrap().lock();
+                        *guard = tcp;
+                        break;
                     }
                     Err(e) => {
-                        log::info!("重连失败:{:?}", e);
+                        log::info!("重连失败,{},{:?}", device_info.connect_server, e);
+                        thread::sleep(Duration::from_secs(3));
                     }
                 }
             }
         }
-        if let Err(e) = tcp_stream.shutdown(Shutdown::Both) {
-            log::info!("tcp链接关闭异常:{:?}", e);
-        }
-        worker.stop_all();
     }
 
     pub async fn start(
         self,
         mut worker: VntWorker,
-        tcp: Option<(TcpStream, std::sync::mpsc::Receiver<Vec<u8>>)>,
+        tcp: Option<TcpStream>,
         head_reserve: usize,          //头部预留字节
         symmetric_channel_num: usize, //对称网络，则再加一组监听，提升打洞成功率
         relay: bool,
@@ -515,7 +484,7 @@ impl Channel {
     ) {
         let handler = self.handler.clone();
         let context = self.context;
-        let main_channel = context.inner.main_channel.clone();
+        let main_channel = context.inner.main_channel.try_clone().unwrap();
         let buf_sender = if parallel > 1 {
             let (buf_sender, buf_receiver) = buf_channel_group(parallel);
             let mut num = 0;
@@ -537,32 +506,26 @@ impl Channel {
         } else {
             None
         };
-        if let Some((tcp_stream, receiver)) = tcp {
+        if let Some(tcp_stream) = tcp {
             let context = context.clone();
             let handler = handler.clone();
             let main_channel_tcp = worker.worker("main_channel_tcp");
             thread::Builder::new()
-                .name("main_channel_tcp".into())
+                .name("channel_tcp".into())
                 .spawn(move || {
-                    Self::start_tcp(
-                        main_channel_tcp,
-                        tcp_stream,
-                        receiver,
-                        context,
-                        handler,
-                        head_reserve,
-                    )
+                    Self::start_tcp(tcp_stream, context, handler, head_reserve);
+                    drop(main_channel_tcp)
                 })
                 .unwrap();
         }
         {
-            let worker = worker.worker("main_channel_1");
+            let worker = worker.worker("main_channel_udp");
             let context = context.clone();
-            let main_channel = main_channel.clone();
+            let main_channel = main_channel.try_clone().unwrap();
             let handler = handler.clone();
             let buf_sender = buf_sender.clone();
             thread::Builder::new()
-                .name("ipv4-recv".into())
+                .name("channel_udp".into())
                 .spawn(move || {
                     log::info!("启动udp v4");
                     Self::main_start_(
@@ -583,6 +546,7 @@ impl Channel {
         }
         let mut cur_status = Status::Cone;
         let mut status_receiver = context.inner.status_receiver.clone();
+        let channel_num = context.inner.channel_num;
         loop {
             tokio::select! {
                 _=worker.stop_wait()=>{
@@ -601,7 +565,7 @@ impl Channel {
                                         continue;
                                     }
                                     cur_status = Status::Symmetric;
-                                    for _ in 0..symmetric_channel_num {
+                                    for _ in 0..symmetric_channel_num - channel_num {
                                         match UdpSocket::bind("0.0.0.0:0").await {
                                             Ok(udp) => {
                                                 let udp = Arc::new(udp);
@@ -632,7 +596,7 @@ impl Channel {
         worker: VntWorker,
         context: Context,
         id: usize,
-        udp: Arc<StdUdpSocket>,
+        udp: StdUdpSocket,
         handler: ChannelDataHandler,
         buf_sender: Option<BufSenderGroup>,
         head_reserve: usize,

vnt/src/channel/punch.rs

@@ -52,7 +52,13 @@ impl NatInfo {
         ipv6_addr: SocketAddrV6,
         mut nat_type: NatType,
     ) -> Self {
-        public_ips.retain(|ip| !ip.is_loopback() && !ip.is_private() && !ip.is_unspecified());
+        public_ips.retain(|ip| {
+            !ip.is_multicast()
+                && !ip.is_broadcast()
+                && !ip.is_unspecified()
+                && !ip.is_loopback()
+                && !ip.is_private()
+        });
         if public_ips.len() > 1 {
             nat_type = NatType::Symmetric;
         }
@@ -65,6 +71,20 @@ impl NatInfo {
             nat_type,
         }
     }
+    pub fn update_addr(&mut self, ip: Ipv4Addr, port: u16) {
+        if !ip.is_multicast()
+            && !ip.is_broadcast()
+            && !ip.is_unspecified()
+            && !ip.is_loopback()
+            && !ip.is_private()
+            && port != 0
+        {
+            self.public_port = port;
+            if !self.public_ips.contains(&ip) {
+                self.public_ips.push(ip);
+            }
+        }
+    }
 }
 
 #[derive(Clone)]
@@ -165,15 +185,17 @@ impl Punch {
                 self.port_index.insert(id, index);
             }
             NatType::Cone => {
-                let is_cone = self.context.is_cone();
-                for ip in nat_info.public_ips {
-                    let addr = SocketAddr::V4(SocketAddrV4::new(ip, nat_info.public_port));
-                    self.context.send_main_udp(buf, addr)?;
-                    if !is_cone {
-                        //只有一方是对称，则对称方要使用全部端口发送数据，符合上述计算的概率
-                        self.context.try_send_all(buf, addr)?;
+                if nat_info.public_port != 0 {
+                    let is_cone = self.context.is_cone();
+                    for ip in nat_info.public_ips {
+                        let addr = SocketAddr::V4(SocketAddrV4::new(ip, nat_info.public_port));
+                        self.context.send_main_udp(buf, addr)?;
+                        if !is_cone {
+                            //只有一方是对称，则对称方要使用全部端口发送数据，符合上述计算的概率
+                            self.context.try_send_all(buf, addr)?;
+                        }
+                        tokio::time::sleep(Duration::from_millis(2)).await;
                     }
-                    tokio::time::sleep(Duration::from_millis(2)).await;
                 }
             }
         }

vnt/src/core/mod.rs

@@ -236,14 +236,13 @@ impl VntUtil {
 
         let (cone_sender, cone_receiver) = channel(3);
         let (symmetric_sender, symmetric_receiver) = channel(2);
-        let (tcp_sender, tcp) = if let Some(main_tcp_channel) = self.main_tcp_channel {
-            let (tcp_sender, tcp_receiver) = std::sync::mpsc::sync_channel::<Vec<u8>>(100);
-            (Some(tcp_sender), Some((main_tcp_channel, tcp_receiver)))
+        let (tcp_sender, tcp_receiver) = if let Some(main_tcp_channel) = self.main_tcp_channel {
+            (Some(main_tcp_channel.try_clone()?), Some(main_tcp_channel))
         } else {
             (None, None)
         };
         let context = Context::new(
-            Arc::new(self.main_channel),
+            self.main_channel,
             tcp_sender,
             current_device.clone(),
             1,
@@ -383,7 +382,7 @@ impl VntUtil {
             let relay = config.relay;
             tokio::spawn(async move {
                 channel
-                    .start(channel_worker, tcp, 14, 65, relay, config.parallel)
+                    .start(channel_worker, tcp_receiver, 14, 65, relay, config.parallel)
                     .await
             });
         }
@@ -600,13 +599,22 @@ impl Config {
         punch_model: PunchModel,
         port: u16,
         first_latency: bool,
-    ) -> Self {
+    ) -> Result<Self, Error> {
         for x in stun_server.iter_mut() {
             if !x.contains(":") {
                 x.push_str(":3478");
             }
         }
-        Self {
+        if token.is_empty() || token.len() > 128 {
+            return Err(Error::Stop(String::from("token too long")));
+        }
+        if device_id.is_empty() || device_id.len() > 128 {
+            return Err(Error::Stop(String::from("device_id too long")));
+        }
+        if name.is_empty() || name.len() > 128 {
+            return Err(Error::Stop(String::from("name too long")));
+        }
+        Ok(Self {
             tap,
             token,
             device_id,
@@ -631,6 +639,6 @@ impl Config {
             punch_model,
             port,
             first_latency,
-        }
+        })
     }
 }

vnt/src/handle/heartbeat_handler.rs

@@ -201,7 +201,7 @@ async fn start_heartbeat_(
 
         let src = current_dev.virtual_ip();
 
-        if count < 7 || count % 7 == 0 {
+        if count % 10 == 7 {
             let mut route_list: Option<Vec<(Ipv4Addr, Vec<Route>)>> = None;
             let peer_list = { device_list.lock().1.clone() };
             for peer in peer_list {
@@ -223,7 +223,7 @@ async fn start_heartbeat_(
                     {
                         log::warn!("virtual_ip:{},route:{:?},e:{:?}", peer.virtual_ip, route, e);
                     }
-                    if route.is_p2p() {
+                    if route.is_p2p() && !sender.is_first_latency() {
                         continue;
                     }
                 } else {

vnt/src/nat/mod.rs

@@ -111,19 +111,8 @@ impl NatTest {
         self.info.lock().clone()
     }
     pub fn update_addr(&self, ip: Ipv4Addr, port: u16) {
-        if !ip.is_multicast()
-            && !ip.is_broadcast()
-            && !ip.is_unspecified()
-            && !ip.is_loopback()
-            && !ip.is_private()
-            && port != 0
-        {
-            let mut guard = self.info.lock();
-            guard.public_port = port;
-            if !guard.public_ips.contains(&ip) {
-                guard.public_ips.push(ip);
-            }
-        }
+        let mut guard = self.info.lock();
+        guard.update_addr(ip, port)
     }
     pub async fn re_test(
         &self,