Commit 3b8eeedb authored by nanahira's avatar nanahira

change

parent f5ffc9fe
Pipeline #37396 passed with stages
in 49 seconds
...@@ -38,7 +38,6 @@ pub struct Config { ...@@ -38,7 +38,6 @@ pub struct Config {
} }
use crossbeam_utils::thread; use crossbeam_utils::thread;
use grouping_by::GroupingBy; use grouping_by::GroupingBy;
use pnet::packet::ipv4::Ipv4Packet;
use socket2::Socket; use socket2::Socket;
fn main() -> Result<(), Box<dyn Error>> { fn main() -> Result<(), Box<dyn Error>> {
...@@ -75,29 +74,52 @@ fn main() -> Result<(), Box<dyn Error>> { ...@@ -75,29 +74,52 @@ fn main() -> Result<(), Box<dyn Error>> {
thread::scope(|s| { thread::scope(|s| {
for router in router_readers.values_mut() { for router in router_readers.values_mut() {
s.spawn(|_| { #[cfg(target_os = "linux")]
let mut buffer = [0u8; 1500 - 20]; // minus typical IP header space let mark_set = std::sync::atomic::AtomicBool::new(false);
s.spawn(move |_| {
// 使用更大的缓冲区以支持巨帧
let mut buffer = vec![0u8; 9000];
let meta_size = size_of::<Meta>(); let meta_size = size_of::<Meta>();
// Pre-initialize with our Meta header (local -> remote) // 预初始化 Meta 头部(local -> remote)
let meta = Meta { let meta = Meta {
src_id: config.local_id, src_id: config.local_id,
dst_id: router.config.remote_id, dst_id: router.config.remote_id,
reversed: 0, reversed: 0,
}; };
// Turn the Meta struct into bytes // 直接写入缓冲区,避免额外的切片操作
let meta_bytes = unsafe { unsafe {
std::slice::from_raw_parts(&meta as *const Meta as *const u8, meta_size) let meta_ptr = buffer.as_mut_ptr() as *mut Meta;
}; *meta_ptr = meta;
buffer[..meta_size].copy_from_slice(meta_bytes); }
loop { loop {
let n = router.tun_reader.read(&mut buffer[meta_size..]).unwrap(); match router.tun_reader.read(&mut buffer[meta_size..]) {
if let Some(ref addr) = *router.endpoint.read().unwrap() { Ok(n) => {
router.encrypt(&mut buffer[meta_size..meta_size + n]); // 使用 try_read 减少锁争用
#[cfg(target_os = "linux")] if let Ok(endpoint_guard) = router.endpoint.try_read() {
let _ = router.socket.set_mark(router.config.mark); if let Some(ref addr) = *endpoint_guard {
let _ = router.socket.send_to(&buffer[..meta_size + n], addr); router.encrypt(&mut buffer[meta_size..meta_size + n]);
#[cfg(target_os = "linux")]
{
// 只在第一次设置 mark
use std::sync::atomic::Ordering;
if !mark_set.load(Ordering::Relaxed) {
let _ = router.socket.set_mark(router.config.mark);
mark_set.store(true, Ordering::Relaxed);
}
}
let _ = router.socket.send_to(&buffer[..meta_size + n], addr);
}
}
}
Err(_) => {
// TUN 读取失败时短暂休眠,避免 CPU 空转
std::thread::sleep(std::time::Duration::from_millis(1));
}
} }
} }
}); });
...@@ -105,32 +127,54 @@ fn main() -> Result<(), Box<dyn Error>> { ...@@ -105,32 +127,54 @@ fn main() -> Result<(), Box<dyn Error>> {
for (socket, mut router_writers) in router_writers3 { for (socket, mut router_writers) in router_writers3 {
s.spawn(move |_| { s.spawn(move |_| {
let mut recv_buf = [MaybeUninit::uninit(); 1500]; // 使用更大的缓冲区和重用内存
let mut recv_buf = vec![MaybeUninit::uninit(); 9000];
let meta_size = size_of::<Meta>();
loop { loop {
let _ = (|| { match socket.recv_from(&mut recv_buf) {
let (len, addr) = socket.recv_from(&mut recv_buf).unwrap(); Ok((len, addr)) => {
let data: &mut [u8] = unsafe { transmute(&mut recv_buf[..len]) }; // 快速边界检查
if len < 20 + meta_size {
let packet = Ipv4Packet::new(data).ok_or("malformed packet")?; continue;
let header_len = packet.get_header_length() as usize * 4; }
let (_ip_header, rest) = data
.split_at_mut_checked(header_len) let data: &mut [u8] = unsafe { transmute(&mut recv_buf[..len]) };
.ok_or("malformed packet")?;
let (meta_bytes, payload) = rest // 优化:直接计算 IP 头部长度,避免创建 Ipv4Packet
.split_at_mut_checked(size_of::<Meta>()) let header_len = ((data[0] & 0x0f) as usize) * 4;
.ok_or("malformed packet")?;
let meta: &Meta = unsafe { transmute(meta_bytes.as_ptr()) }; if len < header_len + meta_size {
if meta.dst_id == config.local_id && meta.reversed == 0 { continue;
let router = router_writers }
.get_mut(&meta.src_id)
.ok_or("missing router")?; // 直接从内存读取 Meta,避免额外的切片操作
*router.endpoint.write().unwrap() = Some(addr); let meta: &Meta = unsafe {
router.decrypt(payload, &local_secret); &*(data.as_ptr().add(header_len) as *const Meta)
router.tun_writer.write_all(payload)?; };
if meta.dst_id == config.local_id && meta.reversed == 0 {
if let Some(router) = router_writers.get_mut(&meta.src_id) {
// 使用 try_write 减少锁争用
if let Ok(mut endpoint) = router.endpoint.try_write() {
*endpoint = Some(addr);
}
let payload_start = header_len + meta_size;
let payload = &mut data[payload_start..];
router.decrypt(payload, &local_secret);
// 忽略写入错误,继续处理下一个数据包
let _ = router.tun_writer.write_all(payload);
}
}
} }
Err(_) => {
Ok::<(), Box<dyn Error>>(()) // Socket 接收失败时短暂休眠
})(); std::thread::sleep(std::time::Duration::from_millis(1));
}
}
} }
}); });
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment