wireshift_fallback/
lib.rs

1#![warn(clippy::pedantic)]
2#![cfg_attr(
3    not(test),
4    deny(
5        clippy::unwrap_used,
6        clippy::expect_used,
7        clippy::todo,
8        clippy::unimplemented,
9        clippy::panic
10    )
11)]
12#![allow(
13    clippy::module_name_repetitions,
14    clippy::must_use_candidate,
15    clippy::missing_errors_doc,
16)]
17#![deny(unsafe_op_in_unsafe_fn)]
18//! Fallback worker backend for `wireshift`.
19//!
20//! This backend preserves the same typed request and completion contracts as
21//! the native `io_uring` path while executing unsupported or unavailable
22//! operations on a blocking worker pool.
23
24#![warn(missing_docs)]
25#![allow(
26    clippy::needless_pass_by_value,
27    clippy::unnecessary_wraps,
28    clippy::fn_params_excessive_bools
29)]
30
31/// Network operation handlers (connect, bind, accept, etc.).
32pub mod net_ops;
33/// File and buffer I/O operation handlers (read, write, fsync, etc.).
34pub mod ops;
35/// Worker thread pool for executing operations off the main thread.
36pub mod pool;
37
38use std::sync::atomic::{AtomicBool, Ordering};
39use std::sync::{Arc, Mutex};
40use std::thread;
41
42use crossbeam_channel::{bounded, Sender, TrySendError};
43use wireshift_core::backend::{
44    disconnected_error, Backend, BackendCompletion, BackendKind, BackendSubmission,
45    CancellationHandle,
46};
47use wireshift_core::{Error, Result, RingConfig};
48
49use crate::pool::{worker_loop, CancelRegistry, WorkerMessage};
50
51/// Blocking worker backend.
52#[derive(Debug)]
53pub struct FallbackBackend {
54    sender: Sender<WorkerMessage>,
55    cancel_registry: Arc<CancelRegistry>,
56    workers: Mutex<Vec<thread::JoinHandle<()>>>,
57    shutting_down: AtomicBool,
58}
59
60impl FallbackBackend {
61    /// Creates a fallback backend with the configured worker count.
62    ///
63    /// # Errors
64    ///
65    /// Returns an error if the completion channel sender cannot be cloned
66    /// or worker thread spawning fails.
67    pub fn new(
68        config: &RingConfig,
69        completion_tx: std::sync::mpsc::Sender<BackendCompletion>,
70    ) -> Result<Self> {
71        let (sender, receiver) = bounded(config.queue_depth as usize);
72        let cancel_registry = Arc::new(CancelRegistry::new());
73        let mut workers = Vec::with_capacity(config.fallback_threads);
74        for _ in 0..config.fallback_threads {
75            let rx = receiver.clone();
76            let completion_tx = completion_tx.clone();
77            let cancel_registry = cancel_registry.clone();
78            workers.push(thread::spawn(move || {
79                worker_loop(rx, completion_tx, cancel_registry);
80            }));
81        }
82        Ok(Self {
83            sender,
84            cancel_registry,
85            workers: Mutex::new(workers),
86            shutting_down: AtomicBool::new(false),
87        })
88    }
89}
90
91impl Backend for FallbackBackend {
92    fn kind(&self) -> BackendKind {
93        BackendKind::Fallback
94    }
95
96    fn submit(&self, submission: BackendSubmission) -> Result<()> {
97        if self.shutting_down.load(Ordering::Acquire) {
98            return Err(Error::completion(
99                "fallback backend is shutting down",
100                "do not submit new requests while shutdown is in progress",
101            ));
102        }
103        let _ = self.cancel_registry.insert(submission.id)?;
104        match self.sender.try_send(WorkerMessage::Submit(submission)) {
105            Ok(()) => Ok(()),
106            Err(TrySendError::Full(WorkerMessage::Submit(submission))) => {
107                let _ = self.cancel_registry.remove(submission.id);
108                Err(Error::submission(
109                    "fallback work queue is full",
110                    "drain completions or increase queue_depth before submitting more work",
111                ))
112            }
113            Err(TrySendError::Disconnected(WorkerMessage::Submit(submission))) => {
114                let _ = self.cancel_registry.remove(submission.id);
115                Err(disconnected_error())
116            }
117            Err(
118                TrySendError::Full(WorkerMessage::Shutdown)
119                | TrySendError::Disconnected(WorkerMessage::Shutdown),
120            ) => Err(disconnected_error()),
121        }
122    }
123
124    fn cancel(&self, cancellation: CancellationHandle) -> Result<()> {
125        if self.cancel_registry.cancel(cancellation.target)? {
126            Ok(())
127        } else {
128            Err(Error::canceled(
129                format!("request {} is no longer cancelable", cancellation.target),
130                "cancel the request before it completes or starts running",
131            ))
132        }
133    }
134
135    fn shutdown(&self) -> Result<()> {
136        self.shutting_down.store(true, Ordering::Release);
137        let mut workers = self.workers.lock().map_err(|_| {
138            Error::completion(
139                "fallback worker set mutex was poisoned",
140                "avoid panicking while the ring is being dropped",
141            )
142        })?;
143        let worker_count = workers.len();
144        for _ in 0..worker_count {
145            self.sender
146                .send(WorkerMessage::Shutdown)
147                .map_err(|_| disconnected_error())?;
148        }
149        for worker in workers.drain(..) {
150            worker.join().map_err(|_| {
151                Error::completion(
152                    "fallback worker thread panicked during shutdown",
153                    "fix the backend worker panic before dropping the ring",
154                )
155            })?;
156        }
157        Ok(())
158    }
159}
wireshift_fallback/lib.rs

wireshift_fallback/
lib.rs
