From d20c14f2ee4354705957e1dfeac1915371c575b9 Mon Sep 17 00:00:00 2001 From: Donat Zenichev Date: Mon, 7 Jul 2025 18:40:09 +0200 Subject: [PATCH] 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 | 30 +++++++++++++++++++++ core/AmEventDispatcher.h | 3 +++ core/AmSession.cpp | 52 +++++++++++++++++++++++++++++++++++++ core/AmSession.h | 26 +++++++++++++++++++ core/AmSessionContainer.cpp | 28 ++++++++++++++++++++ core/AmSessionContainer.h | 5 ++++ 6 files changed, 144 insertions(+) diff --git a/core/AmEventDispatcher.cpp b/core/AmEventDispatcher.cpp index 916a5889..0ca9a57c 100644 --- a/core/AmEventDispatcher.cpp +++ b/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 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; diff --git a/core/AmEventDispatcher.h b/core/AmEventDispatcher.h index e2c650c4..62a5904e 100644 --- a/core/AmEventDispatcher.h +++ b/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(); diff --git a/core/AmSession.cpp b/core/AmSession.cpp index d61bee51..718ead50 100644 --- a/core/AmSession.cpp +++ b/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 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 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 lock(snapshot_lock); + copy = session_snapshot; } const vector& AmSession::getPayloads() @@ -463,6 +508,9 @@ void AmSession::finalize() } void AmSession::setStopped(bool wakeup) { + /* make snapshot safe */ + lock_guard 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 lock(snapshot_lock); + AmSessionContainer::instance()->destroySession(this); } diff --git a/core/AmSession.h b/core/AmSession.h index 9abfd00e..b47f2353 100644 --- a/core/AmSession.h +++ b/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 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 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; diff --git a/core/AmSessionContainer.cpp b/core/AmSessionContainer.cpp index 1149d054..3b16fddc 100644 --- a/core/AmSessionContainer.cpp +++ b/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(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; } diff --git a/core/AmSessionContainer.h b/core/AmSessionContainer.h index 98094a6a..f83be8c1 100644 --- a/core/AmSessionContainer.h +++ b/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.