ThreadEngine: Allow interthread signaling without needing as many hacks

git-svn-id: http://svn.inspircd.org/repository/trunk/inspircd@11251 e03df62e-2008-0410-955e-edbf42e46eb7

7 files changed, 451 insertions, 357 deletions

diff --git a/include/threadengine.h b/include/threadengine.h
index 27060ebe9..d7eaaef69 100644
--- a/include/threadengine.h
+++ b/include/threadengine.h
@@ -30,6 +30,15 @@ class CoreExport Thread : public Extensible
 	/** Set to true when the thread is to exit
 	 */
 	bool ExitFlag;
+// TODO protected:
+ public:
+	/** Get thread's current exit status.
+	 * (are we being asked to exit?)
+	 */
+	bool GetExitFlag()
+	{
+		return ExitFlag;
+	}
  public:
 	/** Opaque thread state managed by threading engine
 	 */
@@ -60,18 +69,108 @@ class CoreExport Thread : public Extensible
 
 	/** Signal the thread to exit gracefully.
 	 */
-	void SetExitFlag(bool value)
+	virtual void SetExitFlag()
 	{
-		ExitFlag = value;
+		ExitFlag = true;
 	}
+};
 
-	/** Get thread's current exit status.
-	 * (are we being asked to exit?)
+
+class CoreExport QueuedThread : public Thread
+{
+	ThreadQueueData queue;
+ protected:
+	/** Waits for an enqueue operation to complete
+	 * You MUST hold the queue lock when you call this.
+	 * It will be unlocked while you wait, and will be relocked
+	 * before the function returns
 	 */
-	bool GetExitFlag()
+	void WaitForQueue()
 	{
-		return ExitFlag;
+		queue.Wait();
+	}
+ public:
+	/** Lock queue.
+	 */
+	void LockQueue()
+	{
+		queue.Lock();
+	}
+	/** Unlock queue.
+	 */
+	void UnlockQueue()
+	{
+		queue.Unlock();
+	}
+	/** Unlock queue and wake up worker
+	 */
+	void UnlockQueueWakeup()
+	{
+		queue.Wakeup();
+		queue.Unlock();
+	}
+	virtual void SetExitFlag()
+	{
+		queue.Lock();
+		Thread::SetExitFlag();
+		queue.Wakeup();
+		queue.Unlock();
+	}
+};
+
+class CoreExport SocketThread : public Thread
+{
+	ThreadQueueData queue;
+	ThreadSignalData signal;
+ protected:
+	/** Waits for an enqueue operation to complete
+	 * You MUST hold the queue lock when you call this.
+	 * It will be unlocked while you wait, and will be relocked
+	 * before the function returns
+	 */
+	void WaitForQueue()
+	{
+		queue.Wait();
+	}
+	/** Notifies parent by making the SignalFD ready to read
+	 * No requirements on locking
+	 */
+	void NotifyParent();
+ public:
+	SocketThread(InspIRCd* SI);
+	virtual ~SocketThread();
+	/** Lock queue.
+	 */
+	void LockQueue()
+	{
+		queue.Lock();
+	}
+	/** Unlock queue.
+	 */
+	void UnlockQueue()
+	{
+		queue.Unlock();
+	}
+	/** Unlock queue and send wakeup to worker
+	 */
+	void UnlockQueueWakeup()
+	{
+		queue.Wakeup();
+		queue.Unlock();
+	}
+	virtual void SetExitFlag()
+	{
+		queue.Lock();
+		Thread::SetExitFlag();
+		queue.Wakeup();
+		queue.Unlock();
 	}
+
+	/**
+	 * Called in the context of the parent thread after a notification
+	 * has passed through the socket
+	 */
+	virtual void OnNotify() = 0;
 };
 
 #endif
diff --git a/include/threadengines/threadengine_pthread.h b/include/threadengines/threadengine_pthread.h
index 2aba4cb15..bad263d18 100644
--- a/include/threadengines/threadengine_pthread.h
+++ b/include/threadengines/threadengine_pthread.h
@@ -107,4 +107,50 @@ class CoreExport Mutex
 	}
 };
 
