1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
use crate::{queue::Queue, task_future::TaskFuture, worker::spawn_worker};
use std::{
num::NonZeroUsize,
sync::Arc,
thread::{self, JoinHandle},
};
pub struct ZeroPool {
queue: Arc<Queue>,
workers: Box<[JoinHandle<()>]>,
}
impl ZeroPool {
/// Creates a new thread pool with worker count equal to available parallelism
///
/// Worker count is determined by `std::thread::available_parallelism()`,
/// falling back to 1 if unavailable. This is usually the optimal choice
/// for CPU-bound workloads.
///
/// # Examples
///
/// ```rust
/// use zero_pool::ZeroPool;
/// let pool = ZeroPool::new();
/// ```
#[must_use]
pub fn new() -> Self {
let worker_count = thread::available_parallelism().unwrap_or(NonZeroUsize::MIN);
Self::with_workers(worker_count)
}
/// Creates a new thread pool with the specified number of workers
///
/// Use this when you need precise control over the worker count,
/// for example when coordinating with other thread pools or
/// when you know the optimal count for your specific workload.
///
/// # Examples
///
/// ```rust
/// use std::num::NonZeroUsize;
/// use zero_pool::ZeroPool;
/// let pool = ZeroPool::with_workers(NonZeroUsize::new(4).unwrap());
/// ```
#[must_use]
pub fn with_workers(worker_count: NonZeroUsize) -> Self {
let worker_count = worker_count.get();
let queue = Arc::new(Queue::new(worker_count));
let workers = (0..worker_count)
.map(|id| {
let handle = spawn_worker(id, queue.clone());
queue.register_thread(id, handle.thread().clone());
handle
})
.collect();
ZeroPool { queue, workers }
}
/// Submit a single typed task
///
/// The parameter struct must remain valid until the future completes.
/// This is the recommended method for submitting individual tasks.
///
/// # Examples
///
/// ```rust
/// use zero_pool::ZeroPool;
///
/// struct MyTaskParams { value: u64, result: *mut u64 }
///
/// fn my_task_fn(params: &MyTaskParams) {
/// unsafe { *params.result = params.value * 2; }
/// }
///
/// let pool = ZeroPool::new();
/// let mut result = 0u64;
/// let task_params = MyTaskParams { value: 42, result: &mut result };
/// let future = pool.submit_task(my_task_fn, &task_params);
/// future.wait();
/// assert_eq!(result, 84);
/// ```
#[inline]
pub fn submit_task<T>(&self, task_fn: fn(&T), params: &T) -> TaskFuture {
let slice = std::slice::from_ref(params);
self.queue.push_task_batch(task_fn, slice)
}
/// Submit a batch of uniform tasks
///
/// All tasks in the batch must be the same type and use the same task function.
/// This method handles the pointer conversion automatically and is the most
/// convenient way to submit large batches of similar work.
///
/// # Examples
///
/// ```rust
/// use zero_pool::ZeroPool;
///
/// struct MyTaskParams { value: u64, result: *mut u64 }
///
/// fn my_task_fn(params: &MyTaskParams) {
/// unsafe { *params.result = params.value * 2; }
/// }
///
/// let pool = ZeroPool::new();
/// let mut results = vec![0u64; 1000];
/// let task_params: Vec<_> = results
/// .iter_mut()
/// .enumerate()
/// .map(|(i, res)| MyTaskParams { value: i as u64, result: res })
/// .collect();
/// let future = pool.submit_batch(my_task_fn, &task_params);
/// future.wait();
/// assert_eq!(results[0], 0);
/// assert_eq!(results[1], 2);
/// assert_eq!(results[999], 1998);
/// ```
#[inline]
pub fn submit_batch<T>(&self, task_fn: fn(&T), params_vec: &[T]) -> TaskFuture {
self.queue.push_task_batch(task_fn, params_vec)
}
}
impl Default for ZeroPool {
/// Creates a new thread pool with default settings
///
/// Equivalent to calling `ZeroPool::new()`. Worker count is determined by
/// `std::thread::available_parallelism()`, falling back to 1 if unavailable.
fn default() -> Self {
Self::new()
}
}
impl Drop for ZeroPool {
fn drop(&mut self) {
self.queue.shutdown();
for handle in std::mem::take(&mut self.workers) {
let _ = handle.join();
}
}
}