diff --git a/vnt/src/channel/context.rs b/vnt/src/channel/context.rs index 869a963..a221ced 100644 --- a/vnt/src/channel/context.rs +++ b/vnt/src/channel/context.rs @@ -28,6 +28,7 @@ impl Context { is_tcp: bool, packet_loss_rate: Option, packet_delay: u32, + use_ipv6: bool, ) -> Self { let channel_num = main_udp_socket.len(); assert_ne!(channel_num, 0, "not channel"); @@ -51,6 +52,7 @@ impl Context { packet_loss_rate, packet_delay, main_index: AtomicUsize::new(0), + use_ipv6, }; Self { inner: Arc::new(inner), @@ -70,7 +72,7 @@ impl Deref for Context { } /// 对称网络增加的udp socket数目,有助于增加打洞成功率 -pub const SYMMETRIC_CHANNEL_NUM: usize = 64; +pub const SYMMETRIC_CHANNEL_NUM: usize = 100; const PACKET_LOSS_RATE_DENOMINATOR: u32 = 100_0000; pub struct ContextInner { // 核心udp socket @@ -90,6 +92,7 @@ pub struct ContextInner { //控制延迟 packet_delay: u32, main_index: AtomicUsize, + use_ipv6: bool, } impl ContextInner { @@ -172,15 +175,16 @@ impl ContextInner { } } pub fn send_main_udp(&self, index: usize, buf: &[u8], mut addr: SocketAddr) -> io::Result<()> { - //核心udp socket都是ipv6模式,如果是v4地址则需要转换成v6 - //只有服务器地址可能需要这样转换 - if let SocketAddr::V4(ipv4) = addr { - addr = SocketAddr::V6(SocketAddrV6::new( - ipv4.ip().to_ipv6_mapped(), - ipv4.port(), - 0, - 0, - )); + if self.use_ipv6 { + //如果是v4地址则需要转换成v6 + if let SocketAddr::V4(ipv4) = addr { + addr = SocketAddr::V6(SocketAddrV6::new( + ipv4.ip().to_ipv6_mapped(), + ipv4.port(), + 0, + 0, + )); + } } self.main_udp_socket[index].send_to(buf, addr)?; Ok(()) @@ -208,17 +212,9 @@ impl ContextInner { thread::sleep(Duration::from_millis(1)); } } - pub fn try_send_all_main(&self, buf: &[u8], mut addr: SocketAddr) { - if let SocketAddr::V4(ipv4) = addr { - addr = SocketAddr::V6(SocketAddrV6::new( - ipv4.ip().to_ipv6_mapped(), - ipv4.port(), - 0, - 0, - )); - } - for udp in &self.main_udp_socket { - if let Err(e) = udp.send_to(buf, addr) { + pub fn try_send_all_main(&self, buf: &[u8], addr: SocketAddr) { + for index in 0..self.channel_num() { + if let Err(e) = self.send_main_udp(index, buf, addr) { log::warn!("{:?},add={:?}", e, addr); } } diff --git a/vnt/src/channel/mod.rs b/vnt/src/channel/mod.rs index 6b0dd28..ebc6cf8 100644 --- a/vnt/src/channel/mod.rs +++ b/vnt/src/channel/mod.rs @@ -154,13 +154,31 @@ pub fn init_context( ) -> io::Result<(Context, mio::net::TcpListener)> { assert!(!ports.is_empty(), "not channel"); let mut udps = Vec::with_capacity(ports.len()); + //检查系统是否支持ipv6 + let use_ipv6 = match socket2::Socket::new(socket2::Domain::IPV6, socket2::Type::DGRAM, None) { + Ok(_) => true, + Err(e) => { + log::warn!("{:?}", e); + false + } + }; for port in &ports { //监听v6+v4双栈 - let address: SocketAddr = format!("[::]:{}", port).parse().unwrap(); - let socket = socket2::Socket::new(socket2::Domain::IPV6, socket2::Type::DGRAM, None)?; - io_convert(socket.set_only_v6(false), |_| { - format!("set_only_v6 failed: {}", &address) - })?; + let (socket, address) = if use_ipv6 { + let address: SocketAddr = format!("[::]:{}", port).parse().unwrap(); + let socket = socket2::Socket::new(socket2::Domain::IPV6, socket2::Type::DGRAM, None)?; + io_convert(socket.set_only_v6(false), |_| { + format!("set_only_v6 failed: {}", &address) + })?; + (socket, address) + } else { + let address: SocketAddr = format!("0.0.0.0:{}", port).parse().unwrap(); + ( + socket2::Socket::new(socket2::Domain::IPV4, socket2::Type::DGRAM, None)?, + address, + ) + }; + io_convert(socket.set_reuse_address(true), |_| { format!("set_reuse_address failed: {}", &address) })?; @@ -184,15 +202,24 @@ pub fn init_context( is_tcp, packet_loss_rate, packet_delay, + use_ipv6, ); let port = context.main_local_udp_port()?[0]; //监听v6+v4双栈,tcp通道使用异步io - let address: SocketAddr = format!("[::]:{}", port).parse().unwrap(); - let socket = socket2::Socket::new(socket2::Domain::IPV6, socket2::Type::STREAM, None)?; - io_convert(socket.set_only_v6(false), |_| { - format!("set_only_v6 failed: {}", &address) - })?; + let (socket, address) = if use_ipv6 { + let address: SocketAddr = format!("[::]:{}", port).parse().unwrap(); + let socket = socket2::Socket::new(socket2::Domain::IPV6, socket2::Type::STREAM, None)?; + io_convert(socket.set_only_v6(false), |_| { + format!("set_only_v6 failed: {}", &address) + })?; + (socket, address) + } else { + let address: SocketAddr = format!("0.0.0.0:{}", port).parse().unwrap(); + let socket = socket2::Socket::new(socket2::Domain::IPV4, socket2::Type::STREAM, None)?; + (socket, address) + }; + io_convert(socket.set_reuse_address(true), |_| { format!("set_reuse_address failed: {}", &address) })?; @@ -200,7 +227,11 @@ pub fn init_context( if ports[0] == 0 { //端口可能冲突,则使用任意端口 log::warn!("监听tcp端口失败 {:?},重试一次", address); - let address: SocketAddr = format!("[::]:{}", 0).parse().unwrap(); + let address: SocketAddr = if use_ipv6 { + format!("[::]:{}", 0).parse().unwrap() + } else { + format!("0.0.0.0:{}", port).parse().unwrap() + }; io_convert(socket.bind(&address.into()), |_| { format!("bind failed: {}", &address) })?;