lib.rs

#![allow(dead_code)]
/// reader-writer lock
/// written by Feng Wang 


use std::sync::Mutex;
use std::sync::Condvar;
use std::cell::UnsafeCell;
use std::ops::{Deref, DerefMut};
//use std::sync::Arc;
use std::rc::Rc;
//use std::{thread, time};
/// Provides a reader-writer lock to protect data of type `T`

pub struct RwLock<T> {
    mutex: Mutex<()>,
    data: UnsafeCell<T>,
    // Hao Chen's requirements
    pref: Preference,
    order: Order,
    // conditional variables 
    //read_go: Condva,
    //write_go: Condvar,
    // state variables
    state_vars: UnsafeCell<Vars>,
}

struct Vars {

    // state variables
    waiting_readers: Vec<Rc<Box<Condvar>>>,
    waiting_writers: Vec<Rc<Box<Condvar>>>,
    active_readers: u32,
    active_writers: u32,
}


#[derive(PartialEq)]
pub enum Preference {
    /// Readers-preferred
    /// * Readers must wait when a writer is active.
    /// * Writers must wait when a reader is active or waiting, or a writer is active.
    Reader,
    /// Writers-preferred: 
    /// * Readers must wait when a writer is active or waiting.
    /// * Writer must wait when a reader or writer is active.
    Writer,
}

/// In which order to schedule threads
pub enum Order {
    /// First in first out
    Fifo,
    /// Last in first out
    Lifo,
}

impl<T> RwLock<T> {
    /// Constructs a new `RwLock`
    ///
    /// data: the shared object to be protected by this lock
    /// pref: which preference
    /// order: in which order to wake up the threads waiting on this lock
    pub fn new(data: T, pref: Preference, order: Order) -> RwLock<T> {

        RwLock{
            mutex: Mutex::new(()),
            data: UnsafeCell::new(data),
            pref: pref,
            order: order,
            //read_go: Condvar::new(),
            //write_go: Condvar::new(),
            state_vars: UnsafeCell::new(
                Vars {
                    waiting_readers: Vec::new(),
                    waiting_writers: Vec::new(),
                    active_writers: 0,
                    active_readers: 0,

                })

        }
    }
    // when should read wait 
    fn read_should_wait(&self) -> bool {
        let  state_vars = self.state_vars.get();// *mut T 
        let  waiting_writers = unsafe{ &((*state_vars).waiting_writers) };// vector
        let  active_writers = unsafe { (*state_vars).active_writers };


        match self.pref {
            Preference::Reader => {
                
                return active_writers > 0;
            },
            Preference::Writer => {

                return active_writers > 0 || waiting_writers.len() > 0;
            }
        }

    }
    // when shold write wait 
    fn write_should_wait(&self) -> bool {
        let state_vars = self.state_vars.get();// *mut Vars
        //let waiting_writers = unsafe{ &((*state_vars).waiting_writers)};// vector
        let waiting_readers = unsafe{ &((*state_vars).waiting_readers) };// vector
        let active_readers = unsafe{ (*state_vars).active_readers };
        let active_writers = unsafe{ (*state_vars).active_writers };
        match self.pref {
            Preference::Reader => {
                return active_readers > 0 || waiting_readers.len() > 0 
                            || active_writers > 0;
            },
            Preference::Writer => {
                //if active_writers > 0 || active_readers > 0 {

                    //println!("write_should_wait satisfied");
                //}
                return active_writers > 0 || active_readers > 0;
            }
        }
    }

    /// Requests a read lock, waits when necessary, and wakes up at the earliest opportunity
    /// 
    /// Always returns Ok(_).
    /// (We declare this return type to be `Result` to be compatible with `std::sync::RwLock`)
    pub fn read(&self) -> Result<RwLockReadGuard<T>, ()> {
        let mut guard = self.mutex.lock().unwrap();
        let cv = Rc::new(Box::new(Condvar::new())); // point to a condvar
        unsafe {
            (*self.state_vars.get()).waiting_readers.push(cv.clone());
        }
        
        while self.read_should_wait() {
            guard = cv.wait(guard).unwrap();
        }
        
        unsafe {

            //(*self.state_vars.get()).waiting_readers.pop();
            match self.order {
                Order::Fifo => {
                    (*self.state_vars.get()).waiting_readers.remove(0);
                },
                Order::Lifo => {
                    (*self.state_vars.get()).waiting_readers.pop();
                }
            }

            (*self.state_vars.get()).active_readers += 1;

        }

        Ok(
            RwLockReadGuard {
                lock: &self
            }
        )
        
    }

