Added kqueue

5 files changed, 164 insertions, 39 deletions

diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index d564a8965..292da5a3f 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -22,11 +22,15 @@ endif(WIN32)
 if(HAVE_EPOLL)
   append_to_list(SRC_SRCS socketengines/socketengine_epoll.cpp)
 else(HAVE_EPOLL)
-  if(HAVE_POLL)
-    append_to_list(SRC_SRCS socketengines/socketengine_poll.cpp)
-  else(HAVE_POLL)
-    append_to_list(SRC_SRCS socketengines/socketengine_select.cpp)
-  endif(HAVE_POLL)
+  if(HAVE_KQUEUE)
+    append_to_list(SRC_SRCS socketengines/socketengine_kqueue.cpp)
+  else(HAVE_KQUEUE)
+    if(HAVE_POLL)
+      append_to_list(SRC_SRCS socketengines/socketengine_poll.cpp)
+    else(HAVE_POLL)
+      append_to_list(SRC_SRCS socketengines/socketengine_select.cpp)
+    endif(HAVE_POLL)
+  endif(HAVE_KQUEUE)
 endif(HAVE_EPOLL)
 
 sort_list(SRC_SRCS)
diff --git a/src/socketengines/socketengine_epoll.cpp b/src/socketengines/socketengine_epoll.cpp
index 73f7e10d6..07a033f02 100644
--- a/src/socketengines/socketengine_epoll.cpp
+++ b/src/socketengines/socketengine_epoll.cpp
@@ -124,13 +124,13 @@ void SocketEngine::Process()
 			continue;
 		Socket *s = it->second;
 
-		if (s->HasFlag(SF_DEAD))
-			continue;
-
 		if (ev->events & (EPOLLHUP | EPOLLERR))
 		{
+			socklen_t sz = sizeof(errno);
+			getsockopt(s->GetFD(), SOL_SOCKET, SO_ERROR, &errno, &sz);
 			s->ProcessError();
 			s->SetFlag(SF_DEAD);
+			delete s;
 			continue;
 		}
 
@@ -142,15 +142,6 @@ void SocketEngine::Process()
 
 		if ((ev->events & EPOLLOUT) && !s->ProcessWrite())
 			s->SetFlag(SF_DEAD);