+class ThreadQueueData
+{
+	pthread_mutex_t mutex;
+	pthread_cond_t cond;
+ public:
+	ThreadQueueData()
+	{
+		pthread_mutex_init(&mutex, NULL);
+		pthread_cond_init(&cond, NULL);
+	}
+
+	~ThreadQueueData()
+	{
+		pthread_mutex_destroy(&mutex);
+		pthread_cond_destroy(&cond);
+	}
+
+	void Lock()
+	{
+		pthread_mutex_lock(&mutex);
+	}
+
+	void Unlock()
+	{
+		pthread_mutex_unlock(&mutex);
+	}
+
+	void Wakeup()
+	{
+		pthread_cond_signal(&cond);
+	}
+
+	void Wait()
+	{
+		pthread_cond_wait(&cond, &mutex);
+	}
+};
+
+class ThreadSignalSocket;
+class ThreadSignalData
+{
+ public:
+	ThreadSignalSocket* sock;
+};
+
+
 #endif
diff --git a/include/threadengines/threadengine_win32.h b/include/threadengines/threadengine_win32.h
index c1f9f2cce..0fc9b27bc 100644
--- a/include/threadengines/threadengine_win32.h
+++ b/include/threadengines/threadengine_win32.h
@@ -95,5 +95,54 @@ class CoreExport Mutex
 	}
 };
 
+class ThreadQueueData
+{
+	CRITICAL_SECTION mutex;
+	HANDLE event;
+ public:
+	ThreadQueueData()
+	{
+		InitializeCriticalSection(&mutex);
+		event = CreateEvent(NULL, false, false, NULL);
+	}
+
+	~ThreadQueueData()
+	{
+		DeleteCriticalSection(&mutex);
+	}
+
+	void Lock()
+	{
+		EnterCriticalSection(&mutex);
+	}
+
+	void Unlock()
+	{
+		LeaveCriticalSection(&mutex);
+	}
+
+	void Wakeup()
+	{
+		PulseEvent(event);
+	}
+
+	void Wait()
+	{
+		LeaveCriticalSection(&mutex);
+		WaitForSingleObject(event, INFINITE);
+		EnterCriticalSection(&mutex);
+	}
+};
+
+class ThreadSignalData
+{
+ public:
+	int connFD;
+	ThreadSignalData()
+	{
+		connFD = -1;
+	}
+};
+
 #endif
 
diff --git a/src/modules/extra/m_mssql.cpp b/src/modules/extra/m_mssql.cpp
index 0f1442e59..f1f155e79 100644
--- a/src/modules/extra/m_mssql.cpp
+++ b/src/modules/extra/m_mssql.cpp
@@ -27,8 +27,6 @@
 
 class SQLConn;
 class MsSQLResult;
-class ResultNotifier;
-class MsSQLListener;
 class ModuleMsSQL;
 
 typedef std::map<std::string, SQLConn*> ConnMap;
@@ -45,86 +43,25 @@ unsigned long count(const char * const str, char a)
 	return n;
 }
 
-ResultNotifier* notifier = NULL;
-MsSQLListener* listener = NULL;
-int QueueFD = -1;
-
 ConnMap connections;
-Mutex* QueueMutex;
 Mutex* ResultsMutex;
 Mutex* LoggingMutex;
 
-class QueryThread : public Thread
+class QueryThread : public SocketThread
 {
   private:
 	ModuleMsSQL* Parent;
 	InspIRCd* ServerInstance;
   public:
 	QueryThread(InspIRCd* si, ModuleMsSQL* mod)
-	: Thread(), Parent(mod), ServerInstance(si)
+	: SocketThread(si), Parent(mod), ServerInstance(si)
 	{
 	}
 	~QueryThread() { }
 	virtual void Run();
+	virtual void OnNotify();
 };
 
