wireshift-fallback 0.1.1

#![warn(clippy::pedantic)]
#![cfg_attr(
    not(test),
    deny(
        clippy::unwrap_used,
        clippy::expect_used,
        clippy::todo,
        clippy::unimplemented,
        clippy::panic
    )
)]
#![allow(
    clippy::module_name_repetitions,
    clippy::must_use_candidate,
    clippy::missing_errors_doc,
)]
#![deny(unsafe_op_in_unsafe_fn)]
//! Fallback worker backend for `wireshift`.
//!
//! This backend preserves the same typed request and completion contracts as
//! the native `io_uring` path while executing unsupported or unavailable
//! operations on a blocking worker pool.

#![warn(missing_docs)]
#![allow(
    clippy::needless_pass_by_value,
    clippy::unnecessary_wraps,
    clippy::fn_params_excessive_bools
)]

/// Network operation handlers (connect, bind, accept, etc.).
pub mod net_ops;
/// File and buffer I/O operation handlers (read, write, fsync, etc.).
pub mod ops;
/// Worker thread pool for executing operations off the main thread.
pub mod pool;

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;

use crossbeam_channel::{bounded, Sender, TrySendError};
use wireshift_core::backend::{
    disconnected_error, Backend, BackendCompletion, BackendKind, BackendSubmission,
    CancellationHandle,
};
use wireshift_core::{Error, Result, RingConfig};

use crate::pool::{worker_loop, CancelRegistry, WorkerMessage};

/// Blocking worker backend.
#[derive(Debug)]
pub struct FallbackBackend {
    sender: Sender<WorkerMessage>,
    cancel_registry: Arc<CancelRegistry>,
    workers: Mutex<Vec<thread::JoinHandle<()>>>,
    shutting_down: AtomicBool,
}

impl FallbackBackend {
    /// Creates a fallback backend with the configured worker count.
    ///
    /// # Errors
    ///
    /// Returns an error if the completion channel sender cannot be cloned
    /// or worker thread spawning fails.
    pub fn new(
        config: &RingConfig,
        completion_tx: std::sync::mpsc::Sender<BackendCompletion>,
    ) -> Result<Self> {
        let (sender, receiver) = bounded(config.queue_depth as usize);
        let cancel_registry = Arc::new(CancelRegistry::new());
        let mut workers = Vec::with_capacity(config.fallback_threads);
        for _ in 0..config.fallback_threads {
            let rx = receiver.clone();
            let completion_tx = completion_tx.clone();
            let cancel_registry = cancel_registry.clone();
            workers.push(thread::spawn(move || {
                worker_loop(rx, completion_tx, cancel_registry);
            }));
        }
        Ok(Self {
            sender,
            cancel_registry,
            workers: Mutex::new(workers),
            shutting_down: AtomicBool::new(false),
        })
    }
}

impl Backend for FallbackBackend {
    fn kind(&self) -> BackendKind {
        BackendKind::Fallback
    }

    fn submit(&self, submission: BackendSubmission) -> Result<()> {
        if self.shutting_down.load(Ordering::Acquire) {
            return Err(Error::completion(
                "fallback backend is shutting down",
                "do not submit new requests while shutdown is in progress",
            ));
        }
        let _ = self.cancel_registry.insert(submission.id)?;
        match self.sender.try_send(WorkerMessage::Submit(submission)) {
            Ok(()) => Ok(()),
            Err(TrySendError::Full(WorkerMessage::Submit(submission))) => {
                let _ = self.cancel_registry.remove(submission.id);
                Err(Error::submission(
                    "fallback work queue is full",
                    "drain completions or increase queue_depth before submitting more work",
                ))
            }
            Err(TrySendError::Disconnected(WorkerMessage::Submit(submission))) => {
                let _ = self.cancel_registry.remove(submission.id);
                Err(disconnected_error())
            }
            Err(
                TrySendError::Full(WorkerMessage::Shutdown)
                | TrySendError::Disconnected(WorkerMessage::Shutdown),
            ) => Err(disconnected_error()),
        }
    }

    fn cancel(&self, cancellation: CancellationHandle) -> Result<()> {
        if self.cancel_registry.cancel(cancellation.target)? {
            Ok(())
        } else {
            Err(Error::canceled(
                format!("request {} is no longer cancelable", cancellation.target),
                "cancel the request before it completes or starts running",
            ))
        }
    }

    fn shutdown(&self) -> Result<()> {
        self.shutting_down.store(true, Ordering::Release);
        let mut workers = self.workers.lock().map_err(|_| {
            Error::completion(
                "fallback worker set mutex was poisoned",
                "avoid panicking while the ring is being dropped",
            )
        })?;
        let worker_count = workers.len();
        for _ in 0..worker_count {
            self.sender
                .send(WorkerMessage::Shutdown)
                .map_err(|_| disconnected_error())?;
        }
        for worker in workers.drain(..) {
            worker.join().map_err(|_| {
                Error::completion(
                    "fallback worker thread panicked during shutdown",
                    "fix the backend worker panic before dropping the ring",
                )
            })?;
        }
        Ok(())
    }
}