@@ -66,6 +66,7 @@ use std::any;
6666use std:: env;
6767use std:: fmt;
6868use std:: io:: { self , Read , Write } ;
69+ use std:: num:: NonZeroUsize ;
6970use std:: process:: { Child , Command , Stdio } ;
7071use std:: time:: { Duration , Instant } ;
7172
@@ -484,6 +485,163 @@ impl<T: Task> Drop for Process<T> {
484485 }
485486}
486487
488+ /// A pool of worker processes for concurrent task execution.
489+ ///
490+ /// `ProcessPool` manages multiple worker processes, distributing tasks across them
491+ /// using round-robin scheduling. Each worker runs in its own isolated process with
492+ /// automatic crash recovery.
493+ ///
494+ /// # Example
495+ ///
496+ /// ```no_run
497+ /// use std::num::NonZeroUsize;
498+ /// use tarnish::{Task, ProcessPool};
499+ ///
500+ /// #[derive(Default)]
501+ /// struct HeavyComputation;
502+ ///
503+ /// impl Task for HeavyComputation {
504+ /// type Input = Vec<u8>;
505+ /// type Output = u64;
506+ /// type Error = String;
507+ ///
508+ /// fn run(&mut self, input: Vec<u8>) -> Result<u64, String> {
509+ /// // Expensive computation here
510+ /// Ok(input.iter().map(|&x| x as u64).sum())
511+ /// }
512+ /// }
513+ ///
514+ /// tarnish::main::<HeavyComputation>(|| {
515+ /// let size = NonZeroUsize::new(4).unwrap();
516+ /// let mut pool = ProcessPool::<HeavyComputation>::new(size)
517+ /// .expect("Failed to create pool");
518+ ///
519+ /// // Process 100 items across 4 workers
520+ /// for i in 0..100 {
521+ /// let result = pool.call(vec![i; 1000]);
522+ /// println!("Result {}: {:?}", i, result);
523+ /// }
524+ /// });
525+ /// ```
526+ pub struct ProcessPool < T : Task > {
527+ workers : Vec < Process < T > > ,
528+ next_worker : std:: sync:: atomic:: AtomicUsize ,
529+ }
530+
531+ impl < T : Task > ProcessPool < T > {
532+ /// Create a new process pool with the specified number of workers.
533+ ///
534+ /// Each worker is a separate process that will be spawned immediately.
535+ /// If any worker fails to spawn, an error is returned and no pool is created.
536+ ///
537+ /// # Arguments
538+ ///
539+ /// * `size` - The number of worker processes to spawn (must be non-zero).
540+ ///
541+ /// # Errors
542+ ///
543+ /// Returns an error if any worker process fails to spawn.
544+ ///
545+ /// # Example
546+ ///
547+ /// ```no_run
548+ /// use std::num::NonZeroUsize;
549+ /// use tarnish::ProcessPool;
550+ /// # use tarnish::Task;
551+ /// # #[derive(Default)]
552+ /// # struct MyTask;
553+ /// # impl Task for MyTask {
554+ /// # type Input = String;
555+ /// # type Output = String;
556+ /// # type Error = String;
557+ /// # fn run(&mut self, input: String) -> Result<String, String> { Ok(input) }
558+ /// # }
559+ ///
560+ /// let size = NonZeroUsize::new(4).unwrap();
561+ /// let pool = ProcessPool::<MyTask>::new(size)?;
562+ /// # Ok::<(), tarnish::ProcessError>(())
563+ /// ```
564+ pub fn new ( size : NonZeroUsize ) -> Result < Self > {
565+ let workers = ( 0 ..size. get ( ) )
566+ . map ( |_| Process :: < T > :: spawn ( ) )
567+ . collect :: < Result < Vec < _ > > > ( ) ?;
568+
569+ Ok ( Self {
570+ workers,
571+ next_worker : std:: sync:: atomic:: AtomicUsize :: new ( 0 ) ,
572+ } )
573+ }
574+
575+ /// Execute a task on the next available worker.
576+ ///
577+ /// This method uses round-robin scheduling to distribute work across workers.
578+ /// The call blocks until the worker returns a result. If the worker crashes,
579+ /// it will be automatically restarted.
580+ ///
581+ /// # Errors
582+ ///
583+ /// Returns an error if:
584+ /// - The worker process crashes and cannot be restarted
585+ /// - Communication with the worker fails
586+ /// - The worker returns a task error
587+ ///
588+ /// # Example
589+ ///
590+ /// ```no_run
591+ /// # use std::num::NonZeroUsize;
592+ /// # use tarnish::{Task, ProcessPool};
593+ /// # #[derive(Default)]
594+ /// # struct MyTask;
595+ /// # impl Task for MyTask {
596+ /// # type Input = String;
597+ /// # type Output = String;
598+ /// # type Error = String;
599+ /// # fn run(&mut self, input: String) -> Result<String, String> { Ok(input) }
600+ /// # }
601+ /// let size = NonZeroUsize::new(4).unwrap();
602+ /// let mut pool = ProcessPool::<MyTask>::new(size)?;
603+ /// let result = pool.call("hello".to_string())?;
604+ /// # Ok::<(), tarnish::ProcessError>(())
605+ /// ```
606+ #[ allow( clippy:: arithmetic_side_effects, clippy:: indexing_slicing) ]
607+ pub fn call ( & mut self , input : T :: Input ) -> Result < T :: Output > {
608+ let idx = self
609+ . next_worker
610+ . fetch_add ( 1 , std:: sync:: atomic:: Ordering :: Relaxed )
611+ % self . workers . len ( ) ;
612+
613+ self . workers
614+ . get_mut ( idx)
615+ . ok_or_else ( || ProcessError :: ProtocolError ( format ! ( "Invalid worker index: {idx}" ) ) ) ?
616+ . call ( input)
617+ }
618+
619+ /// Returns the number of workers in the pool.
620+ ///
621+ /// # Example
622+ ///
623+ /// ```no_run
624+ /// # use std::num::NonZeroUsize;
625+ /// # use tarnish::{Task, ProcessPool};
626+ /// # #[derive(Default)]
627+ /// # struct MyTask;
628+ /// # impl Task for MyTask {
629+ /// # type Input = String;
630+ /// # type Output = String;
631+ /// # type Error = String;
632+ /// # fn run(&mut self, input: String) -> Result<String, String> { Ok(input) }
633+ /// # }
634+ /// let size = NonZeroUsize::new(4).unwrap();
635+ /// let pool = ProcessPool::<MyTask>::new(size)?;
636+ /// assert_eq!(pool.size(), 4);
637+ /// # Ok::<(), tarnish::ProcessError>(())
638+ /// ```
639+ #[ must_use]
640+ pub const fn size ( & self ) -> usize {
641+ self . workers . len ( )
642+ }
643+ }
644+
487645/// Handle worker process mode in your main function
488646///
489647/// Call this at the start of your `main()` function. If it returns `Some(exit_code)`,
0 commit comments