/// Runs a number of epochs where fitness is calculated across n parallel
/// processes. This is useful when the fitness calcuation is an expensive
/// operation.
Runs a number of epochs where fitness is calculated across n parallel
processes. This is useful when the fitness calcuation is an expensive
operation.
Runs a number of epochs where fitness is calculated across n parallel processes . This is useful when the fitness calcuation is an expensive operation .
pub fn epochs_parallel(&mut self, n_epochs: u32, n_processes: u32) -> &mut Self {
        scope(|scope| {
            let cvar_pair = Arc::new((Mutex::new(0), Condvar::new()));
            let (tx, rx) = sync_channel(0);
            let process_queue = Arc::new(Mutex::new(rx));
            let processed_stack = Arc::new(Mutex::new(Vec::new()));
            for _ in 0..n_processes {
                let cvar_pair_clone = cvar_pair.clone();
                let processed_stack_clone = processed_stack.clone();
                let process_queue_clone = process_queue.clone();
                scope.spawn(move || {
                    let &(ref lock, ref cvar) = &*cvar_pair_clone;
                    loop {
                        let mut l_unit: LazyUnit<T> =
                            match process_queue_clone.lock().ok().unwrap().recv() {
                                Ok(u) => u,
                                Err(_) => return,
                            };
                        l_unit.fitness();
                        processed_stack_clone.lock().ok().unwrap().push(l_unit);
                        {
                            let mut processed = lock.lock().unwrap();
                            *processed += 1;
                            cvar.notify_all();
                        }
                    }
                });
            }
            let &(ref lock, ref cvar) = &*cvar_pair;
            let mut active_stack = Vec::new();
            while let Some(unit) = self.units.pop() {
                active_stack.push(LazyUnit::from(unit));
            }
            let seed: &[_] = &[self.seed];
            let mut rng: StdRng = SeedableRng::from_seed(seed);
            for i in 0..(n_epochs + 1) {
                let jobs_total = active_stack.len();
                while let Some(unit) = active_stack.pop() {
                    tx.send(unit).unwrap();
                }
                let mut jobs_processed = lock.lock().unwrap();
                while *jobs_processed != jobs_total {
                    jobs_processed = cvar.wait(jobs_processed).unwrap();
                }
                *jobs_processed = 0;
                // Swap the full processed_stack with the active stack.
                mem::swap(&mut active_stack, &mut processed_stack.lock().ok().unwrap());
                // We want to sort such that highest fitness units are at the
                // end.
                active_stack.sort_by(|a, b| {
                    a.lazy_fitness
                        .unwrap_or(0.0)
                        .partial_cmp(&b.lazy_fitness.unwrap_or(0.0))
                        .unwrap_or(Ordering::Equal)
                });
                // If we have the perfect solution then break early.
                if active_stack.last().unwrap().lazy_fitness.unwrap_or(0.0) == 1.0 {
                    break;
                }
                if i != n_epochs {
                    rng = self.epoch(&mut active_stack, rng);
                }
            }
            // Reverse the order of units such that the first unit is the
            // strongest candidate.
            while let Some(unit) = active_stack.pop() {
                self.units.push(unit.unit);
            }
        });
        self
    }
pub fn epochs_parallel ( & mut self , n_epochs : u32 , n_processes : u32 ) -> & mut Self { scope ( | scope | { let cvar_pair = Arc :: new ( ( Mutex :: new ( 0 ) , Condvar :: new ( ) ) ) ; let ( tx , rx ) = sync_channel ( 0 ) ; let process_queue = Arc :: new ( Mutex :: new ( rx ) ) ; let processed_stack = Arc :: new ( Mutex :: new ( Vec :: new ( ) ) ) ; for _ in 0 .. n_processes { let cvar_pair_clone = cvar_pair . clone ( ) ; let processed_stack_clone = processed_stack . clone ( ) ; let process_queue_clone = process_queue . clone ( ) ; scope . spawn ( move | | { let & ( ref lock , ref cvar ) = & * cvar_pair_clone ; loop { let mut l_unit : LazyUnit < T > = match process_queue_clone . lock ( ) . ok ( ) . unwrap ( ) . recv ( ) { Ok ( u ) => u , Err ( _ ) => return , } ; l_unit . fitness ( ) ; processed_stack_clone . lock ( ) . ok ( ) . unwrap ( ) . push ( l_unit ) ; { let mut processed = lock . lock ( ) . unwrap ( ) ; * processed += 1 ; cvar . notify_all ( ) ; } } } ) ; } let & ( ref lock , ref cvar ) = & * cvar_pair ; let mut active_stack = Vec :: new ( ) ; while let Some ( unit ) = self . units . pop ( ) { active_stack . push ( LazyUnit :: from ( unit ) ) ; } let seed : & [ _ ] = & [ self . seed ] ; let mut rng : StdRng = SeedableRng :: from_seed ( seed ) ; for i in 0 .. ( n_epochs + 1 ) { let jobs_total = active_stack . len ( ) ; while let Some ( unit ) = active_stack . pop ( ) { tx . send ( unit ) . unwrap ( ) ; } let mut jobs_processed = lock . lock ( ) . unwrap ( ) ; while * jobs_processed != jobs_total { jobs_processed = cvar . wait ( jobs_processed ) . unwrap ( ) ; } * jobs_processed = 0 ; // Swap the full processed_stack with the active stack. mem :: swap ( & mut active_stack , & mut processed_stack . lock ( ) . ok ( ) . unwrap ( ) ) ; // We want to sort such that highest fitness units are at the // end. active_stack . sort_by ( | a , b | { a . lazy_fitness . unwrap_or ( 0.0 ) . partial_cmp ( & b . lazy_fitness . unwrap_or ( 0.0 ) ) . unwrap_or ( Ordering :: Equal ) } ) ; // If we have the perfect solution then break early. if active_stack . last ( ) . unwrap ( ) . lazy_fitness . unwrap_or ( 0.0 ) == 1.0 { break ; } if i != n_epochs { rng = self . epoch ( & mut active_stack , rng ) ; } } // Reverse the order of units such that the first unit is the // strongest candidate. while let Some ( unit ) = active_stack . pop ( ) { self . units . push ( unit . unit ) ; } } ) ; self }
Runs a number of epochs where fitness is calculated across n parallel
processes.
Runs a number of epochs where fitness is calculated across n parallel processes .
&mut self
n_epochs
u32
n_processes
u32