-class ResultNotifier : public BufferedSocket
-{
-	ModuleMsSQL* mod;
-
- public:
-	ResultNotifier(ModuleMsSQL* m, InspIRCd* SI, int newfd, char* ip) : BufferedSocket(SI, newfd, ip), mod(m)
-	{
-	}
-
-	virtual bool OnDataReady()
-	{
-		char data = 0;
-		if (ServerInstance->SE->Recv(this, &data, 1, 0) > 0)
-		{
-			Dispatch();
-			return true;
-		}
-		return false;
-	}
-
-	void Dispatch();
-};
-
-class MsSQLListener : public ListenSocketBase
-{
-	ModuleMsSQL* Parent;
-	irc::sockets::insp_sockaddr sock_us;
-	socklen_t uslen;
-	FileReader* index;
-
- public:
-	MsSQLListener(ModuleMsSQL* 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 ResultNotifier(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
-	}
-};
-
-
 class MsSQLResult : public SQLresult
 {
  private:
@@ -289,8 +226,6 @@ class MsSQLResult : public SQLresult
 	{
 		delete fl;
 	}
-
-
 };
 
 class SQLConn : public classbase
@@ -569,7 +504,6 @@ class SQLConn : public classbase
 		ResultsMutex->Lock();
 		results.push_back(res);
 		ResultsMutex->Unlock();
-		SendNotify();
 		return SQLerror();
 	}
 
@@ -693,40 +627,6 @@ class SQLConn : public classbase
 		}
 	}
 
-	void SendNotify()
-	{
-		if (QueueFD < 0)
-		{
-			if ((QueueFD = socket(AF_FAMILY, SOCK_STREAM, 0)) == -1)
-			{
-				/* crap, we're out of sockets... */
-				return;
-			}
-
-			irc::sockets::insp_sockaddr addr;
-
-#ifdef IPV6
-			irc::sockets::insp_aton("::1", &addr.sin6_addr);
-			addr.sin6_family = AF_FAMILY;
-			addr.sin6_port = htons(listener->GetPort());
-#else
-			irc::sockets::insp_inaddr ia;
-			irc::sockets::insp_aton("127.0.0.1", &ia);
-			addr.sin_family = AF_FAMILY;
-			addr.sin_addr = ia;
-			addr.sin_port = htons(listener->GetPort());
-#endif
-
-			if (connect(QueueFD, (sockaddr*)&addr,sizeof(addr)) == -1)
-			{
-				/* wtf, we cant connect to it, but we just created it! */
-				return;
-			}
-		}
-		char id = 0;
-		send(QueueFD, &id, 1, 0);
-	}
-
 	void DoLeadingQuery()
 	{
 		SQLrequest& req = queue.front();
@@ -748,7 +648,6 @@ class ModuleMsSQL : public Module
 	{
 		LoggingMutex = new Mutex();
 		ResultsMutex = new Mutex();
-		QueueMutex = new Mutex();
 
 		ServerInstance->Modules->UseInterface("SQLutils");
 
@@ -757,25 +656,6 @@ class ModuleMsSQL : public Module
 			throw ModuleException("m_mssql: Unable to publish feature 'SQL'");
 		}
 
-		/* Create a socket on a random port. Let the tcp stack allocate us an available port */
-#ifdef IPV6
-		listener = new MsSQLListener(this, ServerInstance, 0, "::1");
-#else
-		listener = new MsSQLListener(this, ServerInstance, 0, "127.0.0.1");
-#endif
-
-		if (listener->GetFd() == -1)
-		{
-			ServerInstance->Modules->DoneWithInterface("SQLutils");
-			throw ModuleException("m_mssql: unable to create ITC pipe");
-		}
-		else
-		{
-			LoggingMutex->Lock();
-			ServerInstance->Logs->Log("m_mssql", DEBUG, "MsSQL: Interthread comms port is %d", listener->GetPort());
-			LoggingMutex->Unlock();
-		}
-
 		ReadConf();
 
 		queryDispatcher = new QueryThread(ServerInstance, this);
@@ -792,32 +672,14 @@ class ModuleMsSQL : public Module
 		ClearQueue();
 		ClearAllConnections();
 
-		ServerInstance->SE->DelFd(listener);
-		ServerInstance->BufferedSocketCull();
-
-		if (QueueFD >= 0)
-		{
-			shutdown(QueueFD, 2);
-			close(QueueFD);
-		}
-
-		if (notifier)
-		{
-			ServerInstance->SE->DelFd(notifier);
-			notifier->Close();
-			ServerInstance->BufferedSocketCull();
-		}
-
 		ServerInstance->Modules->UnpublishInterface("SQL", this);
 		ServerInstance->Modules->UnpublishFeature("SQL");
 		ServerInstance->Modules->DoneWithInterface("SQLutils");
 
 		delete LoggingMutex;
 		delete ResultsMutex;
-		delete QueueMutex;
 	}
 
