Skip to content

vey-daemon: add offline wait time limit for ListenUdpRuntime #67

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zh-jq merged 2 commits into from
Jul 18, 2026
Merged

vey-daemon: add offline wait time limit for ListenUdpRuntime #67

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
eb286ff
vey-daemon: add offline wait time limit for ListenUdpRuntime
zh-jq Jul 18, 2026
649cee6
vey-daemon: simplify ebpf feature cfg guard
zh-jq Jul 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 9 additions & 9 deletions lib/vey-daemon/src/listen/quic.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,15 +8,15 @@ use std::net::{SocketAddr, UdpSocket};
use std::sync::Arc;
use std::time::Duration;

#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
use anyhow::anyhow;
use async_trait::async_trait;
use log::{info, warn};
use quinn::{Connection, Endpoint, Incoming};
use tokio::runtime::Handle;
use tokio::sync::broadcast;

#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
use vey_reuseport::quic::{QuicSocketSelectGuard, QuicSocketSelector};
use vey_socket::RawSocket;
use vey_std_ext::net::SocketAddrExt;
Expand All @@ -43,7 +43,7 @@ pub struct ListenQuicRuntime<S> {
server: S,
listen_config: UdpListenConfig,
listen_stats: Arc<ListenStats>,
#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
socket_selector: Option<QuicSocketSelector>,
}

Expand All @@ -56,7 +56,7 @@ where
server,
listen_config,
listen_stats,
#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
socket_selector: None,
}
}
Expand All @@ -77,7 +77,7 @@ where
}
}

#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
if self
.listen_config
.use_ebpf(rustix::process::getuid().as_raw())
Expand Down Expand Up @@ -107,7 +107,7 @@ where

for i in 0..instance_count {
let socket = vey_socket::udp::new_std_bind_listen(&self.listen_config)?;
#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
let guard = if let Some(selector) = &mut self.socket_selector {
let guard = selector.add_socket(RawSocket::from(&socket))?;
Some(guard)
Expand All @@ -128,7 +128,7 @@ where
instance_id: i,
ingress_net_filter: ingress_net_filter.cloned(),
accept_timeout,
#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
_bpf_guard: guard,
_alive_guard: None,
};
Expand All @@ -139,7 +139,7 @@ where
);
}

