* ---------------------------------------------------
*/
-#include <stdio.h>
-#include <string>
+#include "inspircd.h"
#include <mysql.h>
#include <pthread.h>
#include "users.h"
#include "channels.h"
#include "modules.h"
-#include "inspircd.h"
#include "m_sqlv2.h"
/* VERSION 2 API: With nonblocking (threaded) requests */
/* $ModDesc: SQL Service Provider module for all other m_sql* modules */
-/* $CompileFlags: `mysql_config --include` */
-/* $LinkerFlags: `mysql_config --libs_r` `perl extra/mysql_rpath.pl` */
+/* $CompileFlags: exec("mysql_config --include") */
+/* $LinkerFlags: exec("mysql_config --libs_r") rpath("mysql_config --libs_r") */
/* $ModDep: m_sqlv2.h */
/* THE NONBLOCKING MYSQL API!
typedef std::deque<SQLresult*> ResultQueue;
-/** Represents a mysql query queue
- */
-class QueryQueue : public classbase
-{
-private:
- typedef std::deque<SQLrequest> ReqDeque;
-
- ReqDeque priority; /* The priority queue */
- ReqDeque normal; /* The 'normal' queue */
- enum { PRI, NOR, NON } which; /* Which queue the currently active element is at the front of */
-
-public:
- QueryQueue()
- : which(NON)
- {
- }
-
- void push(const SQLrequest &q)
- {
- if(q.pri)
- priority.push_back(q);
- else
- normal.push_back(q);
- }
-
- void pop()
- {
- if((which == PRI) && priority.size())
- {
- priority.pop_front();
- }
- else if((which == NOR) && normal.size())
- {
- normal.pop_front();
- }
-
- /* Reset this */
- which = NON;
-
- /* Silently do nothing if there was no element to pop() */
- }
-
- SQLrequest& front()
- {
- switch(which)
- {
- case PRI:
- return priority.front();
- case NOR:
- return normal.front();
- default:
- if(priority.size())
- {
- which = PRI;
- return priority.front();
- }
-
- if(normal.size())
- {
- which = NOR;
- return normal.front();
- }
-
- /* This will probably result in a segfault,
- * but the caller should have checked totalsize()
- * first so..meh - moron :p
- */
-
- return priority.front();
- }
- }
-
- std::pair<int, int> size()
- {
- return std::make_pair(priority.size(), normal.size());
- }
-
- int totalsize()
- {
- return priority.size() + normal.size();
- }
-
- void PurgeModule(Module* mod)
- {
- DoPurgeModule(mod, priority);
- DoPurgeModule(mod, normal);
- }
-
-private:
- void DoPurgeModule(Module* mod, ReqDeque& q)
- {
- for(ReqDeque::iterator iter = q.begin(); iter != q.end(); iter++)
- {
- if(iter->GetSource() == mod)
- {
- if(iter->id == front().id)
- {
- /* It's the currently active query.. :x */
- iter->SetSource(NULL);
- }
- else
- {
- /* It hasn't been executed yet..just remove it */
- iter = q.erase(iter);
- }
- }
- }
- }
-};
-
/* A mutex to wrap around queue accesses */
pthread_mutex_t queue_mutex = PTHREAD_MUTEX_INITIALIZER;
class MySQLresult : public SQLresult
{
int currentrow;
- //std::vector<std::map<std::string,std::string> > results;
std::vector<std::string> colnames;
std::vector<SQLfieldList> fieldlists;
SQLfieldMap* fieldmap;
MYSQL connection;
MYSQL_RES *res;
MYSQL_ROW row;
- std::string host;
- std::string user;
- std::string pass;
- std::string db;
+ SQLhost host;
std::map<std::string,std::string> thisrow;
bool Enabled;
- std::string id;
public:
ResultQueue rq;
// This constructor creates an SQLConnection object with the given credentials, but does not connect yet.
- SQLConnection(const std::string &thishost, const std::string &thisuser, const std::string &thispass, const std::string &thisdb, const std::string &myid) : host(thishost), user(thisuser), pass(thispass), db(thisdb), Enabled(true), id(myid)
+ SQLConnection(const SQLhost &hi) : host(hi), Enabled(false)
{
}
unsigned int timeout = 1;
mysql_init(&connection);
mysql_options(&connection,MYSQL_OPT_CONNECT_TIMEOUT,(char*)&timeout);
- return mysql_real_connect(&connection, host.c_str(), user.c_str(), pass.c_str(), db.c_str(), 0, NULL, 0);
+ 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);
}
void DoLeadingQuery()
const std::string& GetID()
{
- return id;
+ return host.id;
}
std::string GetHost()
{
- return host;
+ return host.host;
}
void SetEnable(bool Enable)
mysql_close(&connection);
}
+ const SQLhost& GetConfHost()
+ {
+ return host;
+ }
+
};
ConnMap Connections;
+bool HasHost(const SQLhost &host)
+{
+ for (ConnMap::iterator iter = Connections.begin(); iter != Connections.end(); iter++)
+ {
+ if (host == iter->second->GetConfHost())
+ return true;
+ }
+ return false;
+}
+
+bool HostInConf(ConfigReader* conf, const SQLhost &h)
+{
+ for(int i = 0; i < conf->Enumerate("database"); i++)
+ {
+ SQLhost host;
+ host.id = conf->ReadValue("database", "id", i);
+ host.host = conf->ReadValue("database", "hostname", i);
+ host.port = conf->ReadInteger("database", "port", i, true);
+ host.name = conf->ReadValue("database", "name", i);
+ host.user = conf->ReadValue("database", "username", i);
+ host.pass = conf->ReadValue("database", "password", i);
+ host.ssl = conf->ReadFlag("database", "ssl", i);
+ if (h == host)
+ return true;
+ }
+ return false;
+}
+
+void ClearOldConnections(ConfigReader* conf)
+{
+ ConnMap::iterator i,safei;
+ for (i = Connections.begin(); i != Connections.end(); i++)
+ {
+ if (!HostInConf(conf, i->second->GetConfHost()))
+ {
+ DELETE(i->second);
+ safei = i;
+ --i;
+ Connections.erase(safei);
+ }
+ }
+}
+
+void ClearAllConnections()
+{
+ ConnMap::iterator i;
+ while ((i = Connections.begin()) != Connections.end())
+ {
+ Connections.erase(i);
+ DELETE(i->second);
+ }
+}
+
void ConnectDatabases(InspIRCd* ServerInstance)
{
for (ConnMap::iterator i = Connections.begin(); i != Connections.end(); i++)
{
+ if (i->second->IsEnabled())
+ continue;
+
i->second->SetEnable(true);
if (!i->second->Connect())
{
}
}
-void ClearDatabases()
+void LoadDatabases(ConfigReader* conf, InspIRCd* ServerInstance)
{
- ConnMap::iterator i;
- while ((i = Connections.begin()) != Connections.end())
- {
- Connections.erase(i);
- DELETE(i->second);
+ ClearOldConnections(conf);
+ for (int j =0; j < conf->Enumerate("database"); j++)
+ {
+ SQLhost host;
+ host.id = conf->ReadValue("database", "id", j);
+ host.host = conf->ReadValue("database", "hostname", j);
+ host.port = conf->ReadInteger("database", "port", j, true);
+ host.name = conf->ReadValue("database", "name", j);
+ host.user = conf->ReadValue("database", "username", j);
+ host.pass = conf->ReadValue("database", "password", j);
+ host.ssl = conf->ReadFlag("database", "ssl", j);
+
+ if (HasHost(host))
+ continue;
+
+ if (!host.id.empty() && !host.host.empty() && !host.name.empty() && !host.user.empty() && !host.pass.empty())
+ {
+ SQLConnection* ThisSQL = new SQLConnection(host);
+ Connections[host.id] = ThisSQL;
+ }
}
+ ConnectDatabases(ServerInstance);
}
-void LoadDatabases(ConfigReader* ThisConf, InspIRCd* ServerInstance)
+char FindCharId(const std::string &id)
{
- ClearDatabases();
- for (int j =0; j < ThisConf->Enumerate("database"); j++)
- {
- std::string db = ThisConf->ReadValue("database","name",j);
- std::string user = ThisConf->ReadValue("database","username",j);
- std::string pass = ThisConf->ReadValue("database","password",j);
- std::string host = ThisConf->ReadValue("database","hostname",j);
- std::string id = ThisConf->ReadValue("database","id",j);
- if ((db != "") && (host != "") && (user != "") && (id != "") && (pass != ""))
+ char i = 1;
+ for (ConnMap::iterator iter = Connections.begin(); iter != Connections.end(); ++iter, ++i)
+ {
+ if (iter->first == id)
{
- SQLConnection* ThisSQL = new SQLConnection(host,user,pass,db,id);
- Connections[id] = ThisSQL;
+ return i;
}
}
- ConnectDatabases(ServerInstance);
+ return 0;
+}
+
+ConnMap::iterator GetCharId(char id)
+{
+ char i = 1;
+ for (ConnMap::iterator iter = Connections.begin(); iter != Connections.end(); ++iter, ++i)
+ {
+ if (i == id)
+ return iter;
+ }
+ return Connections.end();
}
void NotifyMainThread(SQLConnection* connection_with_new_result)
* thread, we can just use standard connect(), and we can
* block if we like. We just send the connection id of the
* connection back.
+ *
+ * NOTE: We only send a single char down the connection, this
+ * way we know it wont get a partial read at the other end if
+ * the system is especially congested (see bug #263).
+ * The function FindCharId translates a connection name into a
+ * one character id, and GetCharId translates a character id
+ * back into an iterator.
*/
- send(QueueFD, connection_with_new_result->GetID().c_str(), connection_with_new_result->GetID().length()+1, 0);
+ char id = FindCharId(connection_with_new_result->GetID());
+ send(QueueFD, &id, 1, 0);
}
void* DispatcherThread(void* arg);
/** Used by m_mysql to notify one thread when the other has a result
*/
-class Notifier : public InspSocket
+class Notifier : public BufferedSocket
{
insp_sockaddr sock_us;
socklen_t uslen;
/* Create a socket on a random port. Let the tcp stack allocate us an available port */
#ifdef IPV6
- Notifier(InspIRCd* SI) : InspSocket(SI, "::1", 0, true, 3000)
+ Notifier(InspIRCd* SI) : BufferedSocket(SI, "::1", 0, true, 3000)
#else
- Notifier(InspIRCd* SI) : InspSocket(SI, "127.0.0.1", 0, true, 3000)
+ Notifier(InspIRCd* SI) : BufferedSocket(SI, "127.0.0.1", 0, true, 3000)
#endif
{
uslen = sizeof(sock_us);
}
}
- Notifier(InspIRCd* SI, int newfd, char* ip) : InspSocket(SI, newfd, ip)
+ Notifier(InspIRCd* SI, int newfd, char* ip) : BufferedSocket(SI, newfd, ip)
{
- Instance->Log(DEBUG,"Constructor of new socket");
}
/* Using getsockname and ntohs, we can determine which port number we were allocated */
virtual int OnIncomingConnection(int newsock, char* ip)
{
- Instance->Log(DEBUG,"Inbound connection on fd %d!",newsock);
Notifier* n = new Notifier(this->Instance, newsock, ip);
n = n; /* Stop bitching at me, GCC */
return true;
virtual bool OnDataReady()
{
- Instance->Log(DEBUG,"Inbound data!");
- char* data = this->Read();
- ConnMap::iterator iter;
-
- if (data && *data)
+ char data = 0;
+ /* NOTE: Only a single character is read so we know we
+ * cant get a partial read. (We've been told that theres
+ * data waiting, so we wont ever get EAGAIN)
+ * The function GetCharId translates a single character
+ * back into an iterator.
+ */
+ if (Instance->SE->Recv(this, &data, 1, 0) > 0)
{
- Instance->Log(DEBUG,"Looking for connection %s",data);
- /* We expect to be sent a null terminated string */
- if((iter = Connections.find(data)) != Connections.end())
+ ConnMap::iterator iter = GetCharId(data);
+ if (iter != Connections.end())
{
- Instance->Log(DEBUG,"Found it!");
-
/* Lock the mutex, send back the data */
pthread_mutex_lock(&results_mutex);
ResultQueue::iterator n = iter->second->rq.begin();
pthread_mutex_unlock(&results_mutex);
return true;
}
+ /* No error, but unknown id */
+ return true;
}
+ /* Erk, error on descriptor! */
return false;
}
};
InspIRCd* PublicServerInstance;
pthread_t Dispatcher;
int currid;
+ bool rehashing;
ModuleSQL(InspIRCd* Me)
- : Module::Module(Me)
+ : Module::Module(Me), rehashing(false)
{
- ServerInstance->UseInterface("SQLutils");
+ ServerInstance->Modules->UseInterface("SQLutils");
Conf = new ConfigReader(ServerInstance);
PublicServerInstance = ServerInstance;
SQLModule = this;
MessagePipe = new Notifier(ServerInstance);
- ServerInstance->Log(DEBUG,"Bound notifier to 127.0.0.1:%d",MessagePipe->GetPort());
pthread_attr_t attribs;
pthread_attr_init(&attribs);
throw ModuleException("m_mysql: Failed to create dispatcher thread: " + std::string(strerror(errno)));
}
- if (!ServerInstance->PublishFeature("SQL", this))
+ if (!ServerInstance->Modules->PublishFeature("SQL", this))
{
/* Tell worker thread to exit NOW */
giveup = true;
throw ModuleException("m_mysql: Unable to publish feature 'SQL'");
}
- ServerInstance->PublishInterface("SQL", this);
+ ServerInstance->Modules->PublishInterface("SQL", this);
}
virtual ~ModuleSQL()
{
giveup = true;
- ClearDatabases();
- DELETE(Conf);
- ServerInstance->UnpublishInterface("SQL", this);
- ServerInstance->UnpublishFeature("SQL");
- ServerInstance->DoneWithInterface("SQLutils");
+ ClearAllConnections();
+ delete Conf;
+ ServerInstance->Modules->UnpublishInterface("SQL", this);
+ ServerInstance->Modules->UnpublishFeature("SQL");
+ ServerInstance->Modules->DoneWithInterface("SQLutils");
}
char* returnval = NULL;
- ServerInstance->Log(DEBUG, "Got query: '%s' with %d replacement parameters on id '%s'", req->query.q.c_str(), req->query.p.size(), req->dbid.c_str());
-
if((iter = Connections.find(req->dbid)) != Connections.end())
{
req->id = NewID();
return returnval;
}
- ServerInstance->Log(DEBUG, "Got unsupported API version string: %s", request->GetId());
-
return NULL;
}
- virtual void OnRehash(const std::string ¶meter)
+ virtual void OnRehash(User* user, const std::string ¶meter)
{
- /* TODO: set rehash bool here, which makes the dispatcher thread rehash at next opportunity */
+ rehashing = true;
}
virtual Version GetVersion()
while (!giveup)
{
+ if (thismodule->rehashing)
+ {
+ /* XXX: Lock */
+ pthread_mutex_lock(&queue_mutex);
+ thismodule->rehashing = false;
+ LoadDatabases(thismodule->Conf, thismodule->PublicServerInstance);
+ pthread_mutex_unlock(&queue_mutex);
+ /* XXX: Unlock */
+ }
+
SQLConnection* conn = NULL;
/* XXX: Lock here for safety */
pthread_mutex_lock(&queue_mutex);
return NULL;
}
-
-// stuff down here is the module-factory stuff. For basic modules you can ignore this.
-
-class ModuleSQLFactory : public ModuleFactory
-{
- public:
- ModuleSQLFactory()
- {
- }
-
- ~ModuleSQLFactory()
- {
- }
-
- virtual Module * CreateModule(InspIRCd* Me)
- {
- return new ModuleSQL(Me);
- }
-
-};
-
-
-extern "C" void * init_module( void )
-{
- return new ModuleSQLFactory;
-}
+MODULE_INIT(ModuleSQL)
projects / user / henk / code / inspircd.git / blobdiff