yeyinlaoying/vnt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

优化广播

Browse Source
This commit is contained in:
lubeilin
2024-03-30 00:24:19 +08:00
parent b26c4b97b2
commit 284cf38f0b
3 changed files with 87 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vnt/src/core/conn.rs
+1
View File
@@ -190,6 +190,7 @@ impl Vnt {
server_cipher.clone(),
config.parallel,
up_counter,
device_list.clone(),
)?;
maintain::idle_gateway(
&scheduler,
vnt/src/handle/tun_tap/mod.rs
+76 -42
View File
@@ -1,13 +1,15 @@
use std::io;
use std::net::Ipv4Addr;
use crate::channel::context::Context;
use parking_lot::Mutex;
use packet::ip::ipv4::packet::IpV4Packet;
use packet::ip::ipv4::protocol::Protocol;
use crate::channel::context::Context;
use crate::cipher::Cipher;
use crate::external_route::ExternalRoute;
use crate::handle::{check_dest, CurrentDeviceInfo};
use crate::handle::{check_dest, CurrentDeviceInfo, PeerDeviceInfo};
#[cfg(feature = "ip_proxy")]
use crate::ip_proxy::{IpProxyMap, ProxyHandler};
use crate::protocol;
@@ -23,57 +25,82 @@ fn broadcast(
sender: &Context,
net_packet: &mut NetPacket<&mut [u8]>,
current_device: &CurrentDeviceInfo,
device_list: &Mutex<(u16, Vec<PeerDeviceInfo>)>,
) -> io::Result<()> {
let mut peer_ips = Vec::with_capacity(8);
let vec = sender.route_table.route_table_one();
let mut relay_count = 0;
let list: Vec<Ipv4Addr> = device_list
.lock()
.1
.iter()
.filter(|info| info.status.is_online())
.map(|info| info.virtual_ip)
.collect();
const MAX_COUNT: usize = 8;
for (peer_ip, route) in vec {
if peer_ip == current_device.virtual_gateway {
continue;
}
if peer_ips.len() == MAX_COUNT {
relay_count += 1;
let mut p2p_ips = Vec::with_capacity(8);
let mut relay_ips = Vec::with_capacity(8);
let mut overflow = false;
for (index, peer_ip) in list.into_iter().enumerate() {
if index > MAX_COUNT {
overflow = true;
break;
}
if route.is_p2p()
&& sender
if let Some(route) = sender.route_table.route_one_p2p(&peer_ip) {
if sender
.send_by_key(net_packet.buffer(), route.route_key())
.is_ok()
{
peer_ips.push(peer_ip);
} else {
relay_count += 1;
{
p2p_ips.push(peer_ip);
continue;
}
}
relay_ips.push(peer_ip);
}
if (relay_count == 0 && !peer_ips.is_empty()) || current_device.status.offline() {
//不需要转发
if !overflow && relay_ips.is_empty() {
//全部p2p,不需要服务器中转
return Ok(());
}
//转发到服务端的可选择广播，还要进行服务端加密
if peer_ips.is_empty() {
sender.send_default(net_packet.buffer(), current_device.connect_server)?;
} else {
let buf =
vec![0u8; 12 + 1 + peer_ips.len() * 4 + net_packet.data_len() + ENCRYPTION_RESERVED];
//剩余的发送到服务端，需要告知哪些已发送过
let mut server_packet = NetPacket::new_encrypt(buf)?;
server_packet.set_version(Version::V1);
server_packet.set_gateway_flag(true);
server_packet.first_set_ttl(MAX_TTL);
server_packet.set_source(net_packet.source());
//使用对应的目的地址
server_packet.set_destination(net_packet.destination());
server_packet.set_protocol(protocol::Protocol::IpTurn);
server_packet.set_transport_protocol(ip_turn_packet::Protocol::Ipv4Broadcast.into());
let mut broadcast = BroadcastPacket::unchecked(server_packet.payload_mut());
broadcast.set_address(&peer_ips)?;
broadcast.set_data(net_packet.buffer())?;
server_cipher.encrypt_ipv4(&mut server_packet)?;
sender.send_default(server_packet.buffer(), current_device.connect_server)?;
if p2p_ips.is_empty() {
//都没有p2p则直接由服务器转发
if current_device.status.online() {
sender.send_default(net_packet.buffer(), current_device.connect_server)?;
}
return Ok(());
}
Ok(())
if !overflow && relay_ips.len() == 2 {
// 如果转发的ip数不多就直接发
for peer_ip in relay_ips {
//非直连的广播要改变目的地址，不然服务端收到了会再次广播
net_packet.set_destination(peer_ip);
sender.send_ipv4_by_id(
net_packet.buffer(),
&peer_ip,
current_device.connect_server,
current_device.status.online(),
)?;
}
return Ok(());
}
if current_device.status.offline() {
//离线的不再转发
return Ok(());
}
let buf = vec![0u8; 12 + 1 + p2p_ips.len() * 4 + net_packet.data_len() + ENCRYPTION_RESERVED];
//剩余的发送到服务端，需要告知哪些已发送过
let mut server_packet = NetPacket::new_encrypt(buf)?;
server_packet.set_version(Version::V1);
server_packet.set_gateway_flag(true);
server_packet.first_set_ttl(MAX_TTL);
server_packet.set_source(net_packet.source());
//使用对应的目的地址
server_packet.set_destination(net_packet.destination());
server_packet.set_protocol(protocol::Protocol::IpTurn);
server_packet.set_transport_protocol(ip_turn_packet::Protocol::Ipv4Broadcast.into());
let mut broadcast = BroadcastPacket::unchecked(server_packet.payload_mut());
broadcast.set_address(&p2p_ips)?;
broadcast.set_data(net_packet.buffer())?;
server_cipher.encrypt_ipv4(&mut server_packet)?;
sender.send_default(server_packet.buffer(), current_device.connect_server)
}
/// 实现一个原地发送，必须保证是如下结构
@@ -89,6 +116,7 @@ pub fn base_handle(
#[cfg(feature = "ip_proxy")] proxy_map: &Option<IpProxyMap>,
client_cipher: &Cipher,
server_cipher: &Cipher,
device_list: &Mutex<(u16, Vec<PeerDeviceInfo>)>,
) -> io::Result<()> {
let ipv4_packet = IpV4Packet::new(&buf[12..data_len])?;
let protocol = ipv4_packet.protocol();
@@ -118,7 +146,13 @@ pub fn base_handle(
if dest_ip.is_broadcast() || current_device.broadcast_ip == dest_ip {
// 广播 发送到直连目标
client_cipher.encrypt_ipv4(&mut net_packet)?;
broadcast(server_cipher, context, &mut net_packet, &current_device)?;
broadcast(
server_cipher,
context,
&mut net_packet,
&current_device,
device_list,
)?;
return Ok(());
}
if !check_dest(
vnt/src/handle/tun_tap/tun_handler.rs
+10 -1
View File
@@ -2,6 +2,7 @@ use std::sync::Arc;
use std::{io, thread};
use crossbeam_utils::atomic::AtomicCell;
use parking_lot::Mutex;
use packet::icmp::icmp::IcmpPacket;
use packet::icmp::Kind;
@@ -14,7 +15,7 @@ use crate::channel::context::Context;
use crate::cipher::Cipher;
use crate::external_route::ExternalRoute;
use crate::handle::tun_tap::channel_group::{channel_group, GroupSyncSender};
use crate::handle::CurrentDeviceInfo;
use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo};
#[cfg(feature = "ip_proxy")]
use crate::ip_proxy::IpProxyMap;
use crate::util::{SingleU64Adder, StopManager};
@@ -46,6 +47,7 @@ fn handle(
#[cfg(feature = "ip_proxy")] proxy_map: &Option<IpProxyMap>,
client_cipher: &Cipher,
server_cipher: &Cipher,
device_list: &Mutex<(u16, Vec<PeerDeviceInfo>)>,
) -> io::Result<()> {
//忽略掉结构不对的情况（ipv6数据、win tap会读到空数据），不然日志打印太多了
let ipv4_packet = match IpV4Packet::new(&mut data[12..len]) {
@@ -67,6 +69,7 @@ fn handle(
proxy_map,
client_cipher,
server_cipher,
device_list,
);
}
@@ -81,6 +84,7 @@ pub fn start(
server_cipher: Cipher,
parallel: usize,
mut up_counter: SingleU64Adder,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
) -> io::Result<()> {
let worker = {
#[cfg(target_os = "macos")]
@@ -110,6 +114,7 @@ pub fn start(
let ip_proxy_map = ip_proxy_map.clone();
let client_cipher = client_cipher.clone();
let server_cipher = server_cipher.clone();
let device_list = device_list.clone();
thread::Builder::new()
.name(format!("tunHandler-{}", index))
.spawn(move || {
@@ -129,6 +134,7 @@ pub fn start(
&ip_proxy_map,
&client_cipher,
&server_cipher,
&device_list,
) {
Ok(_) => {}
Err(e) => {
@@ -161,6 +167,7 @@ pub fn start(
client_cipher,
server_cipher,
&mut up_counter,
device_list,
) {
log::warn!("stop:{}", e);
}
@@ -180,6 +187,7 @@ fn start_simple(
client_cipher: Cipher,
server_cipher: Cipher,
up_counter: &mut SingleU64Adder,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
) -> io::Result<()> {
let mut buf = [0; 1024 * 16];
loop {
@@ -204,6 +212,7 @@ fn start_simple(
&ip_proxy_map,
&client_cipher,
&server_cipher,
&device_list,
) {
Ok(_) => {}
Err(e) => {