//! # Thread manager //! //! This module describes the data structure and the methods used for thread scheduling //! in the BurritOS operating system. A struct named [`ThreadManager`] holds the list of //! all existing [`Thread`] instances and synchronization objects, such as //! [`Lock`](crate::kernel::synch::Lock), //! [`Semaphore`](crate::kernel::synch::Semaphore) and //! [`Condition`](crate::kernel::synch::Condition). //! //! ## Purpose //! //! [`ThreadManager`] holds the state of the system processes using the following subcomponents: //! //! ### Two lists of threads //! //! - **ready_list**: The list of threads ready to be executed //! - **g_alive**: The list of currently executing threads //! //! The difference between the two above lists lies in the state of the threads in question. //! Ready threads have just been enqueued. They are not being executed yet. The second list is //! needed because many threads may be executing at a given time. However, only a single thread //! can be handled by the machine at a time. The system thus needs to keep in memory the alive //! threads in case the currently running thread finishes or gets rescheduled. //! //! ### A list of synchronization objects //! //! Locks, Semaphores and Conditions allow resource sharing among running threads. Since resources //! can only be accessed by a single thread at a time, we need data structures to signal other //! threads that a resource may be busy or unavailable; say for example that: //! //! - Thread **A** wants to write to a file while **B** is currently reading said file. //! - Thread **A** mutating the state of the file could cause issues for **B**. //! - Therefore **B** needs to lock the file in question to avoid such issues. //! - Thread **A** will have to wait for **B** to finish reading the file. //! //! These synchronization objects are held in an instance of the ObjAddr structure held by //! ThreadManager. Their state is mutated depending on the actions of the currently running thread //! through methods such as [`ThreadManager::sem_p`]. //! //! ## Usage //! //! [`ThreadManager`] is thought as a subcomponent of the [`System`](crate::kernel::system::System) struct. //! Instanciating [`System`](crate::kernel::system::System) will automatically instanciate a [`ThreadManager`] //! //! Manually loading a [`Thread`] into [`ThreadManager`] to execute a program with BurritOS could look like //! this: //! //! ``` //! fn load_thread_manually(args: ...) { //! let mut system = System::new(args.debug); //! //! let thread_exec = Thread::new(args.executable.as_str()); //! let thread_exec = Rc::new(RefCell::new(thread_exec)); //! system.get_thread_manager().get_g_alive().push(Rc::clone(&thread_exec)); //! //! let owner1 = Process { num_thread: 0 }; //! let owner1 = Rc::new(RefCell::new(owner1)); //! system.get_thread_manager().start_thread(Rc::clone(&thread_exec), owner1, loader.elf_header.entrypoint, ptr, -1); //! //! let to_run = system.get_thread_manager().find_next_to_run().unwrap(); //! system.get_thread_manager().switch_to(&mut machine, Rc::clone(&to_run)); //! //! machine.run(&mut system); //! } //! ``` //! //! ## Imports //! //! The [`List`] and [`ObjAddr`] submodules used in this module are defined in the utility //! module. The source code of [`ObjAddr`] has been decoupled from thread_manager in an effort //! to keep this module concise. use std::{ rc::Rc, cell::{ RefCell, Ref } }; use crate::{ utility::{ list::List, objaddr::ObjAddr }, simulator::{ machine::{ NUM_INT_REGS, NUM_FP_REGS, Machine }, interrupt::InterruptStatus, error::{ MachineOk, MachineError } }, kernel::{ thread::Thread, process::Process } }; /// Using this type alias to simplify struct and method definitions pub type ThreadRef = Rc>; /// # Thread manager /// /// An instance of this struct is responsible for managing threads on behalf of the system #[derive(PartialEq)] pub struct ThreadManager { /// Current running thread pub g_current_thread: Option, /// The list of alive threads pub g_alive: List, /// Thread in ready state waiting to become active ready_list: List, /// List of objects created by the thread manager (such as Locks and Semaphores) obj_addrs: ObjAddr, /// If true, enables debug mode debug: bool } impl ThreadManager { /// Thread manager constructor pub fn new(debug: bool) -> Self { Self { g_current_thread: Option::None, g_alive: List::default(), ready_list: List::default(), obj_addrs: ObjAddr::init(), debug } } /// Mark a thread as aready, but not necessarily running yet. /// /// Put it in the ready list, for later scheduling onto the CPU. /// /// ## Pamameter /// /// **thread** is the thread to be put on the read list pub fn ready_to_run(&mut self, thread: ThreadRef) { self.ready_list.push(thread); } /// Return the next thread to be scheduled onto the CPU. /// If there are no ready threads, return Option::None /// /// Thread is removed from the ready list. /// /// **return** Thread thread to be scheduled pub fn find_next_to_run(&mut self) -> Option { self.ready_list.pop() } /// Dispatch the CPU to next_thread. Save the state of the old thread /// and load the state of the new thread. /// /// We assume the state of the previously running thread has already been changed from running to blocked or ready. /// /// Global variable g_current_thread become next_thread /// /// ## Parameter /// /// **next_thread** thread to dispatch to the CPU pub fn switch_to(&mut self, machine: &mut Machine, next_thread: ThreadRef) { if let Some(old_thread) = self.get_g_current_thread() { let old_thread = old_thread.clone(); self.thread_save_processor_state(machine, old_thread.clone()); // old_thread.save_simulator_state(); if old_thread != next_thread { self.debug(format!("switching from \"{}\" to \"{}\"", old_thread.borrow().get_name(), next_thread.borrow().get_name())); self.thread_restore_processor_state(machine, Rc::clone(&next_thread)); // next_thread.restore_simulator_state(); debug_assert!(!self.ready_list.contains(&next_thread)); self.set_g_current_thread(Some(next_thread)); } } else { self.thread_restore_processor_state(machine, Rc::clone(&next_thread)); // next_thread.restore_simulator_state(); self.set_g_current_thread(Some(next_thread)); } } /// Start a thread, attaching it to a process pub fn start_thread(&mut self, thread: ThreadRef, owner: Rc>, func_pc: u64, sp_loc: u64, argument: i64) { self.debug(format!("starting thread \"{}\"", thread.borrow().get_name())); let mut thread_m = thread.borrow_mut(); assert_eq!(thread_m.process, Option::None); thread_m.process = Option::Some(Rc::clone(&owner)); let ptr = sp_loc; // todo addrspace thread_m.init_thread_context(func_pc, ptr, argument); owner.borrow_mut().num_thread += 1; self.get_g_alive().push(Rc::clone(&thread)); self.ready_to_run(Rc::clone(&thread)); } /// Wait for another thread to finish its execution pub fn thread_join(&mut self, machine: &mut Machine, waiter: ThreadRef, waiting_for: ThreadRef) { let waiting_for = Rc::clone(&waiting_for); if self.get_g_alive().contains(&waiting_for) { waiting_for.borrow_mut().join_thread.push(Rc::clone(&waiter)); self.thread_yield(machine, Rc::clone(&waiter), false); } } /// Relinquish the CPU if any other thread is runnable. /// /// Cannot use yield as a function name -> reserved name in rust /// /// ## Parameters /// /// **is_ready** true if **thread** should be readded to ready_to_run list, false otherwise. Typically false when joining per example pub fn thread_yield(&mut self, machine: &mut Machine, thread: ThreadRef, is_ready: bool) { let old_status = machine.interrupt.set_status(crate::simulator::interrupt::InterruptStatus::InterruptOff); self.debug(format!("Yeilding thread: \"{}\"", thread.borrow().get_name())); debug_assert_eq!(&Option::Some(Rc::clone(&thread)), self.get_g_current_thread()); let next_thread = self.find_next_to_run(); if let Some(next_thread) = next_thread { if is_ready { self.ready_to_run(thread); } self.switch_to(machine, next_thread); } machine.interrupt.set_status(old_status); } /// Put the thread to sleep and relinquish the processor pub fn thread_sleep(&mut self, machine: &mut Machine, thread: ThreadRef) { debug_assert_eq!(Option::Some(Rc::clone(&thread)), self.g_current_thread); debug_assert_eq!(machine.interrupt.get_status(), InterruptStatus::InterruptOff); self.debug(format!("Sleeping thread {}", thread.borrow().get_name())); let mut next_thread = self.find_next_to_run(); while next_thread.is_none() { eprintln!("Nobody to run => idle"); machine.interrupt.idle(); if let Some(t) = self.find_next_to_run() { next_thread = Some(t); } else { panic!("Couldn't find next thread to run.\nShutting down..."); } } self.switch_to(machine, Rc::clone(&next_thread.unwrap())); } /// Finish the execution of the thread and prepare its deallocation pub fn thread_finish(&mut self, machine: &mut Machine, thread: ThreadRef) { let old_status = machine.interrupt.set_status(InterruptStatus::InterruptOff); assert!(self.g_alive.remove(Rc::clone(&thread))); self.debug(format!("Finishing thread {}", thread.borrow().get_name())); // g_objets_addrs->removeObject(self.thread) // a ajouté plus tard for (_, el) in thread.borrow().join_thread.iter().enumerate() { self.ready_to_run(Rc::clone(&el)); } self.thread_sleep(machine, Rc::clone(&thread)); machine.interrupt.set_status(old_status); } /// Save the CPU state of a user program on a context switch. pub fn thread_save_processor_state(&mut self, machine: &mut Machine, thread: ThreadRef) { let mut t = thread.borrow_mut(); for i in 0..NUM_INT_REGS { t.thread_context.int_registers[i] = machine.read_int_register(i); } for i in 0..NUM_FP_REGS { t.thread_context.float_registers[i] = machine.read_fp_register(i); } t.thread_context.pc = machine.pc; } /// Restore the CPU state of a user program on a context switch. pub fn thread_restore_processor_state(&self, machine: &mut Machine, thread: ThreadRef) { let t: Ref<_> = thread.borrow(); for i in 0..NUM_INT_REGS { machine.write_int_register(i, t.thread_context.int_registers[i]); } machine.pc = t.thread_context.pc; } /// 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 /// - *id_sema* id of the semaphore, stored in [`ObjAddr`], id given by user program thought exceptions /// - *machine* Current state of the machine pub fn sem_p(&mut self, id_sema: i32, machine: &mut Machine) -> Result { let old_status = machine.interrupt.set_status(InterruptStatus::InterruptOff); let thread = match self.get_g_current_thread() { Some(thread) => Rc::clone(thread), None => Err("sem_p error: current thread should not be None")? }; let sema = match self.get_obj_addrs().search_semaphore(id_sema) { Some(sema) => sema, None => Err("sem_p error: cannot find semaphore")? }; sema.counter -= 1; if sema.counter < 0 { sema.waiting_queue.push(thread.clone()); self.thread_sleep(machine, thread); } machine.interrupt.set_status(old_status); Ok(MachineOk::Ok) } /// 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 /// - *id_sema* id of the semaphore, stored in [`ObjAddr`], id given by user program thought exceptions /// - **machine** the machine where the threads are executed pub fn sem_v(&mut self, id_sema: i32, machine: &mut Machine) -> Result { let sema = match self.get_obj_addrs().search_semaphore(id_sema) { Some(sema) => sema, None => Err("sem_v error: cannot find semaphore")? }; let old_status = machine.interrupt.set_status(InterruptStatus::InterruptOff); sema.counter += 1; if let Some(thread) = sema.waiting_queue.pop() { self.ready_to_run(thread) } machine.interrupt.set_status(old_status); Ok(MachineOk::Ok) } /// 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 /// - **id** id of the lock, stored in [`ObjAddr`], id given by user program thought exceptions /// - **machine** the machine where the threads are executed pub fn lock_acquire(&mut self, id: i32, machine: &mut Machine) -> Result { let current_thread = match self.get_g_current_thread() { Some(thread) => Rc::clone(thread), None => Err("lock_acquire error: current_thread should not be None.")? }; let old_status = machine.interrupt.set_status(InterruptStatus::InterruptOff); if let Some(lock) = self.get_obj_addrs().search_lock(id) { if lock.free { lock.free = false; lock.owner = Some(current_thread) } else { lock.waiting_queue.push(current_thread.clone()); self.thread_sleep(machine, current_thread); } } else { Err("lock_acquire error: cannot find Lock.")? } machine.interrupt.set_status(old_status); Ok(MachineOk::Ok) } /// Wake up a waiter if necessary, or release it if no thread is waiting. pub fn lock_release(&mut self, id: i32, machine: &mut Machine) -> Result { let old_status = machine.interrupt.set_status(InterruptStatus::InterruptOff); let current_thread = match self.get_g_current_thread() { Some(thread) => Rc::clone(thread), None => Err(String::from("lock_release error: current_thread should not be None."))? }; let mut lock = match self.get_obj_addrs().search_lock(id) { Some(lock) => lock, None => Err(String::from("lock_release error: cannot find lock."))? }; if let Some(lock_owner) = &lock.owner { if current_thread.eq(lock_owner) { // is_held_by_current_thread match lock.waiting_queue.pop() { Some(th) => { lock.owner = Some(Rc::clone(&th)); self.ready_to_run(Rc::clone(&th)); }, None => { lock.free = true; lock.owner = None; } } } }; // self.get_obj_addrs().update_lock(id, lock); machine.interrupt.set_status(old_status); Ok(MachineOk::Ok) } /// Currently running thread pub fn get_g_current_thread(&mut self) -> &Option { &self.g_current_thread } /// List of alive threads pub fn get_g_alive(&mut self) -> &mut List { &mut self.g_alive } /// Set currently running thread pub fn set_g_current_thread(&mut self, thread: Option) { self.g_current_thread = thread } /// Returns a mutable reference to the ObjAddr field of this thread_manager pub fn get_obj_addrs(&mut self) -> &mut ObjAddr { &mut self.obj_addrs } /// Prints debug messages if self.debug is set to true. fn debug(&self, message: String) { if self.debug { println!("{}", message); } } } #[cfg(test)] mod test { use std::{rc::Rc, cell::RefCell}; use crate::{simulator::{machine::Machine, loader}, kernel::{system::System, thread::Thread, process::Process, thread_manager::ThreadManager, synch::Semaphore}, utility::cfg::get_debug_configuration}; use crate::kernel::synch::Lock; #[test] fn test_thread_context() { let mut machine = Machine::new(true, get_debug_configuration()); let (loader, ptr) = loader::Loader::new("./target/guac/halt.guac", &mut machine, 0).expect("IO Error"); let start_pc = loader.elf_header.entrypoint; let system = &mut System::new(true); let thread1 = Thread::new("th1"); let thread1 = Rc::new(RefCell::new(thread1)); system.get_thread_manager().get_g_alive().push(Rc::clone(&thread1)); let owner1 = Process { num_thread: 0 }; let owner1 = Rc::new(RefCell::new(owner1)); system.get_thread_manager().start_thread(Rc::clone(&thread1), owner1, loader.elf_header.entrypoint, ptr + machine.page_size, -1); debug_assert_eq!(thread1.borrow_mut().thread_context.pc, start_pc); debug_assert!(system.get_thread_manager().get_g_alive().contains(&Rc::clone(&thread1))); let to_run = system.get_thread_manager().find_next_to_run().unwrap(); debug_assert_eq!(to_run, Rc::clone(&thread1)); system.get_thread_manager().switch_to(&mut machine, Rc::clone(&to_run)); debug_assert_eq!(system.get_thread_manager().g_current_thread, Option::Some(Rc::clone(&thread1))); debug_assert_eq!(machine.pc, loader.elf_header.entrypoint); machine.run(system); } #[test] fn test_lock_single(){ let mut machine = Machine::new(true, get_debug_configuration()); let mut thread_manager = ThreadManager::new(true); let lock = Lock::new(); let lock_id = thread_manager.get_obj_addrs().add_lock(lock); let thread = Rc::new(RefCell::new(Thread::new("test_lock"))); let thread_test = thread.clone(); thread_manager.ready_to_run(Rc::clone(&thread)); thread_manager.set_g_current_thread(Some(thread)); thread_manager.lock_acquire(lock_id, &mut machine).expect("lock acquire return an error: "); { let lock = thread_manager.get_obj_addrs().search_lock(lock_id).unwrap(); assert_eq!(lock.owner,Some(thread_test)); assert!(!lock.free); assert!(lock.waiting_queue.is_empty()); } thread_manager.lock_release(lock_id, &mut machine).expect("lock release return an error: "); { let lock = thread_manager.get_obj_addrs().search_lock(lock_id).unwrap(); assert_eq!(lock.owner, None); assert!(lock.free); assert!(lock.waiting_queue.is_empty()); } } #[test] fn test_lock_multiple() { let mut machine = Machine::new(true, get_debug_configuration()); let mut thread_manager = ThreadManager::new(true); let lock = Lock::new(); let lock_id = thread_manager.get_obj_addrs().add_lock(lock); let thread_1 = Rc::new(RefCell::new(Thread::new("test_lock_1"))); let thread_2 = Rc::new(RefCell::new(Thread::new("test_lock_2"))); thread_manager.ready_to_run(thread_1.clone()); thread_manager.ready_to_run(thread_2.clone()); thread_manager.set_g_current_thread(Some(thread_1.clone())); thread_manager.lock_acquire(lock_id, &mut machine).expect("lock acquire return an error at first iteration: "); { let lock = thread_manager.get_obj_addrs().search_lock(lock_id).unwrap(); assert_eq!(lock.owner,Some(thread_1.clone())); assert!(!lock.free); assert!(lock.waiting_queue.is_empty()); } thread_manager.set_g_current_thread(Some(thread_2.clone())); thread_manager.lock_acquire(lock_id, &mut machine).expect("lock acquire return an error at second iteration: "); { let lock = thread_manager.get_obj_addrs().search_lock(lock_id).unwrap(); assert_eq!(lock.owner,Some(thread_1.clone())); assert!(!lock.free); assert_eq!(lock.waiting_queue.iter().count(),1); } thread_manager.lock_release(lock_id, &mut machine).expect("lock release return an error at first iteration: "); { let lock = thread_manager.get_obj_addrs().search_lock(lock_id).unwrap(); assert_eq!(lock.owner, Some(thread_2.clone())); assert!(!lock.free); assert!(lock.waiting_queue.is_empty()); } thread_manager.set_g_current_thread(Some(thread_2.clone())); thread_manager.lock_release(lock_id, &mut machine).expect("lock release return an error at second iteration: "); { let lock = thread_manager.get_obj_addrs().search_lock(lock_id).unwrap(); assert!(lock.waiting_queue.is_empty()); assert_eq!(lock.owner, None); assert!(lock.free); } } #[test] fn test_semaphore_single() { // Init let mut machine = Machine::new(true, get_debug_configuration()); let mut thread_manager = ThreadManager::new(true); let semaphore = Semaphore::new(1); let sema_id = thread_manager.get_obj_addrs().add_semaphore(semaphore); 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 thread_manager.sem_p(sema_id, &mut machine).expect("semaphore P return an error: "); { let semaphore = thread_manager.get_obj_addrs().search_semaphore(sema_id).unwrap(); assert_eq!(semaphore.counter, 0); assert!(semaphore.waiting_queue.is_empty()); } // V thread_manager.sem_v(sema_id, &mut machine).expect("semaphore V return an error: "); { let semaphore = thread_manager.get_obj_addrs().search_semaphore(sema_id).unwrap(); assert_eq!(semaphore.counter, 1); assert!(semaphore.waiting_queue.is_empty()); } } #[test] fn test_semaphore_multiple() { // Init let mut tm = ThreadManager::new(true); let mut machine = Machine::new(true, get_debug_configuration()); let semaphore = Semaphore::new(2); let sema_id = tm.get_obj_addrs().add_semaphore(semaphore); 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))); tm.sem_p(sema_id, &mut machine).expect("semaphore P return an error: "); { let semaphore = tm.get_obj_addrs().search_semaphore(sema_id).unwrap(); assert_eq!(semaphore.counter, 1); assert!(semaphore.waiting_queue.is_empty()); } tm.set_g_current_thread(Some(Rc::clone(&thread2))); tm.sem_p(sema_id, &mut machine).expect("semaphore P return an error: "); { let semaphore = tm.get_obj_addrs().search_semaphore(sema_id).unwrap(); assert_eq!(semaphore.counter, 0); assert!(semaphore.waiting_queue.is_empty()); } tm.set_g_current_thread(Some(Rc::clone(&thread3))); tm.sem_p( sema_id, &mut machine).expect("semaphore P return an error: "); { let semaphore = tm.get_obj_addrs().search_semaphore(sema_id).unwrap(); assert_eq!(semaphore.counter, -1); assert!(semaphore.waiting_queue.iter().count() == 1); } // V tm.sem_v(sema_id, &mut machine).expect("semaphore V return an error: "); { let semaphore = tm.get_obj_addrs().search_semaphore(sema_id).unwrap(); assert_eq!(semaphore.counter, 0); assert!(semaphore.waiting_queue.is_empty()); } tm.sem_v(sema_id, &mut machine).expect("semaphore V return an error: "); { let semaphore = tm.get_obj_addrs().search_semaphore(sema_id).unwrap(); assert_eq!(semaphore.counter, 1); assert!(semaphore.waiting_queue.is_empty()); } tm.sem_v(sema_id, &mut machine).expect("semaphore V return an error: "); { let semaphore = tm.get_obj_addrs().search_semaphore(sema_id).unwrap(); assert_eq!(semaphore.counter, 2); assert!(semaphore.waiting_queue.is_empty()); } } }