Add System V semaphore basic support

kernel/aster-nix/src/ipc/mod.rs

@@ -0,0 +1,103 @@
+// SPDX-License-Identifier: MPL-2.0
+
+use crate::{
+    prelude::*,
+    process::{Gid, Uid},
+};
+
+pub mod semaphore;
+
+#[allow(non_camel_case_types)]
+pub type key_t = i32;
+
+bitflags! {
+    pub struct IpcFlags: u32{
+        /// Create key if key does not exist
+        const IPC_CREAT  = 1 << 9;
+        /// Fail if key exists
+        const IPC_EXCL  = 1 << 10;
+        /// Return error on wait
+        const IPC_NOWAIT  = 1 << 11;
+        /// Undo the operation on exit
+        const SEM_UNDO = 1 << 12;
+    }
+}
+
+#[repr(i32)]
+#[derive(Debug, Clone, Copy, TryFromInt)]
+#[allow(non_camel_case_types)]
+pub enum IpcControlCmd {
+    IPC_RMID = 0,
+    IPC_SET = 1,
+    IPC_STAT = 2,
+
+    SEM_GETPID = 11,
+    SEM_GETVAL = 12,
+    SEM_GETALL = 13,
+    SEM_GETNCNT = 14,
+    SEM_GETZCNT = 15,
+    SEM_SETVAL = 16,
+    SEM_SETALL = 17,
+}
+
+#[derive(Debug)]
+#[allow(dead_code)]
+pub struct IpcPermission {
+    key: key_t,
+    /// Owner's UID
+    uid: Uid,
+    /// Owner's GID
+    gid: Gid,
+    /// Creator's UID
+    cuid: Uid,
+    /// Creator's GID
+    cguid: Gid,
+    /// Permission mode
+    mode: u16,
+}
+
+impl IpcPermission {
+    pub fn key(&self) -> key_t {
+        self.key
+    }
+
+    /// Returns owner's UID
+    pub fn uid(&self) -> Uid {
+        self.uid
+    }
+
+    /// Returns owner's GID
+    pub fn gid(&self) -> Gid {
+        self.gid
+    }
+
+    /// Returns creator's UID
+    pub fn cuid(&self) -> Uid {
+        self.cuid
+    }
+
+    /// Returns creator's GID
+    pub fn cguid(&self) -> Gid {
+        self.cguid
+    }
+
+    /// Returns permission mode
+    pub fn mode(&self) -> u16 {
+        self.mode
+    }
+
+    pub(self) fn new_sem_perm(key: key_t, uid: Uid, gid: Gid, mode: u16) -> Self {
+        Self {
+            key,
+            uid,
+            gid,
+            cuid: uid,
+            cguid: gid,
+            mode,
+        }
+    }
+}
+
+pub(super) fn init() {
+    semaphore::init();
+}

kernel/aster-nix/src/ipc/semaphore/mod.rs

@@ -0,0 +1,11 @@
+// SPDX-License-Identifier: MPL-2.0
+
+//! Semaphore for the system, including System V semaphore and
+//! POSIX semaphore.
+
+pub mod posix;
+pub mod system_v;
+
+pub(super) fn init() {
+    system_v::init();
+}

kernel/aster-nix/src/ipc/semaphore/posix/mod.rs

@@ -0,0 +1,3 @@
+// SPDX-License-Identifier: MPL-2.0
+
+//! POSIX semaphore.

kernel/aster-nix/src/ipc/semaphore/system_v/mod.rs

@@ -0,0 +1,20 @@
+// SPDX-License-Identifier: MPL-2.0
+
+//! System V semaphore.
+
+use bitflags::bitflags;
+
+pub mod sem;
+pub mod sem_set;
+
+bitflags! {
+    pub struct PermissionMode: u16{
+        const ALTER  = 0o002;
+        const WRITE  = 0o002;
+        const READ   = 0o004;
+    }
+}
+
+pub(super) fn init() {
+    sem_set::init();
+}

kernel/aster-nix/src/ipc/semaphore/system_v/sem.rs

