MT#62181 rework AmCondition

Use std::condition_variable and std::mutex to implement AmCondition.
Only bools are used for conditions in the code, so make it not a
template.

Change-Id: I57d67492e29c220a5ce941ef67d142b34dcebbff

apps/rtmp/RtmpSender.h

@@ -42,7 +42,7 @@ class RtmpSender
   // sender queue
   queue<RTMPPacket> q_send;
   AmMutex           m_q_send;
-  AmCondition<bool> has_work;
+  AmCondition has_work;
 
   // ptr to RtmpConnection::rtmp
   RTMP* p_rtmp;

apps/sbc/call_control/bl_redis/RedisConnectionPool.h

@@ -21,8 +21,8 @@ class RedisConnectionPool
   unsigned int failed_connections;
   AmMutex connections_mut;
 
-  AmCondition<bool> have_active_connection;
-  AmCondition<bool> try_connect;
+  AmCondition have_active_connection;
+  AmCondition try_connect;
 
   vector<unsigned int> retry_timers;
   unsigned int retry_index;

apps/voicemail/AmMail.h

@@ -108,7 +108,7 @@ class AmMailDeamon: public AmThread
 
   AmMutex      event_fifo_mut;
   std::queue<AmMail*>   event_fifo;
-  AmCondition<bool> _run_cond;
+  AmCondition _run_cond;
 
   AmMailDeamon() : _run_cond(false) {}
   AmMailDeamon(const AmMailDeamon&) : _run_cond(false) {}

apps/webconference/WebConference.h

