sipwise
/
sems
mirror of https://github.com/sipwise/sems.git
1
0
Fork 0
Code Issues Releases Wiki Activity

MT#62181 global: use AmThread stop_requested

Browse Source 
Replace other instances of the same machanism with the one provided by
AmThread.

Change-Id: Id7572361e8f4f630051c6715106e0ca9ea4d61e9
mr13.3.1
Richard Fuchs 1 year ago
parent 9f70206617
commit 8b4e26b32a
23 changed files with 41 additions and 112 deletions
Show Stats Download Patch File Download Diff File

34
apps/db_reg_agent/DBRegAgent.cpp
Unescape View File

@ -260,20 +260,12 @@ int DBRegAgent::onLoad()
}
void DBRegAgent::onUnload() {
if (running) {
running = false;
registration_scheduler.stop();
DBG("unclean shutdown. Waiting for processing thread to stop.\n");
for (int i=0;i<400;i++) {
if (shutdown_finished)
break;
usleep(2000); // 2ms
}
stop();
join();
if (!shutdown_finished) {
WARN("processing thread could not be stopped, process will probably crash\n");
}
}
registration_scheduler.stop();
DBG("unclean shutdown. Waiting for processing thread to stop.\n");
registration_scheduler.join();
DBG("closing main DB connection\n");
MainDBConnection.disconnect();
@ -828,8 +820,8 @@ void DBRegAgent::process(AmEvent* ev) {
if(sys_ev){
DBG("Session received system Event\n");
if (sys_ev->sys_event == AmSystemEvent::ServerShutdown) {
running = false;
registration_scheduler.stop();
registration_scheduler.join();
}
return;
}
@ -1231,8 +1223,6 @@ void DBRegAgent::onSipReplyEvent(AmSipReplyEvent* ev) {
}
void DBRegAgent::run() {
running = shutdown_finished = true;
DBG("DBRegAgent thread: waiting 2 sec for server startup ...\n");
sleep(2);
@ -1244,8 +1234,7 @@ void DBRegAgent::run() {
}
DBG("running DBRegAgent thread...\n");
shutdown_finished = false;
while (running) {
while (!stop_requested()) {
waitForEventTimed(500); // 500 milliseconds
processEvents();
}
@ -1264,12 +1253,10 @@ void DBRegAgent::run() {
mysqlpp::Connection::thread_end();
DBG("DBRegAgent thread stopped.\n");
shutdown_finished = true;
}
void DBRegAgent::on_stop() {
DBG("DBRegAgent on_stop()...\n");
running = false;
}
void DBRegAgent::setRegistrationTimer(long object_id, uint64_t timeout,
@ -1728,7 +1715,7 @@ void DBRegAgent::invoke(const string& method,
// /////////////// processor thread /////////////////
DBRegAgentProcessorThread::DBRegAgentProcessorThread()
: AmEventQueue(this), stopped(false) {
: AmEventQueue(this) {
}
DBRegAgentProcessorThread::~DBRegAgentProcessorThread() {
@ -1785,7 +1772,7 @@ void DBRegAgentProcessorThread::run() {
allowance = DBRegAgent::ratelimit_rate;
reg_agent = DBRegAgent::instance();
while (!stopped) {
while (!stop_requested()) {
waitForEvent();
while (eventPending()) {
rateLimitWait();
@ -1806,7 +1793,8 @@ void DBRegAgentProcessorThread::process(AmEvent* ev) {
DBG("Session received system Event\n");
if (sys_ev->sys_event == AmSystemEvent::ServerShutdown) {
DBG("stopping processor thread\n");
stopped = true;
stop();
join();
}
return;
}

7
apps/db_reg_agent/DBRegAgent.h
Unescape View File

@ -103,7 +103,6 @@ class DBRegAgentProcessorThread
{
DBRegAgent* reg_agent;
bool stopped;
void rateLimitWait();
@ -270,12 +269,6 @@ class DBRegAgent
unsigned int expires;
/** processing thread running? */
bool running;
/** processing thread shutdown finished? */
bool shutdown_finished;
AmDynInvoke* uac_auth_i;
void DIcreateRegistration(int object_id, const string& user,

5
apps/dsm/SystemDSM.cpp
Unescape View File

@ -13,7 +13,7 @@
SystemDSM::SystemDSM(const DSMScriptConfig& config,
const string& startDiagName,
bool reload)
: stop_requested(false), AmEventQueue(this),
: AmEventQueue(this),
startDiagName(startDiagName), dummy_session(this),
reload(reload)
{
@ -51,7 +51,7 @@ void SystemDSM::run() {
return;
}
while (!stop_requested && !dummy_session.getStopped()) {
while (!stop_requested() && !dummy_session.getStopped()) {
waitForEvent();
processEvents();
}
@ -62,7 +62,6 @@ void SystemDSM::run() {
void SystemDSM::on_stop() {
DBG("requesting stop of SystemDSM\n");
stop_requested = true;
}
void SystemDSM::process(AmEvent* event) {

1
apps/dsm/SystemDSM.h
Unescape View File

@ -32,7 +32,6 @@ class SystemDSM
{
EventProxySession dummy_session;
atomic_bool stop_requested;
DSMStateEngine engine;
string startDiagName;

6
apps/monitoring/Monitoring.cpp
Unescape View File

@ -721,8 +721,7 @@ LogBucket& Monitor::getLogBucket(const string& call_id) {
void MonitorGarbageCollector::run() {
DBG("running MonitorGarbageCollector thread\n");
running.set(true);
while (running.get()) {
while (!stop_requested()) {
sleep(Monitor::gcInterval);
Monitor::instance()->clearFinished();
}
@ -735,7 +734,8 @@ void MonitorGarbageCollector::postEvent(AmEvent* e) {
if (sys_ev &&
sys_ev->sys_event == AmSystemEvent::ServerShutdown) {
DBG("stopping MonitorGarbageCollector thread\n");
running.set(false);
stop();
join();
return;
}

2
apps/monitoring/Monitoring.h
Unescape View File

@ -129,8 +129,6 @@ class MonitorGarbageCollector
: public AmThread,
public AmEventQueueInterface
{
atomic_bool running;
public:
void run();
void on_stop();

7
apps/registrar_client/SIPRegistrarClient.cpp
Unescape View File

@ -62,8 +62,7 @@ SIPRegistrarClient* SIPRegistrarClient::instance()
SIPRegistrarClient::SIPRegistrarClient(const string& name)
: AmEventQueue(this),
uac_auth_i(NULL),
AmDynInvokeFactory(MOD_NAME),
stop_requested(false)
AmDynInvokeFactory(MOD_NAME)
{
}
@ -77,7 +76,7 @@ void SIPRegistrarClient::run() {
uac_auth_i = uac_auth_f->getInstance();
}
while (!stop_requested) {
while (!stop_requested()) {
if (registrations.size()) {
unsigned int cnt = 250;
while (cnt > 0) {
@ -144,7 +143,7 @@ void SIPRegistrarClient::onServerShutdown() {
AmEventDispatcher::instance()->delEventQueue(it->first);
}
stop_requested = true;
stop();
//
// setStopped();
// return;

1
apps/registrar_client/SIPRegistrarClient.h
Unescape View File

@ -67,7 +67,6 @@ class SIPRegistrarClient : public AmThread,
AmDynInvoke* uac_auth_i;
atomic_bool stop_requested;
void checkTimeouts();
void onServerShutdown();
public:

8
apps/rtmp/RtmpSender.cpp
Unescape View File

@ -37,8 +37,7 @@
RtmpSender::RtmpSender(RTMP* r)
: has_work(false),
p_rtmp(r),
running(false)
p_rtmp(r)
{
}
@ -73,9 +72,7 @@ int RtmpSender::push_back(const RTMPPacket& p)
void RtmpSender::run()
{
running = true;
while(running){
while (!stop_requested()) {
//wait for some work
// (at most 1s)
@ -112,7 +109,6 @@ void RtmpSender::run()
void RtmpSender::on_stop()
{
running = false;
}
int RtmpSender::SendChangeChunkSize()

3
apps/rtmp/RtmpSender.h
Unescape View File

@ -47,9 +47,6 @@ class RtmpSender
// ptr to RtmpConnection::rtmp
RTMP* p_rtmp;
// execution control
atomic_bool running;
int SendChangeChunkSize();
protected:

9
apps/sbc/RegisterCache.cpp
Unescape View File

@ -225,21 +225,14 @@ void _RegisterCache::gbc(unsigned int bucket_id)
bucket->unlock();
}
void _RegisterCache::on_stop()
{
running = false;
}
void _RegisterCache::run()
{
struct timespec tick,rem;
tick.tv_sec = (REG_CACHE_SINGLE_CYCLE/1000000L);
tick.tv_nsec = (REG_CACHE_SINGLE_CYCLE - (tick.tv_sec)*1000000L) * 1000L;
running = true;
gbc_bucket_id = 0;
while(running) {
while (!stop_requested()) {
gbc(gbc_bucket_id);
gbc_bucket_id = (gbc_bucket_id+1);
gbc_bucket_id &= (REG_CACHE_TABLE_ENTRIES-1);

4
apps/sbc/RegisterCache.h
Unescape View File

@ -200,8 +200,6 @@ class _RegisterCache
unsigned int gbc_bucket_id;
atomic_bool running;
// stats
atomic_int active_regs;
@ -216,7 +214,7 @@ protected:
/* AmThread interface */
void run();
void on_stop();
const char *identify() { return "register cache"; }
/**
* Returns the bucket associated with the passed contact-uri

6
apps/xmlrpc2di/XMLRPC2DI.cpp
Unescape View File

@ -474,7 +474,7 @@ void XMLRPC2DIServer::run() {
s->work(DEF_XMLRPCSERVER_WORK_INTERVAL);
processEvents();
}
while(running.get());
while (!stop_requested());
AmEventDispatcher::instance()->delEventQueue(MOD_NAME);
DBG("Exiting XMLRPC2DIServer.\n");
@ -488,7 +488,8 @@ void XMLRPC2DIServer::process(AmEvent* ev) {
if (sys_ev->sys_event == AmSystemEvent::ServerShutdown) {
DBG("XMLRPC2DIServer received system Event: ServerShutdown, "
"stopping thread\n");
running.set(false);
stop();
join();
}
return;
}
@ -498,7 +499,6 @@ void XMLRPC2DIServer::process(AmEvent* ev) {
void XMLRPC2DIServer::on_stop() {
DBG("on_stop().\n");
running.set(false);
}
void XMLRPC2DIServerCallsMethod::execute(XmlRpcValue& params, XmlRpcValue& result) {

4
core/AmEventQueueProcessor.cpp
Unescape View File

@ -111,8 +111,7 @@ void EventQueueWorker::notify(AmEventQueue* sender)
void EventQueueWorker::run()
{
stop_requested = false;
while(!stop_requested){
while (!stop_requested()) {
runcond.wait_for();
@ -142,7 +141,6 @@ void EventQueueWorker::run()
void EventQueueWorker::on_stop()
{
INFO("requesting worker to stop.\n");
stop_requested = true;
runcond.set(true);
}

2
core/AmEventQueueProcessor.h
Unescape View File

@ -40,8 +40,6 @@ class EventQueueWorker
: public AmThread,
public AmEventNotificationSink
{
atomic_bool stop_requested;
AmCondition runcond;
std::deque<AmEventQueue*> process_queues;
AmMutex process_queues_mut;

7
core/AmMediaProcessor.cpp
Unescape View File

@ -214,7 +214,7 @@ void AmMediaProcessor::dispose()
/* the actual media processing thread */
AmMediaProcessorThread::AmMediaProcessorThread()
: events(this), stop_requested(false)
: events(this)
{
}
AmMediaProcessorThread::~AmMediaProcessorThread()
@ -224,13 +224,10 @@ AmMediaProcessorThread::~AmMediaProcessorThread()
void AmMediaProcessorThread::on_stop()
{
INFO("requesting media processor to stop.\n");
stop_requested = true;
}
void AmMediaProcessorThread::run()
{
stop_requested = false;
// wallclock time
unsigned long long ts = 0;//4294417296;
@ -241,7 +238,7 @@ void AmMediaProcessorThread::run()
uint64_t next_tick = now + tick;
while(!stop_requested){
while (!stop_requested()) {
now = gettimeofday_us();

4
core/AmMediaProcessor.h
Unescape View File

@ -171,8 +171,8 @@ class AmMediaProcessorThread :
// AmThread interface
void run();
void on_stop();
atomic_bool stop_requested;
const char *identify() { return "media processor"; }
// AmEventHandler interface
void process(AmEvent* e);
public:

14
core/AmRtpReceiver.cpp
Unescape View File

@ -50,7 +50,6 @@ _AmRtpReceiver::~_AmRtpReceiver()
}
AmRtpReceiverThread::AmRtpReceiverThread()
: stop_requested(false)
{
// libevent event base
ev_base = event_base_new();
@ -68,20 +67,11 @@ void AmRtpReceiverThread::on_stop()
event_base_loopbreak(ev_base);
}
void AmRtpReceiverThread::stop_and_wait()
{
if(!is_stopped()) {
stop();
while(!is_stopped())
usleep(10000);
}
}
void _AmRtpReceiver::dispose()
{
for(unsigned int i=0; i<n_receivers; i++){
receivers[i].stop_and_wait();
receivers[i].stop();
receivers[i].join();
}
}

4
core/AmRtpReceiver.h
Unescape View File

@ -71,8 +71,6 @@ class AmRtpReceiverThread
Streams streams;
AmMutex streams_mut;
atomic_bool stop_requested;
static void _rtp_receiver_read_cb(evutil_socket_t sd, short what, void* arg);
public:
@ -84,8 +82,6 @@ public:
void addStream(int sd, AmRtpStream* stream);
void removeStream(int sd);
void stop_and_wait();
};
class _AmRtpReceiver

6
core/AmSessionProcessor.cpp
Unescape View File

@ -89,8 +89,7 @@ void AmSessionProcessorThread::notify(AmEventQueue* sender) {
void AmSessionProcessorThread::run() {
stop_requested = false;
while(!stop_requested){
while (!stop_requested()) {
runcond.wait_for();
@ -159,8 +158,7 @@ void AmSessionProcessorThread::run() {
}
void AmSessionProcessorThread::on_stop() {
INFO("requesting session to stop.\n");
stop_requested = true;
runcond.set(true);
}
// AmEventHandler interface

1
core/AmSessionProcessor.h
Unescape View File

@ -68,7 +68,6 @@ class AmSessionProcessorThread
AmEventQueue events;
std::list<AmSession*> sessions;
std::vector<AmSession*> startup_sessions;
atomic_bool stop_requested;
AmCondition runcond;
std::set<AmEventQueue*> process_sessions;

4
core/sip/wheeltimer.cpp
Unescape View File

@ -74,7 +74,7 @@ void _wheeltimer::remove_timer(timer* t, bool del_timer)
void _wheeltimer::run()
{
while(!stopped){
while (!stop_requested()) {
// figure out whether there's anything to run, and for how long to sleep
@ -112,8 +112,6 @@ void _wheeltimer::run()
// all done, remove bucket
buckets.erase(beg);
}
shutdown_finished.set(true);
}
void _wheeltimer::process_current_timers(timer_list& list, std::unique_lock<std::mutex>& lock)

14
core/sip/wheeltimer.h
Unescape View File

@ -124,9 +124,6 @@ class _wheeltimer:
// future (or if no timers exist). Needed not to miss the shutdown flag being set.
const uint64_t max_sleep_time = 500000; // half a second
atomic_bool stopped;
AmCondition shutdown_finished;
void place_timer(timer* t, uint64_t, uint64_t);
void add_timer_to_bucket(timer* t, uint64_t);
@ -139,17 +136,16 @@ class _wheeltimer:
protected:
void run();
void on_stop(){
stopped = true;
shutdown_finished.wait_for();
join();
}
_wheeltimer()
: resolution(20000), // 20 ms == 20000 us
stopped(false), shutdown_finished(false) {}
: resolution(20000) // 20 ms == 20000 us
{}
_wheeltimer(uint64_t _resolution)
: resolution(_resolution),
stopped(false), shutdown_finished(false) {}
: resolution(_resolution)
{}
public:
//clock reference

Write Preview
Loading…
Cancel
Save