Rewrote the MySQL API to use threads. This acts similar to before, but is faster. Removed db_mysql_execute for now.

9 files changed, 254 insertions, 6 deletions

diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 4853591b7..2255b317c 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -2,17 +2,21 @@
 file(GLOB SRC_SRCS_CPP RELATIVE ${CMAKE_CURRENT_SOURCE_DIR} "*.cpp")
 set(SRC_SRCS ${SRC_SRCS_C} ${SRC_SRCS_CPP})
 
-# If using Windows, include windows.cpp, as it's Windows-specific
+# If using Windows, add the windows.cpp, the win32 threading engine, and the socket engine to the list
 if(WIN32)
   append_to_list(SRC_SRCS win32/windows.cpp)
-endif(WIN32)
-
-# If using Windows, add the win32 threading engine to the list
-if(WIN32)
   append_to_list(SRC_SRCS threadengines/threadengine_win32.cpp)
+  append_to_list(SRC_SRCS socketengines/socketengine_win32.cpp)
 # If not using Windows, add the pthread threading engine to the list
 else(WIN32)
   append_to_list(SRC_SRCS threadengines/threadengine_pthread.cpp)
+  # If we have eventfd, use it
+  if(HAVE_EVENTFD)
+    append_to_list(SRC_SRCS socketengines/socketengine_eventfd.cpp)
+  # Else fall back to pipe
+  else(HAVE_EVENTFD)
+    append_to_list(SRC_sRCS socketengines/socketengine_pipe.cpp)
+  endif(HAVE_EVENTFD)
 endif(WIN32)
 
 sort_list(SRC_SRCS)
diff --git a/src/bin/mydbgen b/src/bin/mydbgen
index 69e1b8812..f5663ab61 100755
--- a/src/bin/mydbgen
+++ b/src/bin/mydbgen
@@ -5,7 +5,7 @@
 DBSQL="tables.sql"
 
 # Schema Version
-SVER="1"
+SVER="2"
 
 # Local Version, defaults to 0
 LVER="0"
@@ -132,6 +132,9 @@ rm -f $TFILE
 if test "x$FAILED" = "x" ; then
 	# Try to find out more about this installation
 	SQLPORT="$(mysql_config --port 2> /dev/null)"
+	if test "$SQLPORT" = "0" ; then
+		SQLPORT=3306
+	fi
 	echo ""
 	echo "Your MySQL setup is complete and your Anope schema is up to date. Make"
 	echo "sure you configure MySQL on your services.conf file prior to launching"
diff --git a/src/module.cpp b/src/module.cpp
index aa46a6f63..9431137b7 100644
--- a/src/module.cpp
+++ b/src/module.cpp
@@ -92,3 +92,14 @@ unsigned Version::GetBuild() const
 {
 	return this->Build;
 }
+
+Service::Service(Module *o, const Anope::string &n) : owner(o), name(n)
+{
+	ModuleManager::RegisterService(this);
+}
+
+Service::~Service()
+{
+	ModuleManager::UnregisterService(this);
+}
+
diff --git a/src/modulemanager.cpp b/src/modulemanager.cpp
index 11424b78b..8ab72eef5 100644
--- a/src/modulemanager.cpp
+++ b/src/modulemanager.cpp
@@ -11,6 +11,7 @@
 #include "version.h"
 #include <algorithm> // std::find
 
+std::map<Anope::string, Service *> ModuleManager::ServiceProviders;
 std::vector<Module *> ModuleManager::EventHandlers[I_END];
 
 void ModuleManager::LoadModuleList(std::list<Anope::string> &ModuleList)
@@ -430,3 +431,36 @@ void ModuleManager::UnloadAll(bool unload_proto)
 			DeleteModule(m);
 	}
 }
+
+/** Register a service
+ * @oaram s The service
+ * @return true if it was successfully registeed, else false (service name colision)
+ */
+bool ModuleManager::RegisterService(Service *s)
+{
+	return ModuleManager::ServiceProviders.insert(std::make_pair(s->name, s)).second;
+}
+
+/** Unregister a service
+ * @param s The service
+ * @return true if it was unregistered successfully
+ */
+bool ModuleManager::UnregisterService(Service *s)
+{
+	return ModuleManager::ServiceProviders.erase(s->name);
+}
+
+/** Get a service
+ * @param name The service name
+ * @param s The service
+ * @return The service
+ */
+Service *ModuleManager::GetService(const Anope::string &name)
+{
+	std::map<Anope::string, Service *>::const_iterator it = ModuleManager::ServiceProviders.find(name);
+
+	if (it != ModuleManager::ServiceProviders.end())
+		return it->second;
+	return NULL;
+}
+
diff --git a/src/modules.cpp b/src/modules.cpp
index 5cd9b6b69..c90410896 100644
--- a/src/modules.cpp
+++ b/src/modules.cpp
@@ -381,3 +381,19 @@ Version Module::GetVersion() const
 {
 	return Version(VERSION_MAJOR, VERSION_MINOR, VERSION_BUILD);
 }