@@ -212,7 +212,7 @@ class WebConferenceCleaner
   : public AmThread 
 {
   WebConferenceFactory* factory;
-  AmCondition<bool> is_stopped;
+  AmCondition is_stopped;
 
 public:
   WebConferenceCleaner(WebConferenceFactory* factory)

apps/xmlrpc2di/MultithreadXmlRpcServer.h

@@ -25,9 +25,9 @@ namespace XmlRpc {
   {
 
     MultithreadXmlRpcServer* chief;
-    AmCondition<bool> runcond;
+    AmCondition runcond;
 
-    AmCondition<bool> running;
+    AmCondition running;
 
   public: 
     WorkerThread(MultithreadXmlRpcServer* chief);
@@ -68,7 +68,7 @@ namespace XmlRpc {
   private:
     AmMutex waiting_mut;
     std::queue<WorkerThread*> waiting; 
-    AmCondition<bool> have_waiting;
+    AmCondition have_waiting;
     std::vector<WorkerThread*> workers;
     WorkerThread* getIdleThread();
   }; 

apps/xmlrpc2di/XMLRPC2DI.h

@@ -105,7 +105,7 @@ class XMLRPC2DIServer
   unsigned int port; 
   string bind_ip;
 
-  AmCondition<bool> running;
+  AmCondition running;
 
   XMLRPC2DIServerCallsMethod       calls_method;
   XMLRPC2DIServerSetLoglevelMethod setloglevel_method;

core/AmB2ABSession.h

@@ -251,7 +251,7 @@ class AmB2ABCalleeSession: public AmB2ABSession
  protected:
   void onB2ABEvent(B2ABEvent* ev);
   void onBeforeDestroy();
-  AmCondition<bool>* released;
+  AmCondition* released;
 };
 
 /** BRIDGE_DELAY is needed because of possible different packet sizes */ 
@@ -268,7 +268,7 @@ class AmSessionAudioConnector {
   bool connected[2];
   AmMutex tag_mut;
 
-  AmCondition<bool> released;
+  AmCondition released;
   
  public:
   /** create a connector, connect audio to sess */

core/AmConferenceChannel.h

@@ -43,7 +43,7 @@ class ChannelWritingFile : public async_file
 
   void on_flushed();
 
-  AmCondition<bool> finished;
+  AmCondition finished;
 };
 
 /** 

core/AmEventQueue.h

@@ -69,7 +69,7 @@ protected:
 
   std::queue<AmEvent*>      ev_queue;
   AmMutex                   m_queue;
-  AmCondition<bool>         ev_pending;
+  AmCondition         ev_pending;
 
   bool finalized;
 

core/AmEventQueueProcessor.h

@@ -42,7 +42,7 @@ class EventQueueWorker
 {
   atomic_bool stop_requested;
 
-  AmCondition<bool> runcond;
+  AmCondition runcond;
   std::deque<AmEventQueue*> process_queues;
   AmMutex process_queues_mut;
 

core/AmMediaProcessor.h

@@ -54,7 +54,7 @@ struct SchedRequest;
 class AmMediaSession
 {
   private:
-    AmCondition<bool> processing_media;
+    AmCondition processing_media;
 
   public:
     AmMediaSession(): processing_media(false) { }

core/AmSession.h

@@ -136,7 +136,7 @@ private:
 
 protected:
 
-  AmCondition<bool> sess_stopped;
+  AmCondition sess_stopped;
 
   /** this is the group the media is processed with 
       - by default local tag */

core/AmSessionContainer.h

@@ -61,10 +61,10 @@ class AmSessionContainer : public AmThread
   AmMutex      ds_mut;
 
   /** is container closed for new sessions? */
-  AmCondition<bool> _container_closed;
+  AmCondition _container_closed;
 
   /** the daemon only runs if this is true */
-  AmCondition<bool> _run_cond;
+  AmCondition _run_cond;
 
   /** We are a Singleton ! Avoid people to have their own instance. */
   AmSessionContainer();

core/AmSessionProcessor.h

@@ -70,7 +70,7 @@ class AmSessionProcessorThread
   std::vector<AmSession*> startup_sessions;
   atomic_bool stop_requested;
 
-  AmCondition<bool> runcond;
+  AmCondition runcond;
   std::set<AmEventQueue*> process_sessions;
   AmMutex process_sessions_mut;
 

core/AmThread.h

@@ -36,6 +36,7 @@
 #include <queue>
 #include <mutex>
 #include <atomic>
+#include <condition_variable>
 
 using std::lock_guard;
 using std::atomic_bool;
@@ -52,84 +53,56 @@ public:
 };
 
 /**
- * \brief C++ Wrapper class for pthread condition
+ * \brief Wrapper class for std::condition_variable
  */
-template<class T>
 class AmCondition
 {
-  T               t;
-  pthread_mutex_t m;
-  pthread_cond_t  cond;
-
-  void init_cond() {
-    pthread_mutex_init(&m,NULL);
-    pthread_cond_init(&cond,NULL);
-  }
+  bool                    t = false;
+  std::mutex              m;
+  std::condition_variable cond;
 
 public:
-  AmCondition() : t() { init_cond(); }
-  AmCondition(const T& _t) : t(_t) { init_cond(); }
-    
-  ~AmCondition()
-  {
-    pthread_cond_destroy(&cond);
-    pthread_mutex_destroy(&m);
-  }
-    
+  AmCondition() = default;
+  AmCondition(const bool& _t) : t(_t) {}
+
   /** Change the condition's value. */
-  void set(const T& newval)
+  void set(const bool newval)
   {
-    pthread_mutex_lock(&m);
+    std::lock_guard<std::mutex> l(m);
     t = newval;
     if(t)
-      pthread_cond_broadcast(&cond);
-    pthread_mutex_unlock(&m);
+      cond.notify_all();
   }
     
-  T get()
+  bool get()
   {
-    T val;
-    pthread_mutex_lock(&m);
-    val = t;
-    pthread_mutex_unlock(&m);
-    return val;
+    std::lock_guard<std::mutex> l(m);
+    return t;
   }
     
   /** Waits for the condition to be true. */
   void wait_for()
   {
-    pthread_mutex_lock(&m);
+    std::unique_lock<std::mutex> l(m);
     while(!t){
-      pthread_cond_wait(&cond,&m);
+      cond.wait(l);
     }
-    pthread_mutex_unlock(&m);
   }
   
   /** Waits for the condition to be true or a timeout. */
   bool wait_for_to(unsigned long msec)
   {
-    struct timeval now;
-    struct timespec timeout;
-    int retcode = 0;
-    bool ret = false;
-
-    gettimeofday(&now, NULL);
-    timeout.tv_sec = now.tv_sec + (msec / 1000);
-    timeout.tv_nsec = (now.tv_usec + (msec % 1000)*1000)*1000;
-    if(timeout.tv_nsec >= 1000000000){
-      timeout.tv_sec++;
-      timeout.tv_nsec -= 1000000000;
-    }
+    auto timeout = std::chrono::system_clock::now();
+    timeout += std::chrono::milliseconds(msec);
 
-    pthread_mutex_lock(&m);
-    while(!t && !retcode){
-      retcode = pthread_cond_timedwait(&cond,&m, &timeout);
+    std::unique_lock<std::mutex> l(m);
+    while(!t){
+      auto retcode = cond.wait_until(l, timeout);
+      if (retcode == std::cv_status::timeout)
+        break;
     }
 
-    if(t) ret = true;
-    pthread_mutex_unlock(&m);
-
-    return ret;
+    return t;
   }
 };
 
@@ -188,7 +161,7 @@ class AmThreadWatcher: public AmThread
   AmMutex          q_mut;
 
   /** the daemon only runs if this is true */
-  AmCondition<bool> _run_cond;
+  AmCondition _run_cond;
     
   AmThreadWatcher();
   void run();

core/SipCtrlInterface.h

@@ -62,7 +62,7 @@ class _SipCtrlInterface:
 
     friend class udp_trsp;
 
-    AmCondition<bool> stopped;
+    AmCondition stopped;
     
     unsigned short    nr_udp_sockets;
     udp_trsp_socket** udp_sockets;

core/sems.cpp

@@ -233,7 +233,7 @@ static bool apply_args(std::map<char,string>& args)
 }
 
 /** Flag to mark the shutdown is in progress (in the main process) */
-static AmCondition<bool> is_shutting_down(false);
+static AmCondition is_shutting_down(false);
 
 static void signal_handler(int sig)
 {

core/sip/wheeltimer.h

@@ -122,7 +122,7 @@ class _wheeltimer:
 
     // request backlog lock (insert/remove)
     AmMutex               reqs_m;
-    AmCondition<bool>     reqs_cond; // to wake up worker thread when a request is added
+    AmCondition     reqs_cond; // to wake up worker thread when a request is added
     std::deque<timer_req> reqs_backlog;
     std::deque<timer_req> reqs_process;