src/modules/extra/m_mysql.cpp

   1 /*
   2  * InspIRCd -- Internet Relay Chat Daemon
   3  *
   4  *   Copyright (C) 2009-2010 Daniel De Graaf <danieldg@inspircd.org>
   5  *   Copyright (C) 2006-2007, 2009 Dennis Friis <peavey@inspircd.org>
   6  *   Copyright (C) 2006-2009 Craig Edwards <craigedwards@brainbox.cc>
   7  *   Copyright (C) 2008 Robin Burchell <robin+git@viroteck.net>
   8  *
   9  * This file is part of InspIRCd.  InspIRCd is free software: you can
  10  * redistribute it and/or modify it under the terms of the GNU General Public
  11  * License as published by the Free Software Foundation, version 2.
  12  *
  13  * This program is distributed in the hope that it will be useful, but WITHOUT
  14  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
  15  * FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
  16  * details.
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  20  */
  21
  22 /// $CompilerFlags: execute("mysql_config --include" "MYSQL_CXXFLAGS")
  23 /// $LinkerFlags: execute("mysql_config --libs_r" "MYSQL_LDFLAGS" "-lmysqlclient")
  24
  25 /// $PackageInfo: require_system("arch") mariadb-libs
  26 /// $PackageInfo: require_system("centos" "6.0" "6.99") mysql-devel
  27 /// $PackageInfo: require_system("centos" "7.0") mariadb-devel
  28 /// $PackageInfo: require_system("darwin") mysql-connector-c
  29 /// $PackageInfo: require_system("debian") libmysqlclient-dev
  30 /// $PackageInfo: require_system("ubuntu") libmysqlclient-dev
  31
  32 #ifdef __GNUC__
  33 # pragma GCC diagnostic push
  34 #endif
  35
  36 // Fix warnings about the use of `long long` on C++03.
  37 #if defined __clang__
  38 # pragma clang diagnostic ignored "-Wc++11-long-long"
  39 #elif defined __GNUC__
  40 # pragma GCC diagnostic ignored "-Wlong-long"
  41 #endif
  42
  43 #include "inspircd.h"
  44 #include <mysql.h>
  45 #include "modules/sql.h"
  46
  47 #ifdef __GNUC__
  48 # pragma GCC diagnostic pop
  49 #endif
  50
  51 #ifdef _WIN32
  52 # pragma comment(lib, "libmysql.lib")
  53 #endif
  54
  55 /* VERSION 3 API: With nonblocking (threaded) requests */
  56
  57 /* THE NONBLOCKING MYSQL API!
  58  *
  59  * MySQL provides no nonblocking (asyncronous) API of its own, and its developers recommend
  60  * that instead, you should thread your program. This is what i've done here to allow for
  61  * asyncronous SQL requests via mysql. The way this works is as follows:
  62  *
  63  * The module spawns a thread via class Thread, and performs its mysql queries in this thread,
  64  * using a queue with priorities. There is a mutex on either end which prevents two threads
  65  * adjusting the queue at the same time, and crashing the ircd. Every 50 milliseconds, the
  66  * worker thread wakes up, and checks if there is a request at the head of its queue.
  67  * If there is, it processes this request, blocking the worker thread but leaving the ircd
  68  * thread to go about its business as usual. During this period, the ircd thread is able
  69  * to insert futher pending requests into the queue.
  70  *
  71  * Once the processing of a request is complete, it is removed from the incoming queue to
  72  * an outgoing queue, and initialized as a 'response'. The worker thread then signals the
  73  * ircd thread (via a loopback socket) of the fact a result is available, by sending the
  74  * connection ID through the connection.
  75  *
  76  * The ircd thread then mutexes the queue once more, reads the outbound response off the head
  77  * of the queue, and sends it on its way to the original calling module.
  78  *
  79  * XXX: You might be asking "why doesnt it just send the response from within the worker thread?"
  80  * The answer to this is simple. The majority of InspIRCd, and in fact most ircd's are not
  81  * threadsafe. This module is designed to be threadsafe and is careful with its use of threads,
  82  * however, if we were to call a module's OnRequest even from within a thread which was not the
  83  * one the module was originally instantiated upon, there is a chance of all hell breaking loose
  84  * if a module is ever put in a re-enterant state (stack corruption could occur, crashes, data
  85  * corruption, and worse, so DONT think about it until the day comes when InspIRCd is 100%
  86  * gauranteed threadsafe!)
  87  */
  88
  89 class SQLConnection;
  90 class MySQLresult;
  91 class DispatcherThread;
  92
  93 struct QQueueItem
  94 {
  95         SQL::Query* q;
  96         std::string query;
  97         SQLConnection* c;
  98         QQueueItem(SQL::Query* Q, const std::string& S, SQLConnection* C) : q(Q), query(S), c(C) {}
  99 };
 100
 101 struct RQueueItem
 102 {
 103         SQL::Query* q;
 104         MySQLresult* r;
 105         RQueueItem(SQL::Query* Q, MySQLresult* R) : q(Q), r(R) {}
 106 };
 107
 108 typedef insp::flat_map<std::string, SQLConnection*> ConnMap;
 109 typedef std::deque<QQueueItem> QueryQueue;
 110 typedef std::deque<RQueueItem> ResultQueue;
 111
 112 /** MySQL module
 113  *  */
 114 class ModuleSQL : public Module
 115 {
 116  public:
 117         DispatcherThread* Dispatcher;
 118         QueryQueue qq;       // MUST HOLD MUTEX
 119         ResultQueue rq;      // MUST HOLD MUTEX
 120         ConnMap connections; // main thread only
 121
 122         ModuleSQL();
 123         void init() CXX11_OVERRIDE;
 124         ~ModuleSQL();
 125         void ReadConfig(ConfigStatus& status) CXX11_OVERRIDE;
 126         void OnUnloadModule(Module* mod) CXX11_OVERRIDE;
 127         Version GetVersion() CXX11_OVERRIDE;
 128 };
 129
 130 class DispatcherThread : public SocketThread
 131 {
 132  private:
 133         ModuleSQL* const Parent;
 134  public:
 135         DispatcherThread(ModuleSQL* CreatorModule) : Parent(CreatorModule) { }
 136         ~DispatcherThread() { }
 137         void Run() CXX11_OVERRIDE;
 138         void OnNotify() CXX11_OVERRIDE;
 139 };
 140
 141 #if !defined(MYSQL_VERSION_ID) || MYSQL_VERSION_ID<32224
 142 #define mysql_field_count mysql_num_fields
 143 #endif
 144
 145 /** Represents a mysql result set
 146  */
 147 class MySQLresult : public SQL::Result
 148 {
 149  public:
 150         SQL::Error err;
 151         int currentrow;
 152         int rows;
 153         std::vector<std::string> colnames;
 154         std::vector<SQL::Row> fieldlists;
 155
 156         MySQLresult(MYSQL_RES* res, int affected_rows) : err(SQL::SUCCESS), currentrow(0), rows(0)
 157         {
 158                 if (affected_rows >= 1)
 159                 {
 160                         rows = affected_rows;
 161                         fieldlists.resize(rows);
 162                 }
 163                 unsigned int field_count = 0;
 164                 if (res)
 165                 {
 166                         MYSQL_ROW row;
 167                         int n = 0;
 168                         while ((row = mysql_fetch_row(res)))
 169                         {
 170                                 if (fieldlists.size() < (unsigned int)rows+1)
 171                                 {
 172                                         fieldlists.resize(fieldlists.size()+1);
 173                                 }
 174                                 field_count = 0;
 175                                 MYSQL_FIELD *fields = mysql_fetch_fields(res);
 176                                 if(mysql_num_fields(res) == 0)
 177                                         break;
 178                                 if (fields && mysql_num_fields(res))
 179                                 {
 180                                         colnames.clear();
 181                                         while (field_count < mysql_num_fields(res))
 182                                         {
 183                                                 std::string a = (fields[field_count].name ? fields[field_count].name : "");
 184                                                 if (row[field_count])
 185                                                         fieldlists[n].push_back(SQL::Field(row[field_count]));
 186                                                 else
 187                                                         fieldlists[n].push_back(SQL::Field());
 188                                                 colnames.push_back(a);
 189                                                 field_count++;
 190                                         }
 191                                         n++;
 192                                 }
 193                                 rows++;
 194                         }
 195                         mysql_free_result(res);
 196                 }
 197         }
 198
 199         MySQLresult(SQL::Error& e) : err(e)
 200         {
 201
 202         }
 203
 204         int Rows() CXX11_OVERRIDE
 205         {
 206                 return rows;
 207         }
 208
 209         void GetCols(std::vector<std::string>& result) CXX11_OVERRIDE
 210         {
 211                 result.assign(colnames.begin(), colnames.end());
 212         }
 213
 214         bool HasColumn(const std::string& column, size_t& index) CXX11_OVERRIDE
 215         {
 216                 for (size_t i = 0; i < colnames.size(); ++i)
 217                 {
 218                         if (colnames[i] == column)
 219                         {
 220                                 index = i;
 221                                 return true;
 222                         }
 223                 }
 224                 return false;
 225         }
 226
 227         SQL::Field GetValue(int row, int column)
 228         {
 229                 if ((row >= 0) && (row < rows) && (column >= 0) && (column < (int)fieldlists[row].size()))
 230                 {
 231                         return fieldlists[row][column];
 232                 }
 233                 return SQL::Field();
 234         }
 235
 236         bool GetRow(SQL::Row& result) CXX11_OVERRIDE
 237         {
 238                 if (currentrow < rows)
 239                 {
 240                         result.assign(fieldlists[currentrow].begin(), fieldlists[currentrow].end());
 241                         currentrow++;
 242                         return true;
 243                 }
 244                 else
 245                 {
 246                         result.clear();
 247                         return false;
 248                 }
 249         }
 250 };
 251
 252 /** Represents a connection to a mysql database
 253  */
 254 class SQLConnection : public SQL::Provider
 255 {
 256  private:
 257         bool EscapeString(SQL::Query* query, const std::string& in, std::string& out)
 258         {
 259                 // In the worst case each character may need to be encoded as using two bytes and one
 260                 // byte is the NUL terminator.
 261                 std::vector<char> buffer(in.length() * 2 + 1);
 262
 263                 // The return value of mysql_escape_string() is either an error or the length of the
 264                 // encoded string not including the NUL terminator.
 265                 //
 266                 // Unfortunately, someone genius decided that mysql_escape_string should return an
 267                 // unsigned type even though -1 is returned on error so checking whether an error
 268                 // happened is a bit cursed.
 269                 unsigned long escapedsize = mysql_escape_string(&buffer[0], in.c_str(), in.length());
 270                 if (escapedsize == static_cast<unsigned long>(-1))
 271                 {
 272                         SQL::Error err(SQL::QSEND_FAIL, InspIRCd::Format("%u: %s", mysql_errno(connection), mysql_error(connection)));
 273                         query->OnError(err);
 274                         return false;
 275                 }
 276
 277                 out.append(&buffer[0], escapedsize);
 278                 return true;
 279         }
 280
 281  public:
 282         reference<ConfigTag> config;
 283         MYSQL *connection;
 284         Mutex lock;
 285
 286         // This constructor creates an SQLConnection object with the given credentials, but does not connect yet.
 287         SQLConnection(Module* p, ConfigTag* tag)
 288                 : SQL::Provider(p, tag->getString("id"))
 289                 , config(tag)
 290                 , connection(NULL)
 291         {
 292         }
 293
 294         ~SQLConnection()
 295         {
 296                 Close();
 297         }
 298
 299         // This method connects to the database using the credentials supplied to the constructor, and returns
 300         // true upon success.
 301         bool Connect()
 302         {
 303                 unsigned int timeout = 1;
 304                 connection = mysql_init(connection);
 305                 mysql_options(connection,MYSQL_OPT_CONNECT_TIMEOUT,(char*)&timeout);
 306                 std::string host = config->getString("host");
 307                 std::string user = config->getString("user");
 308                 std::string pass = config->getString("pass");
 309                 std::string dbname = config->getString("name");
 310                 unsigned int port = config->getUInt("port", 3306);
 311                 bool rv = mysql_real_connect(connection, host.c_str(), user.c_str(), pass.c_str(), dbname.c_str(), port, NULL, 0);
 312                 if (!rv)
 313                         return rv;
 314
 315                 // Enable character set settings
 316                 std::string charset = config->getString("charset");
 317                 if ((!charset.empty()) && (mysql_set_character_set(connection, charset.c_str())))
 318                         ServerInstance->Logs->Log(MODNAME, LOG_DEFAULT, "WARNING: Could not set character set to \"%s\"", charset.c_str());
 319
 320                 std::string initquery;
 321                 if (config->readString("initialquery", initquery))
 322                 {
 323                         mysql_query(connection,initquery.c_str());
 324                 }
 325                 return true;
 326         }
 327
 328         ModuleSQL* Parent()
 329         {
 330                 return (ModuleSQL*)(Module*)creator;
 331         }
 332
 333         MySQLresult* DoBlockingQuery(const std::string& query)
 334         {
 335
 336                 /* Parse the command string and dispatch it to mysql */
 337                 if (CheckConnection() && !mysql_real_query(connection, query.data(), query.length()))
 338                 {
 339                         /* Successfull query */
 340                         MYSQL_RES* res = mysql_use_result(connection);
 341                         unsigned long rows = mysql_affected_rows(connection);
 342                         return new MySQLresult(res, rows);
 343                 }
 344                 else
 345                 {
 346                         /* XXX: See /usr/include/mysql/mysqld_error.h for a list of
 347                          * possible error numbers and error messages */
 348                         SQL::Error e(SQL::QREPLY_FAIL, InspIRCd::Format("%u: %s", mysql_errno(connection), mysql_error(connection)));
 349                         return new MySQLresult(e);
 350                 }
 351         }
 352
 353         bool CheckConnection()
 354         {
 355                 if (!connection || mysql_ping(connection) != 0)
 356                         return Connect();
 357                 return true;
 358         }
 359
 360         std::string GetError()
 361         {
 362                 return mysql_error(connection);
 363         }
 364
 365         void Close()
 366         {
 367                 mysql_close(connection);
 368         }
 369
 370         void Submit(SQL::Query* q, const std::string& qs) CXX11_OVERRIDE
 371         {
 372                 ServerInstance->Logs->Log(MODNAME, LOG_DEBUG, "Executing MySQL query: " + qs);
 373                 Parent()->Dispatcher->LockQueue();
 374                 Parent()->qq.push_back(QQueueItem(q, qs, this));
 375                 Parent()->Dispatcher->UnlockQueueWakeup();
 376         }
 377
 378         void Submit(SQL::Query* call, const std::string& q, const SQL::ParamList& p) CXX11_OVERRIDE
 379         {
 380                 std::string res;
 381                 unsigned int param = 0;
 382                 for(std::string::size_type i = 0; i < q.length(); i++)
 383                 {
 384                         if (q[i] != '?')
 385                                 res.push_back(q[i]);
 386                         else if (param < p.size() && !EscapeString(call, p[param++], res))
 387                                 return;
 388                 }
 389                 Submit(call, res);
 390         }
 391
 392         void Submit(SQL::Query* call, const std::string& q, const SQL::ParamMap& p) CXX11_OVERRIDE
 393         {
 394                 std::string res;
 395                 for(std::string::size_type i = 0; i < q.length(); i++)
 396                 {
 397                         if (q[i] != '$')
 398                                 res.push_back(q[i]);
 399                         else
 400                         {
 401                                 std::string field;
 402                                 i++;
 403                                 while (i < q.length() && isalnum(q[i]))
 404                                         field.push_back(q[i++]);
 405                                 i--;
 406
 407                                 SQL::ParamMap::const_iterator it = p.find(field);
 408                                 if (it != p.end() && !EscapeString(call, it->second, res))
 409                                         return;
 410                         }
 411                 }
 412                 Submit(call, res);
 413         }
 414 };
 415
 416 ModuleSQL::ModuleSQL()
 417 {
 418         Dispatcher = NULL;
 419 }
 420
 421 void ModuleSQL::init()
 422 {
 423         if (mysql_library_init(0, NULL, NULL))
 424                 throw ModuleException("Unable to initialise the MySQL library!");
 425
 426         Dispatcher = new DispatcherThread(this);
 427         ServerInstance->Threads.Start(Dispatcher);
 428 }
 429
 430 ModuleSQL::~ModuleSQL()
 431 {
 432         if (Dispatcher)
 433         {
 434                 Dispatcher->join();
 435                 Dispatcher->OnNotify();
 436                 delete Dispatcher;
 437         }
 438
 439         for(ConnMap::iterator i = connections.begin(); i != connections.end(); i++)
 440         {
 441                 delete i->second;
 442         }
 443
 444         mysql_library_end();
 445 }
 446
 447 void ModuleSQL::ReadConfig(ConfigStatus& status)
 448 {
 449         ConnMap conns;
 450         ConfigTagList tags = ServerInstance->Config->ConfTags("database");
 451         for(ConfigIter i = tags.first; i != tags.second; i++)
 452         {
 453                 if (!stdalgo::string::equalsci(i->second->getString("module"), "mysql"))
 454                         continue;
 455                 std::string id = i->second->getString("id");
 456                 ConnMap::iterator curr = connections.find(id);
 457                 if (curr == connections.end())
 458                 {
 459                         SQLConnection* conn = new SQLConnection(this, i->second);
 460                         conns.insert(std::make_pair(id, conn));
 461                         ServerInstance->Modules->AddService(*conn);
 462                 }
 463                 else
 464                 {
 465                         conns.insert(*curr);
 466                         connections.erase(curr);
 467                 }
 468         }
 469
 470         // now clean up the deleted databases
 471         Dispatcher->LockQueue();
 472         SQL::Error err(SQL::BAD_DBID);
 473         for(ConnMap::iterator i = connections.begin(); i != connections.end(); i++)
 474         {
 475                 ServerInstance->Modules->DelService(*i->second);
 476                 // it might be running a query on this database. Wait for that to complete
 477                 i->second->lock.Lock();
 478                 i->second->lock.Unlock();
 479                 // now remove all active queries to this DB
 480                 for (size_t j = qq.size(); j > 0; j--)
 481                 {
 482                         size_t k = j - 1;
 483                         if (qq[k].c == i->second)
 484                         {
 485                                 qq[k].q->OnError(err);
 486                                 delete qq[k].q;
 487                                 qq.erase(qq.begin() + k);
 488                         }
 489                 }
 490                 // finally, nuke the connection
 491                 delete i->second;
 492         }
 493         Dispatcher->UnlockQueue();
 494         connections.swap(conns);
 495 }
 496
 497 void ModuleSQL::OnUnloadModule(Module* mod)
 498 {
 499         SQL::Error err(SQL::BAD_DBID);
 500         Dispatcher->LockQueue();
 501         unsigned int i = qq.size();
 502         while (i > 0)
 503         {
 504                 i--;
 505                 if (qq[i].q->creator == mod)
 506                 {
 507                         if (i == 0)
 508                         {
 509                                 // need to wait until the query is done
 510                                 // (the result will be discarded)
 511                                 qq[i].c->lock.Lock();
 512                                 qq[i].c->lock.Unlock();
 513                         }
 514                         qq[i].q->OnError(err);
 515                         delete qq[i].q;
 516                         qq.erase(qq.begin() + i);
 517                 }
 518         }
 519         Dispatcher->UnlockQueue();
 520         // clean up any result queue entries
 521         Dispatcher->OnNotify();
 522 }
 523
 524 Version ModuleSQL::GetVersion()
 525 {
 526         return Version("Provides MySQL support", VF_VENDOR);
 527 }
 528
 529 void DispatcherThread::Run()
 530 {
 531         this->LockQueue();
 532         while (!this->GetExitFlag())
 533         {
 534                 if (!Parent->qq.empty())
 535                 {
 536                         QQueueItem i = Parent->qq.front();
 537                         i.c->lock.Lock();
 538                         this->UnlockQueue();
 539                         MySQLresult* res = i.c->DoBlockingQuery(i.query);
 540                         i.c->lock.Unlock();
 541
 542                         /*
 543                          * At this point, the main thread could be working on:
 544                          *  Rehash - delete i.c out from under us. We don't care about that.
 545                          *  UnloadModule - delete i.q and the qq item. Need to avoid reporting results.
 546                          */
 547
 548                         this->LockQueue();
 549                         if (!Parent->qq.empty() && Parent->qq.front().q == i.q)
 550                         {
 551                                 Parent->qq.pop_front();
 552                                 Parent->rq.push_back(RQueueItem(i.q, res));
 553                                 NotifyParent();
 554                         }
 555                         else
 556                         {
 557                                 // UnloadModule ate the query
 558                                 delete res;
 559                         }
 560                 }
 561                 else
 562                 {
 563                         /* We know the queue is empty, we can safely hang this thread until
 564                          * something happens
 565                          */
 566                         this->WaitForQueue();
 567                 }
 568         }
 569         this->UnlockQueue();
 570 }
 571
 572 void DispatcherThread::OnNotify()
 573 {
 574         // this could unlock during the dispatch, but OnResult isn't expected to take that long
 575         this->LockQueue();
 576         for(ResultQueue::iterator i = Parent->rq.begin(); i != Parent->rq.end(); i++)
 577         {
 578                 MySQLresult* res = i->r;
 579                 if (res->err.code == SQL::SUCCESS)
 580                         i->q->OnResult(*res);
 581                 else
 582                         i->q->OnError(res->err);
 583                 delete i->q;
 584                 delete i->r;
 585         }
 586         Parent->rq.clear();
 587         this->UnlockQueue();
 588 }
 589
 590 MODULE_INIT(ModuleSQL)