refactor(io): Refactor network types into net::io modules

Author: XAMPPRocky

src/cli/service.rs

@@ -1,6 +1,6 @@
 use std::{future::Future, sync::Arc};
 
-use crate::{components::proxy::SessionPool, config::Config};
+use crate::config::Config;
 
 #[derive(Debug, clap::Parser)]
 #[command(next_help_heading = "Service Options")]
@@ -249,19 +249,23 @@ impl Service {
         let mut shutdown_rx = shutdown_rx.clone();
         let mds_task = self.publish_mds(config)?;
         let (phoenix_task, phoenix_finalizer) = self.publish_phoenix(config)?;
-        // We need to call this before qcmp since if we use XDP we handle QCMP
-        // internally without a separate task
-        let (udp_task, finalizer, session_pool) = self.publish_udp(config)?;
-        let qcmp_task = self.publish_qcmp(&shutdown_rx)?;
+        let (udp_task, finalizer, session_pool) = self.listen_udp(config, &shutdown_rx)?;
         let xds_task = self.publish_xds(config)?;
 
+        tracing::info!(services=?[
+            self.udp_enabled.then_some("udp"),
+            self.qcmp_enabled.then_some("qcmp"),
+            self.phoenix_enabled.then_some("phoenix"),
+            self.xds_enabled.then_some("xds"),
+            self.mds_enabled.then_some("mds"),
+        ].into_iter().flatten().collect::<Vec<&str>>(), "starting service listeners");
+
         Ok(tokio::spawn(async move {
             tokio::task::spawn(async move {
                 let (task, result) = tokio::select! {
                     result = mds_task => ("mds", result),
                     result = phoenix_task => ("phoenix", result),
-                    result = qcmp_task => ("qcmp", result),
-                    result = udp_task => ("udp", result),
+                    result = udp_task => ("udp/qcmp", result),
                     result = xds_task => ("xds", result),
                 };
 
@@ -341,20 +345,6 @@ impl Service {
         Ok((std::future::pending(), None))
     }
 
-    /// Spawns an QCMP server if enabled, otherwise returns a future which never completes.
-    fn publish_qcmp(
-        &self,
-        shutdown_rx: &crate::signal::ShutdownRx,
-    ) -> crate::Result<impl Future<Output = crate::Result<()>> + use<>> {
-        if self.qcmp_enabled {
-            tracing::info!(port=%self.qcmp_port, "starting qcmp service");
-            let qcmp = crate::net::raw_socket_with_reuse(self.qcmp_port)?;
-            crate::codec::qcmp::spawn(qcmp, shutdown_rx.clone())?;
-        }
-
-        Ok(std::future::pending())
-    }
-
     /// Spawns an xDS server if enabled, otherwise returns a future which never completes.
     fn publish_mds(
         &self,
@@ -409,9 +399,10 @@ impl Service {
     }
 
     #[allow(clippy::type_complexity)]
-    pub fn publish_udp(
+    pub fn listen_udp(
         &mut self,
         config: &Arc<Config>,
+        shutdown_rx: &crate::signal::ShutdownRx,
     ) -> eyre::Result<(
         impl Future<Output = crate::Result<()>> + use<>,
         Option<Finalizer>,
@@ -421,166 +412,20 @@ impl Service {
             return Ok((either::Left(std::future::pending()), None, None));
         }
 
-        tracing::info!(port=%self.udp_port, "starting udp service");
-
-        #[cfg(target_os = "linux")]
-        {
-            match self.spawn_xdp(config.clone(), self.xdp.force_xdp) {
-                Ok(xdp) => {
-                    if let Some(xdp) = xdp {
-                        self.qcmp_enabled = false;
-                        return Ok((either::Left(std::future::pending()), Some(xdp), None));
-                    } else if self.xdp.force_xdp {
-                        eyre::bail!("XDP was forced on, but failed to initialize");
-                    }
-                }
-                Err(err) => {
-                    if self.xdp.force_xdp {
-                        return Err(err);
-                    }
-
-                    tracing::warn!(
-                        ?err,
-                        "failed to spawn XDP I/O loop, falling back to io-uring"
-                    );
-                }
-            }
-        }
-
         if !self.udp_enabled {
             return Ok((either::Left(std::future::pending()), None, None));
         }
 
-        self.spawn_user_space_router(config.clone())
-            .map(|(fut, func, sp)| (either::Right(fut), Some(func), Some(sp)))
-    }
-
-    /// Launches the user space implementation of the packet router using
-    /// sockets. This implementation uses a pool of buffers and sockets to
-    /// manage UDP sessions and sockets. On Linux this will use io-uring, where
-    /// as it will use epoll interfaces on non-Linux platforms.
-    #[allow(clippy::type_complexity)]
-    pub fn spawn_user_space_router(
-        &self,
-        config: Arc<Config>,
-    ) -> crate::Result<(
-        impl Future<Output = crate::Result<()>> + use<>,
-        Finalizer,
-        Arc<crate::components::proxy::SessionPool>,
-    )> {
-        // If we're on linux, we're using io-uring, but we're probably running in a container
-        // and may not be allowed to call io-uring related syscalls due to seccomp
-        // profiles, so do a quick check here to validate that we can call io_uring_setup
-        // https://www.man7.org/linux/man-pages/man2/io_uring_setup.2.html
-        #[cfg(target_os = "linux")]
-        {
-            if let Err(err) = io_uring::IoUring::new(2) {
-                fn in_container() -> bool {
-                    let sched = match std::fs::read_to_string("/proc/1/sched") {
-                        Ok(s) => s,
-                        Err(error) => {
-                            tracing::warn!(
-                                %error,
-                                "unable to read /proc/1/sched to determine if quilkin is in a container"
-                            );
-                            return false;
-                        }
-                    };
-                    let Some(line) = sched.lines().next() else {
-                        tracing::warn!("/proc/1/sched was empty");
-                        return false;
-                    };
-                    let Some(proc) = line.split(' ').next() else {
-                        tracing::warn!("first line of /proc/1/sched was empty");
-                        return false;
-                    };
-                    proc != "init" && proc != "systemd"
-                }
-
-                if err.kind() == std::io::ErrorKind::PermissionDenied && in_container() {
-                    eyre::bail!(
-                        "failed to call `io_uring_setup` due to EPERM ({err}), quilkin seems to be running inside a container meaning this is likely due to the seccomp profile not allowing the syscall"
-                    );
-                } else {
-                    eyre::bail!("failed to call `io_uring_setup` due to {err}");
-                }
-            }
-        }
-
-        let socket = crate::net::raw_socket_with_reuse(self.udp_port)?;
-        let workers = self.udp_workers.get();
-        let buffer_pool = Arc::new(crate::collections::BufferPool::new(workers, 2 * 1024));
-
-        let mut worker_sends = Vec::with_capacity(workers);
-        let mut session_sends = Vec::with_capacity(workers);
-        for _ in 0..workers {
-            let queue = crate::net::queue(15)?;
-            session_sends.push(queue.0.clone());
-            worker_sends.push(queue);
-        }
-
-        let sessions = SessionPool::new(config.clone(), session_sends, buffer_pool.clone());
-
-        crate::components::proxy::packet_router::spawn_receivers(
+        let (fut, finaliser, sessions) = crate::net::io::listen(
             config,
-            socket,
-            worker_sends,
-            &sessions,
-            buffer_pool,
+            self.udp_enabled.then_some(self.udp_port),
+            self.qcmp_enabled.then_some(self.qcmp_port),
+            self.udp_workers.get(),
+            self.xdp.clone(),
+            shutdown_rx,
         )?;
 
-        Ok((
-            std::future::pending(),
-            Box::from(move |_shutdown_rx: &crate::signal::ShutdownRx| {}),
-            sessions,
-        ))
-    }
-
-    #[cfg(target_os = "linux")]
-    fn spawn_xdp(&self, config: Arc<Config>, force_xdp: bool) -> eyre::Result<Option<Finalizer>> {
-        use crate::net::xdp;
-        use eyre::{Context as _, ContextCompat as _};
-
-        // TODO: remove this once it's been more stabilized
-        if !force_xdp {
-            return Ok(None);
-        }
-
-        let filters = config
-            .dyn_cfg
-            .filters()
-            .context("XDP requires a filter chain")?
-            .clone();
-        let clusters = config
-            .dyn_cfg
-            .clusters()
-            .context("XDP requires a cluster map")?
-            .clone();
-
-        let config = crate::net::xdp::process::ConfigState { filters, clusters };
-
-        let udp_port = if self.udp_enabled { self.udp_port } else { 0 };
-        let qcmp_port = if self.qcmp_enabled { self.qcmp_port } else { 0 };
-
-        tracing::info!(udp_port, qcmp_port, "setting up xdp module");
-        let workers = xdp::setup_xdp_io(xdp::XdpConfig {
-            nic: self
-                .xdp
-                .network_interface
-                .as_deref()
-                .map_or(xdp::NicConfig::Default, xdp::NicConfig::Name),
-            external_port: udp_port,
-            qcmp_port,
-            maximum_packet_memory: self.xdp.maximum_memory,
-            require_zero_copy: self.xdp.force_zerocopy,
-            require_tx_checksum: self.xdp.force_tx_checksum_offload,
-        })
-        .context("failed to setup XDP")?;
-
-        let io_loop = xdp::spawn(workers, config).context("failed to spawn XDP I/O loop")?;
-        Ok(Some(Box::new(move |srx: &crate::signal::ShutdownRx| {
-            io_loop.shutdown(*srx.borrow() == crate::signal::ShutdownKind::Normal);
-        })))
+        Ok((either::Right(fut), finaliser, sessions))
     }
 }
 

src/components/proxy.rs

@@ -14,9 +14,7 @@
  *  limitations under the License.
  */
 
-pub(crate) mod error;
-pub mod packet_router;
-pub(crate) mod sessions;
+use crate::net::{error, udp::sessions};
 
 use super::RunArgs;
 pub use error::PipelineError;