rw_cell/

mpsc.rs

//!Implements the pattern "multiple producer single consumer"
//!
//! # Example
//!
//!```
//!
//! let (w0, mut r) = rw_cell::mpsc::new("Not good, but ok");
//! let w1 = r.writer();
//! let w2 = w0.clone();
//!
//! assert_eq!(r.read_with_is_new(), (&"Not good, but ok", false));
//!
//! w0.write("Not good");
//! assert_eq!(r.read_with_is_new(), (&"Not good", true));
//!
//! w1.write("ok");
//! assert_eq!(r.read(), &"ok");
//!
//! w2.write("But!!!");
//! assert_eq!(r.read(), &"But!!!");
//! ```

use std::sync::Arc;
use crate::option::OptionCell;

/// Struct for write data in cell with non-copy and non-lock
#[derive(Clone)]
pub struct Writer<T> {
    cell: Arc<OptionCell<T>>
}

impl<T> Writer<T> {
    fn new(cell: Arc<OptionCell<T>>) -> Self {
        Self {
            cell,
        }
    }

    /// Write value in cell
    ///
    /// # Examples
    ///
    /// ```
    /// let (w, mut r) = rw_cell::mpsc::new("Not good");
    /// w.write("Good");
    ///
    /// assert_eq!(r.read(), &"Good");
    /// ```
    pub fn write(&self, val: T) {
        self.cell.replace(val);
    }
}

/// Struct for read data from cell with non-copy and non-lock
pub struct Reader<T> {
    cell: Arc<OptionCell<T>>,
    val: T
}

impl<T> Reader<T> {
    fn new(cell: Arc<OptionCell<T>>) -> Self {

        Self {
            val: cell.take().unwrap(),
            cell,
        }
    }

    /// Return [`Writer`] for write data in cell
    pub fn writer(&self) -> Writer<T> {
        Writer::new(self.cell.clone())
    }

    /// Returns a tuple of value references and a boolean value, whether new or not
    ///
    /// # Examples
    ///
    /// ```
    /// let (w, mut r) = rw_cell::mpsc::new("Not good");
    /// assert_eq!(r.read_with_is_new(), (&"Not good", false));
    ///
    /// w.write("But ok");
    /// assert_eq!(r.read_with_is_new(), (&"But ok", true));
    /// ```
    pub fn read_with_is_new(&mut self) -> (&T, bool) {
        match self.cell.take() {
            None => (&self.val, false),
            Some(val) => {
                self.val = val;
                (&self.val, true)
            }
        }
    }

    /// Return a reference to the value from the cell
    pub fn read(&mut self) -> &T {
        match self.cell.take() {
            None => &self.val,
            Some(val) => {
                self.val = val;
                &self.val
            }
        }
    }
}

/// Create new cell with [`Writer`] and [`Reader`]
///
/// # Examples
///
/// ```
/// let (w, mut r) = rw_cell::mpsc::new("Not good");
/// assert_eq!(r.read(), &"Not good");
///
/// w.write("But ok");
/// assert_eq!(r.read_with_is_new(), (&"But ok", true));
/// ```
pub fn new<T>(val: T) -> (Writer<T>, Reader<T>) {
    let cell = Arc::new(OptionCell::new(val));
    (Writer::new(cell.clone()), Reader::new(cell))
}

pub fn default<T>() -> (Writer<T>, Reader<T>)
    where
        T: Default
{
    let cell = Arc::new(OptionCell::new(T::default()));
    (Writer::new(cell.clone()), Reader::new(cell))
}


#[cfg(test)]
mod test {
    use crate::mpsc;

    #[test]
    fn test_read() {
        let (w, mut r) = mpsc::new(vec!["fffff"; 1000]);
        assert_eq!(r.read(), &vec!["fffff"; 1000]);
        w.write(vec!["Not good, but ok"]);
        assert_eq!(r.read(), &vec!["Not good, but ok"]);
    }

    #[test]
    fn test_read_with_is_new() {
        let (w, mut r) = mpsc::new(vec!["fffff"; 1000]);
        assert_eq!(r.read_with_is_new(), (&vec!["fffff"; 1000], false));
        w.write(vec!["Not good"]);
        assert_eq!(r.read_with_is_new(), (&vec!["Not good"], true));
    }

