proc: add userspace mutex system call

src/proc/elf.rs

@@ -15,7 +15,7 @@ use crate::{
         phys,
         pointer::{PhysicalRef, PhysicalRefMut},
         process::ProcessAddressSpace,
-        table::{EntryLevel, MapAttributes},
+        table::MapAttributes,
     },
     task::process::{ProcessImage, ProcessTlsInfo, ProcessTlsLayout},
 };
@@ -252,7 +252,7 @@ fn tls_segment(
                 source.read_exact(dst.deref_mut())
             },
             data_size,
-        );
+        )?;
     }
 
     if mem_size > data_size {
@@ -264,7 +264,7 @@ fn tls_segment(
                 Ok(())
             },
             mem_size - data_size,
-        );
+        )?;
     }
 
     Ok(tls)
@@ -294,8 +294,6 @@ pub fn load_elf_from_file(
             PT_TLS => {
                 assert!(tls.is_none());
                 tls.replace(tls_segment(space, phdr, &file)?);
-                // tls_master_address = tls_segment(space, phdr, &file)?;
-                // tls_size = phdr.p_memsz as usize;
             }
             _ => (),
         }

src/syscall/mod.rs

@@ -4,7 +4,7 @@ use core::{mem::MaybeUninit, time::Duration};
 use abi::{
     error::Error,
     io::{DeviceRequest, DirectoryEntry, FileAttr, FileMode, OpenOptions, RawFd, SeekFrom},
-    process::{ExitCode, Signal, SpawnOption, SpawnOptions, ThreadSpawnOptions},
+    process::{ExitCode, MutexOperation, Signal, SpawnOption, SpawnOptions, ThreadSpawnOptions},
     syscall::SyscallFunction,
 };
 use alloc::rc::Rc;
@@ -365,6 +365,22 @@ fn syscall_handler(func: SyscallFunction, args: &[u64]) -> Result<usize, Error>
             let id = process.spawn_thread(options)?;
             Ok(id.as_user() as _)
         }
+        SyscallFunction::Mutex => {
+            let address = args[0] as usize;
+            let op = arg_user_ref::<MutexOperation>(args[1] as usize)?;
+
+            let mutex = process.get_or_insert_mutex(address);
+
+            match op {
+                &MutexOperation::Wait(value, _timeout) => {
+                    block! { mutex.wait(value).await }.unwrap()
+                }
+                MutexOperation::Wake => mutex.wake(),
+                MutexOperation::WakeAll => mutex.wake_all(),
+            }
+
+            Ok(0)
+        }
         SyscallFunction::Exit => {
             let code = ExitCode::from(args[0] as i32);
             // TODO separate handlers for process exit and thread exit?

src/task/mod.rs

@@ -17,6 +17,7 @@ pub mod context;
 pub mod process;
 pub mod runtime;
 pub mod sched;
+pub mod sync;
 pub mod thread;
 
 pub use context::{Cpu, TaskContext};

src/task/process.rs

@@ -4,7 +4,7 @@ use core::{
     fmt,
     mem::size_of,
     pin::Pin,
-    sync::atomic::{AtomicU64, Ordering},
+    sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering},
     task::{Context, Poll},
 };
 
@@ -36,6 +36,7 @@ use crate::{
 
 use super::{
     runtime::QueueWaker,
+    sync::UserspaceMutex,
     thread::{Thread, ThreadId, ThreadState},
     TaskContext,
 };
@@ -133,6 +134,7 @@ struct ProcessInner {
     session_terminal: Option<VnodeRef>,
 
     threads: Vec<Arc<Thread>>,
+    mutexes: BTreeMap<usize, Arc<UserspaceMutex>>,
 }
 
 pub struct Process {
@@ -173,6 +175,7 @@ impl Process {
                 group_id: id,
                 session_terminal: None,
                 threads: Vec::new(),
+                mutexes: BTreeMap::new(),
             }),
 
             exit_waker: QueueWaker::new(),
@@ -322,47 +325,12 @@ impl Process {
 
             self.exit_waker.wake_all();
         }
-        //         // Scheduler still holds a lock of this process?
-        //         // TODO cancel Wait if a process was killed while suspended?
-        //         {
-        //             let inner = self.inner.lock();
-        //             let exit_status = ExitCode::from(inner.exit_status);
-        //             debugln!(
-        //                 "Handling exit of #{} with status {:?}",
-        //                 self.id(),
-        //                 exit_status
-        //             );
-        //
-        //             // TODO cleanup address space
-        //             // if let Some(space) = self.get_address_space() {
-        //             // }
-        //
-        //             self.io.lock().handle_exit();
-        //         }
-        //
-        //         // Notify any waiters we're done
-        //         self.exit_waker.wake_all();
     }
 
     /// Raises a signal for the specified process
     pub fn raise_signal(self: &Arc<Self>, signal: Signal) {
         let thread = self.inner.lock().threads[0].clone();
         thread.raise_signal(signal);
-
-        // // TODO if the process does not have any running/ready threads, pick one and wake it up
-        // if inner
-        //     .threads
-        //     .iter()
-        //     .all(|t| t.state.load(Ordering::Acquire) == ThreadState::Suspended)
-        // {
-        //     let thread = inner.threads[0].clone();
-
-        //     drop(inner);
-
-        //     thread.enqueue_somewhere();
-
-        //     return;
-        // }
     }
 
     /// Raises a signal for the specified process group
@@ -378,6 +346,15 @@ impl Process {
         }
     }
 
+    pub fn get_or_insert_mutex(&self, address: usize) -> Arc<UserspaceMutex> {
+        let mut inner = self.inner.lock();
+        inner
+            .mutexes
+            .entry(address)
+            .or_insert_with(|| Arc::new(UserspaceMutex::new(address)))
+            .clone()
+    }
+
     // Process list
     pub fn get(id: ProcessId) -> Option<Arc<Self>> {
         PROCESSES.lock().get(&id).cloned()

src/task/sync.rs

@@ -0,0 +1,67 @@
+use core::{
+    pin::Pin,
+    sync::atomic::{AtomicU32, Ordering},
+    task::{Context, Poll},
+};
+
+use alloc::sync::Arc;
+use futures_util::Future;
+
+use super::runtime::QueueWaker;
+
+pub struct UserspaceMutex {
+    queue: QueueWaker,
+    address: usize,
+}
+
+impl UserspaceMutex {
+    pub fn new(address: usize) -> Self {
+        Self {
+            queue: QueueWaker::new(),
+            address,
+        }
+    }
+
+    pub fn wait(self: Arc<Self>, compare_value: u32) -> impl Future<Output = ()> {
+        struct WaitFuture {
+            mutex: Arc<UserspaceMutex>,
+            compare_value: u32,
+        }
+
+        impl WaitFuture {
+            fn load(&self, ordering: Ordering) -> u32 {
+                let value = unsafe { &*(self.mutex.address as *const AtomicU32) };
+                value.load(ordering)
+            }
+        }
+
+        impl Future for WaitFuture {
+            type Output = ();
+
+            fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+                self.mutex.queue.register(cx.waker());
+
+                // Test the mutex
+                if self.load(Ordering::Acquire) != self.compare_value {
+                    self.mutex.queue.remove(cx.waker());
+                    Poll::Ready(())
+                } else {
+                    Poll::Pending
+                }
+            }
+        }
+
+        WaitFuture {
+            mutex: self,
+            compare_value,
+        }
+    }
+
+    pub fn wake(&self) {
+        self.queue.wake_one();
+    }
+
+    pub fn wake_all(&self) {
+        self.queue.wake_all();
+    }
+}