MT#63071 AmSession: add a snapshot mechanism

A a mechanism to get the AmSession object
stored polymorphically on the EvQueueMap.

Follow the classes hierarchy:
AmEventQueueInterface -> AmEventQueue -> AmSession
to retrieve AmSession by down casting it from base class.

When retrieved, acquire the snapshot dedicated lock
to prevent other thread destroying this gotten AmSession.

As soon as AmSessionSnapshot is ready,
release the lock and let other (real owner) thread do anything with it.

Left corner cases:
- by the time the AmEventQueueInterface is retrieved and down-casted
  into the AmSession, returned to the user and the snapshot lock
  is acquired, this thread can simply destroy the AmSession object
- this is an undefined behavior

Change-Id: Ia62b5cb27618e2c077ae5a247bb914be74e17fa9

core/AmEventDispatcher.cpp

@@ -106,6 +106,36 @@ bool AmEventDispatcher::addEventQueue(const string& local_tag,
   return true;
 }
 
+/**
+ * Get AmSession for RO pruposes only.
+ * Can only be used with locks acquired down the cast:
+ * AmEventQueueInterface -> AmEventQueue -> AmSession (here),
+ * otherwise object can be demolished, meanwhile attempt to be used by other thread.
+ */
+AmEventQueueInterface* AmEventDispatcher::getEventQueue(const string& local_tag)
+{
+  if(local_tag.empty()) {
+    ERROR("local_tag is empty");
+    return NULL;
+  }
+
+  AmEventQueueInterface* sess = NULL;
+  unsigned int queue_bucket = hash(local_tag);
+
+  lock_guard<AmMutex> lock(queues_mut[queue_bucket]);
+
+  EvQueueMapIter qi = queues[queue_bucket].find(local_tag);
+
+  if (qi != queues[queue_bucket].end()) {
+    QueueEntry& qe = qi->second;
+
+    if (!qe.id.empty())
+      return qe.q;
+  }
+
+  return NULL;
+}
+
 AmEventQueueInterface* AmEventDispatcher::delEventQueue(const string& local_tag)
 {
   AmEventQueueInterface* q = NULL;

core/AmEventDispatcher.h

@@ -108,6 +108,9 @@ public:
 		       const string& remote_tag,
 		       const string& via_branch);
 
+
+    AmEventQueueInterface* getEventQueue(const string& local_tag);
+
     AmEventQueueInterface* delEventQueue(const string& local_tag);
 
     bool empty();

core/AmSession.cpp

@@ -91,6 +91,36 @@ AmSession::AmSession(AmSipDialog* p_dlg)
   if(!dlg) dlg = new AmSipDialog(this);
   else dlg->setEventhandler(this);
 
+  /* update snapshot */
+  if (!getCallID().empty())
+  {
+    session_snapshot.call_id = getCallID();
+  }
+  if (!getLocalTag().empty())
+  {
+    session_snapshot.local_tag = getLocalTag();
+  }
+  if (!getRemoteTag().empty())
+  {
+    session_snapshot.remote_tag = getRemoteTag();
+  }
+  if (!getFirstBranch().empty())
+  {
+    session_snapshot.first_branch = getFirstBranch();
+  }
+  if (!app_params.empty())
+  {
+    session_snapshot.app_params = app_params;
+  }
+  if (!dlg->getRemoteParty().empty())
+  {
+    session_snapshot.remote_party = dlg->getRemoteParty();
+  }
+  if (!dlg->getLocalParty().empty())
+  {
+    session_snapshot.local_party = dlg->getLocalParty();
+  }
+
   ILOG_DLG(L_DBG, "dlg = %p",dlg);
 }
 
@@ -225,6 +255,9 @@ void AmSession::setLocalTag()
   if (dlg->getLocalTag().empty()) {
     string new_id = getNewId();
     dlg->setLocalTag(new_id);
+    /* update snapshot */
+    lock_guard<AmMutex> lock(snapshot_lock);
+    session_snapshot.local_tag = new_id;
     ILOG_DLG(L_DBG, "AmSession::setLocalTag() - session id set to %s\n", new_id.c_str());
   }
 }
