feat: implemented semaphore and condition variable based statetracker

Author: flw5469

src/internal_modules/roc_pipeline/state_tracker.cpp

@@ -7,15 +7,74 @@
  */
 
 #include "roc_pipeline/state_tracker.h"
+#include "roc_core/log.h"
 #include "roc_core/panic.h"
 
 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)
+    , waiting_mask_(0)
+    , mutex_()
+    , waiting_con_(mutex_) {
+}
+
+// StateTracker::~StateTracker() {
+//     mutex_.unlock();
+// }
+
+// This method should block until the state becomes any of the states specified by the
+// mask, or deadline expires. E.g. if mask is ACTIVE | PAUSED, it should block until state
+// becomes either ACTIVE or PAUSED. (Currently only two states are used, but later more
+// states will be needed). Deadline should be an absolute timestamp.
+
+// Questions:
+// - When should the function return true vs false
+bool StateTracker::wait_state(unsigned int state_mask, core::nanoseconds_t deadline) {
+    
+    mutex_.lock();
+    for (;;) {
+        // If no state is specified in state_mask, return immediately
+        if (state_mask == 0) {
+            return true;
+        }
+
+        if (static_cast<unsigned>(get_state()) & state_mask) {
+            return true;
+        }
+
+        if (deadline >= 0 && deadline <= core::timestamp(core::ClockMonotonic)) {
+            return false;
+        }
+
+        if (sem_is_occupied_.compare_exchange(0, 1)) {
+
+            if (deadline >= 0) {
+                mutex_.unlock();
+                (void)sem_.timed_wait(deadline);
+
+            } else {
+                mutex_.unlock();
+                sem_.wait();
+            }
+
+            mutex_.lock();
+            sem_is_occupied_ = 0;
+            waiting_con_.broadcast();
+
+        } else {
+            if (deadline >= 0) {
+                (void)waiting_con_.timed_wait(deadline);
+            } else {
+                waiting_con_.wait();
+            }
+        }
+    }
 }
 
 sndio::DeviceState StateTracker::get_state() const {
@@ -65,22 +124,50 @@ 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();
     }
+
+    // if (--active_sessions_ < 0) {
+    //     roc_panic("state tracker: unpaired register/unregister session");
+    // }
 }
 
 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();
+    }
+
+    // if (--pending_packets_ < 0) {
+    //     roc_panic("state tracker: unpaired register/unregister packet");
+    // }
+}
+
+void StateTracker::signal_state_change() {
+    // if (waiting_mask_ != 0 && (static_cast<unsigned>(get_state()) & waiting_mask_)) {
+    //     sem_.post();
+    // }
+    if (sem_is_occupied_) {
+        roc_log(LogDebug, "signaling");
+        sem_.post();
     }
 }
 

src/internal_modules/roc_pipeline/state_tracker.h

@@ -12,9 +12,11 @@
 #ifndef ROC_PIPELINE_STATE_TRACKER_H_
 #define ROC_PIPELINE_STATE_TRACKER_H_
 
-#include "roc_core/atomic_int.h"
+#include "roc_core/atomic.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 +34,9 @@ class StateTracker : public core::NonCopyable<> {
     //! Initialize all counters to zero.
     StateTracker();
 
+    //! Block until state becomes any of the ones specified by state_mask.
+    bool wait_state(unsigned state_mask, core::nanoseconds_t deadline);
+
     //! Compute current state.
     sndio::DeviceState get_state() const;
 
@@ -63,9 +68,16 @@ class StateTracker : public core::NonCopyable<> {
     void unregister_packet();
 
 private:
-    core::AtomicInt<int32_t> halt_state_;
-    core::AtomicInt<int32_t> active_sessions_;
-    core::AtomicInt<int32_t> pending_packets_;
+    core::Semaphore sem_;
+    core::Atomic<int> halt_state_;
+    core::Atomic<int> active_sessions_;
+    core::Atomic<int> pending_packets_;
+    core::Atomic<int> sem_is_occupied_;
+    core::Atomic<unsigned> waiting_mask_;
+    core::Mutex mutex_;
+    core::Cond waiting_con_;
+
+    void signal_state_change();
 };
 
 } // namespace pipeline