diff --git a/SConstruct b/SConstruct index 0b8f87b19..ba1d838fa 100644 --- a/SConstruct +++ b/SConstruct @@ -682,6 +682,10 @@ if meta.platform in ['linux', 'unix']: elif meta.platform in ['android']: meta.gnu_toolchain = True +if conf.CheckFunc('sem_clockwait', + header="#define _GNU_SOURCE\n#include \n"): + conf.env.Append(CPPDEFINES=['ROC_HAVE_SEM_CLOCKWAIT']) + conf.env['ROC_SYSTEM_BINDIR'] = GetOption('bindir') conf.env['ROC_SYSTEM_INCDIR'] = GetOption('incdir') @@ -788,10 +792,15 @@ else: 'target_posix_pc', ]) - if meta.platform in ['linux', 'unix', 'android']: - env.Append(ROC_TARGETS=[ - 'target_posix_ext', - ]) + if meta.platform in ['linux', 'android', 'unix']: + if 'ROC_HAVE_SEM_CLOCKWAIT' in env['CPPDEFINES']: + env.Append(ROC_TARGETS=[ + 'target_posix_sem', + ]) + else: + env.Append(ROC_TARGETS=[ + 'target_nosem', + ]) if meta.platform in ['linux', 'unix', 'macos', 'windows', 'android']: env.Append(ROC_TARGETS=[ diff --git a/docs/sphinx/internals/code_structure.rst b/docs/sphinx/internals/code_structure.rst index 89de35b00..ed52b787b 100644 --- a/docs/sphinx/internals/code_structure.rst +++ b/docs/sphinx/internals/code_structure.rst @@ -135,8 +135,9 @@ target_openfec Enabled if OpenFEC is available target_openssl Enabled if OpenSSL is available target_pc Enabled for PC (like server, desktop, laptop) target_posix Enabled for a POSIX OS -target_posix_ext Enabled for a POSIX OS with POSIX extensions target_posix_pc Enabled for a POSIX OS on PC +target_posix_sem Enabled if sem_clockwait() is available +target_nosem Enabled if sem_clockwait() is not available target_pulseaudio Enabled if PulseAudio is available target_sndfile Enabled if libsndfile is available target_sox Enabled if SoX is available @@ -153,11 +154,11 @@ Example directory structure employing targets: │ ├── ... │ ├── mutex.cpp │ └── mutex.h - ├── target_posix_ext + ├── target_posix_sem │ └── roc_core │ ├── ... - │ ├── time.cpp - │ └── time.h + │ ├── semaphore.cpp + │ └── semaphore.h ├── target_darwin │ └── roc_core │ ├── ... diff --git a/src/internal_modules/roc_core/target_darwin/roc_core/semaphore.cpp b/src/internal_modules/roc_core/target_darwin/roc_core/semaphore.cpp index d33f85d63..4950bea00 100644 --- a/src/internal_modules/roc_core/target_darwin/roc_core/semaphore.cpp +++ b/src/internal_modules/roc_core/target_darwin/roc_core/semaphore.cpp @@ -38,10 +38,8 @@ bool Semaphore::timed_wait(nanoseconds_t deadline) { } for (;;) { - const nanoseconds_t timeout = deadline - timestamp(ClockMonotonic); - if (timeout <= 0) { - return false; - } + const nanoseconds_t remaining = deadline - timestamp(ClockMonotonic); + const nanoseconds_t timeout = remaining > 0 ? remaining : 0; mach_timespec_t ts; ts.tv_sec = unsigned(timeout / Second); diff --git a/src/internal_modules/roc_core/target_nosem/roc_core/semaphore.cpp b/src/internal_modules/roc_core/target_nosem/roc_core/semaphore.cpp new file mode 100644 index 000000000..f511e1955 --- /dev/null +++ b/src/internal_modules/roc_core/target_nosem/roc_core/semaphore.cpp @@ -0,0 +1,58 @@ +/* + * Copyright (c) 2026 Roc Streaming authors + * + * This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. + */ + +#include "roc_core/semaphore.h" +#include "roc_core/panic.h" + +namespace roc { +namespace core { + +Semaphore::Semaphore(unsigned counter) + : cond_(mutex_) + , counter_(counter) { +} + +bool Semaphore::timed_wait(nanoseconds_t deadline) { + if (deadline < 0) { + roc_panic("semaphore: unexpected negative deadline"); + } + + Mutex::Lock lock(mutex_); + + while (counter_ == 0) { + const nanoseconds_t timeout = deadline - timestamp(ClockMonotonic); + if (timeout <= 0) { + return false; + } + (void)cond_.timed_wait(timeout); + } + + counter_--; + + return true; +} + +void Semaphore::wait() { + Mutex::Lock lock(mutex_); + + while (counter_ == 0) { + cond_.wait(); + } + + counter_--; +} + +void Semaphore::post() { + Mutex::Lock lock(mutex_); + + counter_++; + cond_.signal(); +} + +} // namespace core +} // namespace roc diff --git a/src/internal_modules/roc_core/target_nosem/roc_core/semaphore.h b/src/internal_modules/roc_core/target_nosem/roc_core/semaphore.h new file mode 100644 index 000000000..0f1ff609b --- /dev/null +++ b/src/internal_modules/roc_core/target_nosem/roc_core/semaphore.h @@ -0,0 +1,56 @@ +/* + * Copyright (c) 2026 Roc Streaming authors + * + * This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. + */ + +//! @file roc_core/target_nosem/roc_core/semaphore.h +//! @brief Semaphore. + +#ifndef ROC_CORE_SEMAPHORE_H_ +#define ROC_CORE_SEMAPHORE_H_ + +#include "roc_core/attributes.h" +#include "roc_core/cond.h" +#include "roc_core/mutex.h" +#include "roc_core/noncopyable.h" +#include "roc_core/time.h" + +namespace roc { +namespace core { + +//! Semaphore. +//! +//! @remarks +//! This implementation is used on platforms that don't provide sem_clockwait(), +//! and hence can't wait on a POSIX semaphore using monotonic clock. It is based +//! on mutex and condition variable. Unlike other implementations, post() is not +//! lock-free here. +class Semaphore : public NonCopyable<> { +public: + //! Initialize semaphore with given counter. + explicit Semaphore(unsigned counter = 0); + + //! Wait until the counter becomes non-zero, decrement it, and return true. + //! If deadline expires before the counter becomes non-zero, returns false. + //! Deadline is an absolute timestamp in ClockMonotonic domain. + ROC_NODISCARD bool timed_wait(nanoseconds_t deadline); + + //! Wait until the counter becomes non-zero, decrement it, and return. + void wait(); + + //! Increment counter and wake up blocked waits. + void post(); + +private: + Mutex mutex_; + Cond cond_; + unsigned counter_; +}; + +} // namespace core +} // namespace roc + +#endif // ROC_CORE_SEMAPHORE_H_ diff --git a/src/internal_modules/roc_core/target_posix_ext/roc_core/semaphore.cpp b/src/internal_modules/roc_core/target_posix_sem/roc_core/semaphore.cpp similarity index 90% rename from src/internal_modules/roc_core/target_posix_ext/roc_core/semaphore.cpp rename to src/internal_modules/roc_core/target_posix_sem/roc_core/semaphore.cpp index 0d9685390..2469193b0 100644 --- a/src/internal_modules/roc_core/target_posix_ext/roc_core/semaphore.cpp +++ b/src/internal_modules/roc_core/target_posix_sem/roc_core/semaphore.cpp @@ -9,6 +9,7 @@ #include "roc_core/semaphore.h" #include "roc_core/cpu_instructions.h" #include "roc_core/errno_to_str.h" +#include "roc_core/log.h" #include "roc_core/panic.h" #include @@ -38,12 +39,12 @@ bool Semaphore::timed_wait(nanoseconds_t deadline) { roc_panic("semaphore: unexpected negative deadline"); } - for (;;) { - timespec ts; - ts.tv_sec = long(deadline / Second); - ts.tv_nsec = long(deadline % Second); + timespec ts; + ts.tv_sec = long(deadline / Second); + ts.tv_nsec = long(deadline % Second); - if (sem_timedwait(&sem_, &ts) == 0) { + for (;;) { + if (sem_clockwait(&sem_, CLOCK_MONOTONIC, &ts) == 0) { return true; } diff --git a/src/internal_modules/roc_core/target_posix_ext/roc_core/semaphore.h b/src/internal_modules/roc_core/target_posix_sem/roc_core/semaphore.h similarity index 96% rename from src/internal_modules/roc_core/target_posix_ext/roc_core/semaphore.h rename to src/internal_modules/roc_core/target_posix_sem/roc_core/semaphore.h index 4eab2fd84..e768fb21b 100644 --- a/src/internal_modules/roc_core/target_posix_ext/roc_core/semaphore.h +++ b/src/internal_modules/roc_core/target_posix_sem/roc_core/semaphore.h @@ -6,7 +6,7 @@ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */ -//! @file roc_core/target_posix_ext/roc_core/semaphore.h +//! @file roc_core/target_posix_sem/roc_core/semaphore.h //! @brief Semaphore. #ifndef ROC_CORE_SEMAPHORE_H_ diff --git a/src/internal_modules/roc_pipeline/state_tracker.cpp b/src/internal_modules/roc_pipeline/state_tracker.cpp index bc2b6ef6f..aa00685b5 100644 --- a/src/internal_modules/roc_pipeline/state_tracker.cpp +++ b/src/internal_modules/roc_pipeline/state_tracker.cpp @@ -13,9 +13,74 @@ namespace roc { namespace pipeline { StateTracker::StateTracker() - : halt_state_(-1) + : sem_(0) + , halt_state_(-1) , active_sessions_(0) - , pending_packets_(0) { + , pending_packets_(0) + , sem_is_occupied_(0) + , mutex_() + , waiting_con_(mutex_) { +} + +bool StateTracker::wait_state(unsigned state_mask, core::nanoseconds_t deadline) { + if (state_mask == 0) { + return true; + } + + bool sem_owner = false; + bool matched = false; + + mutex_.lock(); + while (true) { + const core::nanoseconds_t now = core::timestamp(core::ClockMonotonic); + const core::nanoseconds_t timeout = deadline - now; + + if (static_cast(get_state()) & state_mask) { + matched = true; + break; + } + + if (deadline > 0 && timeout <= 0) { + break; + } + + if (!sem_owner) { + sem_owner = sem_is_occupied_.compare_exchange(0, 1); + if (sem_owner) { + // Re-check state now that flag is published, otherwise a signal made + // right before the flag was set would be lost. + continue; + } + } + + if (sem_owner) { + mutex_.unlock(); + if (deadline > 0) { + (void)sem_.timed_wait(deadline); + } else { + sem_.wait(); + } + mutex_.lock(); + waiting_con_.broadcast(); + } else { + if (deadline > 0) { + // Unlike Semaphore, Cond expects relative timeout. + (void)waiting_con_.timed_wait(timeout); + } else { + waiting_con_.wait(); + } + } + } + + if (sem_owner) { + // Hand semaphore ownership over to one of the condvar waiters. + sem_is_occupied_ = 0; + waiting_con_.broadcast(); + } + + mutex_.unlock(); + + return matched; } sndio::DeviceState StateTracker::get_state() const { @@ -54,10 +119,12 @@ bool StateTracker::is_closed() const { void StateTracker::set_broken() { halt_state_ = sndio::DeviceState_Broken; + signal_state_change_(); } void StateTracker::set_closed() { halt_state_ = sndio::DeviceState_Closed; + signal_state_change_(); } size_t StateTracker::num_sessions() const { @@ -65,22 +132,38 @@ size_t StateTracker::num_sessions() const { } void StateTracker::register_session() { - active_sessions_++; + if (active_sessions_++ == 0) { + signal_state_change_(); + } } void StateTracker::unregister_session() { - if (--active_sessions_ < 0) { + int prev_sessions = active_sessions_--; + if (prev_sessions == 0) { roc_panic("state tracker: unpaired register/unregister session"); + } else if (prev_sessions == 1 && pending_packets_ == 0) { + signal_state_change_(); } } void StateTracker::register_packet() { - pending_packets_++; + if (pending_packets_++ == 0 && active_sessions_ == 0) { + signal_state_change_(); + } } void StateTracker::unregister_packet() { - if (--pending_packets_ < 0) { + int prev_packets = pending_packets_--; + if (prev_packets == 0) { roc_panic("state tracker: unpaired register/unregister packet"); + } else if (prev_packets == 1 && active_sessions_ == 0) { + signal_state_change_(); + } +} + +void StateTracker::signal_state_change_() { + if (sem_is_occupied_) { + sem_.post(); } } diff --git a/src/internal_modules/roc_pipeline/state_tracker.h b/src/internal_modules/roc_pipeline/state_tracker.h index b09180b3f..8e25d823d 100644 --- a/src/internal_modules/roc_pipeline/state_tracker.h +++ b/src/internal_modules/roc_pipeline/state_tracker.h @@ -13,8 +13,12 @@ #define ROC_PIPELINE_STATE_TRACKER_H_ #include "roc_core/atomic_int.h" +#include "roc_core/cond.h" +#include "roc_core/mutex.h" #include "roc_core/noncopyable.h" +#include "roc_core/semaphore.h" #include "roc_core/stddefs.h" +#include "roc_core/time.h" #include "roc_sndio/device_defs.h" namespace roc { @@ -32,6 +36,28 @@ class StateTracker : public core::NonCopyable<> { //! Initialize all counters to zero. StateTracker(); + //! Wait for state change. + //! + //! @remarks + //! Blocks until the state becomes any of the states specified by the mask, + //! or deadline expires. E.g. if mask is ACTIVE | PAUSED, blocks until + //! state becomes either ACTIVE or PAUSED. + //! + //! Empty mask means that there is nothing to wait for, and returns true + //! immediately. + //! + //! Deadline should be an absolute timestamp in ClockMonotonic domain. + //! Non-positive deadline (zero or negative) means no deadline: blocks + //! until the mask matches, however long that takes. + //! + //! @returns + //! true if state matches the mask and false if deadline expired. + //! + //! @note + //! Remember that pipeline state may be outdated immediately after this + //! method returns (e.g. if new packet arrives concurrently). + bool wait_state(unsigned state_mask, core::nanoseconds_t deadline); + //! Compute current state. sndio::DeviceState get_state() const; @@ -63,9 +89,15 @@ class StateTracker : public core::NonCopyable<> { void unregister_packet(); private: + void signal_state_change_(); + + core::Semaphore sem_; core::AtomicInt halt_state_; core::AtomicInt active_sessions_; core::AtomicInt pending_packets_; + core::AtomicInt sem_is_occupied_; + core::Mutex mutex_; + core::Cond waiting_con_; }; } // namespace pipeline diff --git a/src/tests/roc_core/test_semaphore.cpp b/src/tests/roc_core/test_semaphore.cpp new file mode 100644 index 000000000..2c43bf4af --- /dev/null +++ b/src/tests/roc_core/test_semaphore.cpp @@ -0,0 +1,217 @@ +/* + * Copyright (c) 2026 Roc Streaming authors + * + * This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. + */ + +#include + +#include "roc_core/atomic_bool.h" +#include "roc_core/semaphore.h" +#include "roc_core/thread.h" +#include "roc_core/time.h" + +namespace roc { +namespace core { + +namespace { + +enum { NumThreads = 5, NumPosts = 5 }; + +// Deadline offset for tests that expect the deadline to expire. +const nanoseconds_t ShortTimeout = Millisecond * 10; + +// Upper bound for calls that must not block at all. +const nanoseconds_t MaxImmediate = Millisecond * 100; + +// Time given to a freshly started thread to actually block inside the semaphore. +const nanoseconds_t SettleDelay = Microsecond * 100; + +class BlockingThread : public Thread { +public: + BlockingThread() + : sem_(NULL) { + } + + void init(Semaphore& sem) { + sem_ = &sem; + } + + bool running() const { + return running_; + } + + void wait_running() { + while (!running_) { + sleep_for(ClockMonotonic, Microsecond); + } + } + +private: + virtual void run() { + running_ = true; + sem_->wait(); + running_ = false; + } + + Semaphore* sem_; + AtomicBool running_; +}; + +class TimedThread : public Thread { +public: + TimedThread() + : sem_(NULL) + , deadline_(0) + , result_(false) { + } + + void init(Semaphore& sem, nanoseconds_t deadline) { + sem_ = &sem; + deadline_ = deadline; + } + + // Value returned by timed_wait(); valid only after join(). + bool result() const { + return result_; + } + + void wait_running() { + while (!running_) { + sleep_for(ClockMonotonic, Microsecond); + } + } + +private: + virtual void run() { + running_ = true; + result_ = sem_->timed_wait(deadline_); + running_ = false; + } + + Semaphore* sem_; + nanoseconds_t deadline_; + bool result_; + AtomicBool running_; +}; + +} // namespace + +TEST_GROUP(semaphore) {}; + +TEST(semaphore, post_then_wait) { + Semaphore sem(0); + + const nanoseconds_t start = timestamp(ClockMonotonic); + + sem.post(); + sem.wait(); + + CHECK(timestamp(ClockMonotonic) - start < MaxImmediate); +} + +TEST(semaphore, initial_counter) { + Semaphore sem(2); + + const nanoseconds_t start = timestamp(ClockMonotonic); + + sem.wait(); + sem.wait(); + + CHECK(timestamp(ClockMonotonic) - start < MaxImmediate); +} + +TEST(semaphore, timed_wait_success) { + Semaphore sem(0); + + sem.post(); + + CHECK(sem.timed_wait(timestamp(ClockMonotonic) + ShortTimeout)); +} + +TEST(semaphore, timed_wait_timeout) { + Semaphore sem(0); + + const nanoseconds_t start = timestamp(ClockMonotonic); + CHECK(!sem.timed_wait(start + ShortTimeout)); + + const nanoseconds_t elapsed = timestamp(ClockMonotonic) - start; + CHECK(elapsed >= ShortTimeout / 2); + CHECK(elapsed < Second); +} + +TEST(semaphore, timed_wait_expired_deadline) { + Semaphore sem(0); + + const nanoseconds_t start = timestamp(ClockMonotonic); + CHECK(!sem.timed_wait(start - Second)); + CHECK(timestamp(ClockMonotonic) - start < MaxImmediate); +} + +TEST(semaphore, timed_wait_ready_expired_deadline) { + Semaphore sem(0); + + sem.post(); + + // Like POSIX sem_timedwait(), a ready semaphore succeeds despite the deadline. + CHECK(sem.timed_wait(timestamp(ClockMonotonic) - Second)); +} + +TEST(semaphore, multiple_posts) { + Semaphore sem(0); + + for (int i = 0; i < NumPosts; i++) { + sem.post(); + } + + for (int i = 0; i < NumPosts; i++) { + CHECK(sem.timed_wait(timestamp(ClockMonotonic) + ShortTimeout)); + } + + CHECK(!sem.timed_wait(timestamp(ClockMonotonic) + ShortTimeout)); +} + +TEST(semaphore, blocking_thread) { + Semaphore sem(0); + + BlockingThread thr; + thr.init(sem); + CHECK(thr.start()); + + thr.wait_running(); + sleep_for(ClockMonotonic, SettleDelay); + CHECK(thr.running()); + + sem.post(); + thr.join(); +} + +TEST(semaphore, multiple_threads) { + Semaphore sem(0); + + TimedThread threads[NumThreads]; + + // Deadline is generous on purpose: it is a safety net, not the thing under test. + for (size_t i = 0; i < NumThreads; i++) { + threads[i].init(sem, timestamp(ClockMonotonic) + Second * 5); + CHECK(threads[i].start()); + } + + for (size_t i = 0; i < NumThreads; i++) { + threads[i].wait_running(); + } + + for (size_t i = 0; i < NumThreads; i++) { + sem.post(); + } + + for (size_t i = 0; i < NumThreads; i++) { + threads[i].join(); + CHECK(threads[i].result()); + } +} + +} // namespace core +} // namespace roc diff --git a/src/tests/roc_pipeline/test_state_tracker.cpp b/src/tests/roc_pipeline/test_state_tracker.cpp new file mode 100644 index 000000000..4ecad388f --- /dev/null +++ b/src/tests/roc_pipeline/test_state_tracker.cpp @@ -0,0 +1,412 @@ +/* + * Copyright (c) 2026 Roc Streaming authors + * + * This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. + */ + +#include "test_harness.h" + +#include "roc_core/atomic_bool.h" +#include "roc_core/cpu_instructions.h" +#include "roc_core/thread.h" +#include "roc_core/time.h" +#include "roc_pipeline/state_tracker.h" +#include "roc_sndio/device_defs.h" + +namespace roc { +namespace pipeline { + +namespace { + +enum { NumThreads = 5 }; + +// Deadline value that means "block forever". +const core::nanoseconds_t NoDeadline = -1; + +// Deadline offset for tests that expect the deadline to expire. +const core::nanoseconds_t ShortTimeout = core::Millisecond * 10; + +// Upper bound for calls that must not block at all. +const core::nanoseconds_t MaxImmediate = core::Millisecond * 100; + +// Time given to a freshly started thread to actually block inside wait_state(). +const core::nanoseconds_t SettleDelay = core::Microsecond * 100; + +// Deadline long enough to never expire while the test runs. +const core::nanoseconds_t LongTimeout = core::Minute * 10; + +class WaitThread : public core::Thread { +public: + WaitThread() + : tracker_(NULL) + , state_mask_(0) + , deadline_(0) + , result_(false) { + } + + void init(StateTracker& tracker, unsigned state_mask, core::nanoseconds_t deadline) { + tracker_ = &tracker; + state_mask_ = state_mask; + deadline_ = deadline; + } + + // Value returned by wait_state(); valid only after join(). + bool result() const { + return result_; + } + + void wait_running() { + while (!running_) { + core::sleep_for(core::ClockMonotonic, core::Microsecond); + } + } + +private: + virtual void run() { + running_ = true; + result_ = tracker_->wait_state(state_mask_, deadline_); + running_ = false; + } + + StateTracker* tracker_; + unsigned state_mask_; + core::nanoseconds_t deadline_; + bool result_; + core::AtomicBool running_; +}; + +class SignalThread : public core::Thread { +public: + SignalThread() + : tracker_(NULL) + , delay_(0) { + } + + void init(StateTracker& tracker, unsigned delay) { + tracker_ = &tracker; + delay_ = delay; + } + + void unblock() { + go_ = true; + } + + void wait_spinning() { + while (!spinning_) { + core::cpu_relax(); + } + } + +private: + virtual void run() { + spinning_ = true; + while (!go_) { + core::cpu_relax(); + } + for (unsigned n = 0; n < delay_; n++) { + core::cpu_relax(); + } + tracker_->register_packet(); + } + + StateTracker* tracker_; + unsigned delay_; + core::AtomicBool spinning_; + core::AtomicBool go_; +}; + +} // namespace + +TEST_GROUP(state_tracker) {}; + +TEST(state_tracker, already_active) { + StateTracker tracker; + + tracker.register_packet(); + CHECK(tracker.get_state() == sndio::DeviceState_Active); + + const core::nanoseconds_t start = core::timestamp(core::ClockMonotonic); + CHECK(tracker.wait_state(sndio::DeviceState_Active, NoDeadline)); + CHECK(core::timestamp(core::ClockMonotonic) - start < MaxImmediate); + + tracker.unregister_packet(); +} + +TEST(state_tracker, already_idle) { + StateTracker tracker; + + CHECK(tracker.get_state() == sndio::DeviceState_Idle); + + const core::nanoseconds_t start = core::timestamp(core::ClockMonotonic); + CHECK(tracker.wait_state(sndio::DeviceState_Idle, NoDeadline)); + CHECK(core::timestamp(core::ClockMonotonic) - start < MaxImmediate); +} + +TEST(state_tracker, zero_deadline_match) { + StateTracker tracker; + + tracker.register_packet(); + + const core::nanoseconds_t start = core::timestamp(core::ClockMonotonic); + CHECK(tracker.wait_state(sndio::DeviceState_Active, 0)); + CHECK(core::timestamp(core::ClockMonotonic) - start < MaxImmediate); + + tracker.unregister_packet(); +} + +// Zero deadline behaves like NoDeadline: it blocks until the mask matches. +TEST(state_tracker, zero_deadline_blocks_until_match) { + StateTracker tracker; + + WaitThread thr; + thr.init(tracker, sndio::DeviceState_Active, 0); + CHECK(thr.start()); + + thr.wait_running(); + core::sleep_for(core::ClockMonotonic, SettleDelay); + + tracker.register_packet(); + + thr.join(); + CHECK(thr.result()); + + tracker.unregister_packet(); +} + +TEST(state_tracker, past_deadline) { + StateTracker tracker; + + const core::nanoseconds_t start = core::timestamp(core::ClockMonotonic); + CHECK(!tracker.wait_state(sndio::DeviceState_Active, start - core::Second)); + CHECK(core::timestamp(core::ClockMonotonic) - start < MaxImmediate); +} + +TEST(state_tracker, empty_mask) { + StateTracker tracker; + + const core::nanoseconds_t start = core::timestamp(core::ClockMonotonic); + CHECK(tracker.wait_state(0, NoDeadline)); + CHECK(core::timestamp(core::ClockMonotonic) - start < MaxImmediate); +} + +TEST(state_tracker, timeout) { + StateTracker tracker; + + WaitThread thr; + thr.init(tracker, sndio::DeviceState_Active, + core::timestamp(core::ClockMonotonic) + ShortTimeout); + CHECK(thr.start()); + + // Nothing ever changes state, so the thread can exit only when deadline expires. + thr.join(); + CHECK(!thr.result()); +} + +TEST(state_tracker, wakeup_on_register_packet) { + StateTracker tracker; + + WaitThread thr; + thr.init(tracker, sndio::DeviceState_Active, NoDeadline); + CHECK(thr.start()); + + thr.wait_running(); + core::sleep_for(core::ClockMonotonic, SettleDelay); + + tracker.register_packet(); + + thr.join(); + CHECK(thr.result()); + + tracker.unregister_packet(); +} + +TEST(state_tracker, wakeup_on_register_session) { + StateTracker tracker; + + WaitThread thr; + thr.init(tracker, sndio::DeviceState_Active, NoDeadline); + CHECK(thr.start()); + + thr.wait_running(); + core::sleep_for(core::ClockMonotonic, SettleDelay); + + tracker.register_session(); + + thr.join(); + CHECK(thr.result()); + + tracker.unregister_session(); +} + +TEST(state_tracker, wakeup_on_idle) { + StateTracker tracker; + + tracker.register_packet(); + CHECK(tracker.get_state() == sndio::DeviceState_Active); + + WaitThread thr; + thr.init(tracker, sndio::DeviceState_Idle, NoDeadline); + CHECK(thr.start()); + + thr.wait_running(); + core::sleep_for(core::ClockMonotonic, SettleDelay); + + tracker.unregister_packet(); + + thr.join(); + CHECK(thr.result()); +} + +TEST(state_tracker, wakeup_on_broken) { + StateTracker tracker; + + WaitThread thr; + thr.init(tracker, sndio::DeviceState_Broken, NoDeadline); + CHECK(thr.start()); + + thr.wait_running(); + core::sleep_for(core::ClockMonotonic, SettleDelay); + + tracker.set_broken(); + + thr.join(); + CHECK(thr.result()); +} + +TEST(state_tracker, wakeup_on_closed) { + StateTracker tracker; + + WaitThread thr; + thr.init(tracker, sndio::DeviceState_Closed, NoDeadline); + CHECK(thr.start()); + + thr.wait_running(); + core::sleep_for(core::ClockMonotonic, SettleDelay); + + tracker.set_closed(); + + thr.join(); + CHECK(thr.result()); +} + +TEST(state_tracker, multi_bit_mask) { + StateTracker tracker; + + const unsigned state_mask = + (unsigned)sndio::DeviceState_Active | (unsigned)sndio::DeviceState_Broken; + + WaitThread thr; + thr.init(tracker, state_mask, NoDeadline); + CHECK(thr.start()); + + thr.wait_running(); + core::sleep_for(core::ClockMonotonic, SettleDelay); + + tracker.set_broken(); + + thr.join(); + CHECK(thr.result()); +} + +TEST(state_tracker, concurrent_waiters) { + StateTracker tracker; + + WaitThread threads[NumThreads]; + + for (size_t i = 0; i < NumThreads; i++) { + threads[i].init(tracker, sndio::DeviceState_Active, NoDeadline); + CHECK(threads[i].start()); + } + + for (size_t i = 0; i < NumThreads; i++) { + threads[i].wait_running(); + } + + core::sleep_for(core::ClockMonotonic, SettleDelay); + + tracker.register_packet(); + + for (size_t i = 0; i < NumThreads; i++) { + threads[i].join(); + CHECK(threads[i].result()); + } + + tracker.unregister_packet(); +} + +// Covers the bug where a waiter that lost the semaphore-owner election passed the +// absolute deadline to the condvar as if it were a relative timeout. +TEST(state_tracker, concurrent_waiters_mixed_deadlines) { + StateTracker tracker; + + WaitThread thr_a; + thr_a.init(tracker, sndio::DeviceState_Broken, NoDeadline); + CHECK(thr_a.start()); + + thr_a.wait_running(); + core::sleep_for(core::ClockMonotonic, SettleDelay * 2); + + WaitThread thr_b; + thr_b.init(tracker, sndio::DeviceState_Broken, + core::timestamp(core::ClockMonotonic) + ShortTimeout); + CHECK(thr_b.start()); + + const core::nanoseconds_t start = core::timestamp(core::ClockMonotonic); + thr_b.join(); + CHECK(!thr_b.result()); + CHECK(core::timestamp(core::ClockMonotonic) - start < core::Second); + + tracker.set_broken(); + + thr_a.join(); + CHECK(thr_a.result()); +} + +TEST(state_tracker, long_non_expiring_deadline) { + StateTracker tracker; + + WaitThread thr; + thr.init(tracker, sndio::DeviceState_Active, + core::timestamp(core::ClockMonotonic) + LongTimeout); + CHECK(thr.start()); + + thr.wait_running(); + core::sleep_for(core::ClockMonotonic, SettleDelay); + + tracker.register_packet(); + + thr.join(); + CHECK(thr.result()); + + tracker.unregister_packet(); +} + +// Signaler changes state concurrently with the waiter. The delay is swept so that the +// state change eventually lands exactly in between the moment when the waiter checks +// the state and the moment when it announces itself to the signaler. If that window is +// not handled, the signaler skips the wake up and the waiter blocks forever. +TEST(state_tracker, race_signal_and_wait) { + enum { MaxDelay = 256, NumRepeats = 8 }; + + for (unsigned delay = 0; delay < MaxDelay; delay++) { + for (int rep = 0; rep < NumRepeats; rep++) { + StateTracker tracker; + + SignalThread signaler; + signaler.init(tracker, delay); + CHECK(signaler.start()); + + signaler.wait_spinning(); + signaler.unblock(); + CHECK(tracker.wait_state(sndio::DeviceState_Active, NoDeadline)); + + signaler.join(); + tracker.unregister_packet(); + } + } +} + +} // namespace pipeline +} // namespace roc