Impl std-like thread interface.

src/arch/x86_64/interrupt/trapframe.rs

@@ -35,9 +35,10 @@ pub struct TrapFrame {
 
 /// 用于在内核栈中构造新线程的中断帧
 impl TrapFrame {
-    fn new_kernel_thread(entry: extern fn() -> !, rsp: usize) -> Self {
+    fn new_kernel_thread(entry: extern fn(usize) -> !, arg: usize, rsp: usize) -> Self {
         use arch::gdt;
         let mut tf = TrapFrame::default();
+        tf.rdi = arg;
         tf.cs = gdt::KCODE_SELECTOR.0 as usize;
         tf.rip = entry as usize;
         tf.ss = gdt::KDATA_SELECTOR.0 as usize;
@@ -87,11 +88,11 @@ pub struct InitStack {
 }
 
 impl InitStack {
-    pub fn new_kernel_thread(entry: extern fn() -> !, rsp: usize) -> Self {
+    pub fn new_kernel_thread(entry: extern fn(usize) -> !, arg: usize, rsp: usize) -> Self {
         InitStack {
             context: Context::new(),
             trapret: trap_ret as usize,
-            tf: TrapFrame::new_kernel_thread(entry, rsp),
+            tf: TrapFrame::new_kernel_thread(entry, arg, rsp),
         }
     }
     pub fn new_user_thread(entry_addr: usize, rsp: usize, is32: bool) -> Self {

src/lib.rs

@@ -49,6 +49,7 @@ mod consts;
 mod process;
 mod syscall;
 mod fs;
+mod thread;
 mod sync;
 
 #[allow(dead_code)]

src/process/mod.rs

@@ -15,7 +15,7 @@ mod scheduler;
 pub fn init(mut mc: MemoryController) {
     PROCESSOR.call_once(|| {Mutex::new({
         let initproc = Process::new_init(&mut mc);
-        let idleproc = Process::new("idle", idle_thread, &mut mc);
+        let idleproc = Process::new("idle", idle_thread, 0, &mut mc);
         let mut processor = Processor::new();
         processor.add(initproc);
         processor.add(idleproc);
@@ -27,7 +27,7 @@ pub fn init(mut mc: MemoryController) {
 pub static PROCESSOR: Once<Mutex<Processor>> = Once::new();
 pub static MC: Once<Mutex<MemoryController>> = Once::new();
 
-extern fn idle_thread() -> ! {
+extern fn idle_thread(arg: usize) -> ! {
     loop {
         println!("idle ...");
         let mut i = 0;
@@ -45,6 +45,13 @@ pub fn add_user_process(name: impl AsRef<str>, data: &[u8]) {
     processor.add(new);
 }
 
+pub fn add_kernel_process(entry: extern fn(usize) -> !, arg: usize) -> Pid {
+    let mut processor = PROCESSOR.try().unwrap().lock();
+    let mut mc = MC.try().unwrap().lock();
+    let mut new = Process::new("", entry, arg, &mut mc);
+    processor.add(new)
+}
+
 pub fn print() {
     debug!("{:#x?}", *PROCESSOR.try().unwrap().lock());
 }

src/process/process.rs

@@ -32,9 +32,9 @@ pub enum Status {
 
 impl Process {
     /// Make a new kernel thread
-    pub fn new(name: &str, entry: extern fn() -> !, mc: &mut MemoryController) -> Self {
+    pub fn new(name: &str, entry: extern fn(usize) -> !, arg: usize, mc: &mut MemoryController) -> Self {
         let kstack = mc.alloc_stack(7).unwrap();
-        let data = InitStack::new_kernel_thread(entry, kstack.top());
+        let data = InitStack::new_kernel_thread(entry, arg, kstack.top());
         let rsp = kstack.push_at_top(data);
 
         Process {

src/process/processor.rs

@@ -17,9 +17,10 @@ pub struct Processor {
     /// Choose what on next schedule ?
     next: Option<Pid>,
     // WARNING: if MAX_PROCESS_NUM is too large, will cause stack overflow
-    scheduler: StrideScheduler,
+    scheduler: RRScheduler,
 }
 
+// TODO: 除schedule()外的其它函数，应该只设置进程状态，不应调用schedule
 impl Processor {
     pub fn new() -> Self {
         Processor {
@@ -29,12 +30,13 @@ impl Processor {
             kernel_page_table: None,
             next: None,
             // NOTE: max_time_slice <= 5 to ensure 'priority' test pass
-            scheduler: StrideScheduler::new(5),
+            scheduler: RRScheduler::new(100),
         }
     }
 
     pub fn lab6_set_priority(&mut self, priority: u8) {
-        self.scheduler.set_priority(self.current_pid, priority);
+        unimplemented!();
+//        self.scheduler.set_priority(self.current_pid, priority);
     }
 
     pub fn set_reschedule(&mut self) {
@@ -153,6 +155,7 @@ impl Processor {
 
         info!("Processor: switch from {} to {}\n  rsp: ??? -> {:#x}", pid0, pid, to.rsp);
         unsafe {
+            // FIXME: safely pass MutexGuard
             use core::mem::forget;
             super::PROCESSOR.try().unwrap().force_unlock();
             switch(&mut from.rsp, to.rsp);
@@ -191,6 +194,12 @@ impl Processor {
         self.set_status(pid, Status::Sleeping);
         self.event_hub.push(time, Event::Wakeup(pid));
     }
+    pub fn sleep_(&mut self, pid: Pid) {
+        self.set_status(pid, Status::Sleeping);
+    }
+    pub fn wakeup_(&mut self, pid: Pid) {
+        self.set_status(pid, Status::Ready);
+    }
 
     /// Let current process wait for another
     pub fn current_wait_for(&mut self, pid: Pid) -> WaitResult {

src/syscall.rs

@@ -2,6 +2,7 @@
 
 use arch::interrupt::TrapFrame;
 use process::*;
+use thread;
 use util;
 
 /// 系统调用入口点
@@ -108,9 +109,7 @@ fn sys_wait(pid: usize, code: *mut i32) -> i32 {
 }
 
 fn sys_yield() -> i32 {
-    info!("yield:");
+    thread::yield_now();
-    let mut processor = PROCESSOR.try().unwrap().lock();
-    processor.set_reschedule();
     0
 }
 
@@ -122,7 +121,7 @@ fn sys_kill(pid: usize) -> i32 {
 
 /// Get the current process id
 fn sys_getpid() -> i32 {
-    PROCESSOR.try().unwrap().lock().current_pid() as i32
+    thread::current().id() as i32
 }
 
 /// Exit the current process
@@ -134,10 +133,7 @@ fn sys_exit(error_code: usize) -> i32 {
 }
 
 fn sys_sleep(time: usize) -> i32 {
-    info!("sleep: {} ticks", time);
+    thread::sleep(time);
-    let mut processor = PROCESSOR.try().unwrap().lock();
-    let pid = processor.current_pid();
-    processor.sleep(pid, time);
     0
 }
 

src/thread.rs

@@ -0,0 +1,135 @@
+//! Thread std-like interface
+//!
+//! Based on process mod.
+//! Used in the kernel.
+
+use process::*;
+use core::marker::PhantomData;
+use core::ptr;
+use alloc::boxed::Box;
+
+/// Gets a handle to the thread that invokes it.
+pub fn current() -> Thread {
+    Thread {
+        pid: PROCESSOR.try().unwrap().lock().current_pid(),
+    }
+}
+
+/// Puts the current thread to sleep for the specified amount of time.
+pub fn sleep(time: usize) {
+    // TODO: use core::time::Duration
+    info!("sleep: {} ticks", time);
+    let mut processor = PROCESSOR.try().unwrap().lock();
+    let pid = processor.current_pid();
+    processor.sleep(pid, time);
+    processor.schedule();
+}
+
+/// Spawns a new thread, returning a JoinHandle for it.
+pub fn spawn<F, T>(f: F) -> JoinHandle<T>
+    where
+        F: Send + 'static + FnOnce() -> T,
+        T: Send + 'static,
+{
+    use process;
+    let pid = process::add_kernel_process(kernel_thread_entry::<F, T>, &f as *const _ as usize);
+    return JoinHandle {
+        thread: Thread { pid },
+        mark: PhantomData,
+    };
+
+    extern fn kernel_thread_entry<F, T>(f: usize) -> !
+        where
+            F: Send + 'static + FnOnce() -> T,
+            T: Send + 'static,
+    {
+        debug!("kernel_thread_entry");
+        let f = unsafe { ptr::read(f as *mut F) };
+        let ret = Box::new(f());
+        let mut processor = PROCESSOR.try().unwrap().lock();
+        let pid = processor.current_pid();
+        processor.exit(pid, Box::into_raw(ret) as usize);
+        unreachable!()
+    }
+}
+
+/// Cooperatively gives up a timeslice to the OS scheduler.
+pub fn yield_now() {
+    info!("yield:");
+    let mut processor = PROCESSOR.try().unwrap().lock();
+    processor.set_reschedule();
+    processor.schedule();
+}
+
+/// Blocks unless or until the current thread's token is made available.
+pub fn park() {
+    let mut processor = PROCESSOR.try().unwrap().lock();
+    let pid = processor.current_pid();
+    processor.sleep_(pid);
+    processor.schedule();
+}
+
+/// A handle to a thread.
+pub struct Thread {
+    pid: usize,
+}
+
+impl Thread {
+    /// Atomically makes the handle's token available if it is not already.
+    pub fn unpark(&self) {
+        let mut processor = PROCESSOR.try().unwrap().lock();
+        processor.wakeup_(self.pid);
+    }
+    /// Gets the thread's unique identifier.
+    pub fn id(&self) -> usize {
+        self.pid
+    }
+}
+
+/// An owned permission to join on a thread (block on its termination).
+pub struct JoinHandle<T> {
+    thread: Thread,
+    mark: PhantomData<T>,
+}
+
+impl<T> JoinHandle<T> {
+    /// Extracts a handle to the underlying thread.
+    pub fn thread(&self) -> &Thread {
+        &self.thread
+    }
+    /// Waits for the associated thread to finish.
+    pub fn join(self) -> Result<T, ()> {
+        let mut processor = PROCESSOR.try().unwrap().lock();
+        match processor.current_wait_for(self.thread.pid) {
+            WaitResult::Ok(_, exit_code) => {
+                unsafe {
+                    let value = Box::from_raw(exit_code as *mut T);
+                    Ok(ptr::read(exit_code as *const T))
+                }
+            }
+            WaitResult::NotExist => Err(()),
+        }
+    }
+}
+
+pub mod test {
+    use thread;
+
+    pub fn unpack() {
+        let parked_thread = thread::spawn(|| {
+            println!("Parking thread");
+            thread::park();
+            println!("Thread unparked");
+            5
+        });
+
+        // Let some time pass for the thread to be spawned.
+        thread::sleep(200);
+
+        println!("Unpark the thread");
+        parked_thread.thread().unpark();
+
+        let ret = parked_thread.join().unwrap();
+        assert_eq!(ret, 5);
+    }
+}