Move ThreadSwitchFuture from kernel-hal to zircon-object

kernel-hal/src/bare/thread.rs

@@ -1,30 +1,11 @@
 //! Thread spawning.
 
-use alloc::boxed::Box;
+use core::future::Future;
-use core::task::{Context, Poll};
-use core::{future::Future, pin::Pin};
-
-use spin::Mutex;
 
 hal_fn_impl! {
     impl mod crate::hal_fn::thread {
-        fn spawn(future: Pin<Box<dyn Future<Output = ()> + Send + 'static>>, vmtoken: usize) {
+        fn spawn(future: impl Future<Output = ()> + Send + 'static) {
-            struct PageTableSwitchWrapper {
+            executor::spawn(future);
-                inner: Mutex<Pin<Box<dyn Future<Output = ()> + Send>>>,
-                vmtoken: usize,
-            }
-            impl Future for PageTableSwitchWrapper {
-                type Output = ();
-                fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-                    crate::vm::activate_paging(self.vmtoken);
-                    self.inner.lock().as_mut().poll(cx)
-                }
-            }
-
-            executor::spawn(PageTableSwitchWrapper {
-                inner: Mutex::new(future),
-                vmtoken,
-            });
         }
 
         fn set_tid(_tid: u64, _pid: u64) {}

kernel-hal/src/common/future.rs

@@ -20,7 +20,7 @@ impl Future for YieldFuture {
             Poll::Ready(())
         } else {
             self.flag = true;
-            cx.waker().clone().wake();
+            cx.waker().wake_by_ref();
             Poll::Pending
         }
     }
@@ -46,7 +46,7 @@ impl Future for SleepFuture {
         }
         if self.deadline.as_nanos() < i64::max_value() as u128 {
             let waker = cx.waker().clone();
-            timer::timer_set(self.deadline, Box::new(move |_| waker.wake()));
+            timer::timer_set(self.deadline, Box::new(move |_| waker.wake_by_ref()));
         }
         Poll::Pending
     }
@@ -118,7 +118,7 @@ impl Future for DisplayFlushFuture {
             let frame_time = self.frame_time;
             self.next_flush_time += frame_time;
             let waker = cx.waker().clone();
-            timer::timer_set(self.next_flush_time, Box::new(move |_| waker.wake()));
+            timer::timer_set(self.next_flush_time, Box::new(move |_| waker.wake_by_ref()));
         }
         Poll::Pending
     }

kernel-hal/src/hal_fn.rs

@@ -1,5 +1,5 @@
 use alloc::{boxed::Box, string::String, vec::Vec};
-use core::{future::Future, ops::Range, pin::Pin, time::Duration};
+use core::{future::Future, ops::Range, time::Duration};
 
 use crate::drivers::prelude::{IrqHandler, IrqPolarity, IrqTriggerMode};
 use crate::{common, HalResult, KernelConfig, KernelHandler, MMUFlags, PhysAddr, VirtAddr};
