src/modules/extra/m_mysql.cpp

   1 /*       +------------------------------------+
   2  *       | Inspire Internet Relay Chat Daemon |
   3  *       +------------------------------------+
   4  *
   5  *  InspIRCd: (C) 2002-2010 InspIRCd Development Team
   6  * See: http://wiki.inspircd.org/Credits
   7  *
   8  * This program is free but copyrighted software; see
   9  *          the file COPYING for details.
  10  *
  11  * ---------------------------------------------------
  12  */
  13
  14 /* Stop mysql wanting to use long long */
  15 #define NO_CLIENT_LONG_LONG
  16
  17 #include "inspircd.h"
  18 #include <mysql.h>
  19 #include "m_sqlv2.h"
  20
  21 #ifdef WINDOWS
  22 #pragma comment(lib, "mysqlclient.lib")
  23 #endif
  24
  25 /* VERSION 2 API: With nonblocking (threaded) requests */
  26
  27 /* $ModDesc: SQL Service Provider module for all other m_sql* modules */
  28 /* $CompileFlags: exec("mysql_config --include") */
  29 /* $LinkerFlags: exec("mysql_config --libs_r") rpath("mysql_config --libs_r") */
  30 /* $ModDep: m_sqlv2.h */
  31
  32 /* THE NONBLOCKING MYSQL API!
  33  *
  34  * MySQL provides no nonblocking (asyncronous) API of its own, and its developers recommend
  35  * that instead, you should thread your program. This is what i've done here to allow for
  36  * asyncronous SQL requests via mysql. The way this works is as follows:
  37  *
  38  * The module spawns a thread via class Thread, and performs its mysql queries in this thread,
  39  * using a queue with priorities. There is a mutex on either end which prevents two threads
  40  * adjusting the queue at the same time, and crashing the ircd. Every 50 milliseconds, the
  41  * worker thread wakes up, and checks if there is a request at the head of its queue.
  42  * If there is, it processes this request, blocking the worker thread but leaving the ircd
  43  * thread to go about its business as usual. During this period, the ircd thread is able
  44  * to insert futher pending requests into the queue.
  45  *
  46  * Once the processing of a request is complete, it is removed from the incoming queue to
  47  * an outgoing queue, and initialized as a 'response'. The worker thread then signals the
  48  * ircd thread (via a loopback socket) of the fact a result is available, by sending the
  49  * connection ID through the connection.
  50  *
  51  * The ircd thread then mutexes the queue once more, reads the outbound response off the head
  52  * of the queue, and sends it on its way to the original calling module.
  53  *
  54  * XXX: You might be asking "why doesnt he just send the response from within the worker thread?"
  55  * The answer to this is simple. The majority of InspIRCd, and in fact most ircd's are not
  56  * threadsafe. This module is designed to be threadsafe and is careful with its use of threads,
  57  * however, if we were to call a module's OnRequest even from within a thread which was not the
  58  * one the module was originally instantiated upon, there is a chance of all hell breaking loose
  59  * if a module is ever put in a re-enterant state (stack corruption could occur, crashes, data
  60  * corruption, and worse, so DONT think about it until the day comes when InspIRCd is 100%
  61  * gauranteed threadsafe!)
  62  *
  63  * For a diagram of this system please see http://wiki.inspircd.org/Mysql2
  64  */
  65
  66
  67 class SQLConnection;
  68 class DispatcherThread;
  69
  70 typedef std::map<std::string, SQLConnection*> ConnMap;
  71 typedef std::deque<SQLresult*> ResultQueue;
  72
  73 static unsigned long count(const char * const str, char a)
  74 {
  75         unsigned long n = 0;
  76         for (const char *p = str; *p; ++p)
  77         {
  78                 if (*p == '?')
  79                         ++n;
  80         }
  81         return n;
  82 }
  83
  84
  85 /** MySQL module
  86  *  */
  87 class ModuleSQL : public Module
  88 {
  89  public:
  90          int currid;
  91          bool rehashing;
  92          DispatcherThread* Dispatcher;
  93          Mutex ResultsMutex;
  94          Mutex LoggingMutex;
  95          Mutex ConnMutex;
  96          ServiceProvider sqlserv;
  97
  98          ModuleSQL();
  99          ~ModuleSQL();
 100          unsigned long NewID();
 101          void OnRequest(Request& request);
 102          void OnRehash(User* user);
 103          Version GetVersion();
 104 };
 105
 106
 107 #if !defined(MYSQL_VERSION_ID) || MYSQL_VERSION_ID<32224
 108 #define mysql_field_count mysql_num_fields
 109 #endif
 110
 111 /** Represents a mysql result set
 112  */
 113 class MySQLresult : public SQLresult
 114 {
 115         int currentrow;
 116         std::vector<std::string> colnames;
 117         std::vector<SQLfieldList> fieldlists;
 118         SQLfieldMap* fieldmap;
 119         SQLfieldMap fieldmap2;
 120         SQLfieldList emptyfieldlist;
 121         int rows;
 122  public:
 123
 124         MySQLresult(Module* self, Module* to, MYSQL_RES* res, int affected_rows, unsigned int rid) : SQLresult(self, to, rid), currentrow(0), fieldmap(NULL)
 125         {
 126                 /* A number of affected rows from from mysql_affected_rows.
 127                  */
 128                 fieldlists.clear();
 129                 rows = 0;
 130                 if (affected_rows >= 1)
 131                 {
 132                         rows = affected_rows;
 133                         fieldlists.resize(rows);
 134                 }
 135                 unsigned int field_count = 0;
 136                 if (res)
 137                 {
 138                         MYSQL_ROW row;
 139                         int n = 0;
 140                         while ((row = mysql_fetch_row(res)))
 141                         {
 142                                 if (fieldlists.size() < (unsigned int)rows+1)
 143                                 {
 144                                         fieldlists.resize(fieldlists.size()+1);
 145                                 }
 146                                 field_count = 0;
 147                                 MYSQL_FIELD *fields = mysql_fetch_fields(res);
 148                                 if(mysql_num_fields(res) == 0)
 149                                         break;
 150                                 if (fields && mysql_num_fields(res))
 151                                 {
 152                                         colnames.clear();
 153                                         while (field_count < mysql_num_fields(res))
 154                                         {
 155                                                 std::string a = (fields[field_count].name ? fields[field_count].name : "");
 156                                                 std::string b = (row[field_count] ? row[field_count] : "");
 157                                                 SQLfield sqlf(b, !row[field_count]);
 158                                                 colnames.push_back(a);
 159                                                 fieldlists[n].push_back(sqlf);
 160                                                 field_count++;
 161                                         }
 162                                         n++;
 163                                 }
 164                                 rows++;
 165                         }
 166                         mysql_free_result(res);
 167                         res = NULL;
 168                 }
 169         }
 170
 171         MySQLresult(Module* self, Module* to, SQLerror e, unsigned int rid) : SQLresult(self, to, rid), currentrow(0)
 172         {
 173                 rows = 0;
 174                 error = e;
 175         }
 176
 177         ~MySQLresult()
 178         {
 179         }
 180
 181         virtual int Rows()
 182         {
 183                 return rows;
 184         }
 185
 186         virtual int Cols()
 187         {
 188                 return colnames.size();
 189         }
 190
 191         virtual std::string ColName(int column)
 192         {
 193                 if (column < (int)colnames.size())
 194                 {
 195                         return colnames[column];
 196                 }
 197                 else
 198                 {
 199                         throw SQLbadColName();
 200                 }
 201                 return "";
 202         }
 203
 204         virtual int ColNum(const std::string &column)
 205         {
 206                 for (unsigned int i = 0; i < colnames.size(); i++)
 207                 {
 208                         if (column == colnames[i])
 209                                 return i;
 210                 }
 211                 throw SQLbadColName();
 212                 return 0;
 213         }
 214
 215         virtual SQLfield GetValue(int row, int column)
 216         {
 217                 if ((row >= 0) && (row < rows) && (column >= 0) && (column < Cols()))
 218                 {
 219                         return fieldlists[row][column];
 220                 }
 221
 222                 throw SQLbadColName();
 223
 224                 /* XXX: We never actually get here because of the throw */
 225                 return SQLfield("",true);
 226         }
 227
 228         virtual SQLfieldList& GetRow()
 229         {
 230                 if (currentrow < rows)
 231                         return fieldlists[currentrow++];
 232                 else
 233                         return emptyfieldlist;
 234         }
 235
 236         virtual SQLfieldMap& GetRowMap()
 237         {
 238                 fieldmap2.clear();
 239
 240                 if (currentrow < rows)
 241                 {
 242                         for (int i = 0; i < Cols(); i++)
 243                         {
 244                                 fieldmap2.insert(std::make_pair(colnames[i],GetValue(currentrow, i)));
 245                         }
 246                         currentrow++;
 247                 }
 248
 249                 return fieldmap2;
 250         }
 251
 252         virtual SQLfieldList* GetRowPtr()
 253         {
 254                 SQLfieldList* fieldlist = new SQLfieldList();
 255
 256                 if (currentrow < rows)
 257                 {
 258                         for (int i = 0; i < Rows(); i++)
 259                         {
 260                                 fieldlist->push_back(fieldlists[currentrow][i]);
 261                         }
 262                         currentrow++;
 263                 }
 264                 return fieldlist;
 265         }
 266
 267         virtual SQLfieldMap* GetRowMapPtr()
 268         {
 269                 fieldmap = new SQLfieldMap();
 270
 271                 if (currentrow < rows)
 272                 {
 273                         for (int i = 0; i < Cols(); i++)
 274                         {
 275                                 fieldmap->insert(std::make_pair(colnames[i],GetValue(currentrow, i)));
 276                         }
 277                         currentrow++;
 278                 }
 279
 280                 return fieldmap;
 281         }
 282
 283         virtual void Free(SQLfieldMap* fm)
 284         {
 285                 delete fm;
 286         }
 287
 288         virtual void Free(SQLfieldList* fl)
 289         {
 290                 delete fl;
 291         }
 292 };
 293
 294 /** Represents a connection to a mysql database
 295  */
 296 class SQLConnection : public classbase
 297 {
 298  protected:
 299         MYSQL *connection;
 300         MYSQL_RES *res;
 301         MYSQL_ROW *row;
 302         SQLhost host;
 303         std::map<std::string,std::string> thisrow;
 304         bool Enabled;
 305         ModuleSQL* Parent;
 306         std::string initquery;
 307
 308  public:
 309
 310         QueryQueue queue;
 311         ResultQueue rq;
 312
 313         // This constructor creates an SQLConnection object with the given credentials, but does not connect yet.
 314         SQLConnection(const SQLhost &hi, ModuleSQL* Creator) : connection(NULL), host(hi), Enabled(false), Parent(Creator)
 315         {
 316         }
 317
 318         ~SQLConnection()
 319         {
 320                 Close();
 321         }
 322
 323         // This method connects to the database using the credentials supplied to the constructor, and returns
 324         // true upon success.
 325         bool Connect()
 326         {
 327                 unsigned int timeout = 1;
 328                 connection = mysql_init(connection);
 329                 mysql_options(connection,MYSQL_OPT_CONNECT_TIMEOUT,(char*)&timeout);
 330                 return mysql_real_connect(connection, host.host.c_str(), host.user.c_str(), host.pass.c_str(), host.name.c_str(), host.port, NULL, 0);
 331         }
 332
 333         void DoLeadingQuery()
 334         {
 335                 if (!CheckConnection())
 336                         return;
 337
 338                 if( !initquery.empty() )
 339                         mysql_query(connection,initquery.c_str());
 340
 341                 /* Parse the command string and dispatch it to mysql */
 342                 SQLrequest* req = queue.front();
 343
 344                 /* Pointer to the buffer we screw around with substitution in */
 345                 char* query;
 346
 347                 /* Pointer to the current end of query, where we append new stuff */
 348                 char* queryend;
 349
 350                 /* Total length of the unescaped parameters */
 351                 unsigned long maxparamlen, paramcount;
 352
 353                 /* The length of the longest parameter */
 354                 maxparamlen = 0;
 355
 356                 for(ParamL::iterator i = req->query.p.begin(); i != req->query.p.end(); i++)
 357                 {
 358                         if (i->size() > maxparamlen)
 359                                 maxparamlen = i->size();
 360                 }
 361
 362                 /* How many params are there in the query? */
 363                 paramcount = count(req->query.q.c_str(), '?');
 364
 365                 /* This stores copy of params to be inserted with using numbered params 1;3B*/
 366                 ParamL paramscopy(req->query.p);
 367
 368                 /* To avoid a lot of allocations, allocate enough memory for the biggest the escaped query could possibly be.
 369                  * sizeofquery + (maxtotalparamlength*2) + 1
 370                  *
 371                  * The +1 is for null-terminating the string for mysql_real_escape_string
 372                  */
 373
 374                 query = new char[req->query.q.length() + (maxparamlen*paramcount*2) + 1];
 375                 queryend = query;
 376
 377                 /* Okay, now we have a buffer large enough we need to start copying the query into it and escaping and substituting
 378                  * the parameters into it...
 379                  */
 380
 381                 for(unsigned long i = 0; i < req->query.q.length(); i++)
 382                 {
 383                         if(req->query.q[i] == '?')
 384                         {
 385                                 /* We found a place to substitute..what fun.
 386                                  * use mysql calls to escape and write the
 387                                  * escaped string onto the end of our query buffer,
 388                                  * then we "just" need to make sure queryend is
 389                                  * pointing at the right place.
 390                                  */
 391
 392                                 /* Is it numbered parameter?
 393                                  */
 394
 395                                 bool numbered;
 396                                 numbered = false;
 397
 398                                 /* Numbered parameter number :|
 399                                  */
 400                                 unsigned int paramnum;
 401                                 paramnum = 0;
 402
 403                                 /* Let's check if it's a numbered param. And also calculate it's number.
 404                                  */
 405
 406                                 while ((i < req->query.q.length() - 1) && (req->query.q[i+1] >= '0') && (req->query.q[i+1] <= '9'))
 407                                 {
 408                                         numbered = true;
 409                                         ++i;
 410                                         paramnum = paramnum * 10 + req->query.q[i] - '0';
 411                                 }
 412
 413                                 if (paramnum > paramscopy.size() - 1)
 414                                 {
 415                                         /* index is out of range!
 416                                          */
 417                                         numbered = false;
 418                                 }
 419
 420                                 if (numbered)
 421                                 {
 422                                         unsigned long len = mysql_real_escape_string(connection, queryend, paramscopy[paramnum].c_str(), paramscopy[paramnum].length());
 423
 424                                         queryend += len;
 425                                 }
 426                                 else if (req->query.p.size())
 427                                 {
 428                                         unsigned long len = mysql_real_escape_string(connection, queryend, req->query.p.front().c_str(), req->query.p.front().length());
 429
 430                                         queryend += len;
 431                                         req->query.p.pop_front();
 432                                 }
 433                                 else
 434                                         break;
 435                         }
 436                         else
 437                         {
 438                                 *queryend = req->query.q[i];
 439                                 queryend++;
 440                         }
 441                 }
 442
 443                 *queryend = 0;
 444
 445                 req->query.q = query;
 446
 447                 if (!mysql_real_query(connection, req->query.q.data(), req->query.q.length()))
 448                 {
 449                         /* Successfull query */
 450                         res = mysql_use_result(connection);
 451                         unsigned long rows = mysql_affected_rows(connection);
 452                         MySQLresult* r = new MySQLresult(Parent, req->source, res, rows, req->id);
 453                         r->dbid = this->GetID();
 454                         r->query = req->query.q;
 455                         /* Put this new result onto the results queue.
 456                          * XXX: Remember to mutex the queue!
 457                          */
 458                         Parent->ResultsMutex.Lock();
 459                         rq.push_back(r);
 460                         Parent->ResultsMutex.Unlock();
 461                 }
 462                 else
 463                 {
 464                         /* XXX: See /usr/include/mysql/mysqld_error.h for a list of
 465                          * possible error numbers and error messages */
 466                         SQLerror e(SQL_QREPLY_FAIL, ConvToStr(mysql_errno(connection)) + std::string(": ") + mysql_error(connection));
 467                         MySQLresult* r = new MySQLresult(Parent, req->source, e, req->id);
 468                         r->dbid = this->GetID();
 469                         r->query = req->query.q;
 470
 471                         Parent->ResultsMutex.Lock();
 472                         rq.push_back(r);
 473                         Parent->ResultsMutex.Unlock();
 474                 }
 475
 476                 delete[] query;
 477         }
 478
 479         bool ConnectionLost()
 480         {
 481                 if (&connection)
 482                 {
 483                         return (mysql_ping(connection) != 0);
 484                 }
 485                 else return false;
 486         }
 487
 488         bool CheckConnection()
 489         {
 490                 if (ConnectionLost())
 491                 {
 492                         return Connect();
 493                 }
 494                 else return true;
 495         }
 496
 497         std::string GetError()
 498         {
 499                 return mysql_error(connection);
 500         }
 501
 502         const std::string& GetID()
 503         {
 504                 return host.id;
 505         }
 506
 507         std::string GetHost()
 508         {
 509                 return host.host;
 510         }
 511
 512         void setInitialQuery(std::string init)
 513         {
 514                 initquery = init;
 515         }
 516
 517         void SetEnable(bool Enable)
 518         {
 519                 Enabled = Enable;
 520         }
 521
 522         bool IsEnabled()
 523         {
 524                 return Enabled;
 525         }
 526
 527         void Close()
 528         {
 529                 mysql_close(connection);
 530         }
 531
 532         const SQLhost& GetConfHost()
 533         {
 534                 return host;
 535         }
 536
 537 };
 538
 539 ConnMap Connections;
 540
 541 bool HasHost(const SQLhost &host)
 542 {
 543         for (ConnMap::iterator iter = Connections.begin(); iter != Connections.end(); iter++)
 544         {
 545                 if (host == iter->second->GetConfHost())
 546                         return true;
 547         }
 548         return false;
 549 }
 550
 551 bool HostInConf(const SQLhost &h)
 552 {
 553         ConfigReader conf;
 554         for(int i = 0; i < conf.Enumerate("database"); i++)
 555         {
 556                 SQLhost host;
 557                 host.id         = conf.ReadValue("database", "id", i);
 558                 host.host       = conf.ReadValue("database", "hostname", i);
 559                 host.port       = conf.ReadInteger("database", "port", i, true);
 560                 host.name       = conf.ReadValue("database", "name", i);
 561                 host.user       = conf.ReadValue("database", "username", i);
 562                 host.pass       = conf.ReadValue("database", "password", i);
 563                 host.ssl        = conf.ReadFlag("database", "ssl", i);
 564                 if (h == host)
 565                         return true;
 566         }
 567         return false;
 568 }
 569
 570 void ClearOldConnections()
 571 {
 572         ConnMap::iterator i,safei;
 573         for (i = Connections.begin(); i != Connections.end(); i++)
 574         {
 575                 if (!HostInConf(i->second->GetConfHost()))
 576                 {
 577                         delete i->second;
 578                         safei = i;
 579                         --i;
 580                         Connections.erase(safei);
 581                 }
 582         }
 583 }
 584
 585 void ClearAllConnections()
 586 {
 587         ConnMap::iterator i;
 588         while ((i = Connections.begin()) != Connections.end())
 589         {
 590                 Connections.erase(i);
 591                 delete i->second;
 592         }
 593 }
 594
 595 void ConnectDatabases(ModuleSQL* Parent)
 596 {
 597         for (ConnMap::iterator i = Connections.begin(); i != Connections.end(); i++)
 598         {
 599                 if (i->second->IsEnabled())
 600                         continue;
 601
 602                 i->second->SetEnable(true);
 603                 if (!i->second->Connect())
 604                 {
 605                         /* XXX: MUTEX */
 606                         Parent->LoggingMutex.Lock();
 607                         ServerInstance->Logs->Log("m_mysql",DEFAULT,"SQL: Failed to connect database "+i->second->GetHost()+": Error: "+i->second->GetError());
 608                         i->second->SetEnable(false);
 609                         Parent->LoggingMutex.Unlock();
 610                 }
 611         }
 612 }
 613
 614 void LoadDatabases(ModuleSQL* Parent)
 615 {
 616         ConfigReader conf;
 617         Parent->ConnMutex.Lock();
 618         ClearOldConnections();
 619         for (int j =0; j < conf.Enumerate("database"); j++)
 620         {
 621                 SQLhost host;
 622                 host.id         = conf.ReadValue("database", "id", j);
 623                 host.host       = conf.ReadValue("database", "hostname", j);
 624                 host.port       = conf.ReadInteger("database", "port", j, true);
 625                 host.name       = conf.ReadValue("database", "name", j);
 626                 host.user       = conf.ReadValue("database", "username", j);
 627                 host.pass       = conf.ReadValue("database", "password", j);
 628                 host.ssl        = conf.ReadFlag("database", "ssl", j);
 629                 std::string initquery = conf.ReadValue("database", "initialquery", j);
 630
 631                 if (HasHost(host))
 632                         continue;
 633
 634                 if (!host.id.empty() && !host.host.empty() && !host.name.empty() && !host.user.empty() && !host.pass.empty())
 635                 {
 636                         SQLConnection* ThisSQL = new SQLConnection(host, Parent);
 637                         Connections[host.id] = ThisSQL;
 638
 639                         ThisSQL->setInitialQuery(initquery);
 640                 }
 641         }
 642         ConnectDatabases(Parent);
 643         Parent->ConnMutex.Unlock();
 644 }
 645
 646 char FindCharId(const std::string &id)
 647 {
 648         char i = 1;
 649         for (ConnMap::iterator iter = Connections.begin(); iter != Connections.end(); ++iter, ++i)
 650         {
 651                 if (iter->first == id)
 652                 {
 653                         return i;
 654                 }
 655         }
 656         return 0;
 657 }
 658
 659 ConnMap::iterator GetCharId(char id)
 660 {
 661         char i = 1;
 662         for (ConnMap::iterator iter = Connections.begin(); iter != Connections.end(); ++iter, ++i)
 663         {
 664                 if (i == id)
 665                         return iter;
 666         }
 667         return Connections.end();
 668 }
 669
 670 class DispatcherThread : public SocketThread
 671 {
 672  private:
 673         ModuleSQL* const Parent;
 674  public:
 675         DispatcherThread(ModuleSQL* CreatorModule) : Parent(CreatorModule) { }
 676         ~DispatcherThread() { }
 677         virtual void Run();
 678         virtual void OnNotify();
 679 };
 680
 681 ModuleSQL::ModuleSQL() : rehashing(false), sqlserv(this, "SQL/mysql", SERVICE_DATA)
 682 {
 683         currid = 0;
 684
 685         Dispatcher = new DispatcherThread(this);
 686         ServerInstance->Threads->Start(Dispatcher);
 687
 688         Implementation eventlist[] = { I_OnRehash };
 689         ServerInstance->Modules->Attach(eventlist, this, 1);
 690 }
 691
 692 ModuleSQL::~ModuleSQL()
 693 {
 694         delete Dispatcher;
 695         ClearAllConnections();
 696 }
 697
 698 unsigned long ModuleSQL::NewID()
 699 {
 700         if (currid+1 == 0)
 701                 currid++;
 702         return ++currid;
 703 }
 704
 705 void ModuleSQL::OnRequest(Request& request)
 706 {
 707         if(strcmp(SQLREQID, request.id) == 0)
 708         {
 709                 SQLrequest* req = (SQLrequest*)&request;
 710
 711                 ConnMap::iterator iter;
 712
 713                 Dispatcher->LockQueue();
 714                 ConnMutex.Lock();
 715                 if((iter = Connections.find(req->dbid)) != Connections.end())
 716                 {
 717                         req->id = NewID();
 718                         iter->second->queue.push(new SQLrequest(*req));
 719                 }
 720                 else
 721                 {
 722                         req->error.Id(SQL_BAD_DBID);
 723                 }
 724
 725                 ConnMutex.Unlock();
 726                 Dispatcher->UnlockQueueWakeup();
 727                 /* Yes, it's possible this will generate a spurious wakeup.
 728                  * That's fine, it'll just get ignored.
 729                  */
 730         }
 731 }
 732
 733 void ModuleSQL::OnRehash(User* user)
 734 {
 735         Dispatcher->LockQueue();
 736         rehashing = true;
 737         Dispatcher->UnlockQueueWakeup();
 738 }
 739
 740 Version ModuleSQL::GetVersion()
 741 {
 742         return Version("SQL Service Provider module for all other m_sql* modules", VF_VENDOR);
 743 }
 744
 745 void DispatcherThread::Run()
 746 {
 747         LoadDatabases(Parent);
 748
 749         SQLConnection* conn = NULL;
 750
 751         this->LockQueue();
 752         while (!this->GetExitFlag())
 753         {
 754                 if (Parent->rehashing)
 755                 {
 756                         Parent->rehashing = false;
 757                         LoadDatabases(Parent);
 758                 }
 759
 760                 conn = NULL;
 761                 Parent->ConnMutex.Lock();
 762                 for (ConnMap::iterator i = Connections.begin(); i != Connections.end(); i++)
 763                 {
 764                         if (i->second->queue.totalsize())
 765                         {
 766                                 conn = i->second;
 767                                 break;
 768                         }
 769                 }
 770                 Parent->ConnMutex.Unlock();
 771
 772                 if (conn)
 773                 {
 774                         /* There's an item! */
 775                         this->UnlockQueue();
 776                         conn->DoLeadingQuery();
 777                         this->NotifyParent();
 778                         this->LockQueue();
 779                         conn->queue.pop();
 780                 }
 781                 else
 782                 {
 783                         /* We know the queue is empty, we can safely hang this thread until
 784                          * something happens
 785                          */
 786                         this->WaitForQueue();
 787                 }
 788         }
 789         this->UnlockQueue();
 790 }
 791
 792 void DispatcherThread::OnNotify()
 793 {
 794         SQLConnection* conn;
 795         while (1)
 796         {
 797                 conn = NULL;
 798                 Parent->ConnMutex.Lock();
 799                 for (ConnMap::iterator iter = Connections.begin(); iter != Connections.end(); iter++)
 800                 {
 801                         if (!iter->second->rq.empty())
 802                         {
 803                                 conn = iter->second;
 804                                 break;
 805                         }
 806                 }
 807                 Parent->ConnMutex.Unlock();
 808
 809                 if (!conn)
 810                         break;
 811
 812                 Parent->ResultsMutex.Lock();
 813                 ResultQueue::iterator n = conn->rq.begin();
 814                 Parent->ResultsMutex.Unlock();
 815
 816                 (*n)->Send();
 817                 delete (*n);
 818
 819                 Parent->ResultsMutex.Lock();
 820                 conn->rq.pop_front();
 821                 Parent->ResultsMutex.Unlock();
 822         }
 823 }
 824
 825 MODULE_INIT(ModuleSQL)