-
 	void SendQueue()
 	{
 		for (ConnMap::iterator iter = connections.begin(); iter != connections.end(); iter++)
@@ -929,9 +791,9 @@ class ModuleMsSQL : public Module
 
 	virtual void OnRehash(User* user, const std::string &parameter)
 	{
-		QueueMutex->Lock();
+		queryDispatcher->LockQueue();
 		ReadConf();
-		QueueMutex->Unlock();
+		queryDispatcher->UnlockQueueWakeup();
 	}
 
 	virtual const char* OnRequest(Request* request)
@@ -940,7 +802,7 @@ class ModuleMsSQL : public Module
 		{
 			SQLrequest* req = (SQLrequest*)request;
 
-			QueueMutex->Lock();
+			queryDispatcher->LockQueue();
 
 			ConnMap::iterator iter;
 
@@ -957,7 +819,7 @@ class ModuleMsSQL : public Module
 				req->error.Id(SQL_BAD_DBID);
 			}
 
-			QueueMutex->Unlock();
+			queryDispatcher->UnlockQueueWakeup();
 
 			return returnval;
 		}
@@ -979,17 +841,17 @@ class ModuleMsSQL : public Module
 
 };
 
-void ResultNotifier::Dispatch()
+void QueryThread::OnNotify()
 {
 	mod->SendQueue();
 }
 
 void QueryThread::Run()
 {
+	this->LockQueue();
 	while (this->GetExitFlag() == false)
 	{
 		SQLConn* conn = NULL;
-		QueueMutex->Lock();
 		for (ConnMap::iterator i = connections.begin(); i != connections.end(); i++)
 		{
 			if (i->second->queue.totalsize())
@@ -998,16 +860,20 @@ void QueryThread::Run()
 				break;
 			}
 		}
-		QueueMutex->Unlock();
 		if (conn)
 		{
+			this->UnlockQueue();
 			conn->DoLeadingQuery();
-			QueueMutex->Lock();
+			this->NotifyParent();
+			this->LockQueue();
 			conn->queue.pop();
-			QueueMutex->Unlock();
 		}
-		usleep(1000);
+		else
+		{
+			this->WaitForQueue();
+		}
 	}
+	this->UnlockQueue();
 }
 
 MODULE_INIT(ModuleMsSQL)
diff --git a/src/modules/extra/m_mysql.cpp b/src/modules/extra/m_mysql.cpp
index eefeea74d..4fbf4f1f4 100644
--- a/src/modules/extra/m_mysql.cpp
+++ b/src/modules/extra/m_mysql.cpp
@@ -65,14 +65,10 @@
 
 
 class SQLConnection;
-class MySQLListener;
-
+class DispatcherThread;
 
 typedef std::map<std::string, SQLConnection*> ConnMap;
-static MySQLListener *MessagePipe = NULL;
-int QueueFD = -1;
-
-class DispatcherThread;
+typedef std::deque<SQLresult*> ResultQueue;
 
 unsigned long count(const char * const str, char a)
 {
@@ -97,7 +93,6 @@ class ModuleSQL : public Module
 	 int currid;
 	 bool rehashing;
 	 DispatcherThread* Dispatcher;
-	 Mutex QueueMutex;
 	 Mutex ResultsMutex;
 	 Mutex LoggingMutex;
 	 Mutex ConnMutex;
@@ -111,13 +106,10 @@ class ModuleSQL : public Module
 };
 
 