#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
if let Some(mut selector) = self.socket_selector.take() {
if let Err(e) = selector.load_and_attach() {
if self.listen_config.fail_on_ebpf_error() {
Expand Down Expand Up @@ -180,7 +180,7 @@ pub struct ListenQuicRuntimeInstance<S> {
instance_id: usize,
ingress_net_filter: Option<Arc<AclNetworkRule>>,
accept_timeout: Duration,
#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
_bpf_guard: Option<QuicSocketSelectGuard>,
_alive_guard: Option<ListenAliveGuard>,
}
Expand Down
14 changes: 7 additions & 7 deletions lib/vey-daemon/src/listen/tcp.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
use std::net::SocketAddr;
use std::sync::Arc;

#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
use anyhow::anyhow;
use async_trait::async_trait;
use log::{info, warn};
Expand All @@ -17,7 +17,7 @@ use tokio::sync::broadcast;

use vey_compat::CpuAffinity;
use vey_io_ext::LimitedTcpListener;
#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
use vey_reuseport::tcp::TcpSocketSelector;
use vey_socket::RawSocket;
use vey_std_ext::net::SocketAddrExt;
Expand All @@ -34,7 +34,7 @@ pub trait AcceptTcpServer: BaseServer {
pub struct ListenTcpRuntime<S> {
server: S,
listen_stats: Arc<ListenStats>,
#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
socket_selector: Option<TcpSocketSelector>,
}

Expand All @@ -46,7 +46,7 @@ where
ListenTcpRuntime {
server,
listen_stats,
#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
socket_selector: None,
}
}
Expand Down Expand Up @@ -82,7 +82,7 @@ where
}
}

#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
if listen_config.use_ebpf(rustix::process::getuid().as_raw()) {
match TcpSocketSelector::new(
rustix::process::getpid().as_raw_pid(),
Expand All @@ -109,7 +109,7 @@ where

for i in 0..instance_count {
let listener = vey_socket::tcp::new_std_listener(listen_config)?;
#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
if let Some(selector) = &mut self.socket_selector {
selector.add_socket(RawSocket::from(&listener));
}
Expand All @@ -122,7 +122,7 @@ where
);
}

#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
if let Some(mut selector) = self.socket_selector.take() {
if let Err(e) = selector.load_and_attach() {
if listen_config.fail_on_ebpf_error() {
Expand Down
65 changes: 53 additions & 12 deletions lib/vey-daemon/src/listen/udp/listen.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::task::{Context, Poll};

#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
use anyhow::anyhow;
use async_trait::async_trait;
use bytes::Bytes;
Expand All @@ -32,10 +32,12 @@ use smallvec::SmallVec;
use tokio::net::UdpSocket;
use tokio::runtime::Handle;
use tokio::sync::{broadcast, mpsc};
#[cfg(feature = "ebpf")]
use tokio::time::Instant;

use vey_io_ext::{UdpMoveRecv, UdpMoveSend, UdpSocketExt};
use vey_io_sys::udp::{RecvMsgHdr, SendMsgHdr};
#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
use vey_reuseport::udp::{UdpSocketSelectGuard, UdpSocketSelector};
use vey_socket::RawSocket;
use vey_types::acl::{AclAction, AclNetworkRule};
Expand Down Expand Up @@ -96,7 +98,10 @@ impl UdpMoveRecv for AcceptedUdpPacketReceiver {
fn poll_recv_packet(&mut self, cx: &mut Context<'_>) -> Poll<Result<Bytes, Self::RecvError>> {
match self.inner.poll_recv(cx) {
Poll::Ready(Some(packet)) => Poll::Ready(Ok(packet)),
Poll::Ready(None) => Poll::Ready(Ok(Bytes::new())),
Poll::Ready(None) => Poll::Ready(Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"packet receiver closed",
))),
Poll::Pending => Poll::Pending,
}
}
Expand Down Expand Up @@ -430,7 +435,7 @@ pub struct ListenUdpRuntime<S> {
conn_track: UdpConnectionTrackConfig,
packet_max_size: u16,
listen_stats: Arc<ListenStats>,
#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
socket_selector: Option<UdpSocketSelector>,
}

Expand All @@ -451,7 +456,7 @@ where
conn_track,
packet_max_size,
listen_stats,
#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
socket_selector: None,
}
}
Expand Down Expand Up @@ -528,7 +533,7 @@ where
listen_in_worker,
instance_id: id,
ingress_net_filter: None,
#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
_bpf_guard: None,
_alive_guard: None,

Expand All @@ -550,7 +555,7 @@ where
}
}

#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
if self
.listen_config
.use_ebpf(rustix::process::getuid().as_raw())
Expand Down Expand Up @@ -585,15 +590,15 @@ where

let mut runtime = self.create_instance(i, listen_addr, listen_in_worker);
runtime.ingress_net_filter = ingress_net_filter.cloned();
#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
if let Some(selector) = &mut self.socket_selector {
let guard = selector.add_socket(RawSocket::from(&socket));
runtime._bpf_guard = Some(guard);
}
runtime.into_running(socket, server_reload_sender.subscribe());
}

