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

commit:

Browse Source 
1.去除缓冲池
2.数据处理改为同步方法
3.fmt
This commit is contained in:
lubeilin
2023-09-20 19:54:49 +08:00
parent 3429ee8bd6
commit d7c121a756
9 changed files with 76 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vnt/Cargo.toml
-2
View File
@@ -14,8 +14,6 @@ crossbeam-utils = "0.8"
crossbeam-epoch = "0.9.15"
dashmap = "5.5.1"
parking_lot = "0.12.1"
byte-pool = "0.2.4"
lazy_static = "1.4.0"
rand = "0.8.5"
sha2 = { version = "0.10.6", features = ["oid"] }
thiserror = "1.0.37"
vnt/src/channel/channel.rs
+25 -45
View File
@@ -6,7 +6,6 @@ use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::{Duration, Instant};
use byte_pool::{Block, BytePool};
use crossbeam_epoch::{Atomic, Owned};
use crossbeam_utils::atomic::AtomicCell;
use dashmap::DashMap;
@@ -23,9 +22,6 @@ use crate::handle::recv_handler::ChannelDataHandler;
use crate::handle::CurrentDeviceInfo;
use crate::ip_proxy::DashMapNew;
lazy_static::lazy_static! {
static ref POOL:BytePool = BytePool::new();
}
pub struct ContextInner {
//udp用于打洞、服务端通信(可选)
pub(crate) main_channel: Arc<StdUdpSocket>,
@@ -526,8 +522,10 @@ impl Context {
pub fn update_read_time(&self, id: &Ipv4Addr, route_key: &RouteKey) {
if let Some(mut time) = self.inner.route_table_time.get_mut(&(*route_key, *id)) {
*time.value_mut() = Instant::now();
}else{
self.inner.route_table_time.insert((*route_key,*id),Instant::now());
} else {
self.inner
.route_table_time
.insert((*route_key, *id), Instant::now());
}
}
}
@@ -546,13 +544,13 @@ impl Channel {
#[derive(Clone)]
struct BufSenderGroup(
usize,
Vec<std::sync::mpsc::SyncSender<(Block<'static>, usize, usize, RouteKey)>>,
Vec<std::sync::mpsc::SyncSender<(Vec<u8>, usize, usize, RouteKey)>>,
);
struct BufReceiverGroup(Vec<std::sync::mpsc::Receiver<(Block<'static>, usize, usize, RouteKey)>>);
struct BufReceiverGroup(Vec<std::sync::mpsc::Receiver<(Vec<u8>, usize, usize, RouteKey)>>);
impl BufSenderGroup {
pub fn send(&mut self, val: (Block<'static>, usize, usize, RouteKey)) -> bool {
pub fn send(&mut self, val: (Vec<u8>, usize, usize, RouteKey)) -> bool {
let index = self.0 % self.1.len();
self.0 = self.0.wrapping_add(1);
self.1[index].send(val).is_ok()
@@ -564,7 +562,7 @@ fn buf_channel_group(size: usize) -> (BufSenderGroup, BufReceiverGroup) {
let mut buf_receiver_group = Vec::with_capacity(size);
for _ in 0..size {
let (buf_sender, buf_receiver) =
std::sync::mpsc::sync_channel::<(Block<'static, Vec<u8>>, usize, usize, RouteKey)>(1);
std::sync::mpsc::sync_channel::<(Vec<u8>, usize, usize, RouteKey)>(1);
buf_sender_group.push(buf_sender);
buf_receiver_group.push(buf_receiver);
}
@@ -598,8 +596,7 @@ impl Channel {
.read_exact(&mut buf[head_reserve..head_reserve + len])
.await?;
handler
.handle(&mut buf, head_reserve, head_reserve + len, key, &context)
.await;
.handle(&mut buf, head_reserve, head_reserve + len, key, &context);
}
}
async fn start_tcp(
@@ -686,19 +683,11 @@ impl Channel {
let context = context.clone();
let handler = handler.clone();
std::thread::spawn(move || {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
log::info!("启动异步处理");
runtime.block_on(async move {
while let Ok((mut buf, start, end, route_key)) = buf_receiver.recv() {
handler
.handle(&mut buf, start, end, route_key, &context)
.await;
}
log::warn!("异步处理停止");
});
while let Ok((mut buf, start, end, route_key)) = buf_receiver.recv() {
handler
.handle(&mut buf, start, end, route_key, &context);
}
log::warn!("异步处理停止");
});
}
Some(buf_sender)
@@ -723,12 +712,8 @@ impl Channel {
let handler = handler.clone();
let buf_sender = buf_sender.clone();
std::thread::spawn(move || {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
log::info!("启动udp v6");
runtime.block_on(Self::main_start_(
Self::main_start_(
worker,
context,
UDP_V6_ID,
@@ -736,7 +721,7 @@ impl Channel {
handler,
buf_sender,
head_reserve,
));
)
});
}
{
@@ -746,12 +731,8 @@ impl Channel {
let handler = handler.clone();
let buf_sender = buf_sender.clone();
std::thread::spawn(move || {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
log::info!("启动udp v4");
runtime.block_on(Self::main_start_(
Self::main_start_(
worker,
context,
UDP_ID,
@@ -759,7 +740,7 @@ impl Channel {
handler,
buf_sender,
head_reserve,
));
)
});
}
if relay {
@@ -813,7 +794,7 @@ impl Channel {
}
worker.stop_all();
}
async fn main_start_(
fn main_start_(
worker: VntWorker,
context: Context,
id: usize,
@@ -841,8 +822,7 @@ impl Channel {
end,
RouteKey::new(id, addr),
&context,
)
.await;
);
}
Err(e) => {
log::error!("udp :{:?}", e);
@@ -851,7 +831,7 @@ impl Channel {
}
}
Some(mut buf_sender) => loop {
let mut buf = POOL.alloc(4096);
let mut buf = vec![0; 4096];
match udp.recv_from(&mut buf[head_reserve..]) {
Ok((len, addr)) => {
let end = head_reserve + len;
@@ -884,11 +864,11 @@ impl Channel {
#[cfg(target_os = "windows")]
use std::os::windows::io::AsRawSocket;
#[cfg(target_os = "windows")]
let id = 3 + udp.as_raw_socket() as usize;
let id = 3 + udp.as_raw_socket() as usize;
#[cfg(any(unix))]
use std::os::fd::AsRawFd;
#[cfg(any(unix))]
let id = 3 + udp.as_raw_fd() as usize;
let id = 3 + udp.as_raw_fd() as usize;
context.insert_udp(id, udp.clone());
match buf_sender {
@@ -899,7 +879,7 @@ impl Channel {
rs=udp.recv_from(&mut buf[head_reserve..])=>{
match rs {
Ok((len, addr)) => {
handler.handle(&mut buf, head_reserve, head_reserve + len, RouteKey::new(id, addr), &context).await;
handler.handle(&mut buf, head_reserve, head_reserve + len, RouteKey::new(id, addr), &context);
}
Err(e) => {
log::error!("{:?}",e)
@@ -933,7 +913,7 @@ impl Channel {
}
}
Some(mut buf_sender) => loop {
let mut buf = POOL.alloc(4096);
let mut buf = vec![0; 4096];
tokio::select! {
rs=udp.recv_from(&mut buf[head_reserve..])=>{
match rs {
vnt/src/handle/recv_handler.rs
+39 -43
View File
@@ -98,7 +98,7 @@ impl ChannelDataHandler {
}
impl ChannelDataHandler {
pub async fn handle(
pub fn handle(
&self,
buf: &mut [u8],
start: usize,
@@ -107,14 +107,14 @@ impl ChannelDataHandler {
context: &Context,
) {
assert_eq!(start, 14);
match self.handle0(&mut buf[..end], &route_key, context).await {
match self.handle0(&mut buf[..end], &route_key, context) {
Ok(_) => {}
Err(e) => {
log::warn!("{:?}", e);
}
}
}
async fn handle0(
fn handle0(
&self,
buf: &mut [u8],
route_key: &RouteKey,
@@ -173,8 +173,7 @@ impl ChannelDataHandler {
//服务端解密
self.server_cipher.decrypt_ipv4(&mut net_packet)?;
let data_len = net_packet.data_len();
self.server_packet_handle(context, current_device, buf, data_len, route_key)
.await?;
self.server_packet_handle(context, current_device, buf, data_len, route_key)?;
}
return Ok(());
}
@@ -313,12 +312,10 @@ impl ChannelDataHandler {
Protocol::Service => {}
Protocol::Error => {}
Protocol::Control => {
self.control(context, current_device, source, net_packet, route_key)
.await?;
self.control(context, current_device, source, net_packet, route_key)?;
}
Protocol::OtherTurn => {
self.other_turn(context, current_device, source, net_packet, route_key)
.await?;
self.other_turn(context, current_device, source, net_packet, route_key)?;
}
Protocol::UnKnow(e) => {
log::info!("不支持的协议:{}", e);
@@ -327,7 +324,7 @@ impl ChannelDataHandler {
Ok(())
}
async fn pong_packet(
fn pong_packet(
&self,
gateway: bool,
metric: u8,
@@ -361,7 +358,7 @@ impl ChannelDataHandler {
}
Ok(())
}
async fn control(
fn control(
&self,
context: &Context,
current_device: CurrentDeviceInfo,
@@ -390,8 +387,7 @@ impl ChannelDataHandler {
source,
pong_packet,
route_key,
)
.await?;
)?;
}
ControlPacket::PunchRequest => {
if self.relay {
@@ -437,7 +433,7 @@ impl ChannelDataHandler {
}
Ok(())
}
async fn other_turn(
fn other_turn(
&self,
context: &Context,
current_device: CurrentDeviceInfo,
@@ -518,12 +514,12 @@ impl ChannelDataHandler {
// let _ = context.try_send_main_udp(packet.buffer(),
// SocketAddr::V4(SocketAddrV4::new(peer_nat_info.local_ip, peer_nat_info.local_port)));
// }
if self.punch(source, peer_nat_info).await {
if self.punch(source, peer_nat_info) {
self.client_cipher.encrypt_ipv4(&mut punch_packet)?;
context.try_send_by_key(punch_packet.buffer(), route_key)?;
}
} else {
self.punch(source, peer_nat_info).await;
self.punch(source, peer_nat_info);
}
}
other_turn_packet::Protocol::Unknown(e) => {
@@ -532,7 +528,7 @@ impl ChannelDataHandler {
}
Ok(())
}
async fn punch(&self, peer_ip: Ipv4Addr, peer_nat_info: NatInfo) -> bool {
fn punch(&self, peer_ip: Ipv4Addr, peer_nat_info: NatInfo) -> bool {
match peer_nat_info.nat_type {
NatType::Symmetric => self
.symmetric_sender
@@ -545,7 +541,7 @@ impl ChannelDataHandler {
/// 处理服务端数据
impl ChannelDataHandler {
async fn server_packet_handle(
fn server_packet_handle(
&self,
context: &Context,
current_device: CurrentDeviceInfo,
@@ -557,16 +553,13 @@ impl ChannelDataHandler {
let source = net_packet.source();
match net_packet.protocol() {
Protocol::Service => {
self.service(context, current_device, net_packet, route_key)
.await?;
self.service(context, current_device, net_packet, route_key)?;
}
Protocol::Error => {
self.error(context, current_device, source, net_packet, route_key)
.await?;
self.error(context, current_device, source, net_packet, route_key)?;
}
Protocol::Control => {
self.control_gateway(context, current_device, net_packet, route_key)
.await?;
self.control_gateway(context, current_device, net_packet, route_key)?;
}
Protocol::IpTurn => {
match ip_turn_packet::Protocol::from(net_packet.transport_protocol()) {
@@ -600,7 +593,7 @@ impl ChannelDataHandler {
}
return Ok(());
}
async fn control_gateway(
fn control_gateway(
&self,
context: &Context,
current_device: CurrentDeviceInfo,
@@ -618,8 +611,7 @@ impl ChannelDataHandler {
net_packet.source(),
pong_packet,
route_key,
)
.await?;
)?;
}
ControlPacket::AddrResponse(addr_packet) => self
.nat_test
@@ -628,7 +620,7 @@ impl ChannelDataHandler {
}
Ok(())
}
async fn service(
fn service(
&self,
context: &Context,
current_device: CurrentDeviceInfo,
@@ -643,20 +635,24 @@ impl ChannelDataHandler {
{
let context = context.clone();
let nat_test = self.nat_test.clone();
tokio::spawn(async move {
let local_port = context.main_local_ipv4_port().unwrap_or(0);
let local_ipv4_addr = nat::local_ipv4_addr(local_port);
let local_port = context.main_local_ipv6_port().unwrap_or(0);
let ipv6_addr = nat::local_ipv6_addr(local_port);
let nat_info = nat_test
.re_test(
Ipv4Addr::from(response.public_ip),
response.public_port as u16,
local_ipv4_addr,
ipv6_addr,
)
.await;
context.switch(nat_info.nat_type);
std::thread::spawn(move ||{
tokio::runtime::Builder::new_current_thread()
.enable_all().build().unwrap()
.block_on(async move {
let local_port = context.main_local_ipv4_port().unwrap_or(0);
let local_ipv4_addr = nat::local_ipv4_addr(local_port);
let local_port = context.main_local_ipv6_port().unwrap_or(0);
let ipv6_addr = nat::local_ipv6_addr(local_port);
let nat_info = nat_test
.re_test(
Ipv4Addr::from(response.public_ip),
response.public_port as u16,
local_ipv4_addr,
ipv6_addr,
)
.await;
context.switch(nat_info.nat_type);
})
});
}
let new_ip = Ipv4Addr::from(response.virtual_ip);
@@ -728,7 +724,7 @@ impl ChannelDataHandler {
}
Ok(())
}
async fn error(
fn error(
&self,
_context: &Context,
current_device: CurrentDeviceInfo,
vnt/src/handle/tun_tap/channel_group.rs
+4 -6
View File
@@ -1,15 +1,13 @@
use byte_pool::Block;
#[derive(Clone)]
pub struct BufSenderGroup(
usize,
Vec<std::sync::mpsc::SyncSender<(Block<'static>, usize, usize)>>,
Vec<std::sync::mpsc::SyncSender<(Vec<u8>, usize, usize)>>,
);
pub struct BufReceiverGroup(pub Vec<std::sync::mpsc::Receiver<(Block<'static>, usize, usize)>>);
pub struct BufReceiverGroup(pub Vec<std::sync::mpsc::Receiver<(Vec<u8>, usize, usize)>>);
impl BufSenderGroup {
pub fn send(&mut self, val: (Block<'static>, usize, usize)) -> bool {
pub fn send(&mut self, val: (Vec<u8>, usize, usize)) -> bool {
let index = self.0 % self.1.len();
self.0 = self.0.wrapping_add(1);
self.1[index].send(val).is_ok()
@@ -21,7 +19,7 @@ pub fn buf_channel_group(size: usize) -> (BufSenderGroup, BufReceiverGroup) {
let mut buf_receiver_group = Vec::with_capacity(size);
for _ in 0..size {
let (buf_sender, buf_receiver) =
std::sync::mpsc::sync_channel::<(Block<'static>, usize, usize)>(1);
std::sync::mpsc::sync_channel::<(Vec<u8>, usize, usize)>(1);
buf_sender_group.push(buf_sender);
buf_receiver_group.push(buf_receiver);
}
vnt/src/handle/tun_tap/tap_handler.rs
+1 -6
View File
@@ -1,9 +1,7 @@
use byte_pool::BytePool;
use std::sync::Arc;
use std::{io, thread};
use crossbeam_utils::atomic::AtomicCell;
use lazy_static::lazy_static;
use packet::arp::arp::ArpPacket;
use packet::ethernet;
@@ -22,9 +20,6 @@ use crate::handle::CurrentDeviceInfo;
use crate::igmp_server::IgmpServer;
use crate::ip_proxy::IpProxyMap;
use crate::tun_tap_device::{DeviceReader, DeviceWriter};
lazy_static! {
static ref POOL: BytePool<Vec<u8>> = BytePool::<Vec<u8>>::new();
}
pub fn start(
worker: VntWorker,
@@ -116,7 +111,7 @@ fn start_(
mut buf_sender: BufSenderGroup,
) -> io::Result<()> {
loop {
let mut buf = POOL.alloc(4096);
let mut buf = vec![0; 4096];
if sender.is_close() {
return Ok(());
}
vnt/src/handle/tun_tap/tun_handler.rs
+1 -5
View File
@@ -1,4 +1,3 @@
use byte_pool::BytePool;
use std::sync::Arc;
use std::{io, thread};
@@ -19,9 +18,6 @@ use crate::handle::CurrentDeviceInfo;
use crate::igmp_server::IgmpServer;
use crate::ip_proxy::IpProxyMap;
use crate::tun_tap_device::{DeviceReader, DeviceWriter};
lazy_static::lazy_static! {
static ref POOL:BytePool<Vec<u8>> = BytePool::<Vec<u8>>::new();
}
fn icmp(device_writer: &DeviceWriter, mut ipv4_packet: IpV4Packet<&mut [u8]>) -> Result<()> {
if ipv4_packet.protocol() == ipv4::protocol::Protocol::Icmp {
let mut icmp = IcmpPacket::new(ipv4_packet.payload_mut())?;
@@ -169,7 +165,7 @@ fn start_(
mut buf_sender: BufSenderGroup,
) -> io::Result<()> {
loop {
let mut buf = POOL.alloc(4096);
let mut buf = vec![0; 4096];
buf[..12].fill(0);
if sender.is_close() {
return Ok(());
vnt/src/ip_proxy/tcp_proxy.rs
+3 -3
View File
@@ -5,8 +5,8 @@ use std::sync::Arc;
use std::time::Duration;
use tokio::io::AsyncReadExt;
use tokio::io::AsyncWriteExt;
use tokio::net::{TcpListener, TcpStream};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::net::{TcpListener, TcpStream};
pub struct TcpProxy {
tcp_listener: TcpListener,
@@ -39,7 +39,7 @@ impl TcpProxy {
Duration::from_secs(5),
TcpStream::connect(dest_addr),
)
.await
.await
{
Ok(peer_tcp_stream) => match peer_tcp_stream {
Ok(peer_tcp_stream) => peer_tcp_stream,
@@ -86,7 +86,7 @@ async fn proxy(client: TcpStream, server: TcpStream) -> io::Result<()> {
let (server_read, server_write) = server.into_split();
tokio::spawn(async move {
if let Err(e) = copy(client_read, server_write).await {
log::warn!("{:?}",e);
log::warn!("{:?}", e);
}
});
copy(server_read, client_write).await