    /// Requests a write lock, and waits when necessary.
    /// When the lock becomes available,
    /// * if `order == Order::Fifo`, wakes up the first thread
    /// * if `order == Order::Lifo`, wakes up the last thread
    /// 
    /// Always returns Ok(_).
    pub fn write(&self) -> Result<RwLockWriteGuard<T>, ()> {

        //println!("before guard");
        let mut guard = self.mutex.lock().unwrap();
        //println!("after guard");
        let cv = Rc::new(Box::new(Condvar::new()));
        unsafe {
            (*self.state_vars.get()).waiting_writers.push(cv.clone());
        }
 
        // unsafe {
        //     println!("the length of waiting list = {}", (*self.state_vars.get()).waiting_writers.len());
        // }
        //println!("before writer should wait");
        while self.write_should_wait() {
            //println!("----  waiting  -----");
            guard = cv.wait(guard).unwrap();
            
        }
        

        // pop the writer from the waiting list
        // pop wrong !
        unsafe {
            match self.order {
                Order::Fifo => {
                    (*self.state_vars.get()).waiting_writers.remove(0);
                },
                Order::Lifo => {
                    (*self.state_vars.get()).waiting_writers.pop();
                }
            }
            
            (*self.state_vars.get()).active_writers += 1;
        }

        
        Ok(
            RwLockWriteGuard {
                lock: &self
            }
        )

    }

    pub fn wakeup_other_threads(&self) {
        //unsafe { 
        //    println!("{:?}", (*self.state_vars.get()).waiting_writers);
        //}
        match self.pref {
            Preference::Reader => {
                // firstly weak up readers
                unsafe {
                    let ref mut waiting_readers =  (*self.state_vars.get()).waiting_readers;// vector
                    let ref mut waiting_writers =  (*self.state_vars.get()).waiting_writers;
                    match self.order {
                        Order::Lifo => {
                     
                            if waiting_readers.len() > 0 {
                                
                                let len = waiting_readers.len();
                                for i in 0..len {
                                    waiting_readers[len-1-i].notify_one();
                                }
                                
                            }else if waiting_writers.len() > 0 {
                                 
                                waiting_writers[waiting_writers.len()-1].notify_one();
                                
                            }

                        },
                        Order::Fifo => {
                            if waiting_readers.len() > 0 {
                                
                                let len = waiting_readers.len();
                                for i in 0..len {
                                    waiting_readers[i].notify_one();
                                }
                                

                            }else if waiting_writers.len() > 0 {
                                waiting_writers[0].notify_one();
                                
                            }
                        },
                    }
                }            
            },
            Preference::Writer => {
                // firstly weak up writers
                unsafe {
                    let ref mut waiting_writers = (*self.state_vars.get()).waiting_writers;// vector
                    let ref mut waiting_readers = (*self.state_vars.get()).waiting_readers;// vector

                    match self.order {
                        Order::Lifo => {
                            
                            if waiting_writers.len() > 0 {
                                waiting_writers[waiting_writers.len()-1].notify_one();
                                
                            }else if waiting_readers.len() > 0 {
                                let len = waiting_readers.len();
                                for i in 0..len {
                                    waiting_readers[len-1-i].notify_one();
                                }                     
                                         
                            }
                        },
                        Order::Fifo => {
                            if waiting_writers.len() > 0 {
                                //println!("&&& {} writers are waiting", waiting_writers.len());
                                waiting_writers[0].notify_one();
                                //println!("&&&& wake one writer &&&&");
                                
                            }else if waiting_readers.len() > 0 {
                                let len = waiting_readers.len();
                                for i in 0..len {
                                    waiting_readers[i].notify_one();
                                }  
                                                            
                            }                            
                        },
                    }                           
                }
            }
        }
    }
    // function end 
}

/// Declares that it is safe to send and reference `RwLock` between threads safely
unsafe impl<T: Send + Sync> Send for RwLock<T> {}
unsafe impl<T: Send + Sync> Sync for RwLock<T> {}

