sipwise
/
sems
mirror of https://github.com/sipwise/sems.git
1
0
Fork 0
Code Issues Releases Wiki Activity

adds lock striping for the event dispatcher.

Browse Source 
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
sayer/1.4-spce2.6
Stefan Sayer 17 years ago
parent 972283ff66
commit 7a0a948fa9
2 changed files with 133 additions and 77 deletions
Show Stats Download Patch File Download Diff File

193
core/AmEventDispatcher.cpp
Unescape View File

@ -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;i<EVENT_DISPATCHER_BUCKETS;i++) {
queues_mut[i].lock();
EvQueueMapIter it = queues[i].begin();
while (it != queues[i].end()) {
EvQueueMapIter this_evq = it;
it++;
queues_mut[i].unlock();
this_evq->second->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;i<EVENT_DISPATCHER_BUCKETS;i++) {
queues_mut[i].lock();
res = res&queues[i].empty();
queues_mut[i].unlock();
if (!res)
break;
}
return res;
}
@ -171,23 +208,35 @@ void AmEventDispatcher::dispose()
_instance = NULL;
}
}
/** this function optimizes posting of SIP Requests
- if the session does not exist, no event need to be created (req copied) */
bool AmEventDispatcher::postSipRequest(const string& callid, const string& remote_tag,
const AmSipRequest& req)
{
// get local tag
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(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;
}

17
core/AmEventDispatcher.h
Unescape View File

@ -31,6 +31,9 @@
#include "AmSipMsg.h"
#include <map>
#define EVENT_DISPATCHER_POWER 10
#define EVENT_DISPATCHER_BUCKETS (1<<EVENT_DISPATCHER_POWER)
class AmEventDispatcher
{
public:
@ -50,18 +53,22 @@ private:
* Container for active sessions
* local tag -> 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();

Write Preview
Loading…
Cancel
Save