use crate::utility::list::List; use crate::kernel::thread::Thread; use crate::simulator::interrupt::InterruptStatus::InterruptOff; use crate::simulator::machine::Machine; use std::cell::RefCell; use std::rc::Rc; use super::scheduler::Scheduler; use super::system::System; use super::thread_manager::ThreadManager; /// Structure of a Semaphore used for synchronisation pub struct Semaphore { /// Counter of simultanous Semaphore counter:i32, /// QUeue of Semaphore waiting to be exucated waiting_queue:List>>, /// Thread manager which managing threads thread_manager: Rc> } impl Semaphore { /// Initializes a semaphore, so that it can be used for synchronization. /// /// ### Parameters /// - *counter* initial value of counter /// - *thread_manager* Thread manager which managing threads pub fn new(counter: i32, thread_manager: Rc>) -> Semaphore{ Semaphore { counter, waiting_queue: List::new(), thread_manager} } /// Decrement the value, and wait if it becomes < 0. Checking the /// value and decrementing must be done atomically, so we /// need to disable interrupts before checking the value. /// /// Note that thread_manager::thread_sleep assumes that interrupts are disabled /// when it is called. /// /// ### Parameters TODO Refaire /// - *current_thread* the current thread /// - *machine* the machine where the threads are executed pub fn p(&mut self, system: &mut System) { let old_status = system.get_machine().interrupt.set_status(InterruptOff); self.counter -= 1; if self.counter < 0 { match system.get_thread_manager().get_g_current_thread() { Some(thread) => { self.waiting_queue.push(Rc::clone(thread)); system.get_thread_manager().thread_sleep(system, Rc::clone(thread)); }, None => unreachable!("Current thread should not be None") } } system.get_machine().interrupt.set_status(old_status); } /// Increment semaphore value, waking up a waiting thread if any. /// As with P(), this operation must be atomic, so we need to disable /// interrupts. /// /// scheduler::ready_to_run() assumes that interrupts /// are disabled when it is called. /// /// ### Parameters /// - **machine** the machine where the threads are executed /// - **scheduler** the scheduler which determine which thread to execute pub fn v(&mut self, system: &mut System){ let old_status = system.get_machine().interrupt.set_status(InterruptOff); self.counter += 1; if self.waiting_queue.peek() != None { system.get_thread_manager().g_scheduler.ready_to_run(self.waiting_queue.pop().unwrap()); } system.get_machine().interrupt.set_status(old_status); } } /// Lock used for synchronisation, can be interpreted has a Semaphore with a /// counter of 1 /// It's used for critical parts pub struct Lock{ /// Thread owning the lock owner: Option>>, /// The queue of threads waiting for execution waiting_queue:List>>, /// Thread manager which managing threads thread_manager: Rc>, /// A boolean definig if the lock is free or not free: bool } impl Lock { /// Initialize a Lock, so that it can be used for synchronization. /// The lock is initialy free /// /// ### Parameters /// - **thread_manager** Thread manager which managing threads pub fn new(thread_manager: Rc>) -> Lock { Lock { owner: None, waiting_queue: List::new(), thread_manager, free: true } } /// Wait until the lock become free. Checking the /// state of the lock (free or busy) and modify it must be done /// atomically, so we need to disable interrupts before checking /// the value of free. /// /// Note that thread_manager::thread_seep assumes that interrupts are disabled /// when it is called. /// /// ### Parameters /// - **current_thread** the current thread /// - **machine** the machine where the threads are executed pub fn acquire(&mut self, system: &mut System) { let old_status = system.get_machine().interrupt.set_status(InterruptOff); if self.free { self.free = false; self.owner = Option::Some(match system.get_thread_manager().get_g_current_thread() { Some(th) => { Rc::clone(th) }, None => unreachable!() }); } else { let t = system.get_thread_manager().get_g_current_thread(); match t { Some(x) => { let x = Rc::clone(x); self.waiting_queue.push(Rc::clone(&x)); system.thread_sleep(Rc::clone(&x)); }, None => unreachable!("Current thread should not be None") } } system.get_machine().interrupt.set_status(old_status); } /// Wake up a waiter if necessary, or release it if no thread is waiting. /// We check that the lock is held by the g_current_thread. /// As with Acquire, this operation must be atomic, so we need to disable /// interrupts. scheduler::ready_to_run() assumes that threads /// are disabled when it is called. /// /// ### Parameters /// - **machine** the machine where the code is executed /// - **scheduler** the scheduler which determine which thread to execute pub fn release(&mut self, system: &mut System) { let old_status = system.get_machine().interrupt.set_status(InterruptOff); match system.get_thread_manager().get_g_current_thread() { Some(thread) => { if self.held_by_current_thread(system) { if self.waiting_queue.peek() != None { self.owner = Some(self.waiting_queue.pop().unwrap()); match &self.owner { Some(x) => system.get_thread_manager().g_scheduler.ready_to_run(Rc::clone(&x)), None => () } } else { self.free = true; self.owner = None; } } } None => () } system.get_machine().interrupt.set_status(old_status); } pub fn held_by_current_thread(&mut self, system: &mut System) -> bool { match &self.owner { Some(x) => match system.get_thread_manager().get_g_current_thread() { Some(thread) => Rc::ptr_eq(&x, thread), None => false } None => false } } } /// Structure of a condition used for synchronisation pub struct Condition{ /// The queue of threads waiting for execution waiting_queue:List>>, /// Thread manager which managing threads thread_manager: Rc>, } impl Condition { /// Initializes a Condition, so that it can be used for synchronization. /// /// ### Parameters /// - *thread_manager* Thread manager which managing threads pub fn new(thread_manager: Rc>) -> Condition { Condition{ waiting_queue: List::new(), thread_manager } } /// Block the calling thread (put it in the wait queue). /// This operation must be atomic, so we need to disable interrupts. /// /// ### Parameters /// - **current_thread** the current thread /// - **machine** the machine where threads are executed pub fn wait(&mut self, current_thread: Rc>, machine: &mut Machine, system: &mut System) { let old_status = machine.interrupt.set_status(InterruptOff); self.waiting_queue.push(Rc::clone(¤t_thread)); self.thread_manager.borrow_mut().thread_sleep(system, current_thread); machine.interrupt.set_status(old_status); } /// Wake up the first thread of the wait queue (if any). /// This operation must be atomic, so we need to disable interrupts. /// /// ### Parameters /// - **machine** the machine where the code is executed /// - **scheduler** the scheduler which determine which thread to execute pub fn signal(&mut self, machine: &mut Machine, scheduler: &mut Scheduler) { let old_status = machine.interrupt.set_status(InterruptOff); if self.waiting_queue.peek() != None { scheduler.ready_to_run(self.waiting_queue.pop().unwrap()); } machine.interrupt.set_status(old_status); } /// Wake up all threads waiting in the waitqueue of the condition /// This operation must be atomic, so we need to disable interrupts. /// /// ### Parameters /// - **machine** the machine where the code is executed /// - **scheduler** the scheduler which determine which thread to execute pub fn broadcast(&mut self, machine: &mut Machine, scheduler: &mut Scheduler) { let old_status = machine.interrupt.set_status(InterruptOff); while self.waiting_queue.peek() != None { scheduler.ready_to_run(self.waiting_queue.pop().unwrap()); } machine.interrupt.set_status(old_status); } } #[cfg(test)] mod test { use std::{rc::Rc, cell::RefCell}; use crate::{kernel::{thread::Thread, synch::{Semaphore, Lock}}, init_system, simulator::machine::Machine}; #[test] fn test_semaphore_single() { // Init let system = init_system!(); let mut semaphore = Semaphore::new(1, Rc::clone(&system.borrow_mut().get_thread_manager())); let thread = Rc::new(RefCell::new(Thread::new("test_semaphore"))); // P semaphore.p(thread, Rc::clone(&system)); assert_eq!(semaphore.counter, 0); assert!(semaphore.waiting_queue.is_empty()); // V semaphore.v(Rc::clone(&system)); assert_eq!(semaphore.counter, 1); assert!(semaphore.waiting_queue.is_empty()); } #[test] #[ignore] fn test_semaphore_multiple() { // Init let system = init_system!(); let tm = system.borrow_mut().get_thread_manager(); let mut semaphore = Semaphore::new(2, Rc::clone(&tm)); let thread1 = Rc::new(RefCell::new(Thread::new("test_semaphore_1"))); let thread2 = Rc::new(RefCell::new(Thread::new("test_semaphore_2"))); let thread3 = Rc::new(RefCell::new(Thread::new("test_semaphore_3"))); let mut borrow_tm = tm.borrow_mut(); let scheduler = &mut borrow_tm.g_scheduler; scheduler.ready_to_run(Rc::clone(&thread1)); scheduler.ready_to_run(Rc::clone(&thread2)); scheduler.ready_to_run(Rc::clone(&thread3)); // P borrow_tm.set_g_current_thread(Some(Rc::clone(&thread1))); semaphore.p(thread1, Rc::clone(&system)); assert_eq!(semaphore.counter, 1); assert!(semaphore.waiting_queue.is_empty()); borrow_tm.set_g_current_thread(Some(Rc::clone(&thread2))); semaphore.p(thread2, Rc::clone(&system)); assert_eq!(semaphore.counter, 0); assert!(semaphore.waiting_queue.is_empty()); borrow_tm.set_g_current_thread(Some(Rc::clone(&thread3))); semaphore.p(thread3, Rc::clone(&system)); assert_eq!(semaphore.counter, -1); assert!(semaphore.waiting_queue.iter().count() == 1); // V semaphore.v(Rc::clone(&system)); assert_eq!(semaphore.counter, 0); assert!(semaphore.waiting_queue.is_empty()); semaphore.v(Rc::clone(&system)); assert_eq!(semaphore.counter, 1); assert!(semaphore.waiting_queue.is_empty()); semaphore.v(Rc::clone(&system)); assert_eq!(semaphore.counter, 2); assert!(semaphore.waiting_queue.is_empty()); } #[test] #[ignore] fn test_lock_simple() { let system = init_system!(); let sys = system.borrow_mut(); let tm = sys.get_thread_manager(); let thread = Rc::new(RefCell::new(Thread::new("test_lock"))); tm.borrow_mut().set_g_current_thread(Some(Rc::clone(&thread))); let mut lock = Lock::new(Rc::clone(&tm)); assert!(lock.free); lock.acquire(Some(Rc::clone(&thread)), Rc::clone(&system)); assert!(lock.held_by_current_thread(Rc::clone(&thread))); assert!(!lock.free); lock.release(Rc::clone(&system), Rc::clone(&thread)); assert!(!lock.held_by_current_thread(thread)); assert!(lock.free); } #[test] #[ignore] fn test_lock_multiple() { let system = init_system!(); let thread1 = Rc::new(RefCell::new(Thread::new("test_lock1"))); let thread2 = Rc::new(RefCell::new(Thread::new("test_lock2"))); let thread3 = Rc::new(RefCell::new(Thread::new("test_lock3"))); let tm = system.borrow_mut().get_thread_manager(); tm.borrow_mut().set_g_current_thread(Some(Rc::clone(&thread1))); let mut lock = Lock::new(Rc::clone(&tm)); assert!(lock.free); lock.acquire(Some(Rc::clone(&thread1)), Rc::clone(&system)); assert!(lock.held_by_current_thread(Rc::clone(&thread1))); assert!(!lock.free); tm.borrow_mut().set_g_current_thread(Some(Rc::clone(&thread2))); lock.acquire(Some(Rc::clone(&thread2)), Rc::clone(&system)); tm.borrow_mut().set_g_current_thread(Some(Rc::clone(&thread1))); assert!(lock.held_by_current_thread(Rc::clone(&thread1))); assert!(lock.waiting_queue.iter().count() == 1); assert!(!lock.free); lock.release(Rc::clone(&system), Rc::clone(&thread1)); assert!(!lock.held_by_current_thread(thread1)); assert!(lock.held_by_current_thread(Rc::clone(&thread2))); assert!(!lock.free); tm.borrow_mut().set_g_current_thread(Some(Rc::clone(&thread2))); lock.release(Rc::clone(&system), Rc::clone(&thread2)); assert!(!lock.held_by_current_thread(thread2)); assert!(lock.free); } }