update: add statetracker wait(), implement semaphore using mutex and cond var in platform without sem_clockwait() to prevent semaphore:timed_wait() to hang when changing sysem time, add test cases for state tracker and semaphore

Author: flw5469

SConstruct

@@ -782,10 +782,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=[

src/internal_modules/roc_core/target_nosem/roc_core/semaphore.cpp

@@ -0,0 +1,101 @@
+/*
+ * Copyright (c) 2020 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/cpu_instructions.h"
+#include "roc_core/errno_to_str.h"
+#include "roc_core/log.h"
+#include "roc_core/panic.h"
+
+#include <errno.h>
+#include <time.h>
+
+namespace roc {
+namespace core {
+
+Semaphore::Semaphore(unsigned counter)
+    : mutex_()
+    , counter_(counter)
+    , guard_(0) {
+    if (int err = pthread_cond_init(&cond_, NULL)) {
+        roc_panic("sem: pthread_cond_init(): %s", errno_to_str(err).c_str());
+    }
+}
+
+Semaphore::~Semaphore() {
+    while (guard_) {
+        cpu_relax();
+    }
+
+    int err = 0;
+    if ((err = pthread_cond_destroy(&cond_))) {
+        roc_panic("sem: pthread_cond_destroy(): %s", errno_to_str(err).c_str());
+    }
+}
+
+bool Semaphore::timed_wait(nanoseconds_t deadline) {
+    if (deadline < 0) {
+        roc_panic("semaphore: unexpected negative deadline");
+    }
+
+    roc_log(roc::LogDebug, "origin time is %" PRId64 "\n", deadline);
+    roc_log(roc::LogDebug, "time is %" PRId64 "\n", deadline);
+    roc_log(roc::LogDebug, "now time is %" PRId64 "\n",
+            core::timestamp(core::ClockMonotonic));
+
+    struct timespec ts;
+    ts.tv_sec = time_t(deadline / Second);
+    ts.tv_nsec = long(deadline % Second);
+
+    int err = 0;
+    mutex_.lock();
+    while (err == 0 && counter_ == 0) {
+        err = pthread_cond_clockwait(&cond_, &mutex_.mutex_, CLOCK_MONOTONIC, &ts);
+        printf("finish waiting without sem");
+        // roc_log(roc::LogDebug, "finish waiting without sem");
+
+        if (err != 0 && err != ETIMEDOUT) {
+            roc_panic("semaphore: pthread_cond_timedwait(): %s",
+                      errno_to_str(err).c_str());
+        }
+    }
+
+    if (err == 0) {
+        counter_--;
+    }
+    mutex_.unlock();
+
+    // return false when err == ETIMEDOUT
+    return (err == 0);
+}
+
+void Semaphore::wait() {
+    int err = 0;
+    mutex_.lock();
+    while (err == 0 && counter_ == 0) {
+        if ((err = pthread_cond_wait(&cond_, &mutex_.mutex_))) {
+            roc_panic("semaphore: pthread_cond_wait(): %s", errno_to_str(err).c_str());
+        }
+    }
+    counter_--;
+    mutex_.unlock();
+}
+
+void Semaphore::post() {
+    ++guard_;
+    mutex_.lock();
+    counter_++;
+    if (int err = pthread_cond_broadcast(&cond_)) {
+        roc_panic("cond: pthread_cond_broadcast(): %s", errno_to_str(err).c_str());
+    }
+    mutex_.unlock();
+    --guard_;
+}
+
+} // namespace core
+} // namespace roc

src/internal_modules/roc_core/target_nosem/roc_core/semaphore.h

@@ -0,0 +1,55 @@
+/*
+ * Copyright (c) 2020 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/atomic.h"
+#include "roc_core/attributes.h"
+#include "roc_core/mutex.h"
+#include "roc_core/noncopyable.h"
+#include "roc_core/time.h"
+#include <pthread.h>
+
+namespace roc {
+namespace core {
+
+//! Semaphore.
+class Semaphore : public NonCopyable<> {
+public:
+    //! Initialize semaphore with given counter.
+    explicit Semaphore(unsigned counter = 0);
+
+    ~Semaphore();
+
+    //! Wait until the counter becomes non-zero, decrement it, and return true.
+    //! If deadline expires before the counter becomes non-zero, returns false.
+    //! Deadline should be in the same time domain as core::timestamp().
+    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.
+    //! This method is implemented using mutex for platforms where
+    void post();
+
+private:
+    pthread_cond_t cond_;
+    core::Mutex mutex_;
+    Atomic<unsigned> counter_;
+    Atomic<int> guard_;
+};
+
+} // namespace core
+} // namespace roc
+
+#endif // ROC_CORE_SEMAPHORE_H_

src/internal_modules/roc_core/target_posix/roc_core/mutex.h

@@ -70,6 +70,7 @@ class Mutex : public NonCopyable<> {
 
 private:
     friend class Cond;
+    friend class Semaphore;
 
     mutable pthread_mutex_t mutex_;
     mutable AtomicInt<int32_t> guard_;

…/target_posix_ext/roc_core/semaphore.cpp …/target_posix_sem/roc_core/semaphore.cppsrc/internal_modules/roc_core/target_posix_ext/roc_core/semaphore.cpp renamed to src/internal_modules/roc_core/target_posix_sem/roc_core/semaphore.cpp src/internal_modules/roc_core/target_posix_ext/roc_core/semaphore.cpp renamed to 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>
@@ -34,16 +35,17 @@ Semaphore::~Semaphore() {
 }
 
 bool Semaphore::timed_wait(nanoseconds_t deadline) {
+    printf("Enter function of timed_wait");
     if (deadline < 0) {
         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;
         }
 

…re/target_posix_ext/roc_core/semaphore.h …re/target_posix_sem/roc_core/semaphore.hsrc/internal_modules/roc_core/target_posix_ext/roc_core/semaphore.h renamed to src/internal_modules/roc_core/target_posix_sem/roc_core/semaphore.h src/internal_modules/roc_core/target_posix_ext/roc_core/semaphore.h renamed to 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

@@ -19,7 +19,6 @@ StateTracker::StateTracker()
     , active_sessions_(0)
     , pending_packets_(0)
     , sem_is_occupied_(0)
-    , waiting_mask_(0)
     , mutex_()
     , waiting_con_(mutex_) {
 }
@@ -36,24 +35,24 @@ StateTracker::StateTracker()
 // Questions:
 // - When should the function return true vs false
 bool StateTracker::wait_state(unsigned int state_mask, core::nanoseconds_t deadline) {
-    
+    // If no state is specified in state_mask, return immediately
+    if (state_mask == 0) {
+        return true;
+    }
+
     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) {
+            mutex_.unlock();
             return true;
         }
 
         if (deadline >= 0 && deadline <= core::timestamp(core::ClockMonotonic)) {
+            mutex_.unlock();
             return false;
         }
 
         if (sem_is_occupied_.compare_exchange(0, 1)) {
-
             if (deadline >= 0) {
                 mutex_.unlock();
                 (void)sem_.timed_wait(deadline);

src/internal_modules/roc_pipeline/state_tracker.h

@@ -73,7 +73,6 @@ class StateTracker : public core::NonCopyable<> {
     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_;