From d9884e05d1f198831c23ef2a18d8f31fa7cf87c6 Mon Sep 17 00:00:00 2001 From: Stefan Sayer Date: Tue, 23 Sep 2008 19:03:46 +0000 Subject: [PATCH] first steps on proper shutdown: - active sessions and other event receivers get a SystemEvent::ServerShutdown, default behaviour of AmSession is setStopped() - session container waits for all sessions to be ended - signaling server, rtp receiver, media processor, event dispatcher are stopped and deleted based on a patch by Rui Jin Zheng rjzheng at boronetworks dot com git-svn-id: http://svn.berlios.de/svnroot/repos/sems/trunk@1087 8eb893ce-cfd4-0310-b710-fb5ebe64c474 --- core/AmEvent.cpp | 9 +++ core/AmEvent.h | 30 ++++++++ core/AmEventDispatcher.cpp | 41 ++++++++++ core/AmEventDispatcher.h | 6 ++ core/AmEventQueue.cpp | 5 ++ core/AmEventQueue.h | 1 + core/AmMediaProcessor.cpp | 52 ++++++++++++- core/AmMediaProcessor.h | 3 + core/AmRtpReceiver.cpp | 27 ++++++- core/AmRtpReceiver.h | 7 +- core/AmRtpStream.cpp | 3 +- core/AmServer.cpp | 16 ++++ core/AmServer.h | 7 +- core/AmSession.cpp | 19 ++++- core/AmSession.h | 4 + core/AmSessionContainer.cpp | 150 +++++++++++++++++++++++------------- core/AmSessionContainer.h | 9 ++- core/sems.cpp | 25 ++++++ 18 files changed, 348 insertions(+), 66 deletions(-) diff --git a/core/AmEvent.cpp b/core/AmEvent.cpp index b57c87ac..1aadc2f2 100644 --- a/core/AmEvent.cpp +++ b/core/AmEvent.cpp @@ -5,7 +5,16 @@ AmEvent::AmEvent(int event_id) { } +AmEvent::AmEvent(const AmEvent& rhs) +: event_id(rhs.event_id), processed(rhs.processed) +{ +} + AmEvent::~AmEvent() { } +AmEvent* AmEvent::clone() { + return new AmEvent(*this); +} + diff --git a/core/AmEvent.h b/core/AmEvent.h index 261ec6a0..dac08d61 100644 --- a/core/AmEvent.h +++ b/core/AmEvent.h @@ -34,6 +34,7 @@ using std::string; #define E_PLUGIN 100 +#define E_SYSTEM 101 /** \brief base event class */ struct AmEvent @@ -42,7 +43,11 @@ struct AmEvent bool processed; AmEvent(int event_id); + AmEvent(const AmEvent& rhs); + virtual ~AmEvent(); + + virtual AmEvent* clone(); }; /** @@ -62,6 +67,25 @@ struct AmPluginEvent: public AmEvent : AmEvent(E_PLUGIN), name(n), data(d) {} }; +/** + * \brief named event for system events (e.g. server stopped) + */ +struct AmSystemEvent : public AmEvent +{ + enum EvType { + ServerShutdown = 0 + }; + + EvType sys_event; + + AmSystemEvent(EvType e) + : AmEvent(E_SYSTEM), sys_event(e) { } + + AmSystemEvent(const AmSystemEvent& rhs) + : AmEvent(rhs), sys_event(rhs.sys_event) { } + + AmEvent* clone() { return new AmSystemEvent(*this); }; +}; /** \brief event handler interface */ class AmEventHandler @@ -71,4 +95,10 @@ class AmEventHandler virtual ~AmEventHandler() { }; }; +/* class AmEventFactory */ +/* { */ +/* virtual AmEvent* generateEvent(const string& receiver_id) = 0; */ +/* virtual ~AmEventFactory() { } */ +/* }; */ + #endif diff --git a/core/AmEventDispatcher.cpp b/core/AmEventDispatcher.cpp index 2838c473..b6897661 100644 --- a/core/AmEventDispatcher.cpp +++ b/core/AmEventDispatcher.cpp @@ -107,6 +107,7 @@ bool AmEventDispatcher::post(const string& local_tag, AmEvent* ev) return posted; } + bool AmEventDispatcher::post(const string& callid, const string& remote_tag, AmEvent* ev) { bool posted = false; @@ -126,6 +127,46 @@ bool AmEventDispatcher::post(const string& callid, const string& remote_tag, AmE return posted; } +bool AmEventDispatcher::broadcast(AmEvent* ev) +{ + if (!ev) + return false; + + bool posted = false; + m_queues.lock(); + + for (EvQueueMapIter it = queues.begin(); + it != queues.end(); it++) { + it->second->postEvent(ev->clone()); + posted = true; + } + + m_queues.unlock(); + + delete ev; + + return posted; +} + +bool AmEventDispatcher::empty() { + bool res = false; + + m_queues.lock(); + res = queues.empty(); + m_queues.unlock(); + + return res; +} + +void AmEventDispatcher::dispose() +{ + if(_instance != NULL) { + // todo: add locking here + delete _instance; + _instance = NULL; + } +} + bool AmEventDispatcher::postSipRequest(const string& callid, const string& remote_tag, const AmSipRequest& req) { diff --git a/core/AmEventDispatcher.h b/core/AmEventDispatcher.h index a750123a..4c1dace9 100644 --- a/core/AmEventDispatcher.h +++ b/core/AmEventDispatcher.h @@ -65,6 +65,7 @@ private: public: static AmEventDispatcher* instance(); + static void dispose(); bool postSipRequest(const string& callid, const string& remote_tag, const AmSipRequest& req); @@ -72,6 +73,9 @@ public: bool post(const string& local_tag, AmEvent* ev); bool post(const string& callid, const string& remote_tag, AmEvent* ev); + /* send event to all event queues. Note: event instances will be cloned */ + bool broadcast(AmEvent* ev); + bool addEventQueue(const string& local_tag, AmEventQueueInterface* q, const string& callid="", @@ -80,6 +84,8 @@ public: AmEventQueueInterface* delEventQueue(const string& local_tag, const string& callid="", const string& remote_tag=""); + + bool empty(); }; #endif diff --git a/core/AmEventQueue.cpp b/core/AmEventQueue.cpp index 919a745b..bdc55cc0 100644 --- a/core/AmEventQueue.cpp +++ b/core/AmEventQueue.cpp @@ -82,6 +82,11 @@ void AmEventQueue::waitForEvent() ev_pending.wait_for(); } +void AmEventQueue::wakeup() +{ + ev_pending.set(true); +} + void AmEventQueue::processSingleEvent() { m_queue.lock(); diff --git a/core/AmEventQueue.h b/core/AmEventQueue.h index ac632b67..27344475 100644 --- a/core/AmEventQueue.h +++ b/core/AmEventQueue.h @@ -63,6 +63,7 @@ public: void postEvent(AmEvent*); void processEvents(); void waitForEvent(); + void wakeup(); void processSingleEvent(); }; diff --git a/core/AmMediaProcessor.cpp b/core/AmMediaProcessor.cpp index 7d2110cc..713d6062 100644 --- a/core/AmMediaProcessor.cpp +++ b/core/AmMediaProcessor.cpp @@ -50,14 +50,16 @@ struct SchedRequest : /* session scheduler */ -AmMediaProcessor* AmMediaProcessor::_instance; +AmMediaProcessor* AmMediaProcessor::_instance = NULL; AmMediaProcessor::AmMediaProcessor() + : threads(NULL),num_threads(0) { } AmMediaProcessor::~AmMediaProcessor() { + INFO("Media processor has been recycled.\n"); } void AmMediaProcessor::init() { @@ -167,10 +169,49 @@ void AmMediaProcessor::removeFromProcessor(AmSession* s, threads[sched_thread]->postRequest(new SchedRequest(r_type,s)); } -/* the actual session scheduler thread */ +void AmMediaProcessor::stop() { + assert(threads); + for (unsigned int i=0;istop(); + } + } + bool threads_stopped = true; + do { + usleep(10000); + for (unsigned int i=0;iis_stopped())) { + threads_stopped = false; + break; + } + } + } while(!threads_stopped); + + for (unsigned int i=0;ithreads != NULL) { + _instance->stop(); + } + delete _instance; + _instance = NULL; + } +} + +/* the actual media processing thread */ AmMediaProcessorThread::AmMediaProcessorThread() - : events(this) + : events(this), stop_requested(false) { } AmMediaProcessorThread::~AmMediaProcessorThread() @@ -179,10 +220,13 @@ AmMediaProcessorThread::~AmMediaProcessorThread() void AmMediaProcessorThread::on_stop() { + INFO("requesting media processor to stop.\n"); + stop_requested.set(true); } void AmMediaProcessorThread::run() { + stop_requested = false; struct timeval now,next_tick,diff,tick; // wallclock time unsigned int ts = 0; @@ -193,7 +237,7 @@ void AmMediaProcessorThread::run() gettimeofday(&now,NULL); timeradd(&tick,&now,&next_tick); - while(true){ + while(!stop_requested.get()){ gettimeofday(&now,NULL); diff --git a/core/AmMediaProcessor.h b/core/AmMediaProcessor.h index e32c1f19..64c13787 100644 --- a/core/AmMediaProcessor.h +++ b/core/AmMediaProcessor.h @@ -62,6 +62,7 @@ class AmMediaProcessorThread : // AmThread interface void run(); void on_stop(); + AmSharedVar stop_requested; // AmEventHandler interface void process(AmEvent* e); @@ -120,6 +121,8 @@ public: void changeCallgroup(AmSession* s, const string& new_callgroup); + void stop(); + static void dispose(); }; diff --git a/core/AmRtpReceiver.cpp b/core/AmRtpReceiver.cpp index e3f03cd4..a21d3a79 100644 --- a/core/AmRtpReceiver.cpp +++ b/core/AmRtpReceiver.cpp @@ -57,7 +57,12 @@ AmRtpReceiver* AmRtpReceiver::instance() return _instance; } +bool AmRtpReceiver::haveInstance() { + return NULL != _instance; +} + AmRtpReceiver::AmRtpReceiver() + : stop_requested(false) { fds = new struct pollfd[MAX_RTP_SESSIONS]; nfds = 0; @@ -66,10 +71,28 @@ AmRtpReceiver::AmRtpReceiver() AmRtpReceiver::~AmRtpReceiver() { delete [] (fds); + INFO("RTP receiver has been recycled.\n"); } void AmRtpReceiver::on_stop() { + INFO("requesting RTP receiver to stop.\n"); + stop_requested.set(true); +} + +void AmRtpReceiver::dispose() +{ + if(_instance != NULL) { + if(!_instance->is_stopped()) { + _instance->stop(); + + while(!_instance->is_stopped()) + usleep(10000); + } + // todo: add locking here + delete _instance; + _instance = NULL; + } } void AmRtpReceiver::run() @@ -77,7 +100,7 @@ void AmRtpReceiver::run() unsigned int tmp_nfds = 0; struct pollfd* tmp_fds = new struct pollfd[MAX_RTP_SESSIONS]; - while(true){ + while(!stop_requested.get()){ fds_mut.lock(); tmp_nfds = nfds; @@ -126,6 +149,8 @@ void AmRtpReceiver::run() streams_mut.unlock(); } } + + delete[] (tmp_fds); } void AmRtpReceiver::addStream(int sd, AmRtpStream* stream) diff --git a/core/AmRtpReceiver.h b/core/AmRtpReceiver.h index ed4ff2fa..03b375e9 100644 --- a/core/AmRtpReceiver.h +++ b/core/AmRtpReceiver.h @@ -49,7 +49,6 @@ class AmRtpReceiver: public AmThread { typedef std::map > Streams; static AmRtpReceiver* _instance; - Streams streams; AmMutex streams_mut; @@ -64,11 +63,15 @@ class AmRtpReceiver: public AmThread { void run(); void on_stop(); - + AmSharedVar stop_requested; + public: static AmRtpReceiver* instance(); + static bool haveInstance(); void addStream(int sd, AmRtpStream* stream); void removeStream(int sd); + + static void dispose(); }; #endif diff --git a/core/AmRtpStream.cpp b/core/AmRtpStream.cpp index 6c4c4ac8..c8188f7f 100644 --- a/core/AmRtpStream.cpp +++ b/core/AmRtpStream.cpp @@ -379,7 +379,8 @@ AmRtpStream::AmRtpStream(AmSession* _s) AmRtpStream::~AmRtpStream() { if(l_sd){ - AmRtpReceiver::instance()->removeStream(l_sd); + if (AmRtpReceiver::haveInstance()) + AmRtpReceiver::instance()->removeStream(l_sd); close(l_sd); } } diff --git a/core/AmServer.cpp b/core/AmServer.cpp index 05244ee8..167e749a 100644 --- a/core/AmServer.cpp +++ b/core/AmServer.cpp @@ -42,6 +42,22 @@ AmServer* AmServer::instance() return _instance ? _instance : ((_instance = new AmServer())); } +void AmServer::dispose() +{ + if(_instance != NULL) { + if(_instance->ctrlIface != NULL) { + _instance->ctrlIface->join(); + } + delete _instance; + _instance = NULL; + } +} + +AmServer::~AmServer() +{ + INFO("Signaling Server has been recycled.\n"); +} + void AmServer::run() { ctrlIface->start(); diff --git a/core/AmServer.h b/core/AmServer.h index 52dd97ad..af558db3 100644 --- a/core/AmServer.h +++ b/core/AmServer.h @@ -50,11 +50,12 @@ private: static AmCtrlInterface *ctrlIface; /** Avoid external instantiation. @see instance(). */ - ~AmServer(){} + ~AmServer(); public: /** Get a fifo server instance. */ static AmServer* instance(); + static void dispose(); /** Runs the fifo server. */ void run(); @@ -69,8 +70,8 @@ public: static bool sendRequest(const AmSipRequest &, char* serKey, unsigned int& serKeyLen); static bool sendReply(const AmSipReply &); static string getContact(const string &displayName, - const string &userName, const string &hostName, - const string &uriParams, const string &hdrParams); + const string &userName, const string &hostName, + const string &uriParams, const string &hdrParams); }; #endif diff --git a/core/AmSession.cpp b/core/AmSession.cpp index aef9e02c..ac312a99 100644 --- a/core/AmSession.cpp +++ b/core/AmSession.cpp @@ -371,8 +371,9 @@ void AmSession::run() session_num--; // wait at least until session is out of RtpScheduler - DBG("session is stopped.\n"); //detached.wait_for(); + + DBG("session is stopped.\n"); } void AmSession::on_stop() @@ -484,6 +485,15 @@ void AmSession::process(AmEvent* ev) DBG("AmSession::process\n"); + if (ev->event_id == E_SYSTEM) { + AmSystemEvent* sys_ev = dynamic_cast(ev); + if(sys_ev){ + DBG("Session received system Event\n"); + onSystemEvent(sys_ev); + return; + } + } + AmSipEvent* sip_ev = dynamic_cast(ev); if(sip_ev){ DBG("Session received SIP Event\n"); @@ -752,6 +762,13 @@ int AmSession::acceptAudio(const string& body, return -1; } +void AmSession::onSystemEvent(AmSystemEvent* ev) { + if (ev->sys_event == AmSystemEvent::ServerShutdown) { + setStopped(); + return; + } +} + void AmSession::onSendRequest(const string& method, const string& content_type, const string& body, string& hdrs, int flags, unsigned int cseq) { diff --git a/core/AmSession.h b/core/AmSession.h index 927ee375..0b8de525 100644 --- a/core/AmSession.h +++ b/core/AmSession.h @@ -419,6 +419,10 @@ public: virtual void onSipReply(const AmSipReply& reply); + /** + * entry point for system events + */ + virtual void onSystemEvent(AmSystemEvent* ev); #ifdef WITH_ZRTP /** diff --git a/core/AmSessionContainer.cpp b/core/AmSessionContainer.cpp index e45198a5..5e7db9c8 100644 --- a/core/AmSessionContainer.cpp +++ b/core/AmSessionContainer.cpp @@ -41,80 +41,118 @@ // AmSessionContainer methods -AmSessionContainer* AmSessionContainer::_SessionContainer=0; +AmSessionContainer* AmSessionContainer::_instance=NULL; AmSessionContainer::AmSessionContainer() - : _run_cond(false) + : _run_cond(false), _container_closed(false) { } AmSessionContainer* AmSessionContainer::instance() { - if(!_SessionContainer) - _SessionContainer = new AmSessionContainer(); + if(!_instance) + _instance = new AmSessionContainer(); - return _SessionContainer; + return _instance; } -void AmSessionContainer::on_stop() -{ +void AmSessionContainer::dispose() +{ + if(_instance != NULL) { + if(!_instance->is_stopped()) { + _instance->stop(); + + while(!_instance->is_stopped()) + usleep(10000); + } + // todo: add locking here + delete _instance; + _instance = NULL; + } +} + +bool AmSessionContainer::clean_sessions() { + ds_mut.lock(); + DBG("Session cleaner starting its work\n"); + + try { + SessionQueue n_sessions; + + while(!d_sessions.empty()){ + + AmSession* cur_session = d_sessions.front(); + d_sessions.pop(); + + ds_mut.unlock(); + + if(cur_session->is_stopped() && cur_session->detached.get()){ + + DBG("session %p has been destroyed'\n",(void*)cur_session->_pid); + delete cur_session; + } + else { + DBG("session %p still running\n",(void*)cur_session->_pid); + n_sessions.push(cur_session); + } + + ds_mut.lock(); + } + + swap(d_sessions,n_sessions); + + }catch(std::exception& e){ + ERROR("exception caught in session cleaner: %s\n", e.what()); + throw; /* throw again as this is fatal (because unlocking the mutex fails!! */ + }catch(...){ + ERROR("unknown exception caught in session cleaner!\n"); + throw; /* throw again as this is fatal (because unlocking the mutex fails!! */ + } + bool more = !d_sessions.empty(); + ds_mut.unlock(); + return more; } void AmSessionContainer::run() { - while(1){ + while(!_container_closed.get()){ _run_cond.wait_for(); - // Let some time for the Sessions - // to stop by themselves - sleep(5); - - ds_mut.lock(); - DBG("Session cleaner starting its work\n"); - - try { - SessionQueue n_sessions; + if(_container_closed.get()) + break; - while(!d_sessions.empty()){ - - AmSession* cur_session = d_sessions.front(); - d_sessions.pop(); + // Give the Sessions some time to stop by themselves + sleep(5); - ds_mut.unlock(); + bool more = clean_sessions(); - if(cur_session->is_stopped() && cur_session->detached.get()){ - - DBG("session %p has been destroyed'\n",(void*)cur_session->_pid); - delete cur_session; - } - else { - DBG("session %p still running\n",(void*)cur_session->_pid); - n_sessions.push(cur_session); - } - - ds_mut.lock(); - } + DBG("Session cleaner finished\n"); + if(!more && (!_container_closed.get())) + _run_cond.set(false); + } + DBG("Session cleaner terminating\n"); +} - swap(d_sessions,n_sessions); +void AmSessionContainer::on_stop() +{ + _container_closed.set(true); - }catch(std::exception& e){ - ERROR("exception caught in session cleaner: %s\n", e.what()); - throw; /* throw again as this is fatal (because unlocking the mutex fails!! */ - }catch(...){ - ERROR("unknown exception caught in session cleaner!\n"); - throw; /* throw again as this is fatal (because unlocking the mutex fails!! */ - } + DBG("brodcasting ServerShutdown system event to sessions...\n"); + AmEventDispatcher::instance()-> + broadcast(new AmSystemEvent(AmSystemEvent::ServerShutdown)); + + DBG("waiting for active event queues to stop...\n"); - bool more = !d_sessions.empty(); - ds_mut.unlock(); + while (!AmEventDispatcher::instance()->empty()) + sleep(1); + + DBG("cleaning sessions...\n"); + while (clean_sessions()) + sleep(1); - DBG("Session cleaner finished\n"); - if(!more) - _run_cond.set(false); - } + _run_cond.set(true); // so that thread stops } void AmSessionContainer::stopAndQueue(AmSession* s) @@ -314,14 +352,20 @@ bool AmSessionContainer::addSession(const string& callid, const string& local_tag, AmSession* session) { - return AmEventDispatcher::instance()-> - addEventQueue(local_tag,(AmEventQueue*)session, - callid,remote_tag); + if(_container_closed.get()) + return false; + + return AmEventDispatcher::instance()-> + addEventQueue(local_tag,(AmEventQueue*)session, + callid,remote_tag); } bool AmSessionContainer::addSession(const string& local_tag, AmSession* session) { - return AmEventDispatcher::instance()-> - addEventQueue(local_tag,(AmEventQueue*)session); + if(_container_closed.get()) + return false; + + return AmEventDispatcher::instance()-> + addEventQueue(local_tag,(AmEventQueue*)session); } diff --git a/core/AmSessionContainer.h b/core/AmSessionContainer.h index a2e9081b..e63d13d5 100644 --- a/core/AmSessionContainer.h +++ b/core/AmSessionContainer.h @@ -47,7 +47,7 @@ using std::string; */ class AmSessionContainer : public AmThread { - static AmSessionContainer* _SessionContainer; + static AmSessionContainer* _instance; typedef std::queue SessionQueue; @@ -56,6 +56,9 @@ class AmSessionContainer : public AmThread /** Mutex to protect the dead session container */ AmMutex ds_mut; + /** is container closed for new sessions? */ + AmCondition _container_closed; + /** the daemon only runs if this is true */ AmCondition _run_cond; @@ -72,9 +75,13 @@ class AmSessionContainer : public AmThread /** @see AmThread::on_stop() */ void on_stop(); + bool clean_sessions(); + public: static AmSessionContainer* instance(); + static void dispose(); + /** * Creates a new session. * @param req local request diff --git a/core/sems.cpp b/core/sems.cpp index 2f6f49b4..da3e46d6 100644 --- a/core/sems.cpp +++ b/core/sems.cpp @@ -34,6 +34,7 @@ #include "AmMediaProcessor.h" #include "AmIcmpWatcher.h" #include "AmRtpReceiver.h" +#include "AmEventDispatcher.h" #include "AmZRTP.h" @@ -92,7 +93,31 @@ static void sig_usr_un(int signo) if (!main_pid || (main_pid == getpid())) { unlink(pid_file.c_str()); + + static AmMutex clean_up_mut; + static AmCondition need_clean(true); + + clean_up_mut.lock(); + if(need_clean.get()) { + + need_clean.set(false); + + AmRtpReceiver::dispose(); + + AmSessionContainer::dispose(); + + AmServer::dispose(); + + AmMediaProcessor::dispose(); + + AmEventDispatcher::dispose(); + + } + + clean_up_mut.unlock(); + INFO("Finished.\n"); + exit(0); }