State tracker 2026 01

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 <semaphore.h>\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=[

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
     │       ├── ...

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);

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

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_

src/internal_modules/roc_core/target_posix_ext/roc_core/semaphore.cpp → 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 <errno.h>
@@ -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;
         }
 

src/internal_modules/roc_core/target_posix_ext/roc_core/semaphore.h → 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_

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<unsigned>(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,33 +119,51 @@ 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 {
     return (size_t)active_sessions_;
 }
 
 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();
     }
 }