refactor(dataplane): shutdown signaling + kernel driver scoped threads

Replace the last legacy shutdown signaling (ctrlc handler + mpsc<i32>
exit code + dedicated controller thread) with the single
`lifecycle::Shutdown` path, and migrate the kernel driver to scoped
threads with cancellation observation. After this commit there is one
signaling path: SIGINT/SIGTERM -> shutdown.root, or any subsystem's
report_fatal -> shutdown.root, with the watchdog as the absolute
upper bound.

- `main`: install `spawn_signal_handler` and `spawn_shutdown_watchdog`,
  run everything inside `concurrency::thread::scope`, block on
  `root.cancelled()`, then drain subsystems in canonical order
  (workers -> router -> metrics -> mgmt). Exit code from `is_fatal()`.
- `DriverKernel::start`: takes `&Scope` and `&Subsystem`; workers spawn
  via `spawn_scoped` with an `ExitGuard` Drop pattern that reports
  fatal on panic-unwind, early `?`-return, and unexpected normal exit.
  Reader loops observe cancel between reads. Supervisor joins-and-logs.
- Drop `dataplane/src/drivers/tokio_util.rs` and its
  `run_in_local_tokio_runtime` helper (inlined where needed).
- Drop `ctrlc` and `mio` from `dataplane` dependencies; drop `ctrlc`
  from the workspace.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Daniel Noland <daniel@githedgehog.com>

Author: daniel-noland

Cargo.lock

@@ -116,7 +116,6 @@ clap = { version = "4.6.1", default-features = true, features = [] }
 color-eyre = { version = "0.6.5", default-features = false, features = [] }
 colored = { version = "3.1.1", default-features = false, features = [] }
 crossbeam-utils = { version = "0.8.21", default-features = false, features = [] }
-ctrlc = { version = "3.5.2", default-features = false, features = [] }
 dashmap = { version = "6.2.1", default-features = false, features = [] }
 derive_builder = { version = "0.20.2", default-features = false, features = [] }
 dotenvy = { version = "0.15.7", default-features = false, features = [] }

Cargo.toml

@@ -19,7 +19,6 @@ axum = { workspace = true, features = ["http1", "tokio"] }
 axum-server = { workspace = true }
 concurrency = { workspace = true }
 config = { workspace = true }
-ctrlc = { workspace = true, features = ["termination"] }
 dyn-iter = { workspace = true }
 flow-entry = { workspace = true }
 flow-filter = { workspace = true }
@@ -32,7 +31,6 @@ linkme = { workspace = true }
 metrics = { workspace = true }
 metrics-exporter-prometheus = { workspace = true }
 mgmt = { workspace = true }
-mio = { workspace = true, features = ["os-ext", "net"] }
 nat = { workspace = true }
 net = { workspace = true, features = ["test_buffer"] }
 nix = { workspace = true, features = ["socket", "hostname"] }

dataplane/Cargo.toml

@@ -18,95 +18,97 @@ mod worker;
 
 use concurrency::sync::Arc;
 use concurrency::thread;
+#[allow(unused_imports)] // used under loom/shuttle backends
+use concurrency::thread::BuilderExt;
+use lifecycle::Subsystem;
 use net::buffer::test_buffer::TestBuffer;
 use pipeline::DynPipeline;
 use tracectl::trace_target;
 #[allow(unused)]
 use tracing::{debug, error, info, trace, warn};
 
 use super::DriverError;
-use super::tokio_util::run_in_local_tokio_runtime;
 use kif::{Kif, bring_kifs_up};
 use worker::Worker;
 
 trace_target!("kernel-driver", LevelFilter::INFO, &["driver"]);
 
