Add connmutex to mutex the connections vector, otherwise we can get access from two...

[user/henk/code/inspircd.git] / src / modules / extra / m_mysql.cpp

diff --git a/src/modules/extra/m_mysql.cpp b/src/modules/extra/m_mysql.cpp
index 5e5e97113c11eb3685d3d192160e13aa96bed49f..60a2edbcaf6ff9159502798d2728400ed9546e5d 100644 (file)
--- a/src/modules/extra/m_mysql.cpp
+++ b/src/modules/extra/m_mysql.cpp
@@ -68,11 +68,11 @@
 
 
 class SQLConnection;
-class Notifier;
+class MySQLListener;
 
 
 typedef std::map<std::string, SQLConnection*> ConnMap;
-static Notifier* MessagePipe = NULL;
+static MySQLListener *MessagePipe = NULL;
 int QueueFD = -1;
 
 class DispatcherThread;
@@ -91,6 +91,7 @@ class ModuleSQL : public Module
         Mutex* QueueMutex;
         Mutex* ResultsMutex;
         Mutex* LoggingMutex;
+        Mutex* ConnMutex;
 
         ModuleSQL(InspIRCd* Me);
         ~ModuleSQL();
@@ -574,6 +575,7 @@ void ConnectDatabases(InspIRCd* ServerInstance, ModuleSQL* Parent)
 
 void LoadDatabases(ConfigReader* conf, InspIRCd* ServerInstance, ModuleSQL* Parent)
 {
+       Parent->ConnMutex->Lock();
        ClearOldConnections(conf);
        for (int j =0; j < conf->Enumerate("database"); j++)
        {
@@ -596,6 +598,7 @@ void LoadDatabases(ConfigReader* conf, InspIRCd* ServerInstance, ModuleSQL* Pare
                }
        }
        ConnectDatabases(ServerInstance, Parent);
+       Parent->ConnMutex->Unlock();
 }
 
 char FindCharId(const std::string &id)
@@ -661,46 +664,10 @@ class DispatcherThread : public Thread
  */
 class Notifier : public BufferedSocket
 {
-       insp_sockaddr sock_us;
-       socklen_t uslen;
        ModuleSQL* Parent;
 
  public:
-
-       /* Create a socket on a random port. Let the tcp stack allocate us an available port */
-#ifdef IPV6
-       Notifier(InspIRCd* SI, ModuleSQL* Creator) : BufferedSocket(SI, "::1", 0, true, 3000), Parent(Creator)
-#else
-       Notifier(InspIRCd* SI, ModuleSQL* Creator) : BufferedSocket(SI, "127.0.0.1", 0, true, 3000), Parent(Creator)
-#endif
-       {
-               uslen = sizeof(sock_us);
-               if (getsockname(this->fd,(sockaddr*)&sock_us,&uslen))
-               {
-                       throw ModuleException("Could not create random listening port on localhost");
-               }
-       }
-
-       Notifier(InspIRCd* SI, int newfd, char* ip) : BufferedSocket(SI, newfd, ip)
-       {
-       }
-
-       /* Using getsockname and ntohs, we can determine which port number we were allocated */
-       int GetPort()
-       {
-#ifdef IPV6
-               return ntohs(sock_us.sin6_port);
-#else
-               return ntohs(sock_us.sin_port);
-#endif
-       }
-
-       virtual int OnIncomingConnection(int newsock, char* ip)
-       {
-               Notifier* n = new Notifier(this->Instance, newsock, ip);
-               n = n; /* Stop bitching at me, GCC */
-               return true;
-       }
+       Notifier(ModuleSQL* P, InspIRCd* SI, int newfd, char* ip) : BufferedSocket(SI, newfd, ip), Parent(P) { }
 
        virtual bool OnDataReady()
        {
@@ -711,8 +678,10 @@ class Notifier : public BufferedSocket
                 * The function GetCharId translates a single character
                 * back into an iterator.
                 */
+
                if (Instance->SE->Recv(this, &data, 1, 0) > 0)
                {
+                       Parent->ConnMutex->Lock();
                        ConnMap::iterator iter = GetCharId(data);
                        if (iter != Connections.end())
                        {
@@ -723,9 +692,11 @@ class Notifier : public BufferedSocket
                                delete (*n);
                                iter->second->rq.pop_front();
                                Parent->ResultsMutex->Unlock();
+                               Parent->ConnMutex->Unlock();
                                return true;
                        }
                        /* No error, but unknown id */
+                       Parent->ConnMutex->Unlock();
                        return true;
                }
 
@@ -734,6 +705,40 @@ class Notifier : public BufferedSocket
        }
 };
 