+
+std::list<dynamic_reference_base *> dyn_references;
+
+dynamic_reference_base::dynamic_reference_base()
+{
+	dyn_references.push_back(this);
+}
+
+dynamic_reference_base::~dynamic_reference_base()
+{
+	std::list<dynamic_reference_base *>::iterator it = std::find(dyn_references.begin(), dyn_references.end(), this);
+
+	if (it != dyn_references.end())
+		dyn_references.erase(it);
+}
+
diff --git a/src/socketengines/socketengine_eventfd.cpp b/src/socketengines/socketengine_eventfd.cpp
new file mode 100644
index 000000000..767a6eddb
--- /dev/null
+++ b/src/socketengines/socketengine_eventfd.cpp
@@ -0,0 +1,47 @@
+#include "services.h"
+#include <sys/eventfd.h>
+
+int Pipe::RecvInternal(char *buf, size_t sz) const
+{
+	static eventfd_t dummy;
+	return !eventfd_read(this->Sock, &dummy);
+}
+
+int Pipe::SendInternal(const Anope::string &) const
+{
+	return !eventfd_write(this->Sock, 1);
+}
+
+Pipe::Pipe() : Socket()
+{
+	this->Sock = eventfd(0, EFD_NONBLOCK);
+	if (this->Sock < 0)
+		throw CoreException(Anope::string("Could not create pipe: ") + strerror(errno));
+
+	this->IPv6 = false;
+	this->Type = SOCKTYPE_CLIENT;
+
+	SocketEngine->AddSocket(this);
+}
+
+bool Pipe::ProcessRead()
+{
+	this->RecvInternal(NULL, 0);
+	return this->Read("");
+}
+
+bool Pipe::Read(const Anope::string &)
+{
+	this->OnNotify();
+	return true;
+}
+
+void Pipe::Notify()
+{
+	this->Write("*");
+}
+
+void Pipe::OnNotify()
+{
+}
+
diff --git a/src/socketengines/socketengine_pipe.cpp b/src/socketengines/socketengine_pipe.cpp
new file mode 100644
index 000000000..ee1069d57
--- /dev/null
+++ b/src/socketengines/socketengine_pipe.cpp
@@ -0,0 +1,50 @@
+#include "services.h"
+
+int Pipe::RecvInternal(char *buf, size_t sz) const
+{
+	static char dummy[512];
+	while (read(this->Sock, &dummy, 512) == 512);
+	return 0;
+}
+
+int Pipe::SendInternal(const Anope::string &) const
+{
+	static const char dummy = '*';
+	return write(this->WritePipe, &dummy, 1);
+}
+
+Pipe::Pipe() : Socket()
+{
+	int fds[2];
+	if (pipe2(fds, O_NONBLOCK))
+		throw CoreException(Anope::string("Could not create pipe: ") + strerror(errno));
+
+	this->Sock = fds[0];
+	this->WritePipe = fds[1];
+	this->IPv6 = false;
+	this->Type = SOCKTYPE_CLIENT;
+
+	SocketEngine->AddSocket(this);
+}
+
+bool Pipe::ProcessRead()
+{
+	this->RecvInternal(NULL, 0);
+	return this->Read("");
+}
+
+bool Pipe::Read(const Anope::string &)
+{
+	this->OnNotify();
+	return true;
+}
+
+void Pipe::Notify()
+{
+	this->SendInternal("");
+}
+
+void Pipe::OnNotify()
+{
+}
+
diff --git a/src/socketengines/socketengine_win32.cpp b/src/socketengines/socketengine_win32.cpp
new file mode 100644
index 000000000..1b1a4e50d
--- /dev/null
+++ b/src/socketengines/socketengine_win32.cpp
@@ -0,0 +1,76 @@
+#include "services.h"
+
+static Socket *newsocket = NULL;
+
+class LSocket : public ListenSocket
+{
+ public:
+	LSocket(const Anope::string &host, int port) : ListenSocket(host, port) { }
+
+	bool OnAccept(Socket *s)
+	{
+		newsocket = s;
+		return true;
+	}
+};
+
+int Pipe::RecvInternal(char *buf, size_t sz) const
+{
+	static char dummy[512];
+	return read(this->Sock, &dummy, 512);
+}
+
+int Pipe::SendInternal(const Anope::string &) const
+{
+	static const char dummy = '*';
+	return write(this->WritePipe, &dummy, 1);
+}
+
+Pipe::Pipe() : Socket()
+{
+	LSocket lfs("127.0.0.1", 0);
+
+	int cfd = socket(AF_INET, SOCK_STREAM, 0);
+	if (cfd == -1)
+		throw CoreException("Error accepting new socket for Pipe");
+
+	sockaddr_in addr;
+	socklen_t sz = sizeof(addr);
+	getsockname(lfs.GetSock(), reinterpret_cast<sockaddr *>(&addr), &sz);
+
+	if (connect(cfd, reinterpret_cast<sockaddr *>(&addr), sz))
+		throw CoreException("Error accepting new socket for Pipe");
+	lfs.ProcessRead();
+	if (!newsocket)
+		throw CoreException("Error accepting new socket for Pipe");
+	
+	this->Sock = cfd;
+	this->WritePipe = newsocket->GetSock();
+	this->IPv6 = false;
+	this->Type = SOCKTYPE_CLIENT;
+
+	SocketEngine->AddSocket(this);
+	newsocket = NULL;
+}
+
+bool Pipe::ProcessRead()
+{
+	this->RecvInternal(NULL, 0);
+	return this->Read("");
+}
+
+bool Pipe::Read(const Anope::string &)
+{
+	this->OnNotify();
+	return true;
+}
+
+void Pipe::Notify()
+{
+	this->SendInternal("");
+}
+
+void Pipe::OnNotify()
+{
+}
+
diff --git a/src/sockets.cpp b/src/sockets.cpp
index b85255523..646ee261f 100644
--- a/src/sockets.cpp
+++ b/src/sockets.cpp
@@ -25,11 +25,18 @@ SocketEngineBase::SocketEngineBase()
 
 SocketEngineBase::~SocketEngineBase()
 {
+	for (std::map<int, Socket *>::const_iterator it = this->Sockets.begin(), it_end = this->Sockets.end(); it != it_end; ++it)
+		delete it->second;
+	this->Sockets.clear();
 #ifdef _WIN32
 	WSACleanup();
 #endif
 }
 
+Socket::Socket()
+{
+}
+
 /** Constructor
  * @param nsock The socket
  * @param nIPv6 IPv6?