vey-daemon: add offline wait time limit for ListenUdpRuntime

Author: zh-jq

lib/vey-daemon/src/listen/udp/listen.rs

@@ -32,6 +32,7 @@ use smallvec::SmallVec;
 use tokio::net::UdpSocket;
 use tokio::runtime::Handle;
 use tokio::sync::{broadcast, mpsc};
+use tokio::time::Instant;
 
 use vey_io_ext::{UdpMoveRecv, UdpMoveSend, UdpSocketExt};
 use vey_io_sys::udp::{RecvMsgHdr, SendMsgHdr};
@@ -96,7 +97,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,
         }
     }
@@ -816,6 +820,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;
@@ -825,14 +839,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) => {
@@ -842,6 +874,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());
+                }
             }
         }
     }

lib/vey-types/src/net/udp/listen.rs

@@ -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;
@@ -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 {
@@ -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),
         }
     }
 }
@@ -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;
+    }
 }

lib/vey-yaml/src/value/net/udp.rs

@@ -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}")),
     })?;
 

sphinx/vey-values/configuration/values/network.rst

@@ -771,6 +771,33 @@ It consists of the following fields:
 
   **default**: 16
 
+* offline_wait_time
+
+  **option**, **type**: :ref:`humanize duration <conf_value_humanize_duration>`
+
+  When bpf feature is enabled, the udp socket will be in offline-wait mode, and it will still accept new connections.
+  This config option set the total time of this offline-wait mode.
+
+  **default**: 60s
+
+  .. availability::
+
+    - ``vey-proxy``: available since ``1.13.9``
+
+* offline_quit_time
+
+  **option**, **type**: :ref:`humanize duration <conf_value_humanize_duration>`
+
+  When bpf feature is enabled, the udp socket will be in offline-quit mode after offline-wait mode end, new connections
+  are not allowed in this mode. This config option set the total time of this offline-quit mode. The socket will be
+  closed after offline-quit mode.
+
+  **default**: 1h
+
+  .. availability::
+
+    - ``vey-proxy``: available since ``1.13.9``
+
 .. availability::
 
   - ``vey-proxy``: available since ``1.13.3``