-/// Main structure representing the kernel driver.
-/// This driver:
-///  * receives raw frames via `AF_PACKET`, parses to `Packet<TestBuffer>`
-///  * selects a worker by symmetric flow hash
-///  * workers run independent pipelines and send processed packets back
-///  * dispatcher serializes & transmits on the chosen outgoing interface
+/// AF_PACKET-based kernel driver. Spawns N workers with symmetric-hash
+/// fanout and per-worker pipelines.
 pub struct DriverKernel;
 
 #[allow(clippy::cast_possible_truncation)]
 impl DriverKernel {
-    /// Spawn `workers` processing threads, each with its own pipeline instance.
-    ///
-    /// Returns:
-    ///   - `Arc<Vec<Sender<Packet<TestBuffer>>>>` one sender per worker (dispatcher -> worker)
-    ///   - `Receiver<Packet<TestBuffer>>` a single queue for processed packets (worker -> dispatcher)
-    fn spawn_workers(
+    /// Spawn `num_workers` worker threads into `scope`, each with its own
+    /// pipeline. Bails on the first spawn failure; workers that did spawn
+    /// drain via the scope join.
+    fn spawn_workers_scoped<'scope>(
+        scope: &'scope thread::Scope<'scope, '_>,
+        workers_subsystem: &Subsystem,
         num_workers: usize,
         setup_pipeline: &Arc<dyn Send + Sync + Fn() -> DynPipeline<TestBuffer>>,
         interfaces: &[Kif],
-    ) -> Vec<thread::JoinHandle<Result<(), std::io::Error>>> {
+    ) -> Result<Vec<thread::ScopedJoinHandle<'scope, Result<(), std::io::Error>>>, std::io::Error>
+    {
         info!("Spawning {num_workers} workers");
-        let mut workers = Vec::new();
-        for wid in 0..num_workers {
-            let builder = thread::Builder::new().name(format!("dp-worker-{wid}"));
-            let mut worker = Worker::new(wid, num_workers, setup_pipeline);
-            match worker.start(builder, interfaces) {
-                Ok(handle) => workers.push(handle),
-                Err(e) => {
-                    error!("Failed to start worker {wid}: {e}");
-                }
-            }
-        }
-        workers
+        (0..num_workers)
+            .map(|wid| {
+                let builder = thread::Builder::new().name(format!("dp-worker-{wid}"));
+                Worker::new(wid, num_workers, setup_pipeline, workers_subsystem.clone())
+                    .start(scope, builder, interfaces)
+            })
+            .collect()
     }
 
-    /// Starts the kernel driver, spawns worker threads, and runs the dispatcher loop.
-    ///
-    /// - `args`: kernel driver CLI parameters (e.g., `--interface` list)
-    /// - `workers`: number of worker threads / pipelines
-    /// - `setup_pipeline`: factory returning a **fresh** `DynPipeline<TestBuffer>` per worker
+    /// Spawn worker threads + supervisor into `scope`. The scope joins
+    /// all driver threads on closure return.
     ///
     /// # Errors
-    ///    Returns [`DriverError`] in case the driver fails to start successfully.
-    pub fn start(
-        stop_tx: std::sync::mpsc::Sender<i32>,
+    /// Returns [`DriverError`] on interface setup or thread spawn failure.
+    pub fn start<'scope>(
+        scope: &'scope thread::Scope<'scope, '_>,
+        workers_subsystem: &Subsystem,
         args: impl IntoIterator<Item = impl AsRef<str> + Clone>,
         num_workers: usize,
         setup_pipeline: &Arc<dyn Send + Sync + Fn() -> DynPipeline<TestBuffer>>,
     ) -> Result<(), DriverError> {
+        // A current_thread runtime built inside another tokio runtime
+        // panics; catch nesting in debug.
+        debug_assert!(
+            tokio::runtime::Handle::try_current().is_err(),
+            "DriverKernel::start must not be invoked from within a tokio runtime context"
+        );
+
         info!("Collecting interfaces from config");
         let interfaces = kif::get_interfaces(args)?;
 
-        // ensure that the kernel interfaces for rx/tx are up
-        run_in_local_tokio_runtime(async || bring_kifs_up(interfaces.as_slice()).await)?;
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()?
+            .block_on(bring_kifs_up(interfaces.as_slice()))?;
 
-        // Spawn workers
-        let worker_handles =
-            Self::spawn_workers(num_workers, setup_pipeline, interfaces.as_slice());
+        let worker_handles = Self::spawn_workers_scoped(
+            scope,
+            workers_subsystem,
+            num_workers,
+            setup_pipeline,
+            interfaces.as_slice(),
+        )?;
 
-        let control_builder = thread::Builder::new().name("kernel-driver-controller".to_string());
-        control_builder.spawn(move || {
+        // The supervisor just joins-and-logs; worker fatal reporting is
+        // handled by the `ExitGuard` inside each worker thread.
+        let supervisor_builder =
+            thread::Builder::new().name("kernel-driver-supervisor".to_string());
+        supervisor_builder.spawn_scoped(scope, move || {
             for (id, handle) in worker_handles.into_iter().enumerate() {
-                info!("Waiting for workers to finish");
+                info!("Waiting for worker {id} to finish");
                 match handle.join() {
-                    Ok(result) => match result {
-                        Ok(()) => info!("Worker {id} exited successfully"),
-                        Err(e) => error!("Worker {id} exited with error: {e}"),
-                    },
-                    Err(e) => error!("Unable to spawn worker {id} error: {e:?}"),
+                    Ok(Ok(())) => info!("Worker {id} exited successfully"),
+                    Ok(Err(e)) => error!("Worker {id} exited with error: {e}"),
+                    Err(panic_payload) => error!("Worker {id} panicked: {panic_payload:?}"),
                 }
             }
-
-            // Exiting with error as it's not expected for all workers to finish
-            error!("All workers finished unexpectedly");
-            #[allow(clippy::expect_used)]
-            stop_tx.send(1).expect("Failed to send stop signal");
+            info!("All workers joined");
         })?;
 
         Ok(())

dataplane/src/drivers/kernel/mod.rs

@@ -15,14 +15,16 @@ use tokio::sync::Mutex;
 
 use concurrency::sync::Arc;
 use concurrency::thread;
+#[allow(unused_imports)] // used under loom/shuttle backends
+use concurrency::thread::BuilderExt;
+use lifecycle::Subsystem;
 use net::buffer::test_buffer::TestBuffer;
 use net::interface::InterfaceIndex;
 use net::packet::{DoneReason, Packet};
 use pipeline::{DynPipeline, NetworkFunction};
 
 use crate::drivers::kernel::fanout::{PacketFanoutType, set_packet_fanout};
 use crate::drivers::kernel::kif::Kif;
-use crate::drivers::tokio_util::run_in_local_tokio_runtime;
 
 use tracing::{debug, error, info, trace, warn};
 
@@ -126,35 +128,77 @@ pub struct Worker {
     id: WorkerId,
     total_workers: usize,
     setup_pipeline: Arc<dyn Send + Sync + Fn() -> DynPipeline<TestBuffer>>,
+    subsystem: Subsystem,
 }
 
 impl Worker {
     pub fn new(
         id: WorkerId,
         total_workers: usize,
         setup_pipeline: &Arc<dyn Send + Sync + Fn() -> DynPipeline<TestBuffer>>,
+        subsystem: Subsystem,
     ) -> Self {
         Worker {
             id,
             total_workers,
             setup_pipeline: setup_pipeline.clone(),
+            subsystem,
         }
     }
 
-    pub fn start(
+    #[allow(clippy::too_many_lines)]
+    pub fn start<'scope>(
         &mut self,
+        scope: &'scope thread::Scope<'scope, '_>,
         thread_builder: thread::Builder,
         interfaces: &[Kif],
-    ) -> Result<thread::JoinHandle<Result<(), io::Error>>, io::Error> {
+    ) -> Result<thread::ScopedJoinHandle<'scope, Result<(), io::Error>>, io::Error> {
         let id = self.id;
         let total_workers = self.total_workers;
         let setup = self.setup_pipeline.clone();
-        let interfaces = interfaces.iter().map(Kif::clone).collect::<Vec<_>>();
+        let subsystem = self.subsystem.clone();
+        let cancel = subsystem.cancel_token();
+        let interfaces = interfaces.to_vec();
+
+        let handle_res = thread_builder.spawn_scoped(scope, move || {
+            // Drop-guard so panic-unwind, early-`?`, and unexpected normal
+            // return all reach report_fatal. Disarmed on the graceful path.
+            struct ExitGuard {
+                subsystem: Subsystem,
+                id: WorkerId,
+                armed: bool,
+            }
+            impl ExitGuard {
+                fn disarm(&mut self) {
+                    self.armed = false;
+                }
+            }
+            impl Drop for ExitGuard {
+                fn drop(&mut self) {
+                    if !self.armed || self.subsystem.is_cancelled() {
+                        return;
+                    }
+                    let reason = if std::thread::panicking() {
+                        format!("worker {} panicked", self.id)
+                    } else {
+                        format!("worker {} exited unexpectedly", self.id)
+                    };
+                    self.subsystem.report_fatal(&reason);
+                }
+            }
 
-        let handle_res = thread_builder.spawn(move || {
             info!(worker = id, "Worker started");
+            let mut guard = ExitGuard {
+                subsystem: subsystem.clone(),
+                id,
+                armed: true,
+            };
+
+            let rt = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build_local(tokio::runtime::LocalOptions::default())?;
 
-            run_in_local_tokio_runtime(async || {
+            let result = rt.block_on(async {
                 let (readers, if_table) =
                     match build_interface_table(id, total_workers, interfaces.as_slice()) {
                         Ok(table) => table,
@@ -166,27 +210,39 @@ impl Worker {
 
                 let setup = setup.clone();
                 let if_table = if_table.clone();
+                let cancel = cancel.clone();
 
                 let mut reader_handles = tokio::task::JoinSet::new();
 
                 for intf in readers {
                     let setup = setup.clone();
                     let if_table = if_table.clone();
+                    let cancel = cancel.clone();
                     reader_handles.spawn_local(async move {
                         let intf = intf;
                         let mut pipeline = setup();
                         loop {
-                            tracing::debug!(worker = id, "awaiting packets");
+                            debug!(worker = id, "awaiting packets");
 
-                            let packets_vec = match read_packets_from_interface(id, &intf).await {
-                                Ok(packets) => packets,
-                                Err(e) => {
-                                    error!(
+                            let packets_vec = tokio::select! {
+                                () = cancel.cancelled() => {
+                                    info!(
                                         worker = id,
                                         rx_intf_name = intf.if_name,
-                                        "Error reading packets from interface: {e}"
+                                        "cancellation observed; exiting reader"
                                     );
-                                    vec![]
+                                    break;
+                                }
+                                result = read_packets_from_interface(id, &intf) => match result {
+                                    Ok(packets) => packets,
+                                    Err(e) => {
+                                        error!(
+                                            worker = id,
+                                            rx_intf_name = intf.if_name,
+                                            "Error reading packets from interface: {e}"
+                                        );
+                                        vec![]
+                                    }
                                 }
                             };
 
@@ -198,7 +254,6 @@ impl Worker {
                                 intf.if_name
                             );
 
-                            // Try to receive everything else that is in the buffer
                             let packets = packets_vec.into_iter();
 
                             let mut count = 0;
@@ -238,8 +293,13 @@ impl Worker {
                 }
 
                 Ok::<(), io::Error>(())
-            })?;
-            info!(worker = id, "Worker exited");
+            });
+
+            if subsystem.is_cancelled() {
+                guard.disarm();
+            }
+            info!(worker = id, "worker exited");
+            result?;
             Ok::<(), io::Error>(())
         })?;
         Ok(handle_res)

dataplane/src/drivers/kernel/worker.rs

@@ -4,7 +4,6 @@
 use thiserror::Error;
 
 pub mod kernel;
-mod tokio_util;
 
 #[derive(Error, Debug)]
 pub enum DriverError {