    #[test]
    fn is_work() {
        let (w0, mut r) = mpsc::new(vec!["Ukraine"; 1000]);

        let w1 = r.writer();
        let w2 = r.writer();
        let w3 = r.writer();
        let w4 = r.writer();
        let w5 = r.writer();

        std::thread::spawn(move || loop {
            w0.write(vec!["Slovakia"; 1001])
        });

        std::thread::spawn(move || loop {
            w1.write(vec!["Estonia"; 1002])
        });

        std::thread::spawn(move || loop {
            w2.write(vec!["Czechia"; 1003])
        });

        std::thread::spawn(move || loop {
            w3.write(vec!["United Kingdom"; 1004])
        });

        std::thread::spawn(move || loop {
            w4.write(vec!["Lithuania"; 1004])
        });

        std::thread::spawn(move || loop {
            w5.write(vec!["Latvia"; 1004])
        });

        let mut slovakia        = 0;
        let mut estonia         = 0;
        let mut czechia         = 0;
        let mut united_kingdom  = 0;
        let mut lithuania       = 0;
        let mut latvia          = 0;
        let mut ukraine         = 0;

        for _ in 0..10000000usize {
            match r.read() {
                val if val.first() == Some(&"Slovakia")         => slovakia += 1,
                val if val.first() == Some(&"Estonia")          => estonia += 1,
                val if val.first() == Some(&"Czechia")          => czechia += 1,
                val if val.first() == Some(&"United Kingdom")   => united_kingdom += 1,
                val if val.first() == Some(&"Lithuania")        => lithuania += 1,
                val if val.first() == Some(&"Latvia")           => latvia += 1,
                val if val.first() == Some(&"Ukraine")          => ukraine += 1,
                val => println!("val: {:?}, Not good, but ok", val.first())
            }
        }

        println!("count Slovakia:       {}", slovakia);
        println!("count Estonia:        {}", estonia);
        println!("count Czechia:        {}", czechia);
        println!("count United Kingdom: {}", united_kingdom);
        println!("count Lithuania:      {}", lithuania);
        println!("count Latvia:         {}", latvia);
        println!("count Ukraine:        {}", ukraine);
    }

    // #[test]
    // fn test() {
    //     let read_count = 1_000_000usize;
    //
    //     let rw_cell_avg_time = test_rw_cell(read_count);
    //     let rw_lock_avg_time = test_rwlock(read_count);
    //
    //     println!("rw_cell_avg_time: {rw_cell_avg_time}");
    //     println!("rw_lock_avg_time: {rw_lock_avg_time}");
    // }
    //
    // fn test_rwlock(read_count: usize) -> u128 {
    //     let lock_cell0 = Arc::new(RwLock::new(vec!["fffff"; 1000]));
    //
    //     let lock_cell1 = lock_cell0.clone();
    //     let lock_cell2 = lock_cell0.clone();
    //     let lock_cell3 = lock_cell0.clone();
    //     let lock_cell4 = lock_cell0.clone();
    //     let lock_cell5 = lock_cell0.clone();
    //     let lock_cell6 = lock_cell0.clone();
    //
    //     std::thread::spawn(move || loop {
    //         *lock_cell0.write().unwrap() = vec!["fffff"; 1001];
    //     });
    //
    //     std::thread::spawn(move || loop {
    //         *lock_cell1.write().unwrap() = vec!["fffff"; 1002]
    //     });
    //
    //     std::thread::spawn(move || loop {
    //         *lock_cell2.write().unwrap() = vec!["fffff"; 1003]
    //     });
    //
    //     std::thread::spawn(move || loop {
    //         *lock_cell3.write().unwrap() = vec!["fffff"; 1004]
    //     });
    //
    //     std::thread::spawn(move || loop {
    //         *lock_cell4.write().unwrap() = vec!["fffff"; 1005]
    //     });
    //
    //     std::thread::spawn(move || loop {
    //         *lock_cell5.write().unwrap() = vec!["fffff"; 1006]
    //     });
    //
    //     let mut times = vec![];
    //
    //     for _ in 0..read_count {
    //         let st = Instant::now();
    //         lock_cell6.read().unwrap().len();
    //         times.push(st.elapsed().as_nanos());
    //     }
    //     times.into_iter().sum::<u128>() / read_count as u128
    // }
    //
    // fn test_rw_cell(read_count: usize) -> u128 {
    //     let (w0, mut r) = mwsr::cell(vec!["fffff"; 1000]);
    //
    //     let w1 = w0.clone();
    //     let w2 = w0.clone();
    //     let w3 = w0.clone();
    //     let w4 = w0.clone();
    //     let w5 = w0.clone();
    //
    //     std::thread::spawn(move || loop {
    //         w0.write(vec!["fffff"; 1001])
    //     });
    //
    //     std::thread::spawn(move || loop {
    //         w1.write(vec!["fffff"; 1002])
    //     });
    //
    //     std::thread::spawn(move || loop {
    //         w2.write(vec!["fffff"; 1003])
    //     });
    //
    //     std::thread::spawn(move || loop {
    //         w3.write(vec!["fffff"; 1004])
    //     });
    //
    //     std::thread::spawn(move || loop {
    //         w4.write(vec!["fffff"; 1005])
    //     });
    //
    //     std::thread::spawn(move || loop {
    //         w5.write(vec!["fffff"; 1006])
    //     });
    //
    //     let mut times = vec![];
    //
    //     for _ in 0..read_count {
    //         let st = Instant::now();
    //         r.read().len();
    //         times.push(st.elapsed().as_nanos());
    //     }
    //
    //     times.into_iter().sum::<u128>() / read_count as u128
    // }
}