-
 #if !defined(MYSQL_VERSION_ID) || MYSQL_VERSION_ID<32224
 #define mysql_field_count mysql_num_fields
 #endif
 
-typedef std::deque<SQLresult*> ResultQueue;
-
 /** Represents a mysql result set
  */
 class MySQLresult : public SQLresult
@@ -301,10 +293,6 @@ class MySQLresult : public SQLresult
 	}
 };
 
-class SQLConnection;
-
-void NotifyMainThread(SQLConnection* connection_with_new_result);
-
 /** Represents a connection to a mysql database
  */
 class SQLConnection : public classbase
@@ -452,9 +440,7 @@ class SQLConnection : public classbase
 
 		*queryend = 0;
 
-		Parent->QueueMutex.Lock();
 		req.query.q = query;
-		Parent->QueueMutex.Unlock();
 
 		if (!mysql_real_query(connection, req.query.q.data(), req.query.q.length()))
 		{
@@ -485,13 +471,7 @@ class SQLConnection : public classbase
 			Parent->ResultsMutex.Unlock();
 		}
 
-		/* Now signal the main thread that we've got a result to process.
-		 * Pass them this connection id as what to examine
-		 */
-
 		delete[] query;
-
-		NotifyMainThread(this);
 	}
 
 	bool ConnectionLost()
@@ -675,123 +655,18 @@ ConnMap::iterator GetCharId(char id)
 	return Connections.end();
 }
 
-void NotifyMainThread(SQLConnection* connection_with_new_result)
-{
-	/* Here we write() to the socket the main thread has open
-	 * and we connect()ed back to before our thread became active.
-	 * The main thread is using a nonblocking socket tied into
-	 * the socket engine, so they wont block and they'll receive
-	 * nearly instant notification. Because we're in a seperate
-	 * 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.
-	 */
-	char id = FindCharId(connection_with_new_result->GetID());
-	send(QueueFD, &id, 1, 0);
-}
-
 class ModuleSQL;
 
-class DispatcherThread : public Thread
+class DispatcherThread : public SocketThread
 {
  private:
 	ModuleSQL* Parent;
 	InspIRCd* ServerInstance;
  public:
-	DispatcherThread(InspIRCd* Instance, ModuleSQL* CreatorModule) : Thread(), Parent(CreatorModule), ServerInstance(Instance) { }
+	DispatcherThread(InspIRCd* Instance, ModuleSQL* CreatorModule) : SocketThread(Instance), Parent(CreatorModule), ServerInstance(Instance) { }
 	~DispatcherThread() { }
 	virtual void Run();
-};
-
-/** Used by m_mysql to notify one thread when the other has a result
- */
-class Notifier : public BufferedSocket
-{
-	ModuleSQL* Parent;
-
- public:
-	Notifier(ModuleSQL* P, InspIRCd* SI, int newfd, char* ip) : BufferedSocket(SI, newfd, ip), Parent(P) { }
-
-	virtual bool OnDataReady()
-	{
-		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 (ServerInstance->SE->Recv(this, &data, 1, 0) > 0)
-		{
-			Parent->ConnMutex.Lock();
-			ConnMap::iterator iter = GetCharId(data);
-			Parent->ConnMutex.Unlock();
-			if (iter != Connections.end())
-			{
-				Parent->ResultsMutex.Lock();
-				ResultQueue::iterator n = iter->second->rq.begin();
-				Parent->ResultsMutex.Unlock();
-
-				(*n)->Send();
-				delete (*n);
-
-				Parent->ResultsMutex.Lock();
-				iter->second->rq.pop_front();
-				Parent->ResultsMutex.Unlock();
-
-				return true;
-			}
-			/* No error, but unknown id */
-			return true;
-		}
-
-		/* Erk, error on descriptor! */
-		return false;
-	}
-};
-
-/** Spawn sockets from a listener
- */
-class MySQLListener : public ListenSocketBase
-{
-	ModuleSQL* Parent;
-	irc::sockets::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)
-	{
-		// XXX unsafe casts suck
-		new Notifier(this->Parent, this->ServerInstance, nfd, (char *)ipconnectedto.c_str());
-	}
-
-	/* 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 void OnNotify();
 };
 
 ModuleSQL::ModuleSQL(InspIRCd* Me) : Module(Me), rehashing(false)
@@ -802,25 +677,6 @@ ModuleSQL::ModuleSQL(InspIRCd* Me) : Module(Me), rehashing(false)
 	PublicServerInstance = ServerInstance;
 	currid = 0;
 
-	/* 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
-
-	if (MessagePipe->GetFd() == -1)
-	{
-		ServerInstance->Modules->DoneWithInterface("SQLutils");
-		throw ModuleException("m_mysql: unable to create ITC pipe");
-	}
-	else
-	{
-		LoggingMutex.Lock();
-		ServerInstance->Logs->Log("m_mysql", DEBUG, "MySQL: Interthread comms port is %d", MessagePipe->GetPort());
-		LoggingMutex.Unlock();
-	}
-
 	Dispatcher = new DispatcherThread(ServerInstance, this);
 	ServerInstance->Threads->Start(Dispatcher);
 
@@ -861,13 +717,11 @@ const char* ModuleSQL::OnRequest(Request* request)
 	{
 		SQLrequest* req = (SQLrequest*)request;
 
-		/* XXX: Lock */
-		QueueMutex.Lock();
-
 		ConnMap::iterator iter;
 
 		const char* returnval = NULL;
 
