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

db_reg_agent: added retry on error, delete removed registrations

Browse Source
sayer/1.4-spce2.6
Stefan Sayer 15 years ago committed by Stefan Sayer
parent 6b45d89a3e
commit 8533307c6c
4 changed files with 188 additions and 56 deletions
Show Stats Download Patch File Download Diff File

202
apps/db_reg_agent/DBRegAgent.cpp
Unescape View File

@ -33,8 +33,8 @@
EXPORT_MODULE_FACTORY(DBRegAgent);
DEFINE_MODULE_INSTANCE(DBRegAgent, MOD_NAME);
mysqlpp::Connection DBRegAgent::DBConnection(mysqlpp::use_exceptions);
mysqlpp::Connection DBRegAgent::StatusDBConnection(mysqlpp::use_exceptions);
mysqlpp::Connection DBRegAgent::MainDBConnection(mysqlpp::use_exceptions);
mysqlpp::Connection DBRegAgent::ProcessorDBConnection(mysqlpp::use_exceptions);
string DBRegAgent::joined_query;
string DBRegAgent::registrations_table = "registrations";
@ -46,7 +46,10 @@ bool DBRegAgent::enable_ratelimiting = false;
unsigned int DBRegAgent::ratelimit_rate = 0;
unsigned int DBRegAgent::ratelimit_per = 0;
static void _timer_cb(RegTimer* timer, long subscriber_id, void* data2) {
bool DBRegAgent::delete_removed_registrations = true;
unsigned int DBRegAgent::error_retry_interval = 300;
static void _timer_cb(RegTimer* timer, long subscriber_id, int data2) {
DBRegAgent::instance()->timer_cb(timer, subscriber_id, data2);
}
@ -121,6 +124,14 @@ int DBRegAgent::onLoad()
}
}
delete_removed_registrations =
cfg.getParameter("delete_removed_registrations", "yes") == "yes";
error_retry_interval = cfg.getParameterInt("error_retry_interval", 300);
if (!error_retry_interval) {
WARN("disabled retry on errors!\n");
}
string mysql_server, mysql_user, mysql_passwd, mysql_db;
mysql_server = cfg.getParameter("mysql_server", "localhost");
@ -141,19 +152,23 @@ int DBRegAgent::onLoad()
try {
DBConnection.set_option(new mysqlpp::ReconnectOption(true));
DBConnection.connect(mysql_db.c_str(), mysql_server.c_str(),
MainDBConnection.set_option(new mysqlpp::ReconnectOption(true));
// matched instead of changed rows in result, so we know when to create DB entry
MainDBConnection.set_option(new mysqlpp::FoundRowsOption(true));
MainDBConnection.connect(mysql_db.c_str(), mysql_server.c_str(),
mysql_user.c_str(), mysql_passwd.c_str());
if (!DBConnection) {
ERROR("Database connection failed: %s\n", DBConnection.error());
if (!MainDBConnection) {
ERROR("Database connection failed: %s\n", MainDBConnection.error());
return -1;
}
StatusDBConnection.set_option(new mysqlpp::ReconnectOption(true));
StatusDBConnection.connect(mysql_db.c_str(), mysql_server.c_str(),
ProcessorDBConnection.set_option(new mysqlpp::ReconnectOption(true));
// matched instead of changed rows in result, so we know when to create DB entry
ProcessorDBConnection.set_option(new mysqlpp::FoundRowsOption(true));
ProcessorDBConnection.connect(mysql_db.c_str(), mysql_server.c_str(),
mysql_user.c_str(), mysql_passwd.c_str());
if (!StatusDBConnection) {
ERROR("Database connection failed: %s\n", StatusDBConnection.error());
if (!ProcessorDBConnection) {
ERROR("Database connection failed: %s\n", ProcessorDBConnection.error());
return -1;
}
@ -189,7 +204,7 @@ int DBRegAgent::onLoad()
}
DBG("starting registration timer thread...\n");
registration_timer.start();
registration_scheduler.start();
// run_tests();
@ -202,7 +217,7 @@ bool DBRegAgent::loadRegistrations() {
try {
time_t now_time = time(NULL);
mysqlpp::Query query = DBRegAgent::DBConnection.query();
mysqlpp::Query query = DBRegAgent::MainDBConnection.query();
string query_string, table;
@ -226,7 +241,7 @@ bool DBRegAgent::loadRegistrations() {
else {
DBG("registration status entry for id %ld does not exist, creating...\n",
subscriber_id);
createDBRegistration(subscriber_id, StatusDBConnection);
createDBRegistration(subscriber_id, ProcessorDBConnection);
}
DBG("got subscriber '%s@%s' status %i\n",
@ -478,6 +493,7 @@ void DBRegAgent::process(AmEvent* ev) {
ERROR("unknown event received!\n");
}
// uses ProcessorDBConnection
void DBRegAgent::onRegistrationActionEvent(RegistrationActionEvent* reg_action_ev) {
switch (reg_action_ev->action) {
case RegistrationActionEvent::Register:
@ -491,7 +507,8 @@ void DBRegAgent::onRegistrationActionEvent(RegistrationActionEvent* reg_action_e
reg_action_ev->subscriber_id);
} else {
if (!it->second->doRegistration()) {
updateDBRegistration(reg_action_ev->subscriber_id,
updateDBRegistration(ProcessorDBConnection,
reg_action_ev->subscriber_id,
480, "unable to send request",
true, REG_STATUS_FAILED);
}
@ -509,7 +526,8 @@ void DBRegAgent::onRegistrationActionEvent(RegistrationActionEvent* reg_action_e
reg_action_ev->subscriber_id);
} else {
if (!it->second->doUnregister()) {
updateDBRegistration(reg_action_ev->subscriber_id,
updateDBRegistration(ProcessorDBConnection,
reg_action_ev->subscriber_id,
480, "unable to send request",
true, REG_STATUS_FAILED);
}
@ -540,7 +558,28 @@ void DBRegAgent::createDBRegistration(long subscriber_id, mysqlpp::Connection& c
}
}
void DBRegAgent::updateDBRegistration(long subscriber_id, int last_code,
void DBRegAgent::deleteDBRegistration(long subscriber_id, mysqlpp::Connection& conn) {
string insert_query = "delete from "+registrations_table+
" where subscriber_id=" + long2str(subscriber_id)+";";
try {
mysqlpp::Query query = conn.query();
query << insert_query;
mysqlpp::SimpleResult res = query.execute();
if (!res) {
WARN("removing registration in DB with query '%s' failed: '%s'\n",
insert_query.c_str(), res.info());
}
} catch (const mysqlpp::Exception& er) {
// Catch-all for any MySQL++ exceptions
ERROR("MySQL++ error: %s\n", er.what());
return;
}
}
void DBRegAgent::updateDBRegistration(mysqlpp::Connection& db_connection,
long subscriber_id, int last_code,
const string& last_reason,
bool update_status, int status,
bool update_ts, unsigned int expiry) {
@ -563,7 +602,7 @@ void DBRegAgent::updateDBRegistration(long subscriber_id, int last_code,
DBG("updating registration in DB with query '%s'\n",
update_query.c_str());
mysqlpp::Query query = DBRegAgent::DBConnection.query();
mysqlpp::Query query = db_connection.query();
query << update_query;
mysqlpp::SimpleResult res = query.execute();
@ -574,7 +613,7 @@ void DBRegAgent::updateDBRegistration(long subscriber_id, int last_code,
if (!res.rows()) {
// should not happen - DB entry is created on load or on createRegistration
DBG("creating registration DB entry for subscriber %ld\n", subscriber_id);
createDBRegistration(subscriber_id, DBConnection);
createDBRegistration(subscriber_id, db_connection);
query << update_query;
mysqlpp::SimpleResult res = query.execute();
@ -593,6 +632,7 @@ void DBRegAgent::updateDBRegistration(long subscriber_id, int last_code,
}
// uses MainDBConnection
void DBRegAgent::onSipReplyEvent(AmSipReplyEvent* ev) {
if (!ev) return;
@ -636,6 +676,7 @@ void DBRegAgent::onSipReplyEvent(AmSipReplyEvent* ev) {
int status = 0;
bool update_ts = false;
unsigned int expiry = 0;
bool delete_status = false;
if (ev->reply.code >= 300) {
if (current_state == AmSIPRegistration::RegisterPending) {
@ -645,7 +686,17 @@ void DBRegAgent::onSipReplyEvent(AmSipReplyEvent* ev) {
DBG("registration failed - mark in DB\n");
update_status = true;
status = REG_STATUS_FAILED;
// todo: schedule for retry?
if (error_retry_interval) {
if (registration->getUnregistering()) {
// schedule deregister after error_retry_interval
setRegistrationTimer(subscriber_id, error_retry_interval,
RegistrationActionEvent::Deregister);
} else {
// schedule register-refresh after error_retry_interval
setRegistrationTimer(subscriber_id, error_retry_interval,
RegistrationActionEvent::Register);
}
}
}
} else if (ev->reply.code >= 200) {
// positive reply
@ -658,14 +709,24 @@ void DBRegAgent::onSipReplyEvent(AmSipReplyEvent* ev) {
update_ts = true;
expiry = registration->getExpiresLeft();
} else {
update_status = true;
status = REG_STATUS_REMOVED;
if (delete_removed_registrations) {
delete_status = true;
} else {
update_status = true;
status = REG_STATUS_REMOVED;
}
}
}
DBG("update DB with reply %u %s\n", ev->reply.code, ev->reply.reason.c_str());
updateDBRegistration(subscriber_id, ev->reply.code, ev->reply.reason,
update_status, status, update_ts, expiry);
if (!delete_status) {
DBG("update DB with reply %u %s\n", ev->reply.code, ev->reply.reason.c_str());
updateDBRegistration(MainDBConnection,
subscriber_id, ev->reply.code, ev->reply.reason,
update_status, status, update_ts, expiry);
} else {
DBG("delete DB registration of subscriber %ld\n", subscriber_id);
deleteDBRegistration(subscriber_id, MainDBConnection);
}
} else {
ERROR("internal: inconsistent registration list\n");
@ -682,6 +743,8 @@ void DBRegAgent::run() {
DBG("DBRegAgent thread: waiting 2 sec for server startup ...\n");
sleep(2);
mysqlpp::Connection::thread_start();
if (enable_ratelimiting) {
DBG("starting processor thread\n");
registration_processor.start();
@ -705,6 +768,8 @@ void DBRegAgent::run() {
DBG("removing "MOD_NAME" registrations from Event Dispatcher...\n");
AmEventDispatcher::instance()->delEventQueue(MOD_NAME);
mysqlpp::Connection::thread_end();
DBG("DBRegAgent thread stopped.\n");
}
@ -713,6 +778,35 @@ void DBRegAgent::on_stop() {
running = false;
}
void DBRegAgent::setRegistrationTimer(long subscriber_id, unsigned int timeout,
RegistrationActionEvent::RegAction reg_action) {
DBG("setting Register timer for subscription %ld, timeout %u, reg_action %u\n",
subscriber_id, timeout, reg_action);
RegTimer* timer = NULL;
map<long, RegTimer*>::iterator it=registration_timers.find(subscriber_id);
if (it==registration_timers.end()) {
DBG("timer object for subscription %ld not found\n", subscriber_id);
timer = new RegTimer();
timer->data1 = subscriber_id;
timer->cb = _timer_cb;
DBG("created timer object [%p] for subscription %ld\n", timer, subscriber_id);
} else {
timer = it->second;
DBG("removing scheduled timer...\n");
registration_scheduler.remove_timer(timer);
}
timer->data2 = reg_action;
timer->expires = time(0) + timeout;
DBG("placing timer for %ld in T-%u\n", subscriber_id, timeout);
registration_scheduler.insert_timer(timer);
registration_timers.insert(std::make_pair(subscriber_id, timer));
}
void DBRegAgent::setRegistrationTimer(long subscriber_id,
time_t expiry, time_t reg_start_ts) {
DBG("setting re-Register timer for subscription %ld, expiry %ld, reg_start_t %ld\n",
@ -726,13 +820,14 @@ void DBRegAgent::setRegistrationTimer(long subscriber_id,
timer->data1 = subscriber_id;
timer->cb = _timer_cb;
DBG("created timer object [%p] for subscription %ld\n", timer, subscriber_id);
registration_timers.insert(std::make_pair(subscriber_id, timer));
} else {
timer = it->second;
DBG("removing timer...\n");
registration_timer.remove_timer(timer);
DBG("removing scheduled timer...\n");
registration_scheduler.remove_timer(timer);
}
registration_timers.insert(std::make_pair(subscriber_id, timer));
timer->data2 = RegistrationActionEvent::Register;
if (minimum_reregister_interval>0.0) {
time_t t_expiry_max = reg_start_ts;
@ -748,7 +843,7 @@ void DBRegAgent::setRegistrationTimer(long subscriber_id,
t_expiry_min, t_expiry_max, reg_start_ts, expiry,
reregister_interval, minimum_reregister_interval);
registration_timer.insert_timer_leastloaded(timer, t_expiry_min, t_expiry_max);
registration_scheduler.insert_timer_leastloaded(timer, t_expiry_min, t_expiry_max);
} else {
time_t t_expiry = reg_start_ts;
@ -759,7 +854,7 @@ void DBRegAgent::setRegistrationTimer(long subscriber_id,
t_expiry, reg_start_ts, expiry, reregister_interval);
timer->expires = t_expiry;
registration_timer.insert_timer(timer);
registration_scheduler.insert_timer(timer);
}
}
@ -771,7 +866,8 @@ void DBRegAgent::clearRegistrationTimer(long subscriber_id) {
DBG("timer object for subscription %ld not found\n", subscriber_id);
return;
}
registration_timer.remove_timer(it->second);
DBG("removing timer [%p] from scheduler\n", it->second);
registration_scheduler.remove_timer(it->second);
DBG("deleting timer object [%p]\n", it->second);
delete it->second;
@ -785,7 +881,7 @@ void DBRegAgent::removeRegistrationTimer(long subscriber_id) {
map<long, RegTimer*>::iterator it=registration_timers.find(subscriber_id);
if (it==registration_timers.end()) {
DBG("timer object for subscription %ld not found\n", subscriber_id);
return;
return;
}
DBG("deleting timer object [%p]\n", it->second);
@ -794,14 +890,21 @@ void DBRegAgent::removeRegistrationTimer(long subscriber_id) {
registration_timers.erase(it);
}
void DBRegAgent::timer_cb(RegTimer* timer, long subscriber_id, void* data2) {
DBG("re-registration timer expired: subscriber %ld, timer=[%p]\n", subscriber_id, timer);
scheduleRegistration(subscriber_id);
void DBRegAgent::timer_cb(RegTimer* timer, long subscriber_id, int reg_action) {
DBG("re-registration timer expired: subscriber %ld, timer=[%p], action %d\n",
subscriber_id, timer, reg_action);
registrations_mut.lock();
removeRegistrationTimer(subscriber_id);
registrations_mut.unlock();
switch (reg_action) {
case RegistrationActionEvent::Register:
scheduleRegistration(subscriber_id); break;
case RegistrationActionEvent::Deregister:
scheduleDeregistration(subscriber_id); break;
default: ERROR("internal: unknown reg_action %d for subscriber %ld timer event\n",
reg_action, subscriber_id);
};
}
@ -813,7 +916,6 @@ void DBRegAgent::DIcreateRegistration(int subscriber_id, const string& user,
pass.c_str(), realm.c_str());
createRegistration(subscriber_id, user, pass, realm);
createDBRegistration(subscriber_id, StatusDBConnection);
scheduleRegistration(subscriber_id);
ret.push(200);
ret.push("OK");
@ -834,6 +936,11 @@ void DBRegAgent::DIremoveRegistration(int subscriber_id, AmArg& ret) {
DBG("DI method: removeRegistration(%i)\n",
subscriber_id);
scheduleDeregistration(subscriber_id);
registrations_mut.lock();
clearRegistrationTimer(subscriber_id);
registrations_mut.unlock();
ret.push(200);
ret.push("OK");
}
@ -913,6 +1020,8 @@ void DBRegAgentProcessorThread::run() {
// register us as SIP event receiver for MOD_NAME_processor
AmEventDispatcher::instance()->addEventQueue(MOD_NAME "_processor",this);
mysqlpp::Connection::thread_start();
// initialize ratelimit
gettimeofday(&last_check, NULL);
allowance = DBRegAgent::ratelimit_rate;
@ -925,6 +1034,9 @@ void DBRegAgentProcessorThread::run() {
processSingleEvent();
}
}
mysqlpp::Connection::thread_end();
DBG("DBRegAgentProcessorThread thread stopped\n");
}
@ -969,32 +1081,32 @@ void DBRegAgent::run_tests() {
RegTimer rt;
rt.expires = now.tv_sec + 10;
rt.cb=test_cb;
registration_timer.insert_timer(&rt);
registration_scheduler.insert_timer(&rt);
RegTimer rt2;
rt2.expires = now.tv_sec + 5;
rt2.cb=test_cb;
registration_timer.insert_timer(&rt2);
registration_scheduler.insert_timer(&rt2);
RegTimer rt3;
rt3.expires = now.tv_sec + 15;
rt3.cb=test_cb;
registration_timer.insert_timer(&rt3);
registration_scheduler.insert_timer(&rt3);
RegTimer rt4;
rt4.expires = now.tv_sec - 1;
rt4.cb=test_cb;
registration_timer.insert_timer(&rt4);
registration_scheduler.insert_timer(&rt4);
RegTimer rt5;
rt5.expires = now.tv_sec + 100000;
rt5.cb=test_cb;
registration_timer.insert_timer(&rt5);
registration_scheduler.insert_timer(&rt5);
RegTimer rt6;
rt6.expires = now.tv_sec + 100;
rt6.cb=test_cb;
registration_timer.insert_timer_leastloaded(&rt6, now.tv_sec+5, now.tv_sec+50);
registration_scheduler.insert_timer_leastloaded(&rt6, now.tv_sec+5, now.tv_sec+50);
sleep(30);
@ -1003,7 +1115,7 @@ void DBRegAgent::run_tests() {
RegTimer rt7;
rt6.expires = now.tv_sec + 980;
rt6.cb=test_cb;
registration_timer.insert_timer_leastloaded(&rt6, now.tv_sec+9980, now.tv_sec+9990);
registration_scheduler.insert_timer_leastloaded(&rt6, now.tv_sec+9980, now.tv_sec+9990);
vector<RegTimer*> rts;
@ -1012,7 +1124,7 @@ void DBRegAgent::run_tests() {
rts.push_back(t);
t->expires = now.tv_sec + i;
t->cb=test_cb;
registration_timer.insert_timer_leastloaded(t, now.tv_sec, now.tv_sec+1000);
registration_scheduler.insert_timer_leastloaded(t, now.tv_sec, now.tv_sec+1000);
}
sleep(200);

26
apps/db_reg_agent/DBRegAgent.h
Unescape View File

@ -67,7 +67,7 @@ using std::queue;
struct RegistrationActionEvent : public AmEvent {
enum RegAction { Register, Deregister };
enum RegAction { Register=0, Deregister };
RegistrationActionEvent(RegAction action, int subscriber_id)
: AmEvent(RegistrationActionEventID),
@ -124,26 +124,32 @@ class DBRegAgent
static unsigned int ratelimit_rate;
static unsigned int ratelimit_per;
static bool delete_removed_registrations;
static unsigned int error_retry_interval;
map<long, AmSIPRegistration*> registrations;
map<string, long> registration_ltags;
map<long, RegTimer*> registration_timers;
AmMutex registrations_mut;
// connection used in main DBRegAgent thread
static mysqlpp::Connection DBConnection;
static mysqlpp::Connection MainDBConnection;
// connection used in other threads
static mysqlpp::Connection StatusDBConnection;
// connection used in other thread (processor thread)
static mysqlpp::Connection ProcessorDBConnection;
int onLoad();
RegistrationTimer registration_timer;
RegistrationTimer registration_scheduler;
DBRegAgentProcessorThread registration_processor;
bool loadRegistrations();
void createDBRegistration(long subscriber_id, mysqlpp::Connection& conn);
void updateDBRegistration(long subscriber_id, int last_code,
void deleteDBRegistration(long subscriber_id, mysqlpp::Connection& conn);
void updateDBRegistration(mysqlpp::Connection& db_connection,
long subscriber_id, int last_code,
const string& last_reason,
bool update_status = false, int status = 0,
bool update_ts=false, unsigned int expiry = 0);
@ -168,7 +174,11 @@ class DBRegAgent
/** schedule this subscriber to de-REGISTER imminently*/
void scheduleDeregistration(long subscriber_id);
/** create a timer for that registration
/** create a timer for the registration - fixed expiry + action */
void setRegistrationTimer(long subscriber_id, unsigned int timeout,
RegistrationActionEvent::RegAction reg_action);
/** create a registration refresh timer for that registration
@param subscriber_id - ID of subscription
@param expiry - SIP registration expiry time
@param reg_start_ts - start TS of the SIP registration
@ -223,7 +233,7 @@ class DBRegAgent
void invoke(const string& method,
const AmArg& args, AmArg& ret);
/** re-registration timer callback */
void timer_cb(RegTimer* timer, long subscriber_id, void* data2);
void timer_cb(RegTimer* timer, long subscriber_id, int data2);
friend class DBRegAgentProcessorThread;
};

4
apps/db_reg_agent/RegistrationTimer.h
Unescape View File

@ -41,7 +41,7 @@
#define TIMER_RESOLUTION 100000
class RegTimer;
typedef void (*timer_cb)(RegTimer*, long /*data1*/,void* /*data2*/);
typedef void (*timer_cb)(RegTimer*, long /*data1*/,int /*data2*/);
class RegTimerBucket;
@ -51,7 +51,7 @@ class RegTimer {
timer_cb cb;
long data1;
void* data2;
int data2;
RegTimer()
: expires(0), cb(0), data1(0), data2(0) { }

12
apps/db_reg_agent/etc/db_reg_agent.conf
Unescape View File

@ -20,6 +20,10 @@ joined_query="select subscribers.subscriber_id as subscriber_id, subscribers.use
# default: 7200
# expires=300
# error_retry_interval
#
# default: 300
# reregister_interval: fraction of actual expires after which register is refreshed
# default: reregister_interval=0.5
#reregister_interval=0.5
@ -47,4 +51,10 @@ joined_query="select subscribers.subscriber_id as subscriber_id, subscribers.use
#ratelimit_rate=2
# ratelimit_per=1 : per time unit (in seconds, e.g. 300 REGISTER in 1 second)
#ratelimit_per=1
#ratelimit_per=1
# delete_removed_registrations=yes : delete removed registrations from registrations
# table in DB? (otherwise they will stay with STATUS_REMOVED)
# default: yes
#
#delete_removed_registrations=no

Write Preview
Loading…
Cancel
Save