Function permutator::cartesian_product_sync

source · [−]

pub fn cartesian_product_sync<'a, T>(
    sets: &'a [&[T]], 
    result: Arc<RwLock<Vec<&'a T>>>, 
    cb: impl FnMut()
)

Expand description

Similar to safe cartesian_product function except the way it return the product. It return result through Rc<RefCell<>> to mutable slice of result. It’ll notify caller on each new result via empty callback function.

Parameters

sets A raw sets of data to get a cartesian product.
result An Rc<RefCell<>> contains mutable slice of length equals to sets.len()
cb A callback function which will be called after new product in result is set.

Return

This function return result through function’s parameter result and notify caller that new result is available through cb callback function.

Rationale

The safe cartesian product function return value in callback parameter. It limit the lifetime of return combination to be valid only inside it callback. To use it outside callback scope, it need to copy the value which will have performance penalty. Therefore, jeopardize it own goal of being fast. This function provide alternative safe way to share result which is roughly 50% slower to unsafe counterpart. The performance is on roughly 15%-20% slower than CartesianProduct iterator in uncontrol test environment.

Example

The scenario is we want to get cartesian product from single source of data then distribute the product to two workers which read each combination then do something about it, which in this example, simply print it.

   use std::thread;
   use std::sync::{Arc, RwLock};
   use std::sync::mpsc;
   use std::sync::mpsc::{Receiver, SyncSender};
   use permutator::cartesian_product_sync;
  
   fn start_cartesian_product_process<'a>(data : &'a[&[i32]], cur_result : Arc<RwLock<Vec<&'a i32>>>, notifier : Vec<SyncSender<Option<()>>>, release_recv : Receiver<()>) {
       use std::time::Instant;
       let timer = Instant::now();
       let mut counter = 0;
       cartesian_product_sync(data, cur_result, || {
           notifier.iter().for_each(|n| {
               n.send(Some(())).unwrap(); // notify every thread that new data available
           });

           for _ in 0..notifier.len() {
               release_recv.recv().unwrap(); // block until all thread reading data notify on read completion
           }

           counter += 1;
       });

       notifier.iter().for_each(|n| {n.send(None).unwrap()}); // notify every thread that there'll be no more data.

       println!("Done {} combinations with 2 workers in {:?}", counter, timer.elapsed());
   }
   let k = 7;
   let data : &[&[i32]]= &[&[1, 2, 3], &[4, 5], &[6]];
   let result = vec![&data[0][0]; k];
   let result_sync = Arc::new(RwLock::new(result));

   // workter thread 1
   let (t1_send, t1_recv) = mpsc::sync_channel::<Option<()>>(0);
   let (main_send, main_recv) = mpsc::sync_channel(0);
   let t1_local = main_send.clone();
   let t1_dat = Arc::clone(&result_sync);
   thread::spawn(move || {
       while let Some(_) = t1_recv.recv().unwrap() {
           let result : &Vec<&i32> = &*t1_dat.read().unwrap();
           // println!("Thread1: {:?}", result);
           t1_local.send(()).unwrap(); // notify generator thread that reference is no longer need.
       }
       println!("Thread1 is done");
   });
 
   // worker thread 2
   let (t2_send, t2_recv) = mpsc::sync_channel::<Option<()>>(0);
   let t2_dat = Arc::clone(&result_sync);
   let t2_local = main_send.clone();
   thread::spawn(move || {
       while let Some(_) = t2_recv.recv().unwrap() {
           let result : &Vec<&i32> = &*t2_dat.read().unwrap();
           // println!("Thread2: {:?}", result);
           t2_local.send(()).unwrap(); // notify generator thread that reference is no longer need.
       }
       println!("Thread2 is done");
   });

   // main thread that generate result
   thread::spawn(move || {
       start_cartesian_product_process(data, result_sync, vec![t1_send, t2_send], main_recv);
   }).join().unwrap();

See

cartesian_product function