pebble/threading/
mod.rs

1//! A small, fixed-size worker pool for offloading CPU-bound work off the
2//! main thread — mip/image processing, physics steps, any one-off or
3//! recurring task that shouldn't block a frame.
4//!
5//! Deliberately NOT a full async runtime: no cancellation, no priorities,
6//! no work-stealing. Just a bounded number of OS threads pulling jobs off
7//! one shared, lock-free MPMC queue, with results delivered back via a
8//! channel you poll from an ordinary system. If you outgrow this — need
9//! cancellation, need priority scheduling — that's real, separate
10//! infrastructure to build once you have a concrete case for it, not
11//! something to guess at now.
12//!
13//! Requires the `crossbeam-channel` crate (lock-free MPMC), since
14//! `std::sync::mpsc` only supports a single consumer and would otherwise
15//! force a `Mutex` around the receiver for multiple worker threads.
16
17use crossbeam_channel::{Receiver as CbReceiver, Sender as CbSender, unbounded};
18use std::sync::mpsc::{Receiver, Sender, channel};
19
20use crate::ecs::plugin::Plugin;
21
22type Job = Box<dyn FnOnce() + Send + 'static>;
23
24/// The worker pool itself. Insert as a resource once, at startup; every
25/// system that needs to offload work reaches for `Res<BackgroundTasks>`
26/// and calls `spawn`.
27pub struct BackgroundTasks {
28    job_tx: CbSender<Job>,
29}
30
31impl BackgroundTasks {
32    /// Spawns `worker_count` OS threads, each pulling jobs off one shared,
33    /// lock-free queue until the pool itself is dropped. A worker count
34    /// around your CPU's core count (minus one, to leave room for the
35    /// main thread) is a reasonable default; tune based on actual
36    /// measured load.
37    pub fn new(worker_count: usize) -> Self {
38        let (job_tx, job_rx): (CbSender<Job>, CbReceiver<Job>) = unbounded();
39
40        for _ in 0..worker_count.max(1) {
41            let job_rx = job_rx.clone(); // cheap — crossbeam receivers are natively Clone, no Mutex needed
42            std::thread::spawn(move || {
43                // `recv()` blocks this worker thread only, until a job
44                // arrives or every sender (the pool, plus any clones) is
45                // dropped — no lock contention between workers picking up
46                // jobs concurrently.
47                while let Ok(job) = job_rx.recv() {
48                    job();
49                }
50            });
51        }
52
53        Self { job_tx }
54    }
55
56    /// Queue `work` to run on the pool. Returns a [`TaskHandle`] you can
57    /// poll (non-blocking) from any system to check whether it's done.
58    ///
59    /// `work` runs on whichever worker thread picks it up next — don't
60    /// assume anything about timing or ordering relative to other spawned
61    /// tasks unless you build that coordination yourself.
62    pub fn spawn<T: Send + 'static>(
63        &self,
64        work: impl FnOnce() -> T + Send + 'static,
65    ) -> TaskHandle<T> {
66        let (result_tx, result_rx) = channel::<T>();
67        let job: Job = Box::new(move || {
68            let result = work();
69            let _ = result_tx.send(result); // ignore: receiver may have been dropped, that's fine
70        });
71        // If this fails, every worker thread has panicked and the pool is
72        // effectively dead — surfaced as a permanently-pending TaskHandle
73        // rather than a panic here, since a dead pool shouldn't crash an
74        // unrelated caller trying to queue new work.
75        let _ = self.job_tx.send(job);
76        TaskHandle { rx: result_rx }
77    }
78}
79
80/// A handle to a single in-flight (or already-finished) task. Poll it
81/// from an ordinary system with [`try_recv`](TaskHandle::try_recv) —
82/// never blocks.
83pub struct TaskHandle<T> {
84    rx: Receiver<T>,
85}
86
87impl<T> TaskHandle<T> {
88    /// Returns `Some(result)` once the task has finished, `None`
89    /// otherwise. Never blocks — safe to call every tick.
90    pub fn try_recv(&mut self) -> Option<T> {
91        self.rx.try_recv().ok()
92    }
93}
94
95/// Registers `BackgroundTasks` as a resource with the given worker count.
96///
97/// ```ignore
98/// app.add_plugin(BackgroundTasksPlugin::new(4));
99/// ```
100pub struct BackgroundTasksPlugin {
101    worker_count: usize,
102}
103
104impl BackgroundTasksPlugin {
105    pub fn new(worker_count: usize) -> Self {
106        Self { worker_count }
107    }
108}
109
110impl Plugin for BackgroundTasksPlugin {
111    fn build(&self, app: &mut crate::prelude::App) {
112        app.add_resource(BackgroundTasks::new(self.worker_count));
113    }
114}
pebble/threading/mod.rs

pebble/threading/
mod.rs
