diff --git a/apps/rtmp/RtmpSender.h b/apps/rtmp/RtmpSender.h index 3ca84727..08a6e8b8 100644 --- a/apps/rtmp/RtmpSender.h +++ b/apps/rtmp/RtmpSender.h @@ -42,7 +42,7 @@ class RtmpSender // sender queue queue q_send; AmMutex m_q_send; - AmCondition has_work; + AmCondition has_work; // ptr to RtmpConnection::rtmp RTMP* p_rtmp; diff --git a/apps/sbc/call_control/bl_redis/RedisConnectionPool.h b/apps/sbc/call_control/bl_redis/RedisConnectionPool.h index 94389d5d..b37ebd22 100644 --- a/apps/sbc/call_control/bl_redis/RedisConnectionPool.h +++ b/apps/sbc/call_control/bl_redis/RedisConnectionPool.h @@ -21,8 +21,8 @@ class RedisConnectionPool unsigned int failed_connections; AmMutex connections_mut; - AmCondition have_active_connection; - AmCondition try_connect; + AmCondition have_active_connection; + AmCondition try_connect; vector retry_timers; unsigned int retry_index; diff --git a/apps/voicemail/AmMail.h b/apps/voicemail/AmMail.h index a5499ec4..b72ebe15 100644 --- a/apps/voicemail/AmMail.h +++ b/apps/voicemail/AmMail.h @@ -108,7 +108,7 @@ class AmMailDeamon: public AmThread AmMutex event_fifo_mut; std::queue event_fifo; - AmCondition _run_cond; + AmCondition _run_cond; AmMailDeamon() : _run_cond(false) {} AmMailDeamon(const AmMailDeamon&) : _run_cond(false) {} diff --git a/apps/webconference/WebConference.h b/apps/webconference/WebConference.h index 37f5f7ae..b226a91c 100644 --- a/apps/webconference/WebConference.h +++ b/apps/webconference/WebConference.h @@ -212,7 +212,7 @@ class WebConferenceCleaner : public AmThread { WebConferenceFactory* factory; - AmCondition is_stopped; + AmCondition is_stopped; public: WebConferenceCleaner(WebConferenceFactory* factory) diff --git a/apps/xmlrpc2di/MultithreadXmlRpcServer.h b/apps/xmlrpc2di/MultithreadXmlRpcServer.h index a6be7316..6c981e1e 100644 --- a/apps/xmlrpc2di/MultithreadXmlRpcServer.h +++ b/apps/xmlrpc2di/MultithreadXmlRpcServer.h @@ -25,9 +25,9 @@ namespace XmlRpc { { MultithreadXmlRpcServer* chief; - AmCondition runcond; + AmCondition runcond; - AmCondition running; + AmCondition running; public: WorkerThread(MultithreadXmlRpcServer* chief); @@ -68,7 +68,7 @@ namespace XmlRpc { private: AmMutex waiting_mut; std::queue waiting; - AmCondition have_waiting; + AmCondition have_waiting; std::vector workers; WorkerThread* getIdleThread(); }; diff --git a/apps/xmlrpc2di/XMLRPC2DI.h b/apps/xmlrpc2di/XMLRPC2DI.h index 14977122..17c85a39 100644 --- a/apps/xmlrpc2di/XMLRPC2DI.h +++ b/apps/xmlrpc2di/XMLRPC2DI.h @@ -105,7 +105,7 @@ class XMLRPC2DIServer unsigned int port; string bind_ip; - AmCondition running; + AmCondition running; XMLRPC2DIServerCallsMethod calls_method; XMLRPC2DIServerSetLoglevelMethod setloglevel_method; diff --git a/core/AmB2ABSession.h b/core/AmB2ABSession.h index 073b6c5c..1b3f7cea 100644 --- a/core/AmB2ABSession.h +++ b/core/AmB2ABSession.h @@ -251,7 +251,7 @@ class AmB2ABCalleeSession: public AmB2ABSession protected: void onB2ABEvent(B2ABEvent* ev); void onBeforeDestroy(); - AmCondition* 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 released; + AmCondition released; public: /** create a connector, connect audio to sess */ diff --git a/core/AmConferenceChannel.h b/core/AmConferenceChannel.h index e382e035..501600e4 100644 --- a/core/AmConferenceChannel.h +++ b/core/AmConferenceChannel.h @@ -43,7 +43,7 @@ class ChannelWritingFile : public async_file void on_flushed(); - AmCondition finished; + AmCondition finished; }; /** diff --git a/core/AmEventQueue.h b/core/AmEventQueue.h index f1a3bcff..0790163d 100644 --- a/core/AmEventQueue.h +++ b/core/AmEventQueue.h @@ -69,7 +69,7 @@ protected: std::queue ev_queue; AmMutex m_queue; - AmCondition ev_pending; + AmCondition ev_pending; bool finalized; diff --git a/core/AmEventQueueProcessor.h b/core/AmEventQueueProcessor.h index 2882892d..7efdd4aa 100644 --- a/core/AmEventQueueProcessor.h +++ b/core/AmEventQueueProcessor.h @@ -42,7 +42,7 @@ class EventQueueWorker { atomic_bool stop_requested; - AmCondition runcond; + AmCondition runcond; std::deque process_queues; AmMutex process_queues_mut; diff --git a/core/AmMediaProcessor.h b/core/AmMediaProcessor.h index 0e95fa76..a6681fab 100644 --- a/core/AmMediaProcessor.h +++ b/core/AmMediaProcessor.h @@ -54,7 +54,7 @@ struct SchedRequest; class AmMediaSession { private: - AmCondition processing_media; + AmCondition processing_media; public: AmMediaSession(): processing_media(false) { } diff --git a/core/AmSession.h b/core/AmSession.h index ddac24b8..8c136702 100644 --- a/core/AmSession.h +++ b/core/AmSession.h @@ -136,7 +136,7 @@ private: protected: - AmCondition sess_stopped; + AmCondition sess_stopped; /** this is the group the media is processed with - by default local tag */ diff --git a/core/AmSessionContainer.h b/core/AmSessionContainer.h index c6b73769..348bdd22 100644 --- a/core/AmSessionContainer.h +++ b/core/AmSessionContainer.h @@ -61,10 +61,10 @@ class AmSessionContainer : public AmThread AmMutex ds_mut; /** is container closed for new sessions? */ - AmCondition _container_closed; + AmCondition _container_closed; /** the daemon only runs if this is true */ - AmCondition _run_cond; + AmCondition _run_cond; /** We are a Singleton ! Avoid people to have their own instance. */ AmSessionContainer(); diff --git a/core/AmSessionProcessor.h b/core/AmSessionProcessor.h index 69b40173..6376d6a7 100644 --- a/core/AmSessionProcessor.h +++ b/core/AmSessionProcessor.h @@ -70,7 +70,7 @@ class AmSessionProcessorThread std::vector startup_sessions; atomic_bool stop_requested; - AmCondition runcond; + AmCondition runcond; std::set process_sessions; AmMutex process_sessions_mut; diff --git a/core/AmThread.h b/core/AmThread.h index d005667d..df0e9242 100644 --- a/core/AmThread.h +++ b/core/AmThread.h @@ -36,6 +36,7 @@ #include #include #include +#include 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 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 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 l(m); + return t; } /** Waits for the condition to be true. */ void wait_for() { - pthread_mutex_lock(&m); + std::unique_lock 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 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 _run_cond; + AmCondition _run_cond; AmThreadWatcher(); void run(); diff --git a/core/SipCtrlInterface.h b/core/SipCtrlInterface.h index 80e35379..672085d2 100644 --- a/core/SipCtrlInterface.h +++ b/core/SipCtrlInterface.h @@ -62,7 +62,7 @@ class _SipCtrlInterface: friend class udp_trsp; - AmCondition stopped; + AmCondition stopped; unsigned short nr_udp_sockets; udp_trsp_socket** udp_sockets; diff --git a/core/sems.cpp b/core/sems.cpp index d9058a3e..084cc810 100644 --- a/core/sems.cpp +++ b/core/sems.cpp @@ -233,7 +233,7 @@ static bool apply_args(std::map& args) } /** Flag to mark the shutdown is in progress (in the main process) */ -static AmCondition is_shutting_down(false); +static AmCondition is_shutting_down(false); static void signal_handler(int sig) { diff --git a/core/sip/wheeltimer.h b/core/sip/wheeltimer.h index 068afadc..a87325d1 100644 --- a/core/sip/wheeltimer.h +++ b/core/sip/wheeltimer.h @@ -122,7 +122,7 @@ class _wheeltimer: // request backlog lock (insert/remove) AmMutex reqs_m; - AmCondition 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 reqs_backlog; std::deque reqs_process;