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::thread_manager::ThreadManager; /// Structure of a Semaphore used for synchronisation #[derive(PartialEq)] pub struct Semaphore { /// Counter of simultanous Semaphore pub counter:i32, /// QUeue of Semaphore waiting to be exucated pub waiting_queue:List>>, } 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) -> Semaphore{ Semaphore { counter, waiting_queue: List::default() } } /// 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, machine: &mut Machine, thread_manager: &mut ThreadManager) { let old_status = machine.interrupt.set_status(InterruptOff); self.counter -= 1; if self.counter < 0 { match thread_manager.get_g_current_thread() { Some(thread) => { let rc1_thread = Rc::clone(thread); let rc2_thread = Rc::clone(thread); self.waiting_queue.push(rc1_thread); thread_manager.thread_sleep(machine, rc2_thread); }, None => unreachable!("Current thread should not be None") } } 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, machine: &mut Machine, thread_manager: &mut ThreadManager){ let old_status = machine.interrupt.set_status(InterruptOff); self.counter += 1; match self.waiting_queue.pop() { Some(thread) => thread_manager.ready_to_run(thread), None => () } 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 #[derive(PartialEq)] pub struct Lock { /// Thread owning the lock owner: Option>>, /// The queue of threads waiting for execution waiting_queue:List>>, /// 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() -> Lock { Lock { owner: None, waiting_queue: List::default(), 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, machine: &mut Machine, thread_manager: &mut ThreadManager) { let old_status = machine.interrupt.set_status(InterruptOff); if self.free { self.free = false; self.owner = Option::Some(match thread_manager.get_g_current_thread() { Some(th) => { Rc::clone(&th) }, None => unreachable!() }); } else { match thread_manager.get_g_current_thread() { Some(x) => { let x = Rc::clone(&x); self.waiting_queue.push(Rc::clone(&x)); thread_manager.thread_sleep(machine, Rc::clone(&x)); }, None => unreachable!("Current thread should not be None") } } 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, machine: &mut Machine, thread_manager: &mut ThreadManager) { let old_status = machine.interrupt.set_status(InterruptOff); match thread_manager.get_g_current_thread() { Some(_) => { if self.held_by_current_thread(thread_manager) { match self.waiting_queue.pop() { Some(thread) => { self.owner = Some(thread); match &self.owner { Some(x) => thread_manager.ready_to_run(Rc::clone(&x)), None => () } }, None => { self.free = true; self.owner = None; } } } } None => () } machine.interrupt.set_status(old_status); } /// True if the current thread holds this lock. /// Useful for checking in Release, and in Condition operations below. pub fn held_by_current_thread(&mut self, thread_manager: &mut ThreadManager) -> bool { match &self.owner { Some(x) => match 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>>, } impl Condition { /// Initializes a Condition, so that it can be used for synchronization. /// /// ### Parameters /// - *thread_manager* Thread manager which managing threads pub fn new() -> Condition { Condition{ waiting_queue: List::default()} } /// 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, machine: &mut Machine, thread_manager: &mut ThreadManager) { let old_status = machine.interrupt.set_status(InterruptOff); match thread_manager.get_g_current_thread() { Some(thread) => { let rc1 = Rc::clone(thread); let rc2 = Rc::clone(thread); self.waiting_queue.push(rc1); thread_manager.thread_sleep(machine, rc2); }, None => unreachable!() } 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, thread_manager: &mut ThreadManager) { let old_status = machine.interrupt.set_status(InterruptOff); match self.waiting_queue.pop() { Some(thread) => thread_manager.ready_to_run(thread), None => () } 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, thread_manager: &mut ThreadManager) { let old_status = machine.interrupt.set_status(InterruptOff); match self.waiting_queue.pop() { Some(thread) => thread_manager.ready_to_run(thread), None => () } machine.interrupt.set_status(old_status); } } #[cfg(test)] mod test { use std::{rc::Rc, cell::RefCell}; use crate::{kernel::{thread::Thread, synch::{Semaphore, Lock}, thread_manager::ThreadManager}, simulator::machine::Machine}; #[test] fn test_semaphore_single() { // Init let mut machine = Machine::new(true); let mut thread_manager = ThreadManager::new(); let mut semaphore = Semaphore::new(1); let thread = Rc::new(RefCell::new(Thread::new("test_semaphore"))); thread_manager.ready_to_run(Rc::clone(&thread)); thread_manager.set_g_current_thread(Some(thread)); // P semaphore.p(&mut machine, &mut thread_manager); assert_eq!(semaphore.counter, 0); assert!(semaphore.waiting_queue.is_empty()); // V semaphore.v(&mut machine, &mut thread_manager); assert_eq!(semaphore.counter, 1); assert!(semaphore.waiting_queue.is_empty()); } #[test] fn test_semaphore_multiple() { // Init let mut tm = ThreadManager::new(); let mut machine = Machine::new(true); let mut semaphore = Semaphore::new(2); 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 tm.g_scheduler; tm.ready_to_run(Rc::clone(&thread1)); tm.ready_to_run(Rc::clone(&thread2)); tm.ready_to_run(Rc::clone(&thread3)); // P tm.set_g_current_thread(Some(Rc::clone(&thread1))); semaphore.p(&mut machine, &mut tm); assert_eq!(semaphore.counter, 1); assert!(semaphore.waiting_queue.is_empty()); tm.set_g_current_thread(Some(Rc::clone(&thread2))); semaphore.p(&mut machine, &mut tm); assert_eq!(semaphore.counter, 0); assert!(semaphore.waiting_queue.is_empty()); tm.set_g_current_thread(Some(Rc::clone(&thread3))); semaphore.p(&mut machine, &mut tm); assert_eq!(semaphore.counter, -1); assert!(semaphore.waiting_queue.iter().count() == 1); // V semaphore.v(&mut machine, &mut tm); assert_eq!(semaphore.counter, 0); assert!(semaphore.waiting_queue.is_empty()); semaphore.v(&mut machine, &mut tm); assert_eq!(semaphore.counter, 1); assert!(semaphore.waiting_queue.is_empty()); semaphore.v(&mut machine, &mut tm); assert_eq!(semaphore.counter, 2); assert!(semaphore.waiting_queue.is_empty()); } #[test] fn test_lock_simple() { let mut machine = Machine::new(true); let mut tm = ThreadManager::new(); let thread = Rc::new(RefCell::new(Thread::new("test_lock"))); tm.ready_to_run(Rc::clone(&thread)); tm.set_g_current_thread(Some(Rc::clone(&thread))); let mut lock = Lock::new(); assert!(lock.free); lock.acquire(&mut machine, &mut tm); assert!(lock.held_by_current_thread(&mut tm)); assert!(!lock.free); lock.release(&mut machine, &mut tm); assert!(!lock.held_by_current_thread(&mut tm)); assert!(lock.free); } #[test] fn test_lock_multiple() { let thread1 = Rc::new(RefCell::new(Thread::new("test_lock1"))); let thread2 = Rc::new(RefCell::new(Thread::new("test_lock2"))); let mut machine = Machine::new(true); let mut tm = ThreadManager::new(); tm.ready_to_run(Rc::clone(&thread1)); tm.ready_to_run(Rc::clone(&thread2)); tm.set_g_current_thread(Some(Rc::clone(&thread1))); let mut lock = Lock::new(); assert!(lock.free); lock.acquire(&mut machine, &mut tm); assert!(lock.held_by_current_thread(&mut tm)); assert!(!lock.free); tm.set_g_current_thread(Some(Rc::clone(&thread2))); lock.acquire(&mut machine, &mut tm); tm.set_g_current_thread(Some(Rc::clone(&thread1))); assert!(lock.held_by_current_thread(&mut tm)); assert!(lock.waiting_queue.iter().count() == 1); assert!(!lock.free); lock.release(&mut machine, &mut tm); assert!(!lock.held_by_current_thread(&mut tm)); tm.set_g_current_thread(Some(Rc::clone(&thread2))); assert!(lock.held_by_current_thread(&mut tm)); assert!(!lock.free); lock.release(&mut machine, &mut tm); assert!(!lock.held_by_current_thread(&mut tm)); assert!(lock.free); } }