Add sigtimedwait syscall

src/libos/src/signal/do_sigtimedwait.rs

@@ -0,0 +1,153 @@
+use std::sync::Weak;
+use std::time::Duration;
+
+use super::{siginfo_t, SigNum, SigSet, Signal};
+use crate::events::{Observer, Waiter, WaiterQueueObserver};
+use crate::prelude::*;
+use crate::process::{ProcessRef, TermStatus, ThreadRef};
+
+pub fn do_sigtimedwait(interest: SigSet, timeout: Option<&Duration>) -> Result<siginfo_t> {
+    debug!(
+        "do_rt_sigtimedwait: interest: {:?}, timeout: {:?}",
+        interest, timeout,
+    );
+
+    let thread = current!();
+    let process = thread.process().clone();
+
+    // Interesting, blocked signals
+    let interest = {
+        let blocked = thread.sig_mask().read().unwrap();
+        *blocked & interest
+    };
+
+    let signal = match timeout {
+        None => dequeue_pending_signal(&interest, &thread, &process)
+            .ok_or_else(|| errno!(EAGAIN, "no interesting, pending signal"))?,
+        Some(timeout) => {
+            let pending_sig_waiter = PendingSigWaiter::new(thread, process, interest);
+            pending_sig_waiter.wait(timeout).map_err(|e| {
+                if e.errno() == Errno::EINTR {
+                    return e;
+                }
+                errno!(EAGAIN, "no interesting, pending signal")
+            })?
+        }
+    };
+
+    let siginfo = signal.to_info();
+    Ok(siginfo)
+}
+
+struct PendingSigWaiter {
+    thread: ThreadRef,
+    process: ProcessRef,
+    interest: SigSet,
+    observer: Arc<WaiterQueueObserver<SigNum>>,
+}
+
+impl PendingSigWaiter {
+    pub fn new(thread: ThreadRef, process: ProcessRef, interest: SigSet) -> Arc<Self> {
+        let observer = WaiterQueueObserver::new();
+
+        let weak_observer = Arc::downgrade(&observer) as Weak<dyn Observer<_>>;
+        thread.sig_queues().read().unwrap().notifier().register(
+            weak_observer.clone(),
+            Some(interest),
+            None,
+        );
+        process.sig_queues().read().unwrap().notifier().register(
+            weak_observer,
+            Some(interest),
+            None,
+        );
+
+        Arc::new(Self {
+            thread,
+            process,
+            interest,
+            observer,
+        })
+    }
+
+    pub fn wait(&self, timeout: &Duration) -> Result<Box<dyn Signal>> {
+        let waiter_queue = self.observer.waiter_queue();
+        let waiter = Waiter::new();
+        loop {
+            if *timeout == Duration::new(0, 0) {
+                return_errno!(ETIMEDOUT, "timeout");
+            }
+
+            // Enqueue the waiter so that it can be waken up by the queue later.
+            waiter_queue.reset_and_enqueue(&waiter);
+
+            // Try to dequeue a pending signal from the current process or thread
+            if let Some(signal) =
+                dequeue_pending_signal(&self.interest, &self.thread, &self.process)
+            {
+                return Ok(signal);
+            }
+
+            // As there is no intersting signal to dequeue right now, let's wait
+            // some time to try again later. Most likely, the waiter will keep
+            // waiting until being waken up by the waiter queue, which means
+            // the arrival of an interesting signal.
+            let res = waiter.wait(Some(timeout));
+
+            // Do not try again if some error is encountered. There are only
+            // two possible errors: ETIMEDOUT or EINTR.
+            if let Err(e) = res {
+                // When interrupted, it is possible that the interrupting signal happens
+                // to be an interesting and pending signal. So we attempt to dequeue again.
+                if e.errno() == Errno::EINTR {
+                    if let Some(signal) =
+                        dequeue_pending_signal(&self.interest, &self.thread, &self.process)
+                    {
+                        return Ok(signal);
+                    }
+                }
+                return Err(e);
+            }
+        }
+    }
+}
+
+impl Drop for PendingSigWaiter {
+    fn drop(&mut self) {
+        let weak_observer = Arc::downgrade(&self.observer) as Weak<dyn Observer<_>>;
+        self.thread
+            .sig_queues()
+            .read()
+            .unwrap()
+            .notifier()
+            .unregister(&weak_observer);
+        self.process
+            .sig_queues()
+            .read()
+            .unwrap()
+            .notifier()
+            .unregister(&weak_observer);
+    }
+}
+
+fn dequeue_pending_signal(
+    interest: &SigSet,
+    thread: &ThreadRef,
+    process: &ProcessRef,
+) -> Option<Box<dyn Signal>> {
+    dequeue_process_pending_signal(process, interest)
+        .or_else(|| dequeue_thread_pending_signal(thread, interest))
+}
+
+fn dequeue_process_pending_signal(
+    process: &ProcessRef,
+    interest: &SigSet,
+) -> Option<Box<dyn Signal>> {
+    let blocked = !*interest;
+    process.sig_queues().write().unwrap().dequeue(&blocked)
+}
+
+fn dequeue_thread_pending_signal(thread: &ThreadRef, interest: &SigSet) -> Option<Box<dyn Signal>> {
+    let blocked = !*interest;
+    thread.sig_queues().write().unwrap().dequeue(&blocked)
+}