@@ -63,11 +63,13 @@ hal_fn_def! {
 
     /// Virutal memory operations.
     pub mod vm: common::vm {
-        /// Read current VM token. (e.g. CR3, SATP, ...)
+        /// Read the current VM token, which is the page table root address on
+        /// various architectures. (e.g. CR3, SATP, ...)
         pub fn current_vmtoken() -> PhysAddr;
 
-        /// Activate this page table by given `vmtoken`.
+        /// Activate the page table associated with the `vmtoken` by writing the
-        pub(crate) fn activate_paging(vmtoken: PhysAddr);
+        /// page table root address.
+        pub fn activate_paging(vmtoken: PhysAddr);
 
         /// Flush TLB by the associated `vaddr`, or flush the entire TLB. (`vaddr` is `None`).
         pub(crate) fn flush_tlb(vaddr: Option<VirtAddr>);
@@ -124,7 +126,7 @@ hal_fn_def! {
         pub(crate) fn console_write_early(_s: &str) {}
     }
 
-    /// Context switch.
+    /// User context.
     pub mod context: common::context {
         /// Enter user mode.
         pub fn context_run(context: &mut UserContext) {
@@ -148,7 +150,7 @@ hal_fn_def! {
     /// Thread spawning.
     pub mod thread: common::thread {
         /// Spawn a new thread.
-        pub fn spawn(future: Pin<Box<dyn Future<Output = ()> + Send + 'static>>, vmtoken: usize);
+        pub fn spawn(future: impl Future<Output = ()> + Send + 'static);
 
         /// Set tid and pid of current task.
         pub fn set_tid(tid: u64, pid: u64);

kernel-hal/src/lib.rs

@@ -1,6 +1,7 @@
 //! Hardware Abstraction Layer
 
 #![cfg_attr(not(feature = "libos"), no_std)]
+#![cfg_attr(feature = "libos", feature(thread_id_value))]
 #![feature(asm)]
 #![feature(doc_cfg)]
 #![deny(warnings)]

kernel-hal/src/libos/cpu.rs

@@ -0,0 +1,7 @@
+hal_fn_impl! {
+    impl mod crate::hal_fn::cpu {
+        fn cpu_id() -> u8 {
+            std::thread::current().id().as_u64().get() as u8
+        }
+    }
+}

kernel-hal/src/libos/mod.rs

@@ -4,6 +4,7 @@ mod mock_mem;
 
 pub mod boot;
 pub mod config;
+pub mod cpu;
 pub mod mem;
 pub mod thread;
 pub mod timer;
@@ -14,9 +15,9 @@ pub mod vm;
 #[doc(cfg(feature = "libos"))]
 pub mod libos;
 
-pub use super::hal_fn::{context, cpu, interrupt, rand};
+pub use super::hal_fn::{context, interrupt, rand};
 
-hal_fn_impl_default!(context, cpu, interrupt, rand, super::hal_fn::console);
+hal_fn_impl_default!(context, interrupt, rand, super::hal_fn::console);
 
 #[cfg(target_os = "macos")]
 mod macos;

kernel-hal/src/libos/thread.rs

@@ -1,7 +1,7 @@
 //! Thread spawning.
 
 use async_std::task_local;
-use core::{cell::Cell, future::Future, pin::Pin};
+use core::{cell::Cell, future::Future};
 
 task_local! {
     static TID: Cell<u64> = Cell::new(0);
@@ -10,7 +10,7 @@ task_local! {
 
 hal_fn_impl! {
     impl mod crate::hal_fn::thread {
-        fn spawn(future: Pin<Box<dyn Future<Output = ()> + Send + 'static>>, _vmtoken: usize) {
+        fn spawn(future: impl Future<Output = ()> + Send + 'static) {
             async_std::task::spawn(future);
         }
 

kernel-hal/src/libos/timer.rs

@@ -12,8 +12,8 @@ hal_fn_impl! {
         }
 
         fn timer_set(deadline: Duration, callback: Box<dyn FnOnce(Duration) + Send + Sync>) {
-            let dur = deadline - timer_now();
             task::spawn(async move {
+                let dur = deadline - timer_now();
                 task::sleep(dur).await;
                 callback(timer_now());
             });

kernel-hal/src/libos/vm.rs

@@ -5,9 +5,8 @@ use crate::{addr::is_aligned, MMUFlags, PhysAddr, VirtAddr, PAGE_SIZE};
 
 hal_fn_impl! {
     impl mod crate::hal_fn::vm {
-        fn current_vmtoken() -> PhysAddr {
+        fn current_vmtoken() -> PhysAddr { 0 }
-            0
+        fn activate_paging(_vmtoken: PhysAddr) {}
-        }
     }
 }
 

linux-syscall/src/file/poll.rs

@@ -91,7 +91,7 @@ impl Syscall<'_> {
                             let waker = cx.waker().clone();
                             timer::timer_set(
                                 Duration::from_millis(deadline as u64),
-                                Box::new(move |_| waker.wake()),
+                                Box::new(move |_| waker.wake_by_ref()),
                             );
                         }
                     }
@@ -237,7 +237,7 @@ impl Syscall<'_> {
                             let waker = cx.waker().clone();
                             timer::timer_set(
                                 Duration::from_millis(deadline as u64),
-                                Box::new(move |_| waker.wake()),
+                                Box::new(move |_| waker.wake_by_ref()),
                             );
                         }
                     }

loader/src/linux.rs

@@ -47,6 +47,10 @@ pub fn run(args: Vec<String>, envs: Vec<String>, rootfs: Arc<dyn FileSystem>) ->
     proc
 }
 
+fn thread_fn(thread: CurrentThread) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>> {
+    Box::pin(run_user(thread))
+}
+
 /// The function of a new thread.
 ///
 /// loop:
@@ -55,7 +59,7 @@ pub fn run(args: Vec<String>, envs: Vec<String>, rootfs: Arc<dyn FileSystem>) ->
 /// - enter user mode
 /// - handle trap/interrupt/syscall according to the return value
 /// - return the context to the user thread
-async fn new_thread(thread: CurrentThread) {
+async fn run_user(thread: CurrentThread) {
     kernel_hal::thread::set_tid(thread.id(), thread.proc().id());
     loop {
         // wait
@@ -70,7 +74,7 @@ async fn new_thread(thread: CurrentThread) {
         trace!("back from user: {:#x?}", cx);
         // handle trap/interrupt/syscall
 
-        if let Err(err) = handler_user_trap(&thread, &mut cx).await {
+        if let Err(err) = handle_user_trap(&thread, &mut cx).await {
             thread.exit_linux(err as i32);
         }
 
@@ -78,11 +82,7 @@ async fn new_thread(thread: CurrentThread) {
     }
 }
 
-fn thread_fn(thread: CurrentThread) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>> {
+async fn handle_user_trap(thread: &CurrentThread, cx: &mut UserContext) -> ZxResult {
-    Box::pin(new_thread(thread))
-}
-
-async fn handler_user_trap(thread: &CurrentThread, cx: &mut UserContext) -> ZxResult {
     let pid = thread.proc().id();
 
     #[cfg(target_arch = "x86_64")]

loader/src/zircon.rs

@@ -183,7 +183,11 @@ kcounter!(EXCEPTIONS_USER, "exceptions.user");
 kcounter!(EXCEPTIONS_TIMER, "exceptions.timer");
 kcounter!(EXCEPTIONS_PGFAULT, "exceptions.pgfault");
 
-async fn new_thread(thread: CurrentThread) {
+fn thread_fn(thread: CurrentThread) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>> {
+    Box::pin(run_user(thread))
+}
+
+async fn run_user(thread: CurrentThread) {
     kernel_hal::thread::set_tid(thread.id(), thread.proc().id());
     if thread.is_first_thread() {
         thread
@@ -216,59 +220,60 @@ async fn new_thread(thread: CurrentThread) {
         EXCEPTIONS_USER.add(1);
 
         let trap_num = cx.trap_num;
-        #[cfg(target_arch = "x86_64")]
-        let error_code = cx.error_code;
         thread.end_running(cx);
 
-        #[cfg(target_arch = "aarch64")]
+        if let Err(e) = handler_user_trap(&thread, trap_num).await {
-        match trap_num {
+            thread.handle_exception(e).await;
-            0 => handle_syscall(&thread).await,
-            _ => unimplemented!(),
-        }
-        #[cfg(target_arch = "x86_64")]
-        match trap_num {
-            0x100 => handle_syscall(&thread).await,
-            0x20..=0xff => {
-                kernel_hal::interrupt::handle_irq(trap_num);
-                // TODO: configurable
-                if trap_num == 0xf1 {
-                    EXCEPTIONS_TIMER.add(1);
-                    kernel_hal::thread::yield_now().await;
-                }
-            }
-            0xe => {
-                EXCEPTIONS_PGFAULT.add(1);
-                let (vaddr, flags) = kernel_hal::context::fetch_page_fault_info(error_code);
-                info!(
-                    "page fault from user mode {:#x} {:#x?} {:?}",
-                    vaddr, error_code, flags
-                );
-                let vmar = thread.proc().vmar();
-                if let Err(err) = vmar.handle_page_fault(vaddr, flags) {
-                    error!("handle_page_fault error: {:?}", err);
-                    thread.handle_exception(ExceptionType::FatalPageFault).await;
-                }
-            }
-            0x8 => thread.with_context(|cx| {
-                panic!("Double fault from user mode! {:#x?}", cx);
-            }),
-            num => {
-                let type_ = match num {
-                    0x1 => ExceptionType::HardwareBreakpoint,
-                    0x3 => ExceptionType::SoftwareBreakpoint,
-                    0x6 => ExceptionType::UndefinedInstruction,
-                    0x17 => ExceptionType::UnalignedAccess,
-                    _ => ExceptionType::General,
-                };
-                thread.handle_exception(type_).await;
-            }
         }
     }
     thread.handle_exception(ExceptionType::ThreadExiting).await;
 }
 
-fn thread_fn(thread: CurrentThread) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>> {
+async fn handler_user_trap(thread: &CurrentThread, trap_num: usize) -> Result<(), ExceptionType> {
-    Box::pin(new_thread(thread))
+    #[cfg(target_arch = "aarch64")]
+    match trap_num {
+        0 => handle_syscall(&thread).await,
+        _ => unimplemented!(),
+    }
+    #[cfg(target_arch = "x86_64")]
+    match trap_num {
+        0x100 => handle_syscall(thread).await,
+        0x20..=0xff => {
+            kernel_hal::interrupt::handle_irq(trap_num);
+            // TODO: configurable
+            if trap_num == 0xf1 {
+                EXCEPTIONS_TIMER.add(1);
+                kernel_hal::thread::yield_now().await;
+            }
+        }
+        0xe => {
+            EXCEPTIONS_PGFAULT.add(1);
+            let error_code = thread.with_context(|cx| cx.error_code);
+            let (vaddr, flags) = kernel_hal::context::fetch_page_fault_info(error_code);
+            info!(
+                "page fault from user mode {:#x} {:#x?} {:?}",
+                vaddr, error_code, flags
+            );
+            let vmar = thread.proc().vmar();
+            if let Err(err) = vmar.handle_page_fault(vaddr, flags) {
+                error!("handle_page_fault error: {:?}", err);
+                return Err(ExceptionType::FatalPageFault);
+            }
+        }
+        0x8 => thread.with_context(|cx| {
+            panic!("Double fault from user mode! {:#x?}", cx);
+        }),
+        num => {
+            return Err(match num {
+                0x1 => ExceptionType::HardwareBreakpoint,
+                0x3 => ExceptionType::SoftwareBreakpoint,
+                0x6 => ExceptionType::UndefinedInstruction,
+                0x17 => ExceptionType::UnalignedAccess,
+                _ => ExceptionType::General,
+            })
+        }
+    }
+    Ok(())
 }
 
 async fn handle_syscall(thread: &CurrentThread) {

zircon-object/src/task/job.rs

@@ -299,7 +299,8 @@ pub struct JobInfo {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::task::{CurrentThread, Status, Thread, ThreadState, TASK_RETCODE_SYSCALL_KILL};
+    use crate::task::{Status, Thread, ThreadState, TASK_RETCODE_SYSCALL_KILL};
+    use core::time::Duration;
 
     #[test]
     fn create() {
@@ -439,14 +440,28 @@ mod tests {
         assert!(!job.is_empty());
     }
 
-    #[test]
+    #[async_std::test]
-    fn kill() {
+    async fn kill() {
         let root_job = Job::root();
         let job = Job::create_child(&root_job).expect("failed to create job");
         let proc = Process::create(&root_job, "proc").expect("failed to create process");
         let thread = Thread::create(&proc, "thread").expect("failed to create thread");
-        let current_thread = CurrentThread(thread.clone());
+        thread
+            .start(0, 0, 0, 0, |thread| {
+                std::boxed::Box::pin(async {
+                    println!("should not be killed");
+                    async_std::task::sleep(Duration::from_millis(1000)).await;
+                    {
+                        // FIXME
+                        drop(thread);
+                        async_std::task::sleep(Duration::from_millis(1000)).await;
+                    }
+                    unreachable!("should be killed");
+                })
+            })
+            .expect("failed to start thread");
 
+        async_std::task::sleep(Duration::from_millis(500)).await;
         root_job.kill();
         assert!(root_job.inner.lock().killed);
         assert!(job.inner.lock().killed);
@@ -458,7 +473,8 @@ mod tests {
         assert!(!proc.signal().contains(Signal::PROCESS_TERMINATED));
         assert!(!thread.signal().contains(Signal::THREAD_TERMINATED));
 
-        std::mem::drop(current_thread);
+        // wait for killing...
+        async_std::task::sleep(Duration::from_millis(1000)).await;
         assert!(root_job.inner.lock().killed);
         assert!(job.inner.lock().killed);
         assert_eq!(proc.status(), Status::Exited(TASK_RETCODE_SYSCALL_KILL));

zircon-object/src/task/thread.rs

@@ -178,8 +178,11 @@ bitflags! {
     }
 }
 
+type ThreadFuture = dyn Future<Output = ()> + Send;
+type ThreadFuturePinned = Pin<Box<ThreadFuture>>;
+
 /// The type of a new thread function.
-pub type ThreadFn = fn(thread: CurrentThread) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
+pub type ThreadFn = fn(thread: CurrentThread) -> ThreadFuturePinned;
 
 impl Thread {
     /// Create a new thread.
@@ -270,8 +273,7 @@ impl Thread {
             }
             inner.change_state(ThreadState::Running, &self.base);
         }
-        let vmtoken = self.proc().vmar().table_phys();
+        self.spawn(thread_fn);
-        kernel_hal::thread::spawn(thread_fn(CurrentThread(self.clone())), vmtoken);
         Ok(())
     }
 
@@ -288,8 +290,7 @@ impl Thread {
             }
             inner.change_state(ThreadState::Running, &self.base);
         }
-        let vmtoken = self.proc().vmar().table_phys();
+        self.spawn(thread_fn);
-        kernel_hal::thread::spawn(thread_fn(CurrentThread(self.clone())), vmtoken);
         Ok(())
     }
 
@@ -309,8 +310,7 @@ impl Thread {
             }
             inner.change_state(ThreadState::Running, &self.base);
         }
-        let vmtoken = self.proc().vmar().table_phys();
+        self.spawn(thread_fn);
-        kernel_hal::thread::spawn(thread_fn(CurrentThread(self.clone())), vmtoken);
         Ok(())
     }
 
@@ -337,7 +337,7 @@ impl Thread {
         }
         inner.change_state(ThreadState::Dying, &self.base);
         if let Some(waker) = inner.waker.take() {
-            waker.wake();
+            waker.wake_by_ref();
         }
         // For blocking thread, use the killer
         if let Some(killer) = inner.killer.take() {
@@ -449,6 +449,21 @@ impl Thread {
         context.general.gsbase = gsbase;
         Ok(())
     }
+
+    /// Spawn the future returned by `thread_fn` in this thread.
+    fn spawn(self: &Arc<Self>, thread_fn: ThreadFn) {
+        let current = CurrentThread(self.clone());
+        let future = thread_fn(current);
+        kernel_hal::thread::spawn(ThreadSwitchFuture::new(self.clone(), future));
+    }
+
+    /// Terminate the current running thread.
+    fn terminate(&self) {
+        let mut inner = self.inner.lock();
+        self.exceptionate.shutdown();
+        inner.change_state(ThreadState::Dead, &self.base);
+        self.proc().remove_thread(self.base.id);
+    }
 }
 
 impl Task for Thread {
@@ -471,7 +486,7 @@ impl Task for Thread {
             let state = inner.state;
             inner.change_state(state, &self.base);
             if let Some(waker) = inner.waker.take() {
-                waker.wake();
+                waker.wake_by_ref();
             }
         }
     }
@@ -494,27 +509,14 @@ impl Task for Thread {
 ///
 /// [`Thread`]: crate::task::Thread
 /// [`Thread::start`]: crate::task::Thread::start
-pub struct CurrentThread(pub(super) Arc<Thread>);
+pub struct CurrentThread(Arc<Thread>);
-
-impl Deref for CurrentThread {
-    type Target = Arc<Thread>;
-
-    fn deref(&self) -> &Self::Target {
-        &self.0
-    }
-}
-
-impl Drop for CurrentThread {
-    /// Terminate the current running thread.
-    fn drop(&mut self) {
-        let mut inner = self.inner.lock();
-        self.exceptionate.shutdown();
-        inner.change_state(ThreadState::Dead, &self.base);
-        self.proc().remove_thread(self.base.id);
-    }
-}
 
 impl CurrentThread {
+    /// Returns the inner structure `Arc<Thread>`.
+    pub fn inner(&self) -> Arc<Thread> {
+        self.0.clone()
+    }
+
     /// Exit the current thread.
     ///
     /// The thread do not terminate immediately when exited. It is just made dying.
@@ -687,6 +689,20 @@ impl CurrentThread {
     }
 }
 
+impl Deref for CurrentThread {
+    type Target = Arc<Thread>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl Drop for CurrentThread {
+    fn drop(&mut self) {
+        self.terminate();
+    }
+}
+
 /// `into_result` returns `Self` if the type parameter is already a `ZxResult`,
 /// otherwise wraps the value in an `Ok`.
 ///
@@ -758,6 +774,28 @@ pub struct ThreadInfo {
     cpu_affinity_mask: [u64; 8],
 }
 
+struct ThreadSwitchFuture {
+    thread: Arc<Thread>,
+    future: Mutex<ThreadFuturePinned>,
+}
+
+impl ThreadSwitchFuture {
+    pub fn new(thread: Arc<Thread>, future: ThreadFuturePinned) -> Self {
+        Self {
+            future: Mutex::new(future),
+            thread,
+        }
+    }
+}
+
+impl Future for ThreadSwitchFuture {
+    type Output = ();
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        kernel_hal::vm::activate_paging(self.thread.proc().vmar().table_phys());
+        self.future.lock().as_mut().poll(cx)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::job::Job;

zircon-syscall/src/futex.rs

@@ -4,10 +4,10 @@ use {
 };
 
 impl Syscall<'_> {
-    /// Wait on a futex.  
+    /// Wait on a futex.
-    ///  
+    ///
-    /// This system call function atomically verifies that `value_ptr` still contains the value `current_value`   
+    /// This system call function atomically verifies that `value_ptr` still contains the value `current_value`
-    /// and sleeps until the futex is made available by a call to `zx_futex_wake`  
+    /// and sleeps until the futex is made available by a call to `zx_futex_wake`
     pub async fn sys_futex_wait(
         &self,
         value_ptr: UserInPtr<AtomicI32>,
@@ -30,14 +30,14 @@ impl Syscall<'_> {
         } else {
             Some(proc.get_object::<Thread>(new_futex_owner)?)
         };
-        let future = futex.wait_with_owner(current_value, Some((*self.thread).clone()), new_owner);
+        let future = futex.wait_with_owner(current_value, Some(self.thread.inner()), new_owner);
         self.thread
             .blocking_run(future, ThreadState::BlockedFutex, deadline.into(), None)
             .await?;
         Ok(())
     }
-    /// Wake some waiters and requeue other waiters.   
+    /// Wake some waiters and requeue other waiters.
-    ///   
+    ///
     /// Wake some number of threads waiting on a futex, and move more waiters to another wait queue.
     pub fn sys_futex_requeue(
         &self,
@@ -78,7 +78,7 @@ impl Syscall<'_> {
         Ok(())
     }
 
-    /// Wake some number of threads waiting on a futex.   
+    /// Wake some number of threads waiting on a futex.
     ///
     /// > Waking up zero threads is not an error condition. Passing in an unallocated address for value_ptr is not an error condition.
     pub fn sys_futex_wake(&self, value_ptr: UserInPtr<AtomicI32>, count: u32) -> ZxResult {