/// A read guard for `RwLock`
pub struct RwLockReadGuard<'a, T: 'a> {
    lock: &'a RwLock<T>
}

/// Provides access to the shared object
impl<'a, T> Deref for RwLockReadGuard<'a, T> {
    type Target = T;
    fn deref(&self) -> &T {
        unsafe {
            &*self.lock.data.get()
        }
    }
    
}

/// Releases the read lock
impl<'a, T> Drop for RwLockReadGuard<'a, T> {

    fn drop(&mut self) {
        let guard = self.lock.mutex.lock().unwrap();
        unsafe { 
            if (*self.lock.state_vars.get()).active_readers > 0 {
                (*self.lock.state_vars.get()).active_readers -= 1;
            }
            self.lock.wakeup_other_threads();
        }
        
    }

}

/// A write guard for `RwLock`
pub struct RwLockWriteGuard<'a, T: 'a> {
    lock : &'a  RwLock<T>
}

/// Provides access to the shared object
impl<'a, T> Deref for RwLockWriteGuard<'a, T> {
    type Target = T;

    fn deref(&self) -> &T {
        unsafe { 
            &*self.lock.data.get() 
        }
    }
    
}

/// Provides access to the shared object
impl<'a, T> DerefMut for RwLockWriteGuard<'a, T> {

    fn deref_mut(&mut self) -> &mut T {
        unsafe { 
            &mut *self.lock.data.get() 
        }
    }
}

/// Releases the write lock
impl<'a, T> Drop for RwLockWriteGuard<'a, T> {
    fn drop(&mut self) {
        //println!("***** drop beginning  *****"); 
        let guard = self.lock.mutex.lock().unwrap();
        unsafe { 

            if (*self.lock.state_vars.get()).active_writers > 0 {
                //println!("**** in active_writers -- ***");
                (*self.lock.state_vars.get()).active_writers -= 1;
            }

            assert_eq!((*self.lock.state_vars.get()).active_writers, 0);
            //if (*self.lock.state_vars.get()).active_writers == 0 
            //println!("**** before wake up others ****");
            self.lock.wakeup_other_threads();
            //println!("**** after wake up others ****");

        }  
        //println!("***** drop finished  *****"); 
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::mpsc::channel;
    use std::thread;
    use std::sync::Arc;
    #[test]
    fn test_rw_arc() {
        let arc = Arc::new(RwLock::new(0, Preference::Reader, Order::Fifo));
        let arc2 = arc.clone();
        let (tx, rx) = channel();

        thread::spawn(move || {
            println!("writer spawned!");
            let mut lock = arc2.write().unwrap();
            println!("writer active!");
            for _ in 0..10 {
                let tmp = *lock;
                *lock = -1;
                thread::yield_now();
                *lock = tmp + 1;
            }
            tx.send(()).unwrap();
            println!("writer inactive!")
        });

        // Readers try to catch the writer in the act
        let mut children = Vec::new();
        for _ in 0..5 {
            let arc3 = arc.clone();
            children.push(thread::spawn(move || {
                println!("reader spawned!");
                let lock = arc3.read().unwrap();
                println!("reader active!");
                println!("{}",*lock >= 0);
                assert!(*lock >= 0);

                println!("reader inactive!")
            }));
        }

        // Wait for children to pass their asserts
        for r in children {

            assert!(r.join().is_ok());
        }

        // Wait for writer to finish
        rx.recv().unwrap();
        let lock = arc.read().unwrap();
        assert_eq!(*lock, 10);
    }
    #[test]
    fn test_multi_writers() {
        let arc = Arc::new(RwLock::new(0, Preference::Writer, Order::Fifo));
        
   
        let mut threads = Vec::new();
        for i in 0..3 {
            let arc2 = arc.clone();
            threads.push(thread::spawn( move || {
                println!("writer {} spawned!",i);

                let mut lock = arc2.write().unwrap();
                println!("writer {} active!", i);
                *lock += 2;
                println!("data = {} now in thread {}", *lock,i);
                println!("writer {} end!", i);
            }));             
        }
        for t in threads {
            assert!(t.join().is_ok());
        }
        //println!("new data = {:?}", arc.data);
       
    }
}