src/libos/src/signal/mod.rs

@@ -4,7 +4,7 @@ use crate::prelude::*;
 
 use sig_action::{SigAction, SigActionFlags, SigDefaultAction};
 
-pub use self::c_types::{sigaction_t, sigset_t, stack_t};
+pub use self::c_types::{sigaction_t, siginfo_t, sigset_t, stack_t};
 pub use self::constants::*;
 pub use self::do_kill::do_kill_from_outside_enclave;
 pub use self::do_sigreturn::{deliver_signal, force_signal};
@@ -23,6 +23,7 @@ mod do_sigaltstack;
 mod do_sigpending;
 mod do_sigprocmask;
 mod do_sigreturn;
+mod do_sigtimedwait;
 mod sig_action;
 mod sig_dispositions;
 mod sig_num;

src/libos/src/signal/sig_num.rs

@@ -1,6 +1,7 @@
 use std::fmt;
 
 use super::constants::*;
+use crate::events::Event;
 use crate::prelude::*;
 
 #[repr(C)]
@@ -88,3 +89,5 @@ impl fmt::Debug for SigNum {
         }
     }
 }
+
+impl Event for SigNum {}

src/libos/src/signal/sig_queues.rs

@@ -3,12 +3,14 @@ use std::fmt;
 
 use super::constants::*;
 use super::{SigNum, SigSet, Signal};
