X-Git-Url: https://git.netwichtig.de/gitweb/?a=blobdiff_plain;f=src%2Fmodules%2Fextra%2Fm_mssql.cpp;h=3083dcec2f105e876dc1a1507ba08b869942f412;hb=a4306bc3188148e99245d4e84df7e67949e5a619;hp=93f70a18e48af1ffc7c06e46dd0d635931621070;hpb=d27af79ee06388dc6d1ab31d95348a38cdfbeb91;p=user%2Fhenk%2Fcode%2Finspircd.git diff --git a/src/modules/extra/m_mssql.cpp b/src/modules/extra/m_mssql.cpp index 93f70a18e..3083dcec2 100644 --- a/src/modules/extra/m_mssql.cpp +++ b/src/modules/extra/m_mssql.cpp @@ -1,12 +1,12 @@ -/* +------------------------------------+ - * | Inspire Internet Relay Chat Daemon | - * +------------------------------------+ +/* +------------------------------------+ + * | Inspire Internet Relay Chat Daemon | + * +------------------------------------+ * - * InspIRCd: (C) 2002-2008 InspIRCd Development Team + * InspIRCd: (C) 2002-2008 InspIRCd Development Team * See: http://www.inspircd.org/wiki/index.php/Credits * * This program is free but copyrighted software; see - * the file COPYING for details. + * the file COPYING for details. * * --------------------------------------------------- */ @@ -21,44 +21,85 @@ #include "m_sqlv2.h" /* $ModDesc: MsSQL provider */ +/* $CompileFlags: exec("grep VERSION_NO /usr/include/tdsver.h 2>/dev/null | perl -e 'print "-D_TDSVER=".((<> =~ /freetds v(\d+\.\d+)/i) ? $1*100 : 0);'") */ /* $LinkerFlags: -ltds */ /* $ModDep: m_sqlv2.h */ class SQLConn; class MsSQLResult; class ResultNotifier; +class MsSQLListener; +class ModuleMsSQL; typedef std::map ConnMap; -typedef std::deque paramlist; typedef std::deque ResultQueue; -ResultNotifier* resultnotify = NULL; -ResultNotifier* resultdispatch = NULL; +ResultNotifier* notifier = NULL; +MsSQLListener* listener = NULL; int QueueFD = -1; +ConnMap connections; +Mutex* QueueMutex; +Mutex* ResultsMutex; +Mutex* LoggingMutex; + +class QueryThread : public Thread +{ + private: + ModuleMsSQL* Parent; + InspIRCd* ServerInstance; + public: + QueryThread(InspIRCd* si, ModuleMsSQL* mod) + : Thread(), Parent(mod), ServerInstance(si) + { + } + ~QueryThread() { } + virtual void Run(); +}; + class ResultNotifier : public BufferedSocket { - Module* mod; - insp_sockaddr sock_us; + ModuleMsSQL* mod; + + public: + ResultNotifier(ModuleMsSQL* m, InspIRCd* SI, int newfd, char* ip) : BufferedSocket(SI, newfd, ip), mod(m) + { + } + + virtual bool OnDataReady() + { + char data = 0; + if (ServerInstance->SE->Recv(this, &data, 1, 0) > 0) + { + Dispatch(); + return true; + } + return false; + } + + void Dispatch(); +}; + +class MsSQLListener : public ListenSocketBase +{ + ModuleMsSQL* Parent; + irc::sockets::insp_sockaddr sock_us; socklen_t uslen; + FileReader* index; public: - /* Create a socket on a random port. Let the tcp stack allocate us an available port */ -#ifdef IPV6 - ResultNotifier(InspIRCd* SI, Module* m) : BufferedSocket(SI, "::1", 0, true, 3000), mod(m) -#else - ResultNotifier(InspIRCd* SI, Module* m) : BufferedSocket(SI, "127.0.0.1", 0, true, 3000), mod(m) -#endif + MsSQLListener(ModuleMsSQL* P, InspIRCd* Instance, int port, const std::string &addr) : ListenSocketBase(Instance, port, addr), Parent(P) { uslen = sizeof(sock_us); if (getsockname(this->fd,(sockaddr*)&sock_us,&uslen)) { - throw ModuleException("Could not create random listening port on localhost"); + throw ModuleException("Could not getsockname() to find out port number for ITC port"); } } - ResultNotifier(InspIRCd* SI, Module* m, int newfd, char* ip) : BufferedSocket(SI, newfd, ip), mod(m) + virtual void OnAcceptReady(const std::string &ipconnectedto, int nfd, const std::string &incomingip) { + new ResultNotifier(this->Parent, this->ServerInstance, nfd, (char *)ipconnectedto.c_str()); // XXX unsafe casts suck } /* Using getsockname and ntohs, we can determine which port number we were allocated */ @@ -70,25 +111,6 @@ class ResultNotifier : public BufferedSocket return ntohs(sock_us.sin_port); #endif } - - virtual int OnIncomingConnection(int newsock, char* ip) - { - resultdispatch = new ResultNotifier(Instance, mod, newsock, ip); - return true; - } - - virtual bool OnDataReady() - { - char data = 0; - if (Instance->SE->Recv(this, &data, 1, 0) > 0) - { - Dispatch(); - return true; - } - return false; - } - - void Dispatch(); }; @@ -264,7 +286,7 @@ class SQLConn : public classbase { private: ResultQueue results; - InspIRCd* Instance; + InspIRCd* ServerInstance; Module* mod; SQLhost host; TDSLOGIN* login; @@ -272,8 +294,10 @@ class SQLConn : public classbase TDSCONTEXT* context; public: + QueryQueue queue; + SQLConn(InspIRCd* SI, Module* m, const SQLhost& hi) - : Instance(SI), mod(m), host(hi), login(NULL), sock(NULL), context(NULL) + : ServerInstance(SI), mod(m), host(hi), login(NULL), sock(NULL), context(NULL) { if (OpenDB()) { @@ -282,19 +306,25 @@ class SQLConn : public classbase { if (tds_process_simple_query(sock) != TDS_SUCCEED) { - Instance->Logs->Log("m_mssql",DEFAULT, "WARNING: Could not select database " + host.name + " for DB with id: " + host.id); + LoggingMutex->Lock(); + ServerInstance->Logs->Log("m_mssql",DEFAULT, "WARNING: Could not select database " + host.name + " for DB with id: " + host.id); + LoggingMutex->Unlock(); CloseDB(); } } else { - Instance->Logs->Log("m_mssql",DEFAULT, "WARNING: Could not select database " + host.name + " for DB with id: " + host.id); + LoggingMutex->Lock(); + ServerInstance->Logs->Log("m_mssql",DEFAULT, "WARNING: Could not select database " + host.name + " for DB with id: " + host.id); + LoggingMutex->Unlock(); CloseDB(); } } else { - Instance->Logs->Log("m_mssql",DEFAULT, "WARNING: Could not connect to DB with id: " + host.id); + LoggingMutex->Lock(); + ServerInstance->Logs->Log("m_mssql",DEFAULT, "WARNING: Could not connect to DB with id: " + host.id); + LoggingMutex->Unlock(); CloseDB(); } } @@ -318,7 +348,7 @@ class SQLConn : public classbase /* Total length of the unescaped parameters */ unsigned long paramlen; - /* Total length of query, used for binary-safety in mysql_real_query */ + /* Total length of query, used for binary-safety */ unsigned long querylength = 0; paramlen = 0; @@ -379,12 +409,14 @@ class SQLConn : public classbase *queryend = 0; req.query.q = query; - MsSQLResult* res = new MsSQLResult(mod, req.GetSource(), req.id); + MsSQLResult* res = new MsSQLResult((Module*)mod, req.GetSource(), req.id); res->dbid = host.id; res->query = req.query.q; char* msquery = strdup(req.query.q.data()); - Instance->Logs->Log("m_mssql",DEBUG,"doing Query: %s",msquery); + LoggingMutex->Lock(); + ServerInstance->Logs->Log("m_mssql",DEBUG,"doing Query: %s",msquery); + LoggingMutex->Unlock(); if (tds_submit_query(sock, msquery) != TDS_SUCCEED) { std::string error("failed to execute: "+std::string(req.query.q.data())); @@ -399,8 +431,8 @@ class SQLConn : public classbase int tds_res; while (tds_process_tokens(sock, &tds_res, NULL, TDS_TOKEN_RESULTS) == TDS_SUCCEED) { - //Instance->Logs->Log("m_mssql",DEBUG,"<******> result type: %d", tds_res); - //Instance->Logs->Log("m_mssql",DEBUG,"AFFECTED ROWS: %d", sock->rows_affected); + //ServerInstance->Logs->Log("m_mssql",DEBUG,"<******> result type: %d", tds_res); + //ServerInstance->Logs->Log("m_mssql",DEBUG,"AFFECTED ROWS: %d", sock->rows_affected); switch (tds_res) { case TDS_ROWFMT_RESULT: @@ -438,7 +470,11 @@ class SQLConn : public classbase unsigned char* src; CONV_RESULT dres; ctype = tds_get_conversion_type(col->column_type, col->column_size); - src = &(sock->current_results->current_row[col->column_offset]); +#if _TDSVER >= 82 + src = col->column_data; +#else + src = &(sock->current_results->current_row[col->column_offset]); +#endif srclen = col->column_cur_size; tds_convert(sock->tds_ctx, ctype, (TDS_CHAR *) src, srclen, SYBCHAR, &dres); data[j] = (char*)dres.ib; @@ -452,7 +488,9 @@ class SQLConn : public classbase break; } } + ResultsMutex->Lock(); results.push_back(res); + ResultsMutex->Unlock(); SendNotify(); return SQLerror(); } @@ -460,14 +498,18 @@ class SQLConn : public classbase static int HandleMessage(const TDSCONTEXT * pContext, TDSSOCKET * pTdsSocket, TDSMESSAGE * pMessage) { SQLConn* sc = (SQLConn*)pContext->parent; - sc->Instance->Logs->Log("m_mssql", DEBUG, "Message for DB with id: %s -> %s", sc->host.id.c_str(), pMessage->message); + LoggingMutex->Lock(); + sc->ServerInstance->Logs->Log("m_mssql", DEBUG, "Message for DB with id: %s -> %s", sc->host.id.c_str(), pMessage->message); + LoggingMutex->Unlock(); return 0; } static int HandleError(const TDSCONTEXT * pContext, TDSSOCKET * pTdsSocket, TDSMESSAGE * pMessage) { SQLConn* sc = (SQLConn*)pContext->parent; - sc->Instance->Logs->Log("m_mssql", DEFAULT, "Error for DB with id: %s -> %s", sc->host.id.c_str(), pMessage->message); + LoggingMutex->Lock(); + sc->ServerInstance->Logs->Log("m_mssql", DEFAULT, "Error for DB with id: %s -> %s", sc->host.id.c_str(), pMessage->message); + LoggingMutex->Unlock(); return 0; } @@ -545,6 +587,7 @@ class SQLConn : public classbase while (results.size()) { MsSQLResult* res = results[0]; + ResultsMutex->Lock(); if (res->GetDest()) { res->Send(); @@ -558,6 +601,7 @@ class SQLConn : public classbase delete res; } results.pop_front(); + ResultsMutex->Unlock(); } } @@ -581,18 +625,18 @@ class SQLConn : public classbase return; } - insp_sockaddr addr; + irc::sockets::insp_sockaddr addr; #ifdef IPV6 - insp_aton("::1", &addr.sin6_addr); + irc::sockets::insp_aton("::1", &addr.sin6_addr); addr.sin6_family = AF_FAMILY; - addr.sin6_port = htons(resultnotify->GetPort()); + addr.sin6_port = htons(listener->GetPort()); #else - insp_inaddr ia; - insp_aton("127.0.0.1", &ia); + irc::sockets::insp_inaddr ia; + irc::sockets::insp_aton("127.0.0.1", &ia); addr.sin_family = AF_FAMILY; addr.sin_addr = ia; - addr.sin_port = htons(resultnotify->GetPort()); + addr.sin_port = htons(listener->GetPort()); #endif if (connect(QueueFD, (sockaddr*)&addr,sizeof(addr)) == -1) @@ -605,19 +649,29 @@ class SQLConn : public classbase send(QueueFD, &id, 1, 0); } + void DoLeadingQuery() + { + SQLrequest& req = queue.front(); + req.error = Query(req); + } + }; class ModuleMsSQL : public Module { private: - ConnMap connections; unsigned long currid; + QueryThread* queryDispatcher; public: ModuleMsSQL(InspIRCd* Me) - : Module::Module(Me), currid(0) + : Module(Me), currid(0) { + LoggingMutex = ServerInstance->Mutexes->CreateMutex(); + ResultsMutex = ServerInstance->Mutexes->CreateMutex(); + QueueMutex = ServerInstance->Mutexes->CreateMutex(); + ServerInstance->Modules->UseInterface("SQLutils"); if (!ServerInstance->Modules->PublishFeature("SQL", this)) @@ -625,10 +679,30 @@ class ModuleMsSQL : public Module throw ModuleException("m_mssql: Unable to publish feature 'SQL'"); } - resultnotify = new ResultNotifier(ServerInstance, this); + /* Create a socket on a random port. Let the tcp stack allocate us an available port */ +#ifdef IPV6 + listener = new MsSQLListener(this, ServerInstance, 0, "::1"); +#else + listener = new MsSQLListener(this, ServerInstance, 0, "127.0.0.1"); +#endif + + if (listener->GetFd() == -1) + { + ServerInstance->Modules->DoneWithInterface("SQLutils"); + throw ModuleException("m_mssql: unable to create ITC pipe"); + } + else + { + LoggingMutex->Lock(); + ServerInstance->Logs->Log("m_mssql", DEBUG, "MsSQL: Interthread comms port is %d", listener->GetPort()); + LoggingMutex->Unlock(); + } ReadConf(); + queryDispatcher = new QueryThread(ServerInstance, this); + ServerInstance->Threads->Create(queryDispatcher); + ServerInstance->Modules->PublishInterface("SQL", this); Implementation eventlist[] = { I_OnRequest, I_OnRehash }; ServerInstance->Modules->Attach(eventlist, this, 2); @@ -636,11 +710,11 @@ class ModuleMsSQL : public Module virtual ~ModuleMsSQL() { + delete queryDispatcher; ClearQueue(); ClearAllConnections(); - ServerInstance->SE->DelFd(resultnotify); - resultnotify->Close(); + ServerInstance->SE->DelFd(listener); ServerInstance->BufferedSocketCull(); if (QueueFD >= 0) @@ -649,16 +723,20 @@ class ModuleMsSQL : public Module close(QueueFD); } - if (resultdispatch) + if (notifier) { - ServerInstance->SE->DelFd(resultdispatch); - resultdispatch->Close(); + ServerInstance->SE->DelFd(notifier); + notifier->Close(); ServerInstance->BufferedSocketCull(); } ServerInstance->Modules->UnpublishInterface("SQL", this); ServerInstance->Modules->UnpublishFeature("SQL"); ServerInstance->Modules->DoneWithInterface("SQLutils"); + + delete LoggingMutex; + delete ResultsMutex; + delete QueueMutex; } @@ -733,7 +811,9 @@ class ModuleMsSQL : public Module { if (HasHost(hi)) { + LoggingMutex->Lock(); ServerInstance->Logs->Log("m_mssql",DEFAULT, "WARNING: A MsSQL connection with id: %s already exists. Aborting database open attempt.", hi.id.c_str()); + LoggingMutex->Unlock(); return; } @@ -771,7 +851,9 @@ class ModuleMsSQL : public Module virtual void OnRehash(User* user, const std::string ¶meter) { + QueueMutex->Lock(); ReadConf(); + QueueMutex->Unlock(); } virtual const char* OnRequest(Request* request) @@ -779,18 +861,27 @@ class ModuleMsSQL : public Module if(strcmp(SQLREQID, request->GetId()) == 0) { SQLrequest* req = (SQLrequest*)request; + + QueueMutex->Lock(); + ConnMap::iterator iter; + + const char* returnval = NULL; + if((iter = connections.find(req->dbid)) != connections.end()) { req->id = NewID(); - req->error = iter->second->Query(*req); - return SQLSUCCESS; + iter->second->queue.push(*req); + returnval= SQLSUCCESS; } else { req->error.Id(SQL_BAD_DBID); - return NULL; } + + QueueMutex->Unlock(); + + return returnval; } return NULL; } @@ -812,7 +903,33 @@ class ModuleMsSQL : public Module void ResultNotifier::Dispatch() { - ((ModuleMsSQL*)mod)->SendQueue(); + mod->SendQueue(); +} + +void QueryThread::Run() +{ + while (this->GetExitFlag() == false) + { + SQLConn* conn = NULL; + QueueMutex->Lock(); + for (ConnMap::iterator i = connections.begin(); i != connections.end(); i++) + { + if (i->second->queue.totalsize()) + { + conn = i->second; + break; + } + } + QueueMutex->Unlock(); + if (conn) + { + conn->DoLeadingQuery(); + QueueMutex->Lock(); + conn->queue.pop(); + QueueMutex->Unlock(); + } + usleep(1000); + } } MODULE_INIT(ModuleMsSQL)