From 7a0a948fa9778c0caa9a1d80d74db793168c9ea8 Mon Sep 17 00:00:00 2001 From: Stefan Sayer Date: Thu, 15 Apr 2010 13:22:53 +0000 Subject: [PATCH] adds lock striping for the event dispatcher. this brings some more performance for signaling/CPS/TPS heavy applications, as the single event dispatcher lock can get a bottleneck. ref r30374 git-svn-id: http://svn.berlios.de/svnroot/repos/sems/trunk@1787 8eb893ce-cfd4-0310-b710-fb5ebe64c474 --- core/AmEventDispatcher.cpp | 193 +++++++++++++++++++++++-------------- core/AmEventDispatcher.h | 17 +++- 2 files changed, 133 insertions(+), 77 deletions(-) diff --git a/core/AmEventDispatcher.cpp b/core/AmEventDispatcher.cpp index f4d184ec..ebfaa987 100644 --- a/core/AmEventDispatcher.cpp +++ b/core/AmEventDispatcher.cpp @@ -27,6 +27,23 @@ #include "AmEventDispatcher.h" #include "AmSipEvent.h" +#include "sip/hash.h" + +unsigned int AmEventDispatcher::hash(const string& s1) +{ + return hashlittle(s1.c_str(),s1.length(),0) + & (EVENT_DISPATCHER_BUCKETS-1); +} + +unsigned int AmEventDispatcher::hash(const string& s1, const string s2) +{ + unsigned int h=0; + + h = hashlittle(s1.c_str(),s1.length(),h); + h = hashlittle(s2.c_str(),s2.length(),h); + + return h & (EVENT_DISPATCHER_BUCKETS-1); +} AmEventDispatcher* AmEventDispatcher::_instance=NULL; @@ -41,27 +58,40 @@ bool AmEventDispatcher::addEventQueue(const string& local_tag, const string& callid, const string& remote_tag) { - bool exists = false; + unsigned int queue_bucket = hash(local_tag); - m_queues.lock(); + queues_mut[queue_bucket].lock(); - exists = queues.find(local_tag) != queues.end(); - - if(!callid.empty() && !remote_tag.empty()) { - exists = exists || - (id_lookup.find(callid+remote_tag) != id_lookup.end()); + if (queues[queue_bucket].find(local_tag) != queues[queue_bucket].end()) { + queues_mut[queue_bucket].unlock(); + return false; } - if(!exists){ - queues[local_tag] = q; + unsigned int id_bucket = 0; - if(!callid.empty() && !remote_tag.empty()) - id_lookup[callid+remote_tag] = local_tag; + if(!callid.empty() && !remote_tag.empty()) { + // try to find via id_lookup + id_bucket = hash(callid, remote_tag); + id_lookup_mut[id_bucket].lock(); + + if (id_lookup[id_bucket].find(callid+remote_tag) != + id_lookup[id_bucket].end()) { + id_lookup_mut[id_bucket].unlock(); + queues_mut[queue_bucket].unlock(); + return false; + } } - m_queues.unlock(); + queues[queue_bucket][local_tag] = q; + + if(!callid.empty() && !remote_tag.empty()) { + id_lookup[id_bucket][callid+remote_tag] = local_tag; + id_lookup_mut[id_bucket].unlock(); + } + + queues_mut[queue_bucket].unlock(); - return !exists; + return true; } AmEventQueueInterface* AmEventDispatcher::delEventQueue(const string& local_tag, @@ -70,24 +100,29 @@ AmEventQueueInterface* AmEventDispatcher::delEventQueue(const string& local_tag, { AmEventQueueInterface* q = NULL; - m_queues.lock(); + unsigned int queue_bucket = hash(local_tag); + + queues_mut[queue_bucket].lock(); - EvQueueMapIter qi = queues.find(local_tag); - if(qi != queues.end()) { + EvQueueMapIter qi = queues[queue_bucket].find(local_tag); + if(qi != queues[queue_bucket].end()) { q = qi->second; - queues.erase(qi); + queues[queue_bucket].erase(qi); if(!callid.empty() && !remote_tag.empty()) { + unsigned int id_bucket = hash(callid, remote_tag); + id_lookup_mut[id_bucket].lock(); + + DictIter di = id_lookup[id_bucket].find(callid+remote_tag); + if(di != id_lookup[id_bucket].end()) { + id_lookup[id_bucket].erase(di); + } - DictIter di = id_lookup.find(callid+remote_tag); - if(di != id_lookup.end()) { - - id_lookup.erase(di); - } + id_lookup_mut[id_bucket].unlock(); } } - m_queues.unlock(); + queues_mut[queue_bucket].unlock(); return q; } @@ -95,15 +130,18 @@ AmEventQueueInterface* AmEventDispatcher::delEventQueue(const string& local_tag, bool AmEventDispatcher::post(const string& local_tag, AmEvent* ev) { bool posted = false; - m_queues.lock(); - - EvQueueMapIter it = queues.find(local_tag); - if(it != queues.end()){ + + unsigned int queue_bucket = hash(local_tag); + + queues_mut[queue_bucket].lock(); + + EvQueueMapIter it = queues[queue_bucket].find(local_tag); + if(it != queues[queue_bucket].end()){ it->second->postEvent(ev); posted = true; } - m_queues.unlock(); + queues_mut[queue_bucket].unlock(); return posted; } @@ -111,21 +149,17 @@ bool AmEventDispatcher::post(const string& local_tag, AmEvent* ev) bool AmEventDispatcher::post(const string& callid, const string& remote_tag, AmEvent* ev) { - bool posted = false; - m_queues.lock(); - - DictIter di = id_lookup.find(callid+remote_tag); - if(di != id_lookup.end()) { - - EvQueueMapIter it = queues.find(di->second); - if(it != queues.end()){ - it->second->postEvent(ev); - posted = true; - } + unsigned int id_bucket = hash(callid, remote_tag); + id_lookup_mut[id_bucket].lock(); + DictIter di = id_lookup[id_bucket].find(callid+remote_tag); + if (di == id_lookup[id_bucket].end()) { + id_lookup_mut[id_bucket].unlock(); + return false; } - m_queues.unlock(); - - return posted; + string local_tag = di->second; + id_lookup_mut[id_bucket].unlock(); + + return post(local_tag, ev); } bool AmEventDispatcher::broadcast(AmEvent* ev) @@ -134,32 +168,35 @@ bool AmEventDispatcher::broadcast(AmEvent* ev) return false; bool posted = false; - m_queues.lock(); - - EvQueueMapIter it = queues.begin(); - while (it != queues.end()) { - EvQueueMapIter this_evq = it; - it++; - m_queues.unlock(); - this_evq->second->postEvent(ev->clone()); - m_queues.lock(); - posted = true; + for (size_t i=0;isecond->postEvent(ev->clone()); + queues_mut[i].lock(); + posted = true; + } + queues_mut[i].unlock(); } - m_queues.unlock(); - delete ev; return posted; } bool AmEventDispatcher::empty() { - bool res = false; - - m_queues.lock(); - res = queues.empty(); - m_queues.unlock(); - + bool res = true; + for (size_t i=0;isecond); - if(it != queues.end()){ - it->second->postEvent(new AmSipRequestEvent(req)); - posted = true; - } + unsigned int id_bucket = hash(callid, remote_tag); + id_lookup_mut[id_bucket].lock(); + DictIter di = id_lookup[id_bucket].find(callid+remote_tag); + if (di == id_lookup[id_bucket].end()) { + id_lookup_mut[id_bucket].unlock(); + return false; + } + string local_tag = di->second; + id_lookup_mut[id_bucket].unlock(); + + // post(local_tag) + unsigned int queue_bucket = hash(local_tag); + + queues_mut[queue_bucket].lock(); + + EvQueueMapIter it = queues[queue_bucket].find(local_tag); + if(it != queues[queue_bucket].end()){ + it->second->postEvent(new AmSipRequestEvent(req)); + posted = true; } - m_queues.unlock(); + + queues_mut[queue_bucket].unlock(); return posted; } diff --git a/core/AmEventDispatcher.h b/core/AmEventDispatcher.h index 4c1dace9..5de8a71d 100644 --- a/core/AmEventDispatcher.h +++ b/core/AmEventDispatcher.h @@ -31,6 +31,9 @@ #include "AmSipMsg.h" #include +#define EVENT_DISPATCHER_POWER 10 +#define EVENT_DISPATCHER_BUCKETS (1< event queue */ - EvQueueMap queues; + EvQueueMap queues[EVENT_DISPATCHER_BUCKETS]; + + // mutex for "queues" + AmMutex queues_mut[EVENT_DISPATCHER_BUCKETS]; /** * Call ID + remote tag -> local tag * (needed for CANCELs and some provisionnal answers) * (UAS sessions only) */ - Dictionnary id_lookup; - - // mutex for "queues" and "id_lookup" - AmMutex m_queues; + Dictionnary id_lookup[EVENT_DISPATCHER_BUCKETS]; + // mutex for "id_lookup" + AmMutex id_lookup_mut[EVENT_DISPATCHER_BUCKETS]; + unsigned int hash(const string& s1); + unsigned int hash(const string& s1, const string s2); public: static AmEventDispatcher* instance();