@@ -233,6 +266,18 @@ void AmSession::setLocalTag(const string& tag)
 {
   ILOG_DLG(L_DBG, "AmSession::setLocalTag(%s)\n",tag.c_str());
   dlg->setLocalTag(tag);
+  /* update snapshot */
+  lock_guard<AmMutex> lock(snapshot_lock);
+  session_snapshot.local_tag = tag;
+}
+
+/**
+ * Used by remote unrelated sessions to get a snapshot of this session data primitives.
+ */
+void AmSession::snapshot(AmSessionSnapshot& copy)
+{
+  lock_guard<AmMutex> lock(snapshot_lock);
+  copy = session_snapshot;
 }
 
 const vector<SdpPayload*>& AmSession::getPayloads()
@@ -463,6 +508,9 @@ void AmSession::finalize()
 }
 
 void AmSession::setStopped(bool wakeup) {
+  /* make snapshot safe */
+  lock_guard<AmMutex> lock(snapshot_lock);
+
   if (!sess_stopped.get()) {
     sess_stopped.set(true); 
     onStop();
@@ -484,6 +532,10 @@ string AmSession::getAppParam(const string& param_name) const
 
 void AmSession::destroy() {
   ILOG_DLG(L_DBG, "AmSession::destroy()\n");
+
+  /* make snapshot safe */
+  lock_guard<AmMutex> lock(snapshot_lock);
+
   AmSessionContainer::instance()->destroySession(this);
 }
 

core/AmSession.h

@@ -58,6 +58,21 @@ class AmDtmfEvent;
 /* definition imported from Ser parser/msg_parser.h */
 #define FL_FORCE_ACTIVE 2
 
+/**
+ * A snapshot of the main session's data,
+ * which can be re-used when getting AmSession from other unrelated thread.
+ */
+struct AmSessionSnapshot
+{
+  string local_tag;
+  string remote_tag;
+  string local_party;
+  string remote_party;
+  string call_id;
+  string first_branch;
+  map<string,string> app_params;
+  /* TODO: add more here? */
+};
 
 /**
  * \brief Implements the default behavior of one session
@@ -77,6 +92,13 @@ class AmSession :
 {
   AmMutex      audio_mut;
 
+  /* Used by the snapshot mechanism only.
+   * It is used across multiple threads and is multi-thread visible.
+   *
+   * Purpose: it eliminates a race condition with a destructor,
+   * when AmSession used multi-thread by snapshot */
+  AmMutex snapshot_lock;
+
 protected:
   vector<SdpPayload *>  m_payloads;
   //bool         negotiate_onreply;
@@ -193,6 +215,7 @@ public:
 #endif
 
   AmSipDialog* dlg;
+  AmSessionSnapshot session_snapshot;
 
   /** 
    * \brief Exception occured in a Session
@@ -306,6 +329,9 @@ public:
 
   /* ----         SIP dialog attributes                  ---- */
 
+  /** copies the session's snapshot */
+  void snapshot(AmSessionSnapshot& copy);
+
   /** Gets the Session's call ID */
   const string& getCallID() const;
 

core/AmSessionContainer.cpp

@@ -609,6 +609,34 @@ AmSessionContainer::addSession(const string& local_tag, AmSession* session)
   return AlreadyExist;
 }
 
+/**
+ * Used by AmSession snapshot mechanism only.
+ *
+ * Returns true or false.
+ * Requires: session's tag and reference to existing empty snapshot object.
+ */
+bool AmSessionContainer::getSessionSnapshot(const string& local_tag, AmSessionSnapshot& snap)
+{
+  AmSession * result = NULL;
+
+  if (!_container_closed.get()) {
+
+    result = dynamic_cast<AmSession*>(AmEventDispatcher::instance()->getEventQueue(local_tag));
+    /* TODO: potential unfdefined behavior here, if other thread
+     * manages right now do destroy gotten AmSession,
+     * shortly before the casting is done here, the object is returned
+     * and the `session_snapshot` is acquired by snapshot mechanism.
+     */
+    if (result)
+    {
+      result->snapshot(snap);
+      return true;
+    }
+  }
+
+  return false;
+}
+
 void AmSessionContainer::enableUncleanShutdown() {
   enable_unclean_shutdown = true;
 }

core/AmSessionContainer.h

@@ -121,6 +121,11 @@ class AmSessionContainer : public AmThread
 			   string& app_name,
 			   const AmArg* session_params = NULL);
 
+  /**
+   * Used only by a snapshot mechanism.
+   */
+  bool getSessionSnapshot(const string& local_tag, AmSessionSnapshot& snap);
+
   /**
    * Adds a session to the container (UAS only).
    * @return true if the session is new within the container.