rw_cell/broadcast/

cloneable.rs

//! This approach uses [`Mutex::lock`] when calling the [`Writer::share`],
//! [`Writer::subscribe`] and [`Reader::clone`] methods
//!
//! # Example
//!
//!```
//! let (mut tx, rx) = rw_cell::broadcast::cloneable::new("Not good, but ok");
//!
//! assert_eq!(tx.read(), &"Not good, but ok");
//!
//! let mut rx1 = tx.subscribe();
//! let mut rx2 = rx.clone();
//!
//! assert_eq!(rx1.read_with_is_new(), (&"Not good, but ok", false));
//! assert_eq!(rx2.read_with_is_new(), (&"Not good, but ok", false));
//!
//! tx.share("Not good");
//! assert_eq!(rx1.read_with_is_new(), (&"Not good", true));
//! assert_eq!(rx2.read_with_is_new(), (&"Not good", true));
//!
//! tx.share("ok");
//! assert_eq!(rx1.read(), &"ok");
//! assert_eq!(rx2.read(), &"ok");
//! ```

use std::sync::{Arc, Mutex};
use crate::option::OptionCell;

type Cells<T> = Arc<Mutex<Vec<Cell<T>>>>;
type Cell<T> = Arc<OptionCell<Arc<T>>>;

/// Distribute value between all readers
pub struct Writer<T> {
    cells: Cells<T>,
    val: Arc<T>,
}

impl<T> Writer<T> {
    /// Create new distributor
    fn new(val: T) -> Writer<T> {
        Self {
            cells: Arc::new(Mutex::new(vec![])),
            val: Arc::new(val),
        }
    }

    /// Write val and distribute it between all readers
    ///
    /// # Example
    ///
    /// ```
    /// let (mut tx, rx) = rw_cell::broadcast::cloneable::new("Not good, but ok");
    /// let mut rx1 = tx.subscribe();
    /// let mut rx2 = rx.clone();
    ///
    /// tx.share("Not good");
    ///
    /// assert_eq!(rx1.read_with_is_new(), (&"Not good", true));
    /// assert_eq!(rx2.read_with_is_new(), (&"Not good", true));
    /// ```
    pub fn share(&mut self, val: T) {
        let mut cells = self.cells
            .lock()
            .unwrap();

        self.val = Arc::new(val);

        cells.retain(|cell| {
            let is_single_ref = Arc::strong_count(cell) != 1;
            if is_single_ref {
                cell.replace(self.val.clone());
            }

            is_single_ref
        });
    }

    /// Return a reference to the value from the cell
    ///
    /// # Example
    ///
    /// ```
    /// let (mut tx, rx) = rw_cell::broadcast::cloneable::new("Not good, but ok");
    ///
    /// assert_eq!(tx.read(), &"Not good, but ok");
    /// ```
    pub fn read(&self) -> &T {
        &self.val
    }

    /// Return new [`Reader`] for read data from cell
    pub fn subscribe(&mut self) -> Reader<T>  {
        let cell = Arc::new(OptionCell::new(self.val.clone()));

        self.cells
            .lock()
            .unwrap()
            .push(cell.clone());
        Reader::new(self.cells.clone(), cell)
    }
}

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

impl<T> Clone for Reader<T> {
    fn clone(&self) -> Self {
        let mut cells = self.cells
            .lock()
            .unwrap();

        let self_cell_ptr = &raw const *self.cell;

        let self_cell_idx = cells
            .iter()
            .enumerate()
            .find_map(|(idx, cell)| (self_cell_ptr == &raw const **cell).then_some(idx))
            .expect("Undefined self cell in cells");

        let clone_cell = if let Some(val) = cells[self_cell_idx].take() {
            cells[self_cell_idx].replace(val.clone());
            Arc::new(OptionCell::new(val))
        } else {
            Arc::new(OptionCell::new(self.val.clone()))
        };
        cells.push(clone_cell.clone());

        Reader::new(self.cells.clone(), clone_cell)
    }
}

impl<T> Reader<T> {
    fn new(cells: Cells<T>, cell: Cell<T>) -> Reader<T> {
        Self {
            val: cell.take().unwrap(),
            cells,
            cell,
        }
    }

    /// Returns a tuple of value references and a boolean value, whether new or not
    ///
    /// # Examples
    ///
    /// ```
    /// let (mut tx, mut rx) = rw_cell::broadcast::cloneable::new("Not good, but ok");
    ///
    /// assert_eq!(rx.read_with_is_new(), (&"Not good, but ok", false));
    ///
    /// tx.share("Not good");
    ///
    /// assert_eq!(rx.read_with_is_new(), (&"Not good", 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
    ///
    /// # Examples
    ///
    /// ```
    /// let (mut tx, mut rx) = rw_cell::broadcast::cloneable::new("Not good, but ok");
    ///
    /// assert_eq!(rx.read(), &"Not good, but ok");
    ///
    /// tx.share("ok");
    ///
    /// assert_eq!(rx.read(), &"ok");
    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 (mut tx, mut rx) = rw_cell::broadcast::cloneable::new("Not good");
/// assert_eq!(rx.read(), &"Not good");
///
/// tx.share("But ok");
/// assert_eq!(rx.read_with_is_new(), (&"But ok", true));
/// ```
pub fn new<T>(val: T) -> (Writer<T>, Reader<T>) {
    let mut tx = Writer::new(val);
    let rx = tx.subscribe();
    (tx, rx)
}

pub fn default<T>() -> (Writer<T>, Reader<T>)
where
    T: Default
{
    let mut tx = Writer::new(T::default());
    let rx = tx.subscribe();
    (tx, rx)
}


#[cfg(test)]
mod test {
    use std::thread::JoinHandle;
    use crate::broadcast;
    use crate::broadcast::cloneable::{Writer, Reader};