+		Dispatcher->LockQueue();
 		ConnMutex.Lock();
 		if((iter = Connections.find(req->dbid)) != Connections.end())
 		{
@@ -881,7 +735,10 @@ const char* ModuleSQL::OnRequest(Request* request)
 		}
 
 		ConnMutex.Unlock();
-		QueueMutex.Unlock();
+		Dispatcher->UnlockQueueWakeup();
+		/* Yes, it's possible this will generate a spurious wakeup.
+		 * That's fine, it'll just get ignored.
+		 */
 
 		return returnval;
 	}
@@ -891,7 +748,9 @@ const char* ModuleSQL::OnRequest(Request* request)
 
 void ModuleSQL::OnRehash(User* user, const std::string &parameter)
 {
+	Dispatcher->LockQueue();
 	rehashing = true;
+	Dispatcher->UnlockQueueWakeup();
 }
 
 Version ModuleSQL::GetVersion()
@@ -903,49 +762,15 @@ void DispatcherThread::Run()
 {
 	LoadDatabases(Parent->Conf, Parent->PublicServerInstance, Parent);
 
-	/* Connect back to the Notifier */
-
-	if ((QueueFD = socket(AF_FAMILY, SOCK_STREAM, 0)) == -1)
-	{
-		/* crap, we're out of sockets... */
-		return;
-	}
-
-	irc::sockets::insp_sockaddr addr;
-
-#ifdef IPV6
-	irc::sockets::insp_aton("::1", &addr.sin6_addr);
-	addr.sin6_family = AF_FAMILY;
-	addr.sin6_port = htons(MessagePipe->GetPort());
-#else
-	irc::sockets::insp_inaddr ia;
-	irc::sockets::insp_aton("127.0.0.1", &ia);
-	addr.sin_family = AF_FAMILY;
-	addr.sin_addr = ia;
-	addr.sin_port = htons(MessagePipe->GetPort());
-#endif
-
-	if (connect(QueueFD, (sockaddr*)&addr,sizeof(addr)) == -1)
-	{
-		/* wtf, we cant connect to it, but we just created it! */
-		return;
-	}
-
-	while (this->GetExitFlag() == false)
+	this->LockQueue();
+	while (!this->GetExitFlag())
 	{
 		if (Parent->rehashing)
 		{
-		/* XXX: Lock */
-			Parent->QueueMutex.Lock();
 			Parent->rehashing = false;
 			LoadDatabases(Parent->Conf, Parent->PublicServerInstance, Parent);
-			Parent->QueueMutex.Unlock();
-			/* XXX: Unlock */
 		}
 
-		SQLConnection* conn = NULL;
-		/* XXX: Lock here for safety */
-		Parent->QueueMutex.Lock();
 		Parent->ConnMutex.Lock();
 		for (ConnMap::iterator i = Connections.begin(); i != Connections.end(); i++)
 		{
@@ -956,24 +781,57 @@ void DispatcherThread::Run()
 			}
 		}
 		Parent->ConnMutex.Unlock();
-		Parent->QueueMutex.Unlock();
-		/* XXX: Unlock */
 
-		/* Theres an item! */
 		if (conn)
 		{
+			/* There's an item! */
+			this->UnlockQueue();
 			conn->DoLeadingQuery();
-
-			/* XXX: Lock */
-			Parent->QueueMutex.Lock();
+			this->NotifyParent();
+			this->LockQueue();
 			conn->queue.pop();
-			Parent->QueueMutex.Unlock();
-			/* XXX: Unlock */
 		}
-
-		usleep(1000);
+		else
+		{
+			/* We know the queue is empty, we can safely hang this thread until
+			 * something happens
+			 */
+			this->WaitForQueue();
+		}
 	}
