oxigdal_algorithms/parallel/
batch.rs

1//! Parallel batch processing
2//!
3//! This module provides utilities for processing multiple files or datasets
4//! in parallel with proper error handling and result collection.
5
6use core::sync::atomic::{AtomicUsize, Ordering};
7use rayon::prelude::*;
8
9use crate::error::{AlgorithmError, Result};
10
11/// Configuration for batch processing
12#[derive(Debug, Clone)]
13pub struct BatchConfig {
14    /// Number of threads to use
15    pub num_threads: Option<usize>,
16    /// Maximum number of items to process in parallel
17    pub max_parallel: usize,
18    /// Enable progress tracking
19    pub progress: bool,
20    /// Continue processing on errors
21    pub continue_on_error: bool,
22}
23
24impl Default for BatchConfig {
25    fn default() -> Self {
26        Self {
27            num_threads: None,
28            max_parallel: 100,
29            progress: false,
30            continue_on_error: false,
31        }
32    }
33}
34
35impl BatchConfig {
36    /// Creates a new batch configuration
37    #[must_use]
38    pub const fn new() -> Self {
39        Self {
40            num_threads: None,
41            max_parallel: 100,
42            progress: false,
43            continue_on_error: false,
44        }
45    }
46
47    /// Sets the number of threads
48    #[must_use]
49    pub const fn with_threads(mut self, num_threads: usize) -> Self {
50        self.num_threads = Some(num_threads);
51        self
52    }
53
54    /// Sets the maximum number of parallel items
55    #[must_use]
56    pub const fn with_max_parallel(mut self, max_parallel: usize) -> Self {
57        self.max_parallel = max_parallel;
58        self
59    }
60
61    /// Enables progress tracking
62    #[must_use]
63    pub const fn with_progress(mut self, progress: bool) -> Self {
64        self.progress = progress;
65        self
66    }
67
68    /// Sets continue on error behavior
69    #[must_use]
70    pub const fn with_continue_on_error(mut self, continue_on_error: bool) -> Self {
71        self.continue_on_error = continue_on_error;
72        self
73    }
74}
75
76/// Result of a batch operation
77#[derive(Debug, Clone)]
78pub struct BatchResult<T> {
79    /// Successfully processed items
80    pub successes: Vec<T>,
81    /// Failed items with their errors
82    pub failures: Vec<(usize, String)>,
83    /// Total number of items processed
84    pub total: usize,
85}
86
87impl<T> BatchResult<T> {
88    /// Creates a new batch result
89    #[must_use]
90    pub const fn new(successes: Vec<T>, failures: Vec<(usize, String)>, total: usize) -> Self {
91        Self {
92            successes,
93            failures,
94            total,
95        }
96    }
97
98    /// Returns true if all items succeeded
99    #[must_use]
100    pub fn is_success(&self) -> bool {
101        self.failures.is_empty()
102    }
103
104    /// Returns the number of successful items
105    #[must_use]
106    pub fn success_count(&self) -> usize {
107        self.successes.len()
108    }
109
110    /// Returns the number of failed items
111    #[must_use]
112    pub fn failure_count(&self) -> usize {
113        self.failures.len()
114    }
115
116    /// Returns the success rate as a percentage
117    #[must_use]
118    pub fn success_rate(&self) -> f64 {
119        if self.total == 0 {
120            0.0
121        } else {
122            (self.success_count() as f64 / self.total as f64) * 100.0
123        }
124    }
125}
126
127/// Progress tracker for batch operations
128pub struct BatchProgress {
129    total: usize,
130    processed: AtomicUsize,
131    succeeded: AtomicUsize,
132    failed: AtomicUsize,
133}
134
135impl BatchProgress {
136    /// Creates a new progress tracker
137    #[must_use]
138    pub const fn new(total: usize) -> Self {
139        Self {
140            total,
141            processed: AtomicUsize::new(0),
142            succeeded: AtomicUsize::new(0),
143            failed: AtomicUsize::new(0),
144        }
145    }
146
147    /// Records a successful item
148    pub fn record_success(&self) {
149        self.succeeded.fetch_add(1, Ordering::Relaxed);
150        let current = self.processed.fetch_add(1, Ordering::Relaxed) + 1;
151        self.log_progress(current);
152    }
153
154    /// Records a failed item
155    pub fn record_failure(&self) {
156        self.failed.fetch_add(1, Ordering::Relaxed);
157        let current = self.processed.fetch_add(1, Ordering::Relaxed) + 1;
158        self.log_progress(current);
159    }
160
161    /// Logs progress
162    fn log_progress(&self, current: usize) {
163        if current.is_multiple_of(10) || current == self.total {
164            let percent = (current * 100) / self.total;
165            let succeeded = self.succeeded.load(Ordering::Relaxed);
166            let failed = self.failed.load(Ordering::Relaxed);
167            tracing::info!(
168                "Batch progress: {}/{} ({}%) - Success: {}, Failed: {}",
169                current,
170                self.total,
171                percent,
172                succeeded,
173                failed
174            );
175        }
176    }
177
178    /// Returns current statistics
179    #[must_use]
180    pub fn stats(&self) -> (usize, usize, usize) {
181        (
182            self.processed.load(Ordering::Relaxed),
183            self.succeeded.load(Ordering::Relaxed),
184            self.failed.load(Ordering::Relaxed),
185        )
186    }
187}
188
189/// Process a batch of items in parallel
190///
191/// This function processes multiple items in parallel, collecting both
192/// successes and failures. It provides thread-safe error handling and
193/// optional progress tracking.
194///
195/// # Arguments
196///
197/// * `items` - Items to process
198/// * `config` - Batch configuration
199/// * `func` - Function to apply to each item
200///
201/// # Returns
202///
203/// Batch result containing successes and failures
204///
205/// # Example
206///
207/// ```no_run
208/// # #[cfg(feature = "parallel")]
209/// # {
210/// use oxigdal_algorithms::parallel::{parallel_batch_process, BatchConfig};
211///
212/// let items = vec![1, 2, 3, 4, 5];
213/// let config = BatchConfig::default();
214///
215/// let result = parallel_batch_process(&items, &config, |&item| {
216///     Ok(item * 2)
217/// });
218///
219/// match result {
220///     Ok(batch_result) => {
221///         println!("Success: {}/{}", batch_result.success_count(), batch_result.total);
222///     }
223///     Err(e) => eprintln!("Batch processing failed: {}", e),
224/// }
225/// # }
226/// ```
227pub fn parallel_batch_process<T, R, F>(
228    items: &[T],
229    config: &BatchConfig,
230    func: F,
231) -> Result<BatchResult<R>>
232where
233    T: Sync,
234    R: Send,
235    F: Fn(&T) -> Result<R> + Sync + Send,
236{
237    if items.is_empty() {
238        return Ok(BatchResult::new(Vec::new(), Vec::new(), 0));
239    }
240
241    let progress = if config.progress {
242        Some(BatchProgress::new(items.len()))
243    } else {
244        None
245    };
246
247    // Process items in parallel
248    let results: Vec<(usize, Result<R>)> = items
249        .par_iter()
250        .enumerate()
251        .map(|(idx, item)| {
252            let result = func(item);
253
254            if let Some(ref tracker) = progress {
255                match result {
256                    Ok(_) => tracker.record_success(),
257                    Err(_) => tracker.record_failure(),
258                }
259            }
260
261            (idx, result)
262        })
263        .collect();
264
265    // Separate successes and failures
266    let mut successes = Vec::new();
267    let mut failures = Vec::new();
268
269    for (idx, result) in results {
270        match result {
271            Ok(value) => successes.push(value),
272            Err(e) => {
273                failures.push((idx, e.to_string()));
274                if !config.continue_on_error {
275                    return Err(AlgorithmError::Core(oxigdal_core::OxiGdalError::Internal {
276                        message: format!("Batch processing failed at item {}: {}", idx, e),
277                    }));
278                }
279            }
280        }
281    }
282
283    Ok(BatchResult::new(successes, failures, items.len()))
284}
285
286/// Process a batch of items with indices
287///
288/// Similar to `parallel_batch_process` but also passes the index to the function.
289///
290/// # Arguments
291///
292/// * `items` - Items to process
293/// * `config` - Batch configuration
294/// * `func` - Function to apply to each item (receives index and item)
295///
296/// # Returns
297///
298/// Batch result containing successes and failures
299pub fn parallel_batch_process_indexed<T, R, F>(
300    items: &[T],
301    config: &BatchConfig,
302    func: F,
303) -> Result<BatchResult<R>>
304where
305    T: Sync,
306    R: Send,
307    F: Fn(usize, &T) -> Result<R> + Sync + Send,
308{
309    if items.is_empty() {
310        return Ok(BatchResult::new(Vec::new(), Vec::new(), 0));
311    }
312
313    let progress = if config.progress {
314        Some(BatchProgress::new(items.len()))
315    } else {
316        None
317    };
318
319    let results: Vec<(usize, Result<R>)> = items
320        .par_iter()
321        .enumerate()
322        .map(|(idx, item)| {
323            let result = func(idx, item);
324
325            if let Some(ref tracker) = progress {
326                match result {
327                    Ok(_) => tracker.record_success(),
328                    Err(_) => tracker.record_failure(),
329                }
330            }
331
332            (idx, result)
333        })
334        .collect();
335
336    let mut successes = Vec::new();
337    let mut failures = Vec::new();
338
339    for (idx, result) in results {
340        match result {
341            Ok(value) => successes.push(value),
342            Err(e) => {
343                failures.push((idx, e.to_string()));
344                if !config.continue_on_error {
345                    return Err(AlgorithmError::Core(oxigdal_core::OxiGdalError::Internal {
346                        message: format!("Batch processing failed at item {}: {}", idx, e),
347                    }));
348                }
349            }
350        }
351    }
352
353    Ok(BatchResult::new(successes, failures, items.len()))
354}
355
356/// Process a batch in chunks
357///
358/// This is useful when you want to control memory usage by processing
359/// items in smaller batches.
360///
361/// # Arguments
362///
363/// * `items` - Items to process
364/// * `chunk_size` - Number of items per chunk
365/// * `config` - Batch configuration
366/// * `func` - Function to apply to each chunk
367///
368/// # Returns
369///
370/// Batch result containing successes and failures
371pub fn parallel_batch_process_chunked<T, R, F>(
372    items: &[T],
373    chunk_size: usize,
374    config: &BatchConfig,
375    func: F,
376) -> Result<BatchResult<R>>
377where
378    T: Sync,
379    R: Send,
380    F: Fn(&[T]) -> Result<Vec<R>> + Sync + Send,
381{
382    if items.is_empty() {
383        return Ok(BatchResult::new(Vec::new(), Vec::new(), 0));
384    }
385
386    let chunks: Vec<&[T]> = items.chunks(chunk_size).collect();
387
388    let progress = if config.progress {
389        Some(BatchProgress::new(chunks.len()))
390    } else {
391        None
392    };
393
394    let results: Vec<Result<Vec<R>>> = chunks
395        .par_iter()
396        .map(|chunk| {
397            let result = func(chunk);
398
399            if let Some(ref tracker) = progress {
400                match result {
401                    Ok(_) => tracker.record_success(),
402                    Err(_) => tracker.record_failure(),
403                }
404            }
405
406            result
407        })
408        .collect();
409
410    let mut successes = Vec::new();
411    let mut failures = Vec::new();
412    let mut chunk_idx = 0;
413
414    for result in results {
415        match result {
416            Ok(values) => successes.extend(values),
417            Err(e) => {
418                failures.push((chunk_idx, e.to_string()));
419                if !config.continue_on_error {
420                    return Err(AlgorithmError::Core(oxigdal_core::OxiGdalError::Internal {
421                        message: format!("Chunk processing failed at chunk {}: {}", chunk_idx, e),
422                    }));
423                }
424            }
425        }
426        chunk_idx += 1;
427    }
428
429    Ok(BatchResult::new(successes, failures, items.len()))
430}
431
432/// Parallel map operation
433///
434/// A simpler version that just returns all results or the first error.
435///
436/// # Arguments
437///
438/// * `items` - Items to process
439/// * `func` - Function to apply to each item
440///
441/// # Returns
442///
443/// Vector of results
444///
445/// # Errors
446///
447/// Returns the first error encountered
448pub fn parallel_map<T, R, F>(items: &[T], func: F) -> Result<Vec<R>>
449where
450    T: Sync,
451    R: Send,
452    F: Fn(&T) -> Result<R> + Sync + Send,
453{
454    items.par_iter().map(func).collect()
455}
456
457#[cfg(test)]
458mod tests {
459    #![allow(clippy::expect_used)]
460
461    use super::*;
462
463    #[test]
464    fn test_batch_config() {
465        let config = BatchConfig::default();
466        assert!(config.num_threads.is_none());
467        assert_eq!(config.max_parallel, 100);
468        assert!(!config.progress);
469        assert!(!config.continue_on_error);
470    }
471
472    #[test]
473    fn test_batch_config_builder() {
474        let config = BatchConfig::new()
475            .with_threads(4)
476            .with_max_parallel(50)
477            .with_progress(true)
478            .with_continue_on_error(true);
479
480        assert_eq!(config.num_threads, Some(4));
481        assert_eq!(config.max_parallel, 50);
482        assert!(config.progress);
483        assert!(config.continue_on_error);
484    }
485
486    #[test]
487    fn test_parallel_batch_process_success() {
488        let items = vec![1, 2, 3, 4, 5];
489        let config = BatchConfig::default();
490
491        let result =
492            parallel_batch_process(&items, &config, |&item| Ok(item * 2)).expect("should work");
493
494        assert!(result.is_success());
495        assert_eq!(result.success_count(), 5);
496        assert_eq!(result.failure_count(), 0);
497        assert_eq!(result.successes, vec![2, 4, 6, 8, 10]);
498    }
499
500    #[test]
501    fn test_parallel_batch_process_with_errors() {
502        let items = vec![1, 2, 3, 4, 5];
503        let config = BatchConfig::new().with_continue_on_error(true);
504
505        let result = parallel_batch_process(&items, &config, |&item| {
506            if item % 2 == 0 {
507                Err(AlgorithmError::Core(oxigdal_core::OxiGdalError::Internal {
508                    message: "Even number".to_string(),
509                }))
510            } else {
511                Ok(item * 2)
512            }
513        })
514        .expect("should work");
515
516        assert!(!result.is_success());
517        assert_eq!(result.success_count(), 3); // 1, 3, 5
518        assert_eq!(result.failure_count(), 2); // 2, 4
519        assert!(result.success_rate() > 59.0 && result.success_rate() < 61.0);
520    }
521
522    #[test]
523    fn test_parallel_batch_process_fail_fast() {
524        let items = vec![1, 2, 3, 4, 5];
525        let config = BatchConfig::new().with_continue_on_error(false);
526
527        let result = parallel_batch_process(&items, &config, |&item| {
528            if item % 2 == 0 {
529                Err(AlgorithmError::Core(oxigdal_core::OxiGdalError::Internal {
530                    message: "Even number".to_string(),
531                }))
532            } else {
533                Ok(item * 2)
534            }
535        });
536
537        assert!(result.is_err());
538    }
539
540    #[test]
541    fn test_parallel_batch_process_indexed() {
542        let items = vec!["a", "b", "c"];
543        let config = BatchConfig::default();
544
545        let result = parallel_batch_process_indexed(&items, &config, |idx, &item| {
546            Ok(format!("{}:{}", idx, item))
547        })
548        .expect("should work");
549
550        assert!(result.is_success());
551        assert_eq!(result.success_count(), 3);
552
553        // Results should contain index:item pairs
554        assert!(result.successes.contains(&"0:a".to_string()));
555        assert!(result.successes.contains(&"1:b".to_string()));
556        assert!(result.successes.contains(&"2:c".to_string()));
557    }
558
559    #[test]
560    fn test_parallel_batch_process_chunked() {
561        let items: Vec<i32> = (0..100).collect();
562        let config = BatchConfig::default();
563
564        let result = parallel_batch_process_chunked(&items, 10, &config, |chunk| {
565            Ok(chunk.iter().map(|&x| x * 2).collect())
566        })
567        .expect("should work");
568
569        assert!(result.is_success());
570        assert_eq!(result.success_count(), 100);
571    }
572
573    #[test]
574    fn test_parallel_map() {
575        let items = vec![1, 2, 3, 4, 5];
576        let result = parallel_map(&items, |&item| Ok(item * 2)).expect("should work");
577
578        assert_eq!(result, vec![2, 4, 6, 8, 10]);
579    }
580
581    #[test]
582    fn test_parallel_map_with_error() {
583        let items = vec![1, 2, 3, 4, 5];
584        let result = parallel_map(&items, |&item| {
585            if item == 3 {
586                Err(AlgorithmError::Core(oxigdal_core::OxiGdalError::Internal {
587                    message: "Error".to_string(),
588                }))
589            } else {
590                Ok(item * 2)
591            }
592        });
593
594        assert!(result.is_err());
595    }
596
597    #[test]
598    fn test_batch_result() {
599        let result = BatchResult::new(vec![1, 2, 3], vec![(1, "error".to_string())], 4);
600
601        assert!(!result.is_success());
602        assert_eq!(result.success_count(), 3);
603        assert_eq!(result.failure_count(), 1);
604        assert_eq!(result.total, 4);
605        assert_eq!(result.success_rate(), 75.0);
606    }
607
608    #[test]
609    fn test_batch_progress() {
610        let progress = BatchProgress::new(10);
611
612        progress.record_success();
613        progress.record_success();
614        progress.record_failure();
615
616        let (processed, succeeded, failed) = progress.stats();
617        assert_eq!(processed, 3);
618        assert_eq!(succeeded, 2);
619        assert_eq!(failed, 1);
620    }
621
622    #[test]
623    fn test_empty_batch() {
624        let items: Vec<i32> = Vec::new();
625        let config = BatchConfig::default();
626
627        let result =
628            parallel_batch_process(&items, &config, |&item| Ok(item * 2)).expect("should work");
629
630        assert!(result.is_success());
631        assert_eq!(result.total, 0);
632        assert_eq!(result.success_count(), 0);
633    }
634}
oxigdal_algorithms/parallel/batch.rs

oxigdal_algorithms/parallel/
batch.rs
