Implementation

use std::sync::mpsc::{channel, Receiver, Sender};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};

pub type TaskResult = Result<usize, String>;
pub type Job = Box<dyn FnOnce() -> TaskResult + Send + 'static>;

pub struct WorkerPool {
    workers: Vec<Worker>,
    sender: Option<Sender<Job>>,
}

struct Worker {
    id: usize,
    handle: Option<JoinHandle<()>>,
}

impl WorkerPool {
    pub fn new(size: usize) -> Self {
        assert!(size > 0, "Worker pool size must be greater than 0");
        let (sender, receiver) = channel::<Job>();
        let receiver = Arc::new(Mutex::new(receiver));
        let mut workers = Vec::with_capacity(size);

        for id in 0..size {
            let rx = Arc::clone(&receiver);
            let handle = thread::spawn(move || loop {
                let job = {
                    let lock = rx.lock().unwrap();
                    lock.recv()
                };
                match job {
                    Ok(task) => {
                        let _ = task();
                    }
                    Err(_) => break, // Channel disconnected, terminate thread
                }
            });
            workers.push(Worker {
                id,
                handle: Some(handle),
            });
        }

        WorkerPool {
            workers,
            sender: Some(sender),
        }
    }

    pub fn execute<F>(&self, f: F) -> Receiver<TaskResult>
    where
        F: FnOnce() -> TaskResult + Send + 'static,
    {
        let (res_tx, res_rx) = channel();
        let job = Box::new(move || {
            let res = f();
            let _ = res_tx.send(res.clone());
            res
        });
        if let Some(ref sender) = self.sender {
            sender.send(job).expect("Failed to send job to worker pool");
        }
        res_rx
    }
}