#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
if let Some(mut selector) = self.socket_selector.take() {
if let Err(e) = selector.load_and_attach() {
if self.listen_config.fail_on_ebpf_error() {
Expand Down Expand Up @@ -635,7 +640,7 @@ struct ListenUdpRuntimeInstance<S> {
listen_in_worker: bool,
instance_id: usize,
ingress_net_filter: Option<Arc<AclNetworkRule>>,
#[cfg(all(target_os = "linux", feature = "ebpf"))]
#[cfg(feature = "ebpf")]
_bpf_guard: Option<UdpSocketSelectGuard>,
_alive_guard: Option<ListenAliveGuard>,

Expand Down Expand Up @@ -816,6 +821,16 @@ where
mut rt_state: RuntimeState,
mut ct_table: LruCache<ClientConnectionKey, StreamDispatcher, FixedState>,
) {
let mut wait_sleep = Box::pin(tokio::time::sleep(self.conn_track.offline_wait_time()));
let mut allow_new = true;

info!(
"SRT[{}_v{}#{}] enters offline-wait mode",
self.server.name(),
self.server_version,
self.instance_id
);

loop {
tokio::select! {
biased;
Expand All @@ -825,14 +840,32 @@ where
self.handle_events(&mut event_recv_buf, &mut ct_table).await;
event_recv_buf.clear();
if ct_table.is_empty() {
break;
return;
}
}
r = self.recv_packets(&rt_state.socket) => {
match r {
Ok(packets) => {
for (cc_info, data) in packets {
self.handle_packet(cc_info, data, &rt_state, &mut ct_table);
if allow_new {
self.handle_packet(cc_info, data, &rt_state, &mut ct_table);
} else {
let key = cc_info.connection_key();
if let Some(dispatcher) = ct_table.get(&key) {
match dispatcher.sender.try_send(data) {
Ok(_) => {}
Err(mpsc::error::TrySendError::Full(_)) => {
dispatcher.state.add_recv_dropped();
}
Err(mpsc::error::TrySendError::Closed(_)) => {
dispatcher.state.add_recv_dropped();
ct_table.pop(&key);
}
}
} else {
self.listen_stats.add_dropped();
}
}
}
}
Err(e) => {
Expand All @@ -842,6 +875,14 @@ where
}
}
}
_ = &mut wait_sleep => {
if !allow_new {
break;
}
info!("SRT[{}_v{}#{}] enters offline-quit mode", self.server.name(), self.server_version, self.instance_id);
allow_new = false;
wait_sleep.as_mut().reset(Instant::now() + self.conn_track.offline_quit_time());
}
}
}
}
Expand Down
25 changes: 25 additions & 0 deletions lib/vey-types/src/net/udp/listen.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

use std::net::{IpAddr, Ipv6Addr, SocketAddr};
use std::num::{NonZeroU32, NonZeroUsize};
use std::time::Duration;

use anyhow::anyhow;
use num_traits::ToPrimitive;
Expand Down Expand Up @@ -260,6 +261,8 @@ pub struct UdpConnectionTrackConfig {
dispatch_queue_size: NonZeroUsize,
send_queue_size: NonZeroUsize,
batch_recv_size: NonZeroUsize,
offline_wait_time: Duration,
offline_quit_time: Duration,
}

impl Default for UdpConnectionTrackConfig {
Expand All @@ -270,6 +273,8 @@ impl Default for UdpConnectionTrackConfig {
dispatch_queue_size: unsafe { NonZeroUsize::new_unchecked(32) },
send_queue_size: unsafe { NonZeroUsize::new_unchecked(512) },
batch_recv_size: unsafe { NonZeroUsize::new_unchecked(16) },
offline_wait_time: Duration::from_secs(60),
offline_quit_time: Duration::from_hours(1),
}
}
}
Expand Down Expand Up @@ -324,4 +329,24 @@ impl UdpConnectionTrackConfig {
pub fn set_batch_recv_size(&mut self, batch_recv_size: NonZeroUsize) {
self.batch_recv_size = batch_recv_size;
}

#[inline]
pub fn offline_wait_time(&self) -> Duration {
self.offline_wait_time
}

#[inline]
pub fn set_offline_wait_time(&mut self, wait_time: Duration) {
self.offline_wait_time = wait_time;
}

#[inline]
pub fn offline_quit_time(&self) -> Duration {
self.offline_quit_time
}

#[inline]
pub fn set_offline_quit_time(&mut self, quit_time: Duration) {
self.offline_quit_time = quit_time;
}
}
12 changes: 12 additions & 0 deletions lib/vey-yaml/src/value/net/udp.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -223,6 +223,18 @@ pub fn as_udp_conn_track_config(value: &Yaml) -> anyhow::Result<UdpConnectionTra
config.set_batch_recv_size(batch_size);
Ok(())
}
"offline_wait_time" => {
let wait_time = crate::humanize::as_duration(v)
.context(format!("invalid humanize duration value for key {k}"))?;
config.set_offline_wait_time(wait_time);
Ok(())
}
"offline_quit_time" => {
let wait_time = crate::humanize::as_duration(v)
.context(format!("invalid humanize duration value for key {k}"))?;
config.set_offline_quit_time(wait_time);
Ok(())
}
_ => Err(anyhow!("invalid key {k}")),
})?;

Expand Down
Loading
Loading