+	this->UnlockQueue();
 }
 
+void DispatcherThread::OnNotify()
+{
+	while (1)
+	{
+		SQLConnection* conn = NULL;
+		Parent->ConnMutex.Lock();
+		for (ConnMap::iterator iter = Connections.begin(); iter != Connections.end(); iter++)
+		{
+			if (!iter->second->rq.empty())
+			{
+				conn = iter->second;
+				break;
+			}
+		}
+		Parent->ConnMutex.Unlock();
+
+		if (!conn)
+			break;
+
+		Parent->ResultsMutex.Lock();
+		ResultQueue::iterator n = conn->rq.begin();
+		Parent->ResultsMutex.Unlock();
+
+		(*n)->Send();
+		delete (*n);
+
+		Parent->ResultsMutex.Lock();
+		conn->rq.pop_front();
+		Parent->ResultsMutex.Unlock();
+	}
+}
 
 MODULE_INIT(ModuleSQL)
diff --git a/src/threadengines/threadengine_pthread.cpp b/src/threadengines/threadengine_pthread.cpp
index f5de69a6a..5aef0f247 100644
--- a/src/threadengines/threadengine_pthread.cpp
+++ b/src/threadengines/threadengine_pthread.cpp
@@ -53,6 +53,96 @@ ThreadEngine::~ThreadEngine()
 
 void ThreadData::FreeThread(Thread* thread)
 {
-	thread->SetExitFlag(true);
+	thread->SetExitFlag();
 	pthread_join(pthread_id, NULL);
 }