+use crate::events::Notifier;
 use crate::prelude::*;
 
 pub struct SigQueues {
     count: usize,
     std_queues: Vec<Option<Box<dyn Signal>>>,
     rt_queues: Vec<VecDeque<Box<dyn Signal>>>,
+    notifier: Notifier<SigNum, SigSet>,
 }
 
 impl SigQueues {
@@ -16,10 +18,12 @@ impl SigQueues {
         let count = 0;
         let std_queues = (0..COUNT_STD_SIGS).map(|_| None).collect();
         let rt_queues = (0..COUNT_RT_SIGS).map(|_| Default::default()).collect();
+        let notifier = Notifier::new();
         SigQueues {
             count,
             std_queues,
             rt_queues,
+            notifier,
         }
     }
 
@@ -56,6 +60,8 @@ impl SigQueues {
             queue.push_back(signal);
             self.count += 1;
         }
+
+        self.notifier.broadcast(&signum);
     }
 
     pub fn dequeue(&mut self, blocked: &SigSet) -> Option<Box<dyn Signal>> {
@@ -120,6 +126,10 @@ impl SigQueues {
         None
     }
 
+    pub fn notifier(&self) -> &Notifier<SigNum, SigSet> {
+        &self.notifier
+    }
+
     pub fn pending(&self) -> SigSet {
         let mut pending_sigs = SigSet::new_empty();
         for signum in MIN_STD_SIG_NUM..=MAX_STD_SIG_NUM {

src/libos/src/signal/sig_set.rs

@@ -4,6 +4,7 @@ use std::ops::{Add, AddAssign, BitAnd, BitAndAssign, BitOr, BitOrAssign, Not, Su
 
 use super::constants::MIN_STD_SIG_NUM;
 use super::{sigset_t, SigNum};
+use crate::events::EventFilter;
 use crate::prelude::*;
 
 #[derive(Copy, Clone, Default, PartialEq, Eq)]
@@ -191,3 +192,9 @@ impl fmt::Debug for SigSet {
         write!(f, " }}")
     }
 }
+
+impl EventFilter<SigNum> for SigSet {
+    fn filter(&self, event: &SigNum) -> bool {
+        self.contains(*event)
+    }
+}

src/libos/src/signal/syscalls.rs

@@ -1,10 +1,13 @@
+use std::time::Duration;
+
 use super::constants::*;
 use super::do_sigprocmask::MaskOp;
 use super::signals::FaultSignal;
-use super::{sigaction_t, sigset_t, stack_t, SigAction, SigNum, SigSet, SigStack};
+use super::{sigaction_t, siginfo_t, sigset_t, stack_t, SigAction, SigNum, SigSet, SigStack};
 use crate::prelude::*;
 use crate::process::ProcessFilter;
 use crate::syscall::CpuContext;
+use crate::time::timespec_t;
 use crate::util::mem_util::from_user;
 
 pub fn do_rt_sigaction(
@@ -157,3 +160,37 @@ pub fn do_sigaltstack(
     }
     Ok(0)
 }
+
+pub fn do_rt_sigtimedwait(
+    mask_ptr: *const sigset_t,
+    info_ptr: *mut siginfo_t,
+    timeout_ptr: *const timespec_t,
+    mask_size: usize,
+) -> Result<isize> {
+    let mask: SigSet = {
+        if mask_size < std::mem::size_of::<sigset_t>() {
+            return_errno!(EINVAL, "mask size is not big enough");
+        }
+        if mask_ptr.is_null() {
+            return_errno!(EINVAL, "ptr must not be null");
+        }
+        SigSet::from_c(unsafe { *mask_ptr })
+    };
+    let info: &mut siginfo_t = {
+        if info_ptr.is_null() {
+            return_errno!(EINVAL, "ptr must not be null");
+        }
+        unsafe { &mut *info_ptr }
+    };
+    let timeout: Option<Duration> = {
+        if timeout_ptr.is_null() {
+            None
+        } else {
+            let timeout = timespec_t::from_raw_ptr(timeout_ptr)?;
+            Some(timeout.as_duration())
+        }
+    };
+
+    *info = super::do_sigtimedwait::do_sigtimedwait(mask, timeout.as_ref())?;
+    Ok(0)
+}

src/libos/src/syscall/mod.rs

@@ -47,8 +47,9 @@ use crate::process::{
 };
 use crate::sched::{do_getcpu, do_sched_getaffinity, do_sched_setaffinity, do_sched_yield};
 use crate::signal::{
-    do_kill, do_rt_sigaction, do_rt_sigpending, do_rt_sigprocmask, do_rt_sigreturn, do_sigaltstack,
-    do_tgkill, do_tkill, sigaction_t, sigset_t, stack_t,
+    do_kill, do_rt_sigaction, do_rt_sigpending, do_rt_sigprocmask, do_rt_sigreturn,
+    do_rt_sigtimedwait, do_sigaltstack, do_tgkill, do_tkill, sigaction_t, siginfo_t, sigset_t,
+    stack_t,
 };
 use crate::vm::{MMapFlags, MRemapFlags, MSyncFlags, VMPerms};
 use crate::{fs, process, std, vm};
@@ -212,7 +213,7 @@ macro_rules! process_syscall_table_with_callback {
             (Capget = 125) => handle_unsupported(),
             (Capset = 126) => handle_unsupported(),
             (RtSigpending = 127) => do_rt_sigpending(buf_ptr: *mut sigset_t, buf_size: usize),
-            (RtSigtimedwait = 128) => handle_unsupported(),
+            (RtSigtimedwait = 128) => do_rt_sigtimedwait(mask_ptr: *const sigset_t, info_ptr: *mut siginfo_t, timeout_ptr: *const timespec_t, mask_size: usize),
             (RtSigqueueinfo = 129) => handle_unsupported(),
             (RtSigsuspend = 130) => handle_unsupported(),
             (Sigaltstack = 131) => do_sigaltstack(ss: *const stack_t, old_ss: *mut stack_t, context: *const CpuContext),

test/signal/main.c

@@ -13,6 +13,9 @@
 #include <string.h>
 #include <fcntl.h>
 #include <signal.h>
+#include <pthread.h>
+#include <errno.h>
+#include <time.h>
 #include "test.h"
 
 // ============================================================================
@@ -230,7 +233,7 @@ int div_maybe_zero(int x, int y) {
     return x / y;
 }
 
-#define	fxsave(addr) __asm __volatile("fxsave %0" : "=m" (*(addr)))
+#define fxsave(addr) __asm __volatile("fxsave %0" : "=m" (*(addr)))
 
 int test_handle_sigfpe() {
 #ifdef SGX_MODE_SIM
@@ -318,6 +321,7 @@ int test_handle_sigsegv() {
 
     int *addr = NULL;
     volatile int val = read_maybe_null(addr);
+    (void)val; // to suppress "unused variables" warning
 
     printf("Signal handler successfully jumped over a null-dereferencing instruction\n");
 
@@ -424,6 +428,96 @@ int test_sigchld() {
     return 0;
 }
 
+// ============================================================================
+// Test sigtimedwait syscall
+// ============================================================================
+
+struct send_signal_data {
+    pthread_t           target;
+    int                 signum;
+    struct timespec     delay;
+};
+
+static void *send_signal_with_delay(void *_data) {
+    int ret;
+    struct send_signal_data *data = _data;
+
+    // Ensure data->delay time elapsed
+    while ((ret = nanosleep(&data->delay, NULL) < 0 && errno == EINTR)) ;
+    // Send the signal to the target thread
+    pthread_kill(data->target, data->signum);
+
+    free(data);
+
+    return NULL;
+}
+
+// Raise a signal for the current thread asynchronously by spawning another
+// thread to send the signal to the parent thread after some specified delay.
+// The delay is meant to ensure some operation in the parent thread is
+// completed by the time the signal is sent.
+static void raise_async(int signum, const struct timespec *delay) {
+    pthread_t thread;
+    int ret;
+    struct send_signal_data *data = malloc(sizeof(*data));
+    data->target = pthread_self();
+    data->signum = signum;
+    data->delay = *delay;
+    if ((ret = pthread_create(&thread, NULL, send_signal_with_delay, (void *)data)) < 0) {
+        printf("ERROR: pthread_create failed unexpectedly\n");
+        abort();
+    }
+    pthread_detach(thread);
+}
+
+int test_sigtimedwait() {
+    int ret;
+    siginfo_t info;
+    sigset_t new_mask, old_mask;
+    struct timespec delay, timeout;
+
+    // Update signal mask to block SIGIO
+    sigemptyset(&new_mask);
+    sigaddset(&new_mask, SIGIO);
+    if ((ret = sigprocmask(SIG_BLOCK, &new_mask, &old_mask)) < 0) {
+        THROW_ERROR("sigprocmask failed unexpectedly");
+    }
+
+    // There is no pending signal, yet; so the syscall must return EAGAIN error
+    ret = sigtimedwait(&new_mask, &info, NULL);
+    if (ret == 0 || (ret < 0 && errno != EAGAIN)) {
+        THROW_ERROR("sigprocmask must return with EAGAIN error");
+    }
+
+    // Let's generate a pending signal and then get it
+    raise(SIGIO);
+    if ((ret = sigtimedwait(&new_mask, &info, NULL)) < 0 || info.si_signo != SIGIO) {
+        THROW_ERROR("sigtimedwait should return the SIGIO");
+    }
+
+    // Now let's generate a pending signal in an async way. The pending signal
+    // does not exist yet at the time when sigtimedwait is called. So the
+    // current thread will be put to sleep and waken up until the
+    // asynchronously raised signal is sent to the current thread and becomes
+    // pending.
+    delay.tv_sec = 0;
+    delay.tv_nsec = 10 * 1000 * 1000; // 10ms
+    raise_async(SIGIO, &delay);
+
+    timeout.tv_sec = 0;
+    timeout.tv_nsec = 2 * delay.tv_nsec;
+
+    if ((ret = sigtimedwait(&new_mask, &info, &timeout)) < 0 || info.si_signo != SIGIO) {
+        THROW_ERROR("sigtimedwait should return the SIGIO");
+    }
+
+    // Restore the signal mask
+    if ((ret = sigprocmask(SIG_SETMASK, &old_mask, NULL)) < 0) {
+        THROW_ERROR("sigprocmask failed unexpectedly");
+    }
+    return 0;
+}
+
 // ============================================================================
 // Test suite main
 // ============================================================================
@@ -437,6 +531,7 @@ static test_case_t test_cases[] = {
     TEST_CASE(test_handle_sigsegv),
     TEST_CASE(test_sigaltstack),
     TEST_CASE(test_sigchld),
+    TEST_CASE(test_sigtimedwait),
 };
 
 int main(int argc, const char *argv[]) {