-	}
-
-	for (int i = 0; i < total; ++i)
-	{
-		epoll_event *ev = &events[i];
-		std::map<int, Socket *>::iterator it = Sockets.find(ev->data.fd);
-		if (it == Sockets.end())
-			continue;
-		Socket *s = it->second;
 
 		if (s->HasFlag(SF_DEAD))
 			delete s;
diff --git a/src/socketengines/socketengine_kqueue.cpp b/src/socketengines/socketengine_kqueue.cpp
new file mode 100644
index 000000000..ef592646f
--- /dev/null
+++ b/src/socketengines/socketengine_kqueue.cpp
@@ -0,0 +1,144 @@
+#include "module.h"
+#include <sys/types.h>
+#include <sys/event.h>
+#include <sys/time.h>
+
+static int kq_fd, max_fds;
+static struct kevent *change_events, *event_events;
+static int change_count;
+
+static struct kevent *GetChangeEvent()
+{
+	if (change_count == max_fds)
+	{
+		timespec zero_timespec = { 0, 0 };
+		for (int i = 0; i < change_count; ++i)
+			kevent(kq_fd, &change_events[i], 1, NULL, 0, &zero_timespec);
+		change_count = 0;
+	}
+
+	return &change_events[change_count++];
+}
+
+void SocketEngine::Init()
+{
+	kq_fd = kqueue();
+	max_fds = getdtablesize();
+
+	if (kq_fd < 0)
+		throw SocketException("Unable to create kqueue engine: " + Anope::LastError());
+	else if (max_fds <= 0)
+		throw SocketException("Can't determine maximum number of open sockets");
+
+	change_events = new struct kevent[max_fds];
+	event_events = new struct kevent[max_fds];
+
+	change_count = 0;
+}
+
+void SocketEngine::Shutdown()
+{
+	Process();
+
+	for (std::map<int, Socket *>::const_iterator it = Sockets.begin(), it_end = Sockets.end(); it != it_end;)
+	{
+		Socket *s = it->second;
+		++it;
+		delete s;
+	}
+	Sockets.clear();
+
+	delete [] change_events;
+	delete [] event_events;
+}
+
+void SocketEngine::AddSocket(Socket *s)
+{
+	struct kevent *event = GetChangeEvent();
+	EV_SET(event, s->GetFD(), EVFILT_READ, EV_ADD, 0, 0, NULL);
+
+	Sockets.insert(std::make_pair(s->GetFD(), s));
+}
+
+void SocketEngine::DelSocket(Socket *s)
+{
+	struct kevent *event = GetChangeEvent();
+	EV_SET(event, s->GetFD(), EVFILT_READ, EV_DELETE, 0, 0, NULL);
+
+	event = GetChangeEvent();
+	EV_SET(event, s->GetFD(), EVFILT_WRITE, EV_DELETE, 0, 0, NULL);
+
+	Sockets.erase(s->GetFD());
+}
+
+void SocketEngine::MarkWritable(Socket *s)
+{
+	if (s->HasFlag(SF_WRITABLE))
+		return;
+
+	struct kevent *event = GetChangeEvent();
+	EV_SET(event, s->GetFD(), EVFILT_WRITE, EV_ADD, 0, 0, NULL);
+
+	s->SetFlag(SF_WRITABLE);
+}
+
+void SocketEngine::ClearWritable(Socket *s)
+{
+	if (!s->HasFlag(SF_WRITABLE))
+		return;
+
+	struct kevent *event = GetChangeEvent();
+	EV_SET(event, s->GetFD(), EVFILT_WRITE, EV_DELETE, 0, 0, NULL);
+
+	s->UnsetFlag(SF_WRITABLE);
+}
+
+void SocketEngine::Process()
+{
+	static timespec kq_timespec = { Config->ReadTimeout, 0 };
+	int total = kevent(kq_fd, change_events, change_count, event_events, max_fds, &kq_timespec);
+	change_count = 0;
+	Anope::CurTime = time(NULL);
+
+	/* EINTR can be given if the read timeout expires */
+	if (total == -1)
+	{
+		if (errno != EINTR)
+			Log() << "SockEngine::Process(): error: " << Anope::LastError();
+		return;
+	}
+
+	for (int i = 0; i < total; ++i)
+	{
+		struct kevent *event = &event_events[i];
+		if (event->flags & EV_ERROR)
+			continue;
+
+		std::map<int, Socket *>::iterator it = Sockets.find(event->ident);
+		if (it == Sockets.end())
+			continue;
+		Socket *s = it->second;
+
+		if (event->flags & EV_EOF)
+		{
+			socklen_t sz = sizeof(errno);
+			getsockopt(s->GetFD(), SOL_SOCKET, SO_ERROR, &errno, &sz);
+			s->ProcessError();
+			s->SetFlag(SF_DEAD);
+			delete s;
+			continue;
+		}
+
+		if (!s->Process())
+			continue;
+
+		if (event->filter == EVFILT_READ && !s->ProcessRead())
+			s->SetFlag(SF_DEAD);
+		else if (event->filter == EVFILT_WRITE && !s->ProcessWrite())
+			s->SetFlag(SF_DEAD);
+
+		if (s->HasFlag(SF_DEAD))
+			delete s;
+	}
+}
+
diff --git a/src/socketengines/socketengine_poll.cpp b/src/socketengines/socketengine_poll.cpp
index ba111659a..07e150d84 100644
--- a/src/socketengines/socketengine_poll.cpp
+++ b/src/socketengines/socketengine_poll.cpp
@@ -142,13 +142,13 @@ void SocketEngine::Process()
 			continue;
 		Socket *s = it->second;
 
-		if (s->HasFlag(SF_DEAD))
-			continue;
-
 		if (ev->revents & (POLLERR | POLLRDHUP))
 		{
+			socklen_t sz = sizeof(errno);
+			getsockopt(s->GetFD(), SOL_SOCKET, SO_ERROR, &errno, &sz);
 			s->ProcessError();
 			s->SetFlag(SF_DEAD);
+			delete s;
 			continue;
 		}
 
@@ -160,15 +160,6 @@ void SocketEngine::Process()
 
 		if ((ev->revents & POLLOUT) && !s->ProcessWrite())
 			s->SetFlag(SF_DEAD);
-	}
-
-	for (int i = 0; i < SocketCount; ++i)
-	{
-		pollfd *ev = &events[i];
-		std::map<int, Socket *>::iterator it = Sockets.find(ev->fd);
-		if (it == Sockets.end())
-			continue;
-		Socket *s = it->second;
 
 		if (s->HasFlag(SF_DEAD))
 			delete s;
diff --git a/src/socketengines/socketengine_select.cpp b/src/socketengines/socketengine_select.cpp
index ad6e04ca9..cd755278e 100644
--- a/src/socketengines/socketengine_select.cpp
+++ b/src/socketengines/socketengine_select.cpp
@@ -92,21 +92,22 @@ void SocketEngine::Process()
 	else if (sresult)
 	{
 		int processed = 0;
-		for (std::map<int, Socket *>::const_iterator it = Sockets.begin(), it_end = Sockets.end(); it != it_end && processed != sresult; ++it)
+		for (std::map<int, Socket *>::const_iterator it = Sockets.begin(), it_end = Sockets.end(); it != it_end && processed != sresult;)
 		{
 			Socket *s = it->second;
+			++it;
 
 			bool has_read = FD_ISSET(s->GetFD(), &rfdset), has_write = FD_ISSET(s->GetFD(), &wfdset), has_error = FD_ISSET(s->GetFD(), &efdset);
 			if (has_read || has_write || has_error)
 				++processed;
 
-			if (s->HasFlag(SF_DEAD))
-				continue;
-
 			if (has_error)
 			{
+				socklen_t sz = sizeof(errno);
+				getsockopt(s->GetFD(), SOL_SOCKET, SO_ERROR, &errno, &sz);
 				s->ProcessError();
 				s->SetFlag(SF_DEAD);
+				delete s;
 				continue;
 			}
 
@@ -118,12 +119,6 @@ void SocketEngine::Process()
 
 			if (has_write && !s->ProcessWrite())
 				s->SetFlag(SF_DEAD);
-		}
-
-		for (std::map<int, Socket *>::iterator it = Sockets.begin(), it_end = Sockets.end(); it != it_end; )
-		{
-			Socket *s = it->second;
-			++it;
 
 			if (s->HasFlag(SF_DEAD))
 				delete s;