diff --git a/src/header/pthread/cond.rs b/src/header/pthread/cond.rs index 73dcac6c97..91b9763775 100644 --- a/src/header/pthread/cond.rs +++ b/src/header/pthread/cond.rs @@ -1,5 +1,7 @@ // Used design from https://www.remlab.net/op/futex-condvar.shtml +use crate::header::time::CLOCK_REALTIME; + use super::*; // PTHREAD_COND_INITIALIZER is defined manually in bits_pthread/cbindgen.toml @@ -19,8 +21,19 @@ pub unsafe extern "C" fn pthread_cond_destroy(cond: *mut pthread_cond_t) -> c_in #[unsafe(no_mangle)] pub unsafe extern "C" fn pthread_cond_init( cond: *mut pthread_cond_t, - _attr: *const pthread_condattr_t, + attr: *const pthread_condattr_t, ) -> c_int { + let attr = attr + .cast::() + .as_ref() + .copied() + .unwrap_or_default(); + + if attr.clock != CLOCK_REALTIME { + // As monotonic clock always smaller than realtime clock, this always result in instant timeout. + println!("TODO: pthread_cond_init with monotonic clock"); + } + cond.cast::().write(RlctCond::new()); 0 @@ -40,6 +53,16 @@ pub unsafe extern "C" fn pthread_cond_timedwait( e((&*cond.cast::()).timedwait(&*mutex.cast::(), &*timeout)) } +#[unsafe(no_mangle)] +pub unsafe extern "C" fn pthread_cond_clockwait( + cond: *mut pthread_cond_t, + mutex: *mut pthread_mutex_t, + clock_id: clockid_t, + timeout: *const timespec, +) -> c_int { + e((&*cond.cast::()).clockwait(&*mutex.cast::(), &*timeout, clock_id)) +} + #[unsafe(no_mangle)] pub unsafe extern "C" fn pthread_cond_wait( cond: *mut pthread_cond_t, @@ -100,20 +123,6 @@ pub unsafe extern "C" fn pthread_condattr_setpshared( 0 } -pub(crate) struct RlctCondAttr { - clock: clockid_t, - pshared: c_int, -} +pub(crate) type RlctCondAttr = crate::sync::cond::CondAttr; pub(crate) type RlctCond = crate::sync::cond::Cond; - -impl Default for RlctCondAttr { - fn default() -> Self { - Self { - // FIXME: system clock - clock: 0, - // Default - pshared: PTHREAD_PROCESS_PRIVATE, - } - } -} diff --git a/src/header/time/mod.rs b/src/header/time/mod.rs index 7854f01fea..7a1bbd7845 100644 --- a/src/header/time/mod.rs +++ b/src/header/time/mod.rs @@ -24,7 +24,6 @@ use crate::{ }, sync::{Mutex, MutexGuard}, }; -use __libc_only_for_layout_checks::EINVAL; use alloc::{boxed::Box, collections::BTreeSet, string::String, vec::Vec}; use chrono::{ DateTime, Datelike, FixedOffset, NaiveDate, NaiveDateTime, Offset, ParseError, TimeZone, @@ -34,6 +33,7 @@ use chrono_tz::{OffsetComponents, OffsetName, Tz}; use core::{ cell::OnceCell, convert::{TryFrom, TryInto}, + fmt::Debug, mem, ptr, }; @@ -48,11 +48,12 @@ pub use strptime::strptime; const YEARS_PER_ERA: time_t = 400; const DAYS_PER_ERA: time_t = 146097; const SECS_PER_DAY: time_t = 24 * 60 * 60; +const NANOSECONDS: i64 = 1_000_000_000; const UTC_STR: &core::ffi::CStr = c"UTC"; /// See . #[repr(C)] -#[derive(Clone, Copy, Default)] +#[derive(Clone, Copy, Default, Debug)] pub struct timespec { pub tv_sec: time_t, pub tv_nsec: c_long, @@ -60,36 +61,36 @@ pub struct timespec { impl timespec { // TODO: Write test - pub fn add(base: timespec, interval: timespec) -> Option { - let base_nsec = c_ulong::try_from(base.tv_nsec).ok()?; - let interval_nsec = c_ulong::try_from(interval.tv_nsec).ok()?; - Some(if base_nsec.checked_add(interval_nsec)? < 1_000_000_000 { - timespec { - tv_sec: base.tv_sec.checked_add(interval.tv_sec)?, - tv_nsec: (base_nsec + interval_nsec) as _, - } - } else { - timespec { - tv_sec: base.tv_sec.checked_add(interval.tv_sec)?.checked_add(1)?, - tv_nsec: ((interval_nsec + base_nsec) - 1_000_000_000) as c_long, - } + /// similar logic with timeradd + pub fn add(base: timespec, interval: timespec) -> Option { + let delta_sec = base.tv_sec + interval.tv_sec; + let delta_nsec = base.tv_nsec + interval.tv_nsec; + + if delta_sec < 0 || delta_nsec < 0 { + return None; + } + + Some(Self { + tv_sec: delta_sec + (delta_nsec / NANOSECONDS), + tv_nsec: delta_nsec % NANOSECONDS, }) } - // TODO: Write test + /// similar logic with timersub pub fn subtract(later: timespec, earlier: timespec) -> Option { - let later_nsec = c_ulong::try_from(later.tv_nsec).ok()?; - let earlier_nsec = c_ulong::try_from(earlier.tv_nsec).ok()?; + let delta_sec = later.tv_sec - earlier.tv_sec; + let delta_nsec = later.tv_nsec - earlier.tv_nsec; - let time = if later_nsec > earlier_nsec { + let time = if delta_nsec < 0 { + let roundup_sec = -delta_nsec / NANOSECONDS + 1; timespec { - tv_sec: later.tv_sec.checked_sub(earlier.tv_sec)?, - tv_nsec: (later_nsec - earlier_nsec) as _, + tv_sec: delta_sec - roundup_sec, + tv_nsec: roundup_sec * NANOSECONDS - delta_nsec, } } else { timespec { - tv_sec: later.tv_sec.checked_sub(earlier.tv_sec)?.checked_sub(1)?, - tv_nsec: 1_000_000_000 - (earlier_nsec - later_nsec) as c_long, + tv_sec: delta_sec + (delta_nsec / NANOSECONDS), + tv_nsec: delta_nsec % NANOSECONDS, } }; diff --git a/src/sync/cond.rs b/src/sync/cond.rs index 1b3d846067..ae4c20235d 100644 --- a/src/sync/cond.rs +++ b/src/sync/cond.rs @@ -2,11 +2,32 @@ use crate::{ error::Errno, - header::{pthread::*, time::timespec}, + header::{ + errno::{EINVAL, ENOMEM, ETIMEDOUT}, + pthread::*, + time::{CLOCK_MONOTONIC, CLOCK_REALTIME, clock_gettime, timespec}, + }, + platform::types::clockid_t, }; use core::sync::atomic::{AtomicU32 as AtomicUint, Ordering}; +#[derive(Clone, Copy)] +pub struct CondAttr { + pub clock: clockid_t, + pub pshared: i32, +} + +impl Default for CondAttr { + fn default() -> Self { + Self { + // defaults according to POSIX + clock: CLOCK_REALTIME, // for timedwait + pshared: PTHREAD_PROCESS_PRIVATE, // TODO + } + } +} + pub struct Cond { cur: AtomicUint, prev: AtomicUint, @@ -14,6 +35,12 @@ pub struct Cond { type Result = core::result::Result; +impl Default for Cond { + fn default() -> Self { + Self::new() + } +} + impl Cond { pub fn new() -> Self { Self { @@ -36,8 +63,51 @@ impl Cond { self.broadcast() //self.wake(1) } + pub fn clockwait( + &self, + mutex: &RlctMutex, + timeout: ×pec, + clock_id: clockid_t, + ) -> Result<(), Errno> { + // adjusted timeout similar in semaphore.rs + + let relative = match clock_id { + // FUTEX expect monotonic clock + CLOCK_MONOTONIC => timeout.clone(), + CLOCK_REALTIME => { + let mut realtime = timespec::default(); + unsafe { clock_gettime(CLOCK_REALTIME, &mut realtime) }; + let mut monotonic = timespec::default(); + unsafe { clock_gettime(CLOCK_MONOTONIC, &mut monotonic) }; + let Some(delta) = timespec::subtract(timeout.clone(), realtime) else { + return Err(Errno(ETIMEDOUT)); + }; + let Some(relative) = timespec::add(monotonic, delta) else { + return Err(Errno(ENOMEM)); + }; + relative + } + _ => return Err(Errno(EINVAL)), + }; + + match self.wait_inner(mutex, Some(&relative)) { + Err(Errno(ETIMEDOUT)) => { + // TODO: this is a workaround that SYS_FUTEX returns ETIMEDOUT but actually it signaled in time + let mut monotonic = timespec::default(); + unsafe { clock_gettime(CLOCK_MONOTONIC, &mut monotonic) }; + + if timespec::subtract(relative, monotonic).is_some() { + Ok(()) + } else { + Err(Errno(ETIMEDOUT)) + } + } + r => r, + } + } pub fn timedwait(&self, mutex: &RlctMutex, timeout: ×pec) -> Result<(), Errno> { - self.wait_inner(mutex, Some(timeout)) + // TODO: The clock can be other than CLOCK_REALTIME depends on CondAttr + self.clockwait(mutex, timeout, CLOCK_REALTIME) } fn wait_inner(&self, mutex: &RlctMutex, timeout: Option<×pec>) -> Result<(), Errno> { self.wait_inner_generic( @@ -83,17 +153,22 @@ impl Cond { unlock(); - match deadline { + let futex_r = match deadline { Some(deadline) => { - crate::sync::futex_wait(&self.cur, current, Some(&deadline)); - lock_with_timeout(deadline); + lock_with_timeout(deadline)?; + crate::sync::futex_wait(&self.cur, current, Some(&deadline)) } None => { - crate::sync::futex_wait(&self.cur, current, None); - lock(); + lock()?; + crate::sync::futex_wait(&self.cur, current, None) } + }; + + match futex_r { + super::FutexWaitResult::Waited => Ok(()), + super::FutexWaitResult::Stale => Ok(()), + super::FutexWaitResult::TimedOut => Err(Errno(ETIMEDOUT)), } - Ok(()) } pub fn wait(&self, mutex: &RlctMutex) -> Result<(), Errno> { self.wait_inner(mutex, None) diff --git a/src/sync/semaphore.rs b/src/sync/semaphore.rs index 080175a09e..54b2691f0f 100644 --- a/src/sync/semaphore.rs +++ b/src/sync/semaphore.rs @@ -2,7 +2,7 @@ //TODO: improve implementation use crate::{ - header::time::{CLOCK_MONOTONIC, clock_gettime, timespec}, + header::time::{CLOCK_MONOTONIC, CLOCK_REALTIME, clock_gettime, timespec}, platform::types::{c_uint, clockid_t}, }; @@ -60,28 +60,31 @@ impl Semaphore { } if let Some(timeout) = timeout_opt { - let mut time = timespec::default(); - unsafe { clock_gettime(clock_id, &mut time) }; - if (time.tv_sec > timeout.tv_sec) - || (time.tv_sec == timeout.tv_sec && time.tv_nsec >= timeout.tv_nsec) - { - //Timeout happened, return error - return Err(()); - } else { - // Use futex to wait for the next change, with a relative timeout - let mut relative = timespec { - tv_sec: timeout.tv_sec, - tv_nsec: timeout.tv_nsec, - }; - while relative.tv_nsec < time.tv_nsec { - relative.tv_sec -= 1; - relative.tv_nsec += 1_000_000_000; - } - relative.tv_sec -= time.tv_sec; - relative.tv_nsec -= time.tv_nsec; + let relative = match clock_id { + // FUTEX expect monotonic clock + CLOCK_MONOTONIC => timeout.clone(), + CLOCK_REALTIME => { + let mut realtime = timespec::default(); + unsafe { clock_gettime(CLOCK_REALTIME, &mut realtime) }; - crate::sync::futex_wait(&self.count, value, Some(&relative)); - } + let mut monotonic = timespec::default(); + unsafe { clock_gettime(CLOCK_MONOTONIC, &mut monotonic) }; + + let Some(delta) = timespec::subtract(timeout.clone(), realtime) else { + //Timeout happened + return Err(()); + }; + + let Some(relative) = timespec::add(monotonic, delta) else { + return Err(()); + }; + + relative + } + _ => return Err(()), + }; + + crate::sync::futex_wait(&self.count, value, Some(&relative)); } else { // Use futex to wait for the next change, without a timeout crate::sync::futex_wait(&self.count, value, None); diff --git a/tests/pthread/timedwait.c b/tests/pthread/timedwait.c new file mode 100644 index 0000000000..c265b48045 --- /dev/null +++ b/tests/pthread/timedwait.c @@ -0,0 +1,86 @@ +#include +#include +#include +#include +#include +#include + +#include "test_helpers.h" + +pthread_mutex_t lock; +pthread_cond_t cond; +struct timespec start_time; +int ready = 0; + +struct timespec get_timeout(long msec) { + struct timespec ts; + clock_gettime(CLOCK_REALTIME, &ts); + ts.tv_nsec += msec * 1000000; + return ts; +} + +void print_timed(char* msg) { + struct timespec end_time; + clock_gettime(CLOCK_MONOTONIC, &end_time); + printf("[%.3f]: ", (end_time.tv_nsec - start_time.tv_nsec) / 1000000000.0 + + (end_time.tv_sec - start_time.tv_sec)); + printf("%s\n", msg); + fflush(NULL); +} + +void* signaler_thread(void* arg) { + (void)arg; + + usleep(150000); // 150ms + ready = 1; + print_timed("signaler_thread"); + pthread_cond_signal(&cond); + return NULL; +} + +void test_timeout_case() { + pthread_mutex_lock(&lock); + struct timespec ts = get_timeout(500); + print_timed("test_timeout_case start"); + int rc = pthread_cond_timedwait(&cond, &lock, &ts); + UNEXP_IF(pthread_cond_timedwait, rc, != ETIMEDOUT); + print_timed("test_timeout_case end"); + pthread_mutex_unlock(&lock); +} + +void test_success_case() { + ready = 0; + pthread_t thread; + pthread_create(&thread, NULL, signaler_thread, NULL); + pthread_mutex_lock(&lock); + struct timespec ts = get_timeout(50000); + print_timed("test_success_case start"); + while (!ready) { + print_timed("test_success_case waiting"); + int rc = pthread_cond_timedwait(&cond, &lock, &ts); + UNEXP_IF(pthread_cond_timedwait, rc, != 0); + } + print_timed("test_success_case end"); + pthread_mutex_unlock(&lock); + pthread_join(thread, NULL); +} + +int main() { + clock_gettime(CLOCK_MONOTONIC, &start_time); + if (pthread_mutex_init(&lock, NULL) != 0) { + perror("mutex init failed"); + return 1; + } + if (pthread_cond_init(&cond, NULL) != 0) { + perror("cond init failed"); + return 1; + } + test_timeout_case(); + // TODO: "rlct_clone not implemented for aarch64 yet" +#if defined(__linux__) && defined(__aarch64__) + printf("test_success_case skipped") +#else + test_success_case(); +#endif + return 0; +}