+/** Spawn sockets from a listener
+ */
+class MySQLListener : public ListenSocketBase
+{
+       ModuleSQL* Parent;
+       insp_sockaddr sock_us;
+       socklen_t uslen;
+       FileReader* index;
+
+ public:
+       MySQLListener(ModuleSQL* P, InspIRCd* Instance, int port, const std::string &addr) : ListenSocketBase(Instance, port, addr), Parent(P)
+       {
+               uslen = sizeof(sock_us);
+               if (getsockname(this->fd,(sockaddr*)&sock_us,&uslen))
+               {
+                       throw ModuleException("Could not getsockname() to find out port number for ITC port");
+               }
+       }
+
+       virtual void OnAcceptReady(const std::string &ipconnectedto, int nfd, const std::string &incomingip)
+       {
+               new Notifier(this->Parent, this->ServerInstance, nfd, (char *)ipconnectedto.c_str()); // XXX unsafe casts suck
+       }
+
+       /* Using getsockname and ntohs, we can determine which port number we were allocated */
+       int GetPort()
+       {
+#ifdef IPV6
+               return ntohs(sock_us.sin6_port);
+#else
+               return ntohs(sock_us.sin_port);
+#endif
+       }
+};
 
 ModuleSQL::ModuleSQL(InspIRCd* Me) : Module(Me), rehashing(false)
 {
@@ -743,12 +748,31 @@ ModuleSQL::ModuleSQL(InspIRCd* Me) : Module(Me), rehashing(false)
        PublicServerInstance = ServerInstance;
        currid = 0;
 
-       MessagePipe = new Notifier(ServerInstance, this);
+       /* Create a socket on a random port. Let the tcp stack allocate us an available port */
+#ifdef IPV6
+       MessagePipe = new MySQLListener(this, ServerInstance, 0, "::1");
+#else
+       MessagePipe = new MySQLListener(this, ServerInstance, 0, "127.0.0.1");
+#endif
+
+       LoggingMutex = ServerInstance->Mutexes->CreateMutex();
+       ConnMutex = ServerInstance->Mutexes->CreateMutex();
+
+       if (MessagePipe->GetFd() == -1)
+       {
+               delete ConnMutex;
+               throw ModuleException("m_mysql: unable to create ITC pipe");
+       }
+       else
+       {
+               Parent->LoggingMutex->Lock();
+               ServerInstance->Logs->Log("m_mysql", DEBUG, "MySQL: Interthread comms port is %d", MessagePipe->GetPort());
+               Parent->LoggingMutex->Unlock();
+       }
 
        Dispatcher = new DispatcherThread(ServerInstance, this);
        ServerInstance->Threads->Create(Dispatcher);
 
-       LoggingMutex = ServerInstance->Mutexes->CreateMutex();
        ResultsMutex = ServerInstance->Mutexes->CreateMutex();
        QueueMutex = ServerInstance->Mutexes->CreateMutex();
 
@@ -776,6 +800,7 @@ ModuleSQL::~ModuleSQL()
        delete LoggingMutex;
        delete ResultsMutex;
        delete QueueMutex;
+       delete ConnMutex;
 }
 
 unsigned long ModuleSQL::NewID()
@@ -798,6 +823,7 @@ const char* ModuleSQL::OnRequest(Request* request)
 
                const char* returnval = NULL;
 
+               Parent->ConnMutex->Lock();
                if((iter = Connections.find(req->dbid)) != Connections.end())
                {
                        req->id = NewID();
@@ -809,8 +835,8 @@ const char* ModuleSQL::OnRequest(Request* request)
                        req->error.Id(SQL_BAD_DBID);
                }
 
+               Parent->ConnMutex->Unlock();
                QueueMutex->Unlock();
-               /* XXX: Unlock */
 
                return returnval;
        }
@@ -875,6 +901,7 @@ void DispatcherThread::Run()
                SQLConnection* conn = NULL;
                /* XXX: Lock here for safety */
                Parent->QueueMutex->Lock();
+               Parent->ConnMutex->Lock();
                for (ConnMap::iterator i = Connections.begin(); i != Connections.end(); i++)
                {
                        if (i->second->queue.totalsize())
@@ -883,6 +910,7 @@ void DispatcherThread::Run()
                                break;
                        }
                }
+               Parent->ConnMutex->Unlock();
                Parent->QueueMutex->Unlock();
                /* XXX: Unlock */