src/modules/extra/m_mysql.cpp

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