    #[test]
    fn test_remove_reader() {
        let (mut tx, rx) = broadcast::cloneable::new("Good");

        let _rx1 = tx.subscribe();
        let _rx2 = rx.clone();

        assert_eq!(tx.cells.lock().unwrap().len(), 3);
        drop(_rx1);

        tx.share("Only one");

        assert_eq!(tx.cells.lock().unwrap().len(), 2);
    }

    #[test]
    fn test_clone() {
        let (mut tx, mut rx0) = broadcast::cloneable::new("Good");
        let mut rx1 = rx0.clone();

        assert_eq!(rx1.read(), &"Good");

        let mut rx2 = rx0.clone();
        let mut rx3 = rx1.clone();

        tx.share("Ok");
        let mut rx4 = rx3.clone();
        assert_eq!(rx0.read(), &"Ok");
        assert_eq!(rx1.read(), &"Ok");
        assert_eq!(rx2.read(), &"Ok");
        assert_eq!(rx3.read(), &"Ok");
        assert_eq!(rx4.read(), &"Ok");
    }

    #[test]
    fn test_write_read() {
        let mut tx = Writer::new("Not good, but ok");
        let mut rx1 = tx.subscribe();
        let mut rx2 = tx.subscribe();

        assert_eq!(tx.read(), &"Not good, but ok");

        assert_eq!(rx1.read_with_is_new(), (&"Not good, but ok", false));
        assert_eq!(rx2.read_with_is_new(), (&"Not good, but ok", false));

        tx.share("Not good");
        assert_eq!(rx1.read_with_is_new(), (&"Not good", true));
        assert_eq!(rx2.read_with_is_new(), (&"Not good", true));

        tx.share("ok");
        let mut rx3 = rx2.clone();
        assert_eq!(rx1.read(), &"ok");
        assert_eq!(rx2.read(), &"ok");
        assert_eq!(rx3.read(), &"ok");
    }

    #[test]
    fn is_work() -> Result<(), Box<dyn std::any::Any + Send + 'static>> {
        let (mut tx, rx0) = broadcast::cloneable::new(vec!["Ukraine".to_string(); 1000]);

        let rx1 = rx0.clone();
        let rx2 = tx.subscribe();
        let rx3 = tx.subscribe();

        std::thread::spawn(move || loop {
            for i in 0usize.. {
                match i % 6 {
                    0 => tx.share(vec!["Slovakia".to_string(); 1001]),
                    1 => tx.share(vec!["Estonia".to_string(); 1001]),
                    2 => tx.share(vec!["Czechia".to_string(); 1001]),
                    3 => tx.share(vec!["United Kingdom".to_string(); 1001]),
                    4 => tx.share(vec!["Lithuania".to_string(); 1001]),
                    5 => tx.share(vec!["Latvia".to_string(); 1001]),
                    val => println!("val: {:?}, Not good, but ok", val)
                }
            }
        });
        let count_iter = 1_000_000;
        let h0 = create_thread_read(rx0, count_iter);
        let h1 = create_thread_read(rx1, count_iter);
        let h2 = create_thread_read(rx2, count_iter);
        let h3 = create_thread_read(rx3, count_iter);

        let res0 = h0.join()?;
        let res1 = h1.join()?;
        let res2 = h2.join()?;
        let res3 = h3.join()?;

        println!("Slovakia:         {}", res0.0 + res1.0 + res2.0 + res3.0);
        println!("Estonia:          {}", res0.1 + res1.1 + res2.1 + res3.1);
        println!("Czechia:          {}", res0.2 + res1.2 + res2.2 + res3.2);
        println!("United Kingdom:   {}", res0.3 + res1.3 + res2.3 + res3.3);
        println!("Lithuania:        {}", res0.4 + res1.4 + res2.4 + res3.4);
        println!("Latvia:           {}", res0.5 + res1.5 + res2.5 + res3.5);
        println!("Ukraine:          {}", res0.6 + res1.6 + res2.6 + res3.6);

        Ok(())
    }

    fn create_thread_read(
        mut rw: Reader<Vec<String>>,
        count_iter: usize
    ) -> JoinHandle<(i32, i32, i32, i32, i32, i32, i32)> {
        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;

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