+
+#if 0
+/* TODO this is a linux-specific syscall that allows signals to be
+ * sent using a single file descriptor, rather than 2 for a pipe.
+ * Requires glibc 2.8, kernel 2.6.22+
+ */
+#include <sys/eventfd.h>
+
+class ThreadSignalSocket : public BufferedSocket
+{
+	SocketThread* parent;
+ public:
+	ThreadSignalSocket(SocketThread* p, InspIRCd* SI, int newfd) :
+		BufferedSocket(SI, newfd, const_cast<char*>("0.0.0.0")), parent(p) {}
+
+	~ThreadSignalSocket()
+	{
+	}
+
+	void Notify()
+	{
+		eventfd_write(fd, 1);
+	}
+
+	virtual bool OnDataReady()
+	{
+		eventfd_t data;
+		if (eventfd_read(fd, &data))
+			return false;
+		parent->OnNotify();
+		return true;
+	}
+};
+
+SocketThread::SocketThread(InspIRCd* SI)
+{
+	int fd = eventfd(0, 0); // TODO nonblock
+	if (fd < 0)
+		throw new CoreException("Could not create pipe " + std::string(strerror(errno)));
+	signal.sock = new ThreadSignalSocket(this, SI, fd);
+}
+#else
+
+class ThreadSignalSocket : public BufferedSocket
+{
+	SocketThread* parent;
+	int send_fd;
+ public:
+	ThreadSignalSocket(SocketThread* p, InspIRCd* SI, int recvfd, int sendfd) :
+		BufferedSocket(SI, recvfd, const_cast<char*>("0.0.0.0")), parent(p), send_fd(sendfd)  {}
+
+	~ThreadSignalSocket()
+	{
+		close(send_fd);
+	}
+
+	void Notify()
+	{
+		char dummy = '*';
+		send(send_fd, &dummy, 1, 0);
+	}
+
+	virtual bool OnDataReady()
+	{
+		char data;
+		if (ServerInstance->SE->Recv(this, &data, 1, 0) <= 0)
+			return false;
+		parent->OnNotify();
+		return true;
+	}
+};
+
+SocketThread::SocketThread(InspIRCd* SI)
+{
+	int fds[2];
+	if (pipe(fds))
+		throw new CoreException("Could not create pipe " + std::string(strerror(errno)));
+	signal.sock = new ThreadSignalSocket(this, SI, fds[0], fds[1]);
+}
+#endif
+
+void SocketThread::NotifyParent()
+{
+	signal.sock->Notify();
+}
+
+SocketThread::~SocketThread()
+{
+	delete signal.sock;
+}
diff --git a/src/threadengines/threadengine_win32.cpp b/src/threadengines/threadengine_win32.cpp
index 6ebd6a10a..5a0635a09 100644
--- a/src/threadengines/threadengine_win32.cpp
+++ b/src/threadengines/threadengine_win32.cpp
@@ -51,3 +51,89 @@ void ThreadData::FreeThread(Thread* thread)
 	WaitForSingleObject(handle,INFINITE);
 }
 
+class ThreadSignalSocket : public BufferedSocket
+{
+	SignalThread* parent;
+ public:
+	ThreadSignalSocket(SignalThread* t, InspIRCd* SI, int newfd, char* ip)
+		: BufferedSocket(SI, newfd, ip), parent(t)
+	{
+		parent->results = this;
+	}
+	
+	virtual bool OnDataReady()
+	{
+		char data = 0;
+		if (ServerInstance->SE->Recv(this, &data, 1, 0) > 0)
+		{
+			parent->OnNotify();
+			return true;
+		}
+		return false;
+	}
+};
+
+class ThreadSignalListener : public ListenSocketBase
+{
+	SocketThread* parent;
+	irc::sockets::insp_sockaddr sock_us;
+ public:
+	ThreadSignalListener(SocketThread* t, InspIRCd* Instance, int port, const std::string &addr) : ListenSocketBase(Instance, port, addr), parent(t)
+	{
+		socklen_t 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 ThreadSignalSocket(parent, ServerInstance, nfd, const_cast<char*>(ipconnectedto.c_str()));
+		ServerInstance->SE->DelFd(this);
+		// 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
+	}
+};
+
+SocketThread::SocketThread(InspIRCd* SI)
+{
+	ThreadSignalListener* listener = new ThreadSignalListener(this, ServerInstance, 0, "127.0.0.1");
+	if (listener->GetFd() == -1)
+		throw CoreException("Could not create ITC pipe");
+	int connFD = socket(AF_INET, SOCK_STREAM, 0);
+	if (connFD == -1)
+		throw CoreException("Could not create ITC pipe");
+	
+	irc::sockets::sockaddrs addr;
+	irc::sockets::insp_aton("127.0.0.1", &addr.in4.sin_addr);
+	addr.in4.sin_family = AF_INET;
+	addr.in4.sin_port = htons(listener->GetPort());
+
+	if (connect(connFD, &addr.sa, sizeof(addr.in4)) == -1)
+	{
+		ServerInstance->SE->DelFd(listener);
+		close(connFD);
+		throw CoreException("Could not connet to ITC pipe");
+	}
+	this->signal.connFD = connFD;
+}
+
+void SocketThread::NotifyParent()
+{
+	char dummy = '*';
+	send(signal.connFD, &dummy, 1, 0);
+}
+
+SocketThread::~SocketThread()
+{
+	close(signal.connFD);
+}