From a5dc76ec96a07211e4b7345e98c919aaca53ca32 Mon Sep 17 00:00:00 2001 From: danieldg Date: Sat, 13 Mar 2010 16:35:07 +0000 Subject: Add OnUnloadModule hook to MySQL git-svn-id: http://svn.inspircd.org/repository/trunk/inspircd@12632 e03df62e-2008-0410-955e-edbf42e46eb7 --- src/modules/extra/m_mysql.cpp | 139 +++++++++++++++++++++++++++++------------- 1 file changed, 95 insertions(+), 44 deletions(-) diff --git a/src/modules/extra/m_mysql.cpp b/src/modules/extra/m_mysql.cpp index 08f71c929..ae219df70 100644 --- a/src/modules/extra/m_mysql.cpp +++ b/src/modules/extra/m_mysql.cpp @@ -67,16 +67,23 @@ class SQLConnection; class MySQLresult; class DispatcherThread; -struct QueueItem +struct QQueueItem { SQLQuery* q; SQLConnection* c; - QueueItem(SQLQuery* Q, SQLConnection* C) : q(Q), c(C) {} + QQueueItem(SQLQuery* Q, SQLConnection* C) : q(Q), c(C) {} +}; + +struct RQueueItem +{ + SQLQuery* q; + MySQLresult* r; + RQueueItem(SQLQuery* Q, MySQLresult* R) : q(Q), r(R) {} }; typedef std::map ConnMap; -typedef std::deque QueryQueue; -typedef std::deque ResultQueue; +typedef std::deque QueryQueue; +typedef std::deque ResultQueue; /** MySQL module * */ @@ -84,14 +91,15 @@ class ModuleSQL : public Module { public: DispatcherThread* Dispatcher; - QueryQueue qq; - ResultQueue rq; - ConnMap connections; + QueryQueue qq; // MUST HOLD MUTEX + ResultQueue rq; // MUST HOLD MUTEX + ConnMap connections; // main thread only ModuleSQL(); void init(); ~ModuleSQL(); void OnRehash(User* user); + void OnUnloadModule(Module* mod); Version GetVersion(); }; @@ -115,14 +123,13 @@ class DispatcherThread : public SocketThread class MySQLresult : public SQLResult { public: - SQLQuery* query; SQLerror err; int currentrow; int rows; std::vector colnames; std::vector fieldlists; - MySQLresult(SQLQuery* q, MYSQL_RES* res, int affected_rows) : query(q), err(SQL_NO_ERROR), currentrow(0), rows(0) + MySQLresult(MYSQL_RES* res, int affected_rows) : err(SQL_NO_ERROR), currentrow(0), rows(0) { if (affected_rows >= 1) { @@ -166,7 +173,7 @@ class MySQLresult : public SQLResult } } - MySQLresult(SQLQuery* q, SQLerror& e) : query(q), err(e) + MySQLresult(SQLerror& e) : err(e) { } @@ -217,11 +224,11 @@ class SQLConnection : public SQLProvider public: reference config; MYSQL *connection; - bool active; + Mutex lock; // This constructor creates an SQLConnection object with the given credentials, but does not connect yet. SQLConnection(Module* p, ConfigTag* tag) : SQLProvider(p, "SQL/" + tag->getString("id")), - config(tag), active(false) + config(tag) { } @@ -310,30 +317,23 @@ class SQLConnection : public SQLProvider return (ModuleSQL*)(Module*)creator; } - void DoBlockingQuery(SQLQuery* req) + MySQLresult* DoBlockingQuery(SQLQuery* req) { + /* Parse the command string and dispatch it to mysql */ if (CheckConnection() && !mysql_real_query(connection, req->query.data(), req->query.length())) { /* Successfull query */ MYSQL_RES* res = mysql_use_result(connection); unsigned long rows = mysql_affected_rows(connection); - MySQLresult* r = new MySQLresult(req, res, rows); - Parent()->Dispatcher->LockQueue(); - Parent()->rq.push_back(r); - Parent()->Dispatcher->NotifyParent(); - Parent()->Dispatcher->UnlockQueue(); + return new MySQLresult(res, rows); } else { /* XXX: See /usr/include/mysql/mysqld_error.h for a list of * possible error numbers and error messages */ SQLerror e(SQL_QREPLY_FAIL, ConvToStr(mysql_errno(connection)) + std::string(": ") + mysql_error(connection)); - MySQLresult* r = new MySQLresult(req, e); - Parent()->Dispatcher->LockQueue(); - Parent()->rq.push_back(r); - Parent()->Dispatcher->NotifyParent(); - Parent()->Dispatcher->UnlockQueue(); + return new MySQLresult(e); } } @@ -359,7 +359,7 @@ class SQLConnection : public SQLProvider void submit(SQLQuery* q) { Parent()->Dispatcher->LockQueue(); - Parent()->qq.push_back(QueueItem(q, this)); + Parent()->qq.push_back(QQueueItem(q, this)); Parent()->Dispatcher->UnlockQueueWakeup(); } }; @@ -374,8 +374,8 @@ void ModuleSQL::init() Dispatcher = new DispatcherThread(this); ServerInstance->Threads->Start(Dispatcher); - Implementation eventlist[] = { I_OnRehash }; - ServerInstance->Modules->Attach(eventlist, this, 1); + Implementation eventlist[] = { I_OnRehash, I_OnUnloadModule }; + ServerInstance->Modules->Attach(eventlist, this, 2); } ModuleSQL::~ModuleSQL() @@ -394,7 +394,6 @@ ModuleSQL::~ModuleSQL() void ModuleSQL::OnRehash(User* user) { - Dispatcher->LockQueue(); ConnMap conns; ConfigTagList tags = ServerInstance->Config->ConfTags("database"); for(ConfigIter i = tags.first; i != tags.second; i++) @@ -415,21 +414,56 @@ void ModuleSQL::OnRehash(User* user) connections.erase(curr); } } + + // now clean up the deleted databases + Dispatcher->LockQueue(); + SQLerror err(SQL_BAD_DBID); for(ConnMap::iterator i = connections.begin(); i != connections.end(); i++) { - if (i->second->active) + ServerInstance->Modules->DelService(*i->second); + // it might be running a query on this database. Wait for that to complete + i->second->lock.Lock(); + i->second->lock.Unlock(); + // now remove all active queries to this DB + for(unsigned int j = qq.size() - 1; j >= 0; j--) { - // can't delete it now. Next rehash will try to kill it again - conns.insert(*i); + if (qq[j].c == i->second) + { + qq[j].q->OnError(err); + delete qq[j].q; + qq.erase(qq.begin() + j); + } } - else + // finally, nuke the connection + delete i->second; + } + Dispatcher->UnlockQueue(); + connections.swap(conns); +} + +void ModuleSQL::OnUnloadModule(Module* mod) +{ + SQLerror err(SQL_BAD_DBID); + Dispatcher->LockQueue(); + for(unsigned int i = qq.size() - 1; i >= 0; i--) + { + if (qq[i].q->creator == mod) { - ServerInstance->Modules->DelService(*i->second); - delete i->second; + if (i == 0) + { + // need to wait until the query is done + // (the result will be discarded) + qq[i].c->lock.Lock(); + qq[i].c->lock.Unlock(); + } + qq[i].q->OnError(err); + delete qq[i].q; + qq.erase(qq.begin() + i); } } - connections.swap(conns); Dispatcher->UnlockQueue(); + // clean up any result queue entries + Dispatcher->OnNotify(); } Version ModuleSQL::GetVersion() @@ -444,13 +478,30 @@ void DispatcherThread::Run() { if (!Parent->qq.empty()) { - QueueItem i = Parent->qq.front(); - Parent->qq.pop_front(); - i.c->active = true; + QQueueItem i = Parent->qq.front(); + i.c->lock.Lock(); this->UnlockQueue(); - i.c->DoBlockingQuery(i.q); + MySQLresult* res = i.c->DoBlockingQuery(i.q); + i.c->lock.Unlock(); + + /* + * At this point, the main thread could be working on: + * Rehash - delete i.c out from under us. We don't care about that. + * UnloadModule - delete i.q and the qq item. Need to avoid reporting results. + */ + this->LockQueue(); - i.c->active = false; + if (Parent->qq.front().q == i.q) + { + Parent->qq.pop_front(); + Parent->rq.push_back(RQueueItem(i.q, res)); + NotifyParent(); + } + else + { + // UnloadModule ate the query + delete res; + } } else { @@ -469,13 +520,13 @@ void DispatcherThread::OnNotify() this->LockQueue(); for(ResultQueue::iterator i = Parent->rq.begin(); i != Parent->rq.end(); i++) { - MySQLresult* res = *i; + MySQLresult* res = i->r; if (res->err.id == SQL_NO_ERROR) - res->query->OnResult(*res); + i->q->OnResult(*res); else - res->query->OnError(res->err); - delete res->query; - delete res; + i->q->OnError(res->err); + delete i->q; + delete i->r; } Parent->rq.clear(); this->UnlockQueue(); -- cgit v1.2.3