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

@@ -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);
+}
+

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

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)
 {

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

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();

core/AmEventQueue.h

@@ -63,6 +63,7 @@ public:
   void postEvent(AmEvent*);
   void processEvents();
   void waitForEvent();
+  void wakeup();
   void processSingleEvent();
 };
 

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;i<num_threads;i++) {
+    if(threads[i] != NULL) {
+      threads[i]->stop();
+    }
+  }
+  bool threads_stopped = true;
+  do {
+    usleep(10000);
+    for (unsigned int i=0;i<num_threads;i++) {
+      if((threads[i] != NULL) &&(!threads[i]->is_stopped())) {
+        threads_stopped = false;
+        break;
+      }
+    }
+  } while(!threads_stopped);
+  
+  for (unsigned int i=0;i<num_threads;i++) {
+    if(threads[i] != NULL) {
+      delete threads[i];
+      threads[i] = NULL;
+    }
+  }
+  delete []  threads;
+  threads = NULL;
+}
+
+void AmMediaProcessor::dispose() 
+{
+  if(_instance != NULL) {
+    if(_instance->threads != 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);
 

core/AmMediaProcessor.h

@@ -62,6 +62,7 @@ class AmMediaProcessorThread :
   // AmThread interface
   void run();
   void on_stop();
+  AmSharedVar<bool> 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();
 };
 
 

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)

core/AmRtpReceiver.h

@@ -49,7 +49,6 @@ class AmRtpReceiver: public AmThread {
   typedef std::map<int, AmRtpStream*, greater<int> > Streams;
 
   static AmRtpReceiver* _instance;
-
     
   Streams  streams;
   AmMutex  streams_mut;
@@ -64,11 +63,15 @@ class AmRtpReceiver: public AmThread {
     
   void run();
   void on_stop();
-    
+  AmSharedVar<bool> stop_requested;
+
 public:
   static AmRtpReceiver* instance();
+  static bool haveInstance();
   void addStream(int sd, AmRtpStream* stream);
   void removeStream(int sd);
+
+  static void dispose();
 };
 
 #endif

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);
   }
 }

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();

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

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<AmSystemEvent*>(ev);
+    if(sys_ev){	
+      DBG("Session received system Event\n");
+      onSystemEvent(sys_ev);
+      return;
+    }
+  }
+
   AmSipEvent* sip_ev = dynamic_cast<AmSipEvent*>(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)
 {

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
   /**

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);
 }

core/AmSessionContainer.h

@@ -47,7 +47,7 @@ using std::string;
  */
 class AmSessionContainer : public AmThread
 {
-  static AmSessionContainer* _SessionContainer;
+  static AmSessionContainer* _instance;
 
   typedef std::queue<AmSession*>      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<bool> _container_closed;
+
   /** the daemon only runs if this is true */
   AmCondition<bool> _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

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<bool>  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);
   }