impl Drop for WorkerPool {
    fn drop(&mut self) {
        // Drop sender first so receivers get EOF signal
        drop(self.sender.take());
        for worker in &mut self.workers {
            if let Some(handle) = worker.handle.take() {
                let _ = handle.join();
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    #[test]
    fn test_worker_pool_task_execution() {
        let pool = WorkerPool::new(4);
        let counter = Arc::new(AtomicUsize::new(0));
        let mut receivers = Vec::new();

        for i in 0..10 {
            let counter_clone = Arc::clone(&counter);
            let rx = pool.execute(move || {
                counter_clone.fetch_add(1, Ordering::SeqCst);
                Ok(i * 2)
            });
            receivers.push((i, rx));
        }

        for (i, rx) in receivers {
            let res = rx.recv().expect("Failed to receive task output");
            assert_eq!(res, Ok(i * 2));
        }

        assert_eq!(counter.load(Ordering::SeqCst), 10);
    }

    #[test]
    fn test_worker_pool_graceful_shutdown() {
        let pool = WorkerPool::new(2);
        let rx = pool.execute(|| Ok(42));
        assert_eq!(rx.recv().unwrap(), Ok(42));
        // Dropping pool explicitly triggers worker thread joins
        drop(pool);
    }
}

Technical Explanation

1. Closure Trait Bounds (F: Send + 'static): thread::spawn requires closures to capture only thread-safe ownership (Send) and live for the 'static lifetime. Standard functions or closures capturing non-Send types like Rc or RefCell fail compilation.
2. Result Synchronization: Enqueueing a wrapper closure that calls f() and sends output over a separate mpsc::channel allows asynchronous task dispatch and synchronous response collection.
3. Teardown Mechanics: When WorkerPool drops, clearing the sender closes the channel. Worker loops receiving Err(_) exit cleanly, enabling handle.join() to clean up background OS threads.

Implementation

use std::alloc::{alloc, dealloc, Layout};
use std::ptr::NonNull;
use std::slice;
use std::thread;

pub struct OffHeapBuffer {
    ptr: NonNull<u8>,
    capacity: usize,
    layout: Layout,
}

// SAFETY: OffHeapBuffer exclusively owns its heap allocation.
// Transferring ownership across thread boundaries is sound because no alias references
// survive across moves, and move semantics guarantee that only one thread can access
// or mutate the buffer pointer at any point in time.
unsafe impl Send for OffHeapBuffer {}

impl OffHeapBuffer {
    pub fn new(capacity: usize) -> Result<Self, String> {
        if capacity == 0 {
            return Err("Buffer capacity must be strictly positive".to_string());
        }
        let layout = Layout::array::<u8>(capacity).map_err(|e| e.to_string())?;
        let raw_ptr = unsafe { alloc(layout) };
        let ptr = NonNull::new(raw_ptr).ok_or_else(|| "Heap allocation failed".to_string())?;

        Ok(OffHeapBuffer { ptr, capacity, layout })
    }

    pub fn capacity(&self) -> usize {
        self.capacity
    }

    pub fn write(&mut self, offset: usize, data: &[u8]) -> Result<(), String> {
        if offset + data.len() > self.capacity {
            return Err("Buffer overflow: write out of bounds".to_string());
        }
        unsafe {
            std::ptr::copy_nonoverlapping(data.as_ptr(), self.ptr.as_ptr().add(offset), data.len());
        }
        Ok(())
    }

    pub fn read(&self, offset: usize, len: usize) -> Result<Vec<u8>, String> {
        if offset + len > self.capacity {
            return Err("Buffer overflow: read out of bounds".to_string());
        }
        let data_slice = unsafe { slice::from_raw_parts(self.ptr.as_ptr().add(offset), len) };
        Ok(data_slice.to_vec())
    }
}

impl Drop for OffHeapBuffer {
    fn drop(&mut self) {
        unsafe {
            dealloc(self.ptr.as_ptr(), self.layout);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_off_heap_buffer_thread_transfer() {
        let mut buffer = OffHeapBuffer::new(512).expect("Allocation failed");
        buffer.write(0, b"Main Thread Payload").unwrap();

        // Move buffer into background worker thread
        let handle = thread::spawn(move || {
            let payload = buffer.read(0, 19).unwrap();
            assert_eq!(&payload, b"Main Thread Payload");

            buffer.write(19, b" -> Processed").unwrap();
            buffer
        });

        let mut returned_buffer = handle.join().expect("Worker thread panicked");
        let full_payload = returned_buffer.read(0, 32).unwrap();
        assert_eq!(&full_payload, b"Main Thread Payload -> Processed");
    }

    #[test]
    fn test_off_heap_buffer_bounds_checking() {
        let mut buffer = OffHeapBuffer::new(32).unwrap();
        assert!(buffer.write(20, &[0u8; 20]).is_err());
        assert!(buffer.read(25, 10).is_err());
    }
}

Explanation & Safety Invariants:

1. Why *mut T is !Send: Raw pointers do not encode ownership or aliasing semantics. Rust conservatively marks them !Send to prevent unchecked concurrent access across threads.
2. Soundness of unsafe impl Send: By wrapping the pointer inside OffHeapBuffer and exposing safe move-only semantics without shared references (&self mutates nothing without interior mutability), transferring ownership to another thread is 100% data-race free.
3. RAII Deallocation: The Drop implementation safely reclaims raw memory via std::alloc::dealloc, preventing heap leaks regardless of which thread owns the buffer when it falls out of scope.

Implementation

use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};

pub struct StateContainer<T> {
    pub payload: T,
    pub metadata: String,
}

impl<T> StateContainer<T> {
    pub fn new(payload: T, metadata: impl Into<String>) -> Self {
        Self {
            payload,
            metadata: metadata.into(),
        }
    }
}

pub fn dispatch_to_thread<T, F>(
    mut container: StateContainer<T>,
    task: F,
) -> JoinHandle<StateContainer<T>>
where
    T: Send + 'static,
    F: FnOnce(&mut T) + Send + 'static,
{
    thread::spawn(move || {
        task(&mut container.payload);
        container
    })
}

pub fn assert_send<T: Send>() {}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicBool, Ordering};

    #[test]
    fn test_auto_trait_send_propagation() {
        // Verify auto-trait Send implementations
        assert_send::<StateContainer<i32>>();
        assert_send::<StateContainer<String>>();
        assert_send::<StateContainer<Vec<u8>>>();
        assert_send::<StateContainer<Arc<Mutex<usize>>>>();
    }

    #[test]
    fn test_generic_state_dispatch_and_mutation() {
        let container = StateContainer::new(Arc::new(Mutex::new(100)), "TelemetryPipeline");
        let executed_flag = Arc::new(AtomicBool::new(false));

        let flag_clone = Arc::clone(&executed_flag);
        let handle = dispatch_to_thread(container, move |payload| {
            let mut val = payload.lock().expect("Failed to lock mutex");
            *val += 50;
            flag_clone.store(true, Ordering::SeqCst);
        });

        let processed_container = handle.join().expect("Worker thread panicked");
        assert_eq!(processed_container.metadata, "TelemetryPipeline");
        assert_eq!(*processed_container.payload.lock().unwrap(), 150);
        assert!(executed_flag.load(Ordering::SeqCst));
    }
}

Explanation & Structural Mechanics:

1. Auto-Trait Mechanics: Rust auto-traits recursively inspect struct definitions. StateContainer<T> inherits Send automatically as long as T: Send. If T is replaced with Rc<u32> or *mut u8, StateContainer<T> automatically becomes !Send.
2. Generic Bounds (T: Send + 'static): Explicitly placing T: Send on dispatch_to_thread prevents user code from attempting to transfer thread-unsafe generic payloads into worker threads.
3. Static Trait Verification: assert_send::<T>() is a zero-cost compile-time check ensuring types fulfill thread boundary constraints before runtime instantiation.