@@ -0,0 +1,322 @@
+// SPDX-License-Identifier: MPL-2.0
+
+use core::{
+    sync::atomic::{AtomicU16, AtomicU64, Ordering},
+    time::Duration,
+};
+
+use ostd::sync::{Mutex, Waiter, Waker};
+
+use super::sem_set::SEMVMX;
+use crate::{
+    ipc::{key_t, semaphore::system_v::sem_set::sem_sets, IpcFlags},
+    prelude::*,
+    process::{Pid, Process},
+    time::{
+        clocks::{RealTimeCoarseClock, JIFFIES_TIMER_MANAGER},
+        timer::Timeout,
+    },
+};
+
+#[derive(Clone, Copy, Debug, Pod)]
+#[repr(C)]
+pub struct SemBuf {
+    sem_num: u16,
+    sem_op: i16,
+    sem_flags: i16,
+}
+
+impl SemBuf {
+    pub fn sem_num(&self) -> u16 {
+        self.sem_num
+    }
+
+    pub fn sem_op(&self) -> i16 {
+        self.sem_op
+    }
+
+    pub fn sem_flags(&self) -> i16 {
+        self.sem_flags
+    }
+}
+
+#[repr(u16)]
+#[derive(Debug, TryFromInt, Clone, Copy)]
+enum Status {
+    Normal = 0,
+    Pending = 1,
+    Removed = 2,
+}
+
+struct AtomicStatus(AtomicU16);
+
+impl AtomicStatus {
+    fn new(status: Status) -> Self {
+        Self(AtomicU16::new(status as u16))
+    }
+
+    fn status(&self) -> Status {
+        Status::try_from(self.0.load(Ordering::Relaxed)).unwrap()
+    }
+
+    fn set_status(&self, status: Status) {
+        self.0.store(status as u16, Ordering::Relaxed);
+    }
+}
+
+struct PendingOp {
+    sem_buf: SemBuf,
+    status: Arc<AtomicStatus>,
+    waker: Arc<Waker>,
+    pid: Pid,
+    process: Weak<Process>,
+}
+
+impl Debug for PendingOp {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        f.debug_struct("PendingOp")
+            .field("sem_buf", &self.sem_buf)
+            .field("status", &(self.status.status()))
+            .field("pid", &self.pid)
+            .finish()
+    }
+}
+
+#[derive(Debug)]
+pub struct Semaphore {
+    val: Mutex<i32>,
+    /// PID of the process that last modified semaphore.
+    /// - through semop with op != 0
+    /// - through semctl with SETVAL and SETALL
+    /// - through SEM_UNDO when task exit
+    latest_modified_pid: RwMutex<Pid>,
+    /// Pending alter operations. For each pending operation, it has `sem_op < 0`.
+    pending_alters: Mutex<LinkedList<Box<PendingOp>>>,
+    /// Pending zeros operations. For each pending operation, it has `sem_op = 0`.
+    pending_const: Mutex<LinkedList<Box<PendingOp>>>,
+    /// Last semop time.
+    sem_otime: AtomicU64,
+}
+
+impl Semaphore {
+    pub fn set_val(&self, val: i32) -> Result<()> {
+        if !(0..SEMVMX).contains(&val) {
+            return_errno!(Errno::ERANGE);
+        }
+
+        let mut current_val = self.val.lock();
+        *current_val = val;
+        *self.latest_modified_pid.write() = current!().pid();
+
+        self.update_pending_ops(current_val);
+        Ok(())
+    }
+
+    pub fn val(&self) -> i32 {
+        *self.val.lock()
+    }
+
+    pub fn last_modified_pid(&self) -> Pid {
+        *self.latest_modified_pid.read()
+    }
+
+    pub fn sem_otime(&self) -> Duration {
+        Duration::from_secs(self.sem_otime.load(Ordering::Relaxed))
+    }
+
+    pub fn pending_zero_count(&self) -> usize {
+        self.pending_const.lock().len()
+    }
+
+    pub fn pending_alter_count(&self) -> usize {
+        self.pending_alters.lock().len()
+    }
+
+    /// Notifies the semaphore that the semaphore sets it belongs to have been removed.
+    pub(super) fn removed(&self) {
+        let mut pending_alters = self.pending_alters.lock();
+        for pending_alter in pending_alters.iter_mut() {
+            pending_alter.status.set_status(Status::Removed);
+            pending_alter.waker.wake_up();
+        }
+        pending_alters.clear();
+
+        let mut pending_const = self.pending_const.lock();
+        for pending_const in pending_const.iter_mut() {
+            pending_const.status.set_status(Status::Removed);
+            pending_const.waker.wake_up();
+        }
+        pending_const.clear();
+    }
+
+    pub(super) fn new(val: i32) -> Self {
+        Self {
+            val: Mutex::new(val),
+            latest_modified_pid: RwMutex::new(current!().pid()),
+            pending_alters: Mutex::new(LinkedList::new()),
+            pending_const: Mutex::new(LinkedList::new()),
+            sem_otime: AtomicU64::new(0),
+        }
+    }
+
+    fn update_otime(&self) {
+        self.sem_otime.store(
+            RealTimeCoarseClock::get().read_time().as_secs(),
+            Ordering::Relaxed,
+        );
+    }
+
+    fn sem_op(&self, sem_buf: SemBuf, timeout: Option<Duration>) -> Result<()> {
+        let mut val = self.val.lock();
+        let sem_op = sem_buf.sem_op;
+
+        let flags = IpcFlags::from_bits(sem_buf.sem_flags as u32).unwrap();
+        if flags.contains(IpcFlags::SEM_UNDO) {
+            todo!()
+        }
+
+        // Operate val
+        let positive_condition = sem_op.is_positive();
+        let negative_condition = sem_op.is_negative() && sem_op.abs() as i32 <= *val;
+        let zero_condition = sem_op == 0 && *val == 0;
+
+        if positive_condition || negative_condition {
+            let new_val = val
+                .checked_add(i32::from(sem_op))
+                .ok_or(Error::new(Errno::ERANGE))?;
+            if new_val > SEMVMX {
+                return_errno!(Errno::ERANGE);
+            }
+
+            *val = new_val;
+            *self.latest_modified_pid.write() = current!().pid();
+            self.update_otime();
+
+            self.update_pending_ops(val);
+            return Ok(());
+        } else if zero_condition {
+            return Ok(());
+        }
+
+        // Need to wait for the semaphore
+        if flags.contains(IpcFlags::IPC_NOWAIT) {
+            return_errno!(Errno::EAGAIN);
+        }
+
+        // Add current to pending list
+        let (waiter, waker) = Waiter::new_pair();
+        let status = Arc::new(AtomicStatus::new(Status::Pending));
+        let current = current!();
+        let pid = current.pid();
+        let pending_op = Box::new(PendingOp {
+            sem_buf,
+            status: status.clone(),
+            waker: waker.clone(),
+            process: Arc::downgrade(&current),
+            pid,
+        });
+        if sem_op == 0 {
+            self.pending_const.lock().push_back(pending_op);
+        } else {
+            self.pending_alters.lock().push_back(pending_op);
+        }
+        drop(val);
+
+        // Wait
+        if let Some(timeout) = timeout {
+            let jiffies_timer = JIFFIES_TIMER_MANAGER.get().unwrap().create_timer(move || {
+                waker.wake_up();
+            });
+            jiffies_timer.set_timeout(Timeout::After(timeout));
+        }
+        waiter.wait();
+
+        // Check status and return
+        match status.status() {
+            Status::Normal => Ok(()),
+            Status::Removed => Err(Error::new(Errno::EIDRM)),
+            Status::Pending => {
+                let mut pending_ops = if sem_op == 0 {
+                    self.pending_const.lock()
+                } else {
+                    self.pending_alters.lock()
+                };
+                pending_ops.retain(|op| op.pid != pid);
+                Err(Error::new(Errno::EAGAIN))
+            }
+        }
+    }
+
+    /// Updates pending ops after the val changed.
+    fn update_pending_ops(&self, mut val: MutexGuard<i32>) {
+        debug_assert!(*val >= 0);
+        trace!("Updating pending ops, semaphore before: {:?}", *val);
+
+        // Two steps:
+        // 1. Remove the pending_alters with `sem_op < 0` if it can.
+        // 2. If val is equal to 0, then clear pending_const
+
+        // Step one:
+        let mut pending_alters = self.pending_alters.lock();
+        pending_alters.retain_mut(|op| {
+            if *val == 0 {
+                return true;
+            }
+            // Check if the process alive.
+            if op.process.upgrade().is_none() {
+                return false;
+            }
+            debug_assert!(op.sem_buf.sem_op < 0);
+
+            if op.sem_buf.sem_op.abs() as i32 <= *val {
+                trace!(
+                    "Found removable pending op, op: {:?}, pid:{:?}",
+                    op.sem_buf.sem_op,
+                    op.pid
+                );
+
+                *val += i32::from(op.sem_buf.sem_op);
+                *self.latest_modified_pid.write() = op.pid;
+                self.update_otime();
+                op.status.set_status(Status::Normal);
+                op.waker.wake_up();
+                false
+            } else {
+                true
+            }
+        });
+
+        // Step two:
+        if *val == 0 {
+            let mut pending_const = self.pending_const.lock();
+            pending_const.iter().for_each(|op| {
+                op.status.set_status(Status::Normal);
+                if op.process.upgrade().is_some() {
+                    trace!("Found removable pending op, op: 0, pid:{:?}", op.pid);
+                    op.waker.wake_up();
+                }
+            });
+            pending_const.clear();
+        }
+        trace!("Updated pending ops, semaphore after: {:?}", *val);
+    }
+}
+
+pub fn sem_op(sem_id: key_t, sem_buf: SemBuf, timeout: Option<Duration>) -> Result<()> {
+    debug_assert!(sem_id > 0);
+    debug!("[semop] sembuf: {:?}", sem_buf);
+
+    let sem = {
+        let sem_sets = sem_sets();
+        let sem_set = sem_sets.get(&sem_id).ok_or(Error::new(Errno::EINVAL))?;
+        // TODO: Support permission check
+        warn!("Semaphore operation doesn't support permission check now");
+
+        sem_set
+            .get(sem_buf.sem_num as usize)
+            .ok_or(Error::new(Errno::EFBIG))?
+            .clone()
+    };
+
+    sem.sem_op(sem_buf, timeout)
+}

