/home/noah/src/realizar/src/inference/thread.rs
Line | Count | Source |
1 | | //! Thread configuration for inference operations |
2 | | //! |
3 | | //! Provides dynamic thread allocation following llama.cpp patterns: |
4 | | //! - Prefill/batch: Use all cores for maximum throughput |
5 | | //! - Decode: Use fewer cores to reduce cache thrashing |
6 | | //! |
7 | | //! ## Performance Notes |
8 | | //! |
9 | | //! Profiling shows optimal thread counts for memory-bound quantized matmuls: |
10 | | //! - 48 threads: 11.9 tok/s (too much sync overhead) |
11 | | //! - 24 threads: 18.7 tok/s |
12 | | //! - 16 threads: 25.3 tok/s (optimal) |
13 | | //! - 12 threads: 25.0 tok/s |
14 | | //! - 8 threads: 21.9 tok/s |
15 | | |
16 | | use crate::error::{RealizarError, Result}; |
17 | | |
18 | | /// Thread configuration for dynamic thread allocation |
19 | | /// |
20 | | /// Per llama.cpp: batch processing uses more threads than single-token decode. |
21 | | /// This reduces cache thrashing during decode phase. |
22 | | /// |
23 | | /// # Example |
24 | | /// |
25 | | /// ``` |
26 | | /// use realizar::inference::ThreadConfig; |
27 | | /// |
28 | | /// let config = ThreadConfig::auto(); |
29 | | /// assert!(config.n_threads_batch >= 1); |
30 | | /// assert!(config.n_threads_decode >= 1); |
31 | | /// |
32 | | /// // Batch uses more threads than decode |
33 | | /// assert!(config.n_threads_batch >= config.n_threads_decode); |
34 | | /// ``` |
35 | | #[derive(Debug, Clone, Copy, PartialEq, Eq)] |
36 | | pub struct ThreadConfig { |
37 | | /// Threads for batch/prefill operations (uses all cores) |
38 | | pub n_threads_batch: usize, |
39 | | /// Threads for single-token decode (uses fewer cores) |
40 | | pub n_threads_decode: usize, |
41 | | } |
42 | | |
43 | | impl ThreadConfig { |
44 | | /// Create optimal thread config based on available cores |
45 | | /// |
46 | | /// - Batch: Uses all available cores |
47 | | /// - Decode: Uses half cores (min 1) to reduce cache contention |
48 | | #[must_use] |
49 | 4 | pub fn auto() -> Self { |
50 | 4 | let num_cpus = rayon::current_num_threads(); |
51 | 4 | Self { |
52 | 4 | n_threads_batch: num_cpus, |
53 | 4 | n_threads_decode: (num_cpus / 2).max(1), |
54 | 4 | } |
55 | 4 | } |
56 | | |
57 | | /// Create with explicit thread counts |
58 | | /// |
59 | | /// Both values are clamped to minimum of 1. |
60 | | #[must_use] |
61 | 10 | pub fn new(n_threads_batch: usize, n_threads_decode: usize) -> Self { |
62 | 10 | Self { |
63 | 10 | n_threads_batch: n_threads_batch.max(1), |
64 | 10 | n_threads_decode: n_threads_decode.max(1), |
65 | 10 | } |
66 | 10 | } |
67 | | |
68 | | /// Get the number of threads for the current operation mode |
69 | | /// |
70 | | /// Returns `n_threads_batch` for prefill, `n_threads_decode` otherwise. |
71 | | #[must_use] |
72 | 6 | pub fn threads_for(&self, is_prefill: bool) -> usize { |
73 | 6 | if is_prefill { |
74 | 3 | self.n_threads_batch |
75 | | } else { |
76 | 3 | self.n_threads_decode |
77 | | } |
78 | 6 | } |
79 | | } |
80 | | |
81 | | impl Default for ThreadConfig { |
82 | 1 | fn default() -> Self { |
83 | 1 | Self::auto() |
84 | 1 | } |
85 | | } |
86 | | |
87 | | /// Execution mode for controlling parallelism |
88 | | #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] |
89 | | pub enum InferenceMode { |
90 | | /// Prefill/prompt processing - use maximum threads |
91 | | Prefill, |
92 | | /// Single-token decode - use fewer threads to reduce cache thrashing |
93 | | Decode, |
94 | | } |
95 | | |
96 | | impl InferenceMode { |
97 | | /// Returns true if this is prefill mode |
98 | | #[must_use] |
99 | 4 | pub fn is_prefill(self) -> bool { |
100 | 4 | matches!2 (self, Self::Prefill) |
101 | 4 | } |
102 | | |
103 | | /// Returns true if this is decode mode |
104 | | #[must_use] |
105 | 2 | pub fn is_decode(self) -> bool { |
106 | 2 | matches!1 (self, Self::Decode) |
107 | 2 | } |
108 | | } |
109 | | |
110 | | /// Configure the global rayon thread pool with optimal thread count for inference |
111 | | /// |
112 | | /// Uses ~16 threads by default for optimal memory bandwidth utilization on |
113 | | /// modern multi-core CPUs. This can be overridden with `RAYON_NUM_THREADS` env var. |
114 | | /// |
115 | | /// # Errors |
116 | | /// |
117 | | /// Returns `InvalidConfiguration` if the thread pool has already been initialized. |
118 | | /// |
119 | | /// # Example |
120 | | /// |
121 | | /// ```ignore |
122 | | /// // Call once at startup |
123 | | /// realizar::inference::configure_optimal_thread_pool()?; |
124 | | /// ``` |
125 | 0 | pub fn configure_optimal_thread_pool() -> Result<()> { |
126 | 0 | let optimal_threads = std::env::var("RAYON_NUM_THREADS") |
127 | 0 | .ok() |
128 | 0 | .and_then(|s| s.parse().ok()) |
129 | 0 | .unwrap_or(16); |
130 | | |
131 | 0 | configure_thread_pool(optimal_threads) |
132 | 0 | } |
133 | | |
134 | | /// Configure the global rayon thread pool for inference |
135 | | /// |
136 | | /// NOTE: This should be called once at startup. Rayon's global pool cannot |
137 | | /// be resized dynamically. |
138 | | /// |
139 | | /// # Errors |
140 | | /// |
141 | | /// Returns `InvalidConfiguration` if the thread pool has already been initialized. |
142 | 0 | pub fn configure_thread_pool(num_threads: usize) -> Result<()> { |
143 | 0 | rayon::ThreadPoolBuilder::new() |
144 | 0 | .num_threads(num_threads) |
145 | 0 | .build_global() |
146 | 0 | .map_err(|e| { |
147 | 0 | RealizarError::InvalidConfiguration(format!("Failed to configure thread pool: {e}")) |
148 | 0 | }) |
149 | 0 | } |
150 | | |
151 | | // ============================================================================ |
152 | | // EXTREME TDD: Comprehensive Tests |
153 | | // ============================================================================ |
154 | | |
155 | | #[cfg(test)] |
156 | | mod tests { |
157 | | use super::*; |
158 | | |
159 | | // ------------------------------------------------------------------------ |
160 | | // ThreadConfig Tests |
161 | | // ------------------------------------------------------------------------ |
162 | | |
163 | | #[test] |
164 | 1 | fn test_thread_config_auto_returns_valid_config() { |
165 | 1 | let config = ThreadConfig::auto(); |
166 | 1 | assert!(config.n_threads_batch >= 1, "batch threads must be >= 1"0 ); |
167 | 1 | assert!(config.n_threads_decode >= 1, "decode threads must be >= 1"0 ); |
168 | 1 | } |
169 | | |
170 | | #[test] |
171 | 1 | fn test_thread_config_auto_batch_gte_decode() { |
172 | 1 | let config = ThreadConfig::auto(); |
173 | 1 | assert!( |
174 | 1 | config.n_threads_batch >= config.n_threads_decode, |
175 | 0 | "batch threads should be >= decode threads" |
176 | | ); |
177 | 1 | } |
178 | | |
179 | | #[test] |
180 | 1 | fn test_thread_config_new_clamps_to_minimum() { |
181 | 1 | let config = ThreadConfig::new(0, 0); |
182 | 1 | assert_eq!(config.n_threads_batch, 1, "0 should be clamped to 1"0 ); |
183 | 1 | assert_eq!(config.n_threads_decode, 1, "0 should be clamped to 1"0 ); |
184 | 1 | } |
185 | | |
186 | | #[test] |
187 | 1 | fn test_thread_config_new_preserves_valid_values() { |
188 | 1 | let config = ThreadConfig::new(8, 4); |
189 | 1 | assert_eq!(config.n_threads_batch, 8); |
190 | 1 | assert_eq!(config.n_threads_decode, 4); |
191 | 1 | } |
192 | | |
193 | | #[test] |
194 | 1 | fn test_thread_config_threads_for_prefill() { |
195 | 1 | let config = ThreadConfig::new(16, 8); |
196 | 1 | assert_eq!(config.threads_for(true), 16); |
197 | 1 | } |
198 | | |
199 | | #[test] |
200 | 1 | fn test_thread_config_threads_for_decode() { |
201 | 1 | let config = ThreadConfig::new(16, 8); |
202 | 1 | assert_eq!(config.threads_for(false), 8); |
203 | 1 | } |
204 | | |
205 | | #[test] |
206 | 1 | fn test_thread_config_default_uses_auto() { |
207 | 1 | let default_config = ThreadConfig::default(); |
208 | 1 | let auto_config = ThreadConfig::auto(); |
209 | 1 | assert_eq!(default_config, auto_config); |
210 | 1 | } |
211 | | |
212 | | #[test] |
213 | 1 | fn test_thread_config_clone() { |
214 | 1 | let config = ThreadConfig::new(12, 6); |
215 | 1 | let cloned = config; |
216 | 1 | assert_eq!(config, cloned); |
217 | 1 | } |
218 | | |
219 | | #[test] |
220 | 1 | fn test_thread_config_debug() { |
221 | 1 | let config = ThreadConfig::new(4, 2); |
222 | 1 | let debug_str = format!("{:?}", config); |
223 | 1 | assert!(debug_str.contains("ThreadConfig")); |
224 | 1 | assert!(debug_str.contains("4")); |
225 | 1 | assert!(debug_str.contains("2")); |
226 | 1 | } |
227 | | |
228 | | // ------------------------------------------------------------------------ |
229 | | // InferenceMode Tests |
230 | | // ------------------------------------------------------------------------ |
231 | | |
232 | | #[test] |
233 | 1 | fn test_inference_mode_is_prefill() { |
234 | 1 | assert!(InferenceMode::Prefill.is_prefill()); |
235 | 1 | assert!(!InferenceMode::Decode.is_prefill()); |
236 | 1 | } |
237 | | |
238 | | #[test] |
239 | 1 | fn test_inference_mode_is_decode() { |
240 | 1 | assert!(InferenceMode::Decode.is_decode()); |
241 | 1 | assert!(!InferenceMode::Prefill.is_decode()); |
242 | 1 | } |
243 | | |
244 | | #[test] |
245 | 1 | fn test_inference_mode_equality() { |
246 | 1 | assert_eq!(InferenceMode::Prefill, InferenceMode::Prefill); |
247 | 1 | assert_eq!(InferenceMode::Decode, InferenceMode::Decode); |
248 | 1 | assert_ne!(InferenceMode::Prefill, InferenceMode::Decode); |
249 | 1 | } |
250 | | |
251 | | #[test] |
252 | 1 | fn test_inference_mode_clone() { |
253 | 1 | let mode = InferenceMode::Prefill; |
254 | 1 | let cloned = mode; |
255 | 1 | assert_eq!(mode, cloned); |
256 | 1 | } |
257 | | |
258 | | #[test] |
259 | 1 | fn test_inference_mode_debug() { |
260 | 1 | assert_eq!(format!("{:?}", InferenceMode::Prefill), "Prefill"); |
261 | 1 | assert_eq!(format!("{:?}", InferenceMode::Decode), "Decode"); |
262 | 1 | } |
263 | | |
264 | | #[test] |
265 | 1 | fn test_inference_mode_hash() { |
266 | | use std::collections::HashSet; |
267 | 1 | let mut set = HashSet::new(); |
268 | 1 | set.insert(InferenceMode::Prefill); |
269 | 1 | set.insert(InferenceMode::Decode); |
270 | 1 | assert_eq!(set.len(), 2); |
271 | 1 | assert!(set.contains(&InferenceMode::Prefill)); |
272 | 1 | assert!(set.contains(&InferenceMode::Decode)); |
273 | 1 | } |
274 | | |
275 | | // ------------------------------------------------------------------------ |
276 | | // Integration Tests: ThreadConfig + InferenceMode |
277 | | // ------------------------------------------------------------------------ |
278 | | |
279 | | #[test] |
280 | 1 | fn test_config_with_mode() { |
281 | 1 | let config = ThreadConfig::new(16, 4); |
282 | | |
283 | 1 | let prefill_threads = config.threads_for(InferenceMode::Prefill.is_prefill()); |
284 | 1 | let decode_threads = config.threads_for(InferenceMode::Decode.is_prefill()); |
285 | | |
286 | 1 | assert_eq!(prefill_threads, 16); |
287 | 1 | assert_eq!(decode_threads, 4); |
288 | 1 | } |
289 | | |
290 | | // ------------------------------------------------------------------------ |
291 | | // Edge Case Tests |
292 | | // ------------------------------------------------------------------------ |
293 | | |
294 | | #[test] |
295 | 1 | fn test_thread_config_with_one_thread() { |
296 | 1 | let config = ThreadConfig::new(1, 1); |
297 | 1 | assert_eq!(config.threads_for(true), 1); |
298 | 1 | assert_eq!(config.threads_for(false), 1); |
299 | 1 | } |
300 | | |
301 | | #[test] |
302 | 1 | fn test_thread_config_large_values() { |
303 | 1 | let config = ThreadConfig::new(1024, 512); |
304 | 1 | assert_eq!(config.n_threads_batch, 1024); |
305 | 1 | assert_eq!(config.n_threads_decode, 512); |
306 | 1 | } |
307 | | |
308 | | #[test] |
309 | 1 | fn test_thread_config_decode_larger_than_batch() { |
310 | | // This is allowed, though unusual |
311 | 1 | let config = ThreadConfig::new(4, 8); |
312 | 1 | assert_eq!(config.n_threads_batch, 4); |
313 | 1 | assert_eq!(config.n_threads_decode, 8); |
314 | 1 | } |
315 | | } |