kernel/aster-nix/src/ipc/semaphore/system_v/sem_set.rs

@@ -0,0 +1,188 @@
+// SPDX-License-Identifier: MPL-2.0
+
+use alloc::{collections::btree_map::BTreeMap, vec::Vec};
+use core::{
+    sync::atomic::{AtomicU64, Ordering},
+    time::Duration,
+};
+
+use aster_rights::ReadOp;
+use id_alloc::IdAlloc;
+use ostd::sync::{Mutex, RwMutex, RwMutexReadGuard, RwMutexWriteGuard};
+use spin::Once;
+
+use super::PermissionMode;
+use crate::{
+    ipc::{key_t, semaphore::system_v::sem::Semaphore, IpcPermission},
+    prelude::*,
+    process::Credentials,
+    time::clocks::RealTimeCoarseClock,
+};
+
+// The following constant values are derived from the default values in Linux.
+
+/// Maximum number of semaphore sets.
+pub const SEMMNI: usize = 32000;
+/// Maximum number of semaphores per semaphore ID.
+pub const SEMMSL: usize = 32000;
+/// Maximum number of seaphores in all semaphore sets.
+pub const SEMMNS: usize = SEMMNI * SEMMSL;
+/// Maximum number of operations for semop.
+pub const SEMOPM: usize = 500;
+/// MAximum semaphore value.
+pub const SEMVMX: i32 = 32767;
+/// Maximum value that can be recored for semaphore adjustment (SEM_UNDO).
+pub const SEMAEM: i32 = SEMVMX;
+
+#[derive(Debug)]
+pub struct SemaphoreSet {
+    /// Number of semaphores in the set
+    nsems: usize,
+    /// Semaphores
+    sems: Box<[Arc<Semaphore>]>,
+    /// Semaphore permission
+    permission: IpcPermission,
+    /// Creation time or last modification via `semctl`
+    sem_ctime: AtomicU64,
+}
+
+impl SemaphoreSet {
+    pub fn nsems(&self) -> usize {
+        self.nsems
+    }
+
+    pub fn get(&self, index: usize) -> Option<&Arc<Semaphore>> {
+        self.sems.get(index)
+    }
+
+    pub fn permission(&self) -> &IpcPermission {
+        &self.permission
+    }
+
+    pub fn sem_ctime(&self) -> Duration {
+        Duration::from_secs(self.sem_ctime.load(Ordering::Relaxed))
+    }
+
+    pub fn update_ctime(&self) {
+        self.sem_ctime.store(
+            RealTimeCoarseClock::get().read_time().as_secs(),
+            Ordering::Relaxed,
+        );
+    }
+
+    fn new(key: key_t, nsems: usize, mode: u16, credentials: Credentials<ReadOp>) -> Result<Self> {
+        debug_assert!(nsems <= SEMMSL);
+
+        let mut sems = Vec::with_capacity(nsems);
+        for _ in 0..nsems {
+            sems.push(Arc::new(Semaphore::new(0)));
+        }
+
+        let permission =
+            IpcPermission::new_sem_perm(key, credentials.euid(), credentials.egid(), mode);
+
+        Ok(Self {
+            nsems,
+            sems: sems.into_boxed_slice(),
+            permission,
+            sem_ctime: AtomicU64::new(RealTimeCoarseClock::get().read_time().as_secs()),
+        })
+    }
+}
+
+impl Drop for SemaphoreSet {
+    fn drop(&mut self) {
+        for sem in self.sems.iter() {
+            sem.removed();
+        }
+
+        ID_ALLOCATOR
+            .get()
+            .unwrap()
+            .lock()
+            .free(self.permission.key() as usize);
+    }
+}
+
+pub fn create_sem_set_with_id(
+    id: key_t,
+    nsems: usize,
+    mode: u16,
+    credentials: Credentials<ReadOp>,
+) -> Result<()> {
+    debug_assert!(nsems <= SEMMSL);
+    debug_assert!(id > 0);
+
+    ID_ALLOCATOR
+        .get()
+        .unwrap()
+        .lock()
+        .alloc_specific(id as usize)
+        .ok_or(Error::new(Errno::EEXIST))?;
+
+    let mut sem_sets = SEMAPHORE_SETS.write();
+    sem_sets.insert(id, SemaphoreSet::new(id, nsems, mode, credentials)?);
+
+    Ok(())
+}
+
+/// Checks the semaphore. Return Ok if the semaphore exists and pass the check.
+pub fn check_sem(id: key_t, nsems: Option<usize>, required_perm: PermissionMode) -> Result<()> {
+    debug_assert!(id > 0);
+
+    let sem_sets = SEMAPHORE_SETS.read();
+    let sem_set = sem_sets.get(&id).ok_or(Error::new(Errno::ENOENT))?;
+
+    if let Some(nsems) = nsems {
+        debug_assert!(nsems <= SEMMSL);
+        if nsems > sem_set.nsems() {
+            return_errno!(Errno::EINVAL);
+        }
+    }
+
+    if !required_perm.is_empty() {
+        // TODO: Support permission check
+        warn!("Semaphore doesn't support permission check now");
+    }
+
+    Ok(())
+}
+
+pub fn create_sem_set(nsems: usize, mode: u16, credentials: Credentials<ReadOp>) -> Result<key_t> {
+    debug_assert!(nsems <= SEMMSL);
+
+    let id = ID_ALLOCATOR
+        .get()
+        .unwrap()
+        .lock()
+        .alloc()
+        .ok_or(Error::new(Errno::ENOSPC))? as i32;
+
+    let mut sem_sets = SEMAPHORE_SETS.write();
+    sem_sets.insert(id, SemaphoreSet::new(id, nsems, mode, credentials)?);
+
+    Ok(id)
+}
+
+pub fn sem_sets<'a>() -> RwMutexReadGuard<'a, BTreeMap<key_t, SemaphoreSet>> {
+    SEMAPHORE_SETS.read()
+}
+
+pub fn sem_sets_mut<'a>() -> RwMutexWriteGuard<'a, BTreeMap<key_t, SemaphoreSet>> {
+    SEMAPHORE_SETS.write()
+}
+
+static ID_ALLOCATOR: Once<Mutex<IdAlloc>> = Once::new();
+
+/// Semaphore sets in system
+static SEMAPHORE_SETS: RwMutex<BTreeMap<key_t, SemaphoreSet>> = RwMutex::new(BTreeMap::new());
+
+pub(super) fn init() {
+    ID_ALLOCATOR.call_once(|| {
+        let mut id_alloc = IdAlloc::with_capacity(SEMMNI + 1);
+        // Remove the first index 0
+        id_alloc.alloc();
+
+        Mutex::new(id_alloc)
+    });
+}

kernel/aster-nix/src/lib.rs

@@ -23,6 +23,7 @@
 #![feature(step_trait)]
 #![feature(trait_alias)]
 #![feature(trait_upcasting)]
+#![feature(linked_list_retain)]
 #![register_tool(component_access_control)]
 
 use ostd::{
@@ -55,6 +56,7 @@ pub mod driver;
 pub mod error;
 pub mod events;
 pub mod fs;
+pub mod ipc;
 pub mod net;
 pub mod prelude;
 mod process;
@@ -91,6 +93,7 @@ fn init_thread() {
     thread::work_queue::init();
     net::lazy_init();
     fs::lazy_init();
+    ipc::init();
     // driver::pci::virtio::block::block_device_test();
     let thread = Thread::spawn_kernel_thread(ThreadOptions::new(|| {
         println!("[kernel] Hello world from kernel!");