WARNING: This commit breaks kqueue and select -- work in progress!

epoll now allows both a write and a read event on a socket at the same time.


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

6 files changed, 54 insertions, 13 deletions

diff --git a/include/socketengine.h b/include/socketengine.h
index 1b68f1685..9aa651499 100644
--- a/include/socketengine.h
+++ b/include/socketengine.h
@@ -179,6 +179,8 @@ public:
 	 */
 	virtual bool AddFd(EventHandler* eh);
 
+	virtual void WantWrite(EventHandler* eh);
+
 	/** Returns the maximum number of file descriptors
 	 * you may store in the socket engine at any one time.
 	 * @return The maximum fd value
diff --git a/include/socketengine_epoll.h b/include/socketengine_epoll.h
index 8e4d6308f..5eb69cf65 100644
--- a/include/socketengine_epoll.h
+++ b/include/socketengine_epoll.h
@@ -51,6 +51,7 @@ public:
 	virtual bool DelFd(EventHandler* eh);
 	virtual int DispatchEvents();
 	virtual std::string GetName();
+	virtual void WantWrite(EventHandler* eh);
 };
 
 /** Creates a SocketEngine
diff --git a/src/inspsocket.cpp b/src/inspsocket.cpp
index c7e68df6c..b185ccb35 100644
--- a/src/inspsocket.cpp
+++ b/src/inspsocket.cpp
@@ -129,15 +129,7 @@ void InspSocket::WantWrite()
 	 *
 	 * This behaviour may be fixed in a later version.
 	 */
-	this->Instance->SE->DelFd(this);
-	this->WaitingForWriteEvent = true;
-	if (!this->Instance->SE->AddFd(this))
-	{
-		this->Close();
-		this->fd = -1;
-		this->state = I_ERROR;
-		this->OnError(I_ERR_NOMOREFDS);
-	}
+	this->Instance->SE->WantWrite(this);
 }
 
 void InspSocket::SetQueues(int nfd)
diff --git a/src/socketengine.cpp b/src/socketengine.cpp
index a10d905fb..5c05e232f 100644
--- a/src/socketengine.cpp
+++ b/src/socketengine.cpp
@@ -37,6 +37,10 @@ bool EventHandler::Writeable()
 	return false;
 }
 
+void SocketEngine::WantWrite(EventHandler* eh)
+{
+}
+
 SocketEngine::SocketEngine(InspIRCd* Instance) : ServerInstance(Instance)
 {
 	memset(ref, 0, sizeof(ref));
diff --git a/src/socketengine_epoll.cpp b/src/socketengine_epoll.cpp
index 3d0b7dee0..0e27e5248 100644
--- a/src/socketengine_epoll.cpp
+++ b/src/socketengine_epoll.cpp
@@ -77,6 +77,22 @@ bool EPollEngine::AddFd(EventHandler* eh)
 	return true;
 }
 
+void EPollEngine::WantWrite(EventHandler* eh)
+{
+	struct epoll_event ev;
+	ev.events = EPOLLOUT | EPOLLIN | EPOLLONESHOT;
+	ev.data.fd = eh->GetFd();
+	int i = epoll_ctl(EngineHandle, EPOLL_CTL_MOD, eh->GetFd(), &ev);
+	if (i < 0)
+	{
+		ServerInstance->Log(DEBUG,"epoll: Could not set want write on fd %d!",eh->GetFd());
+	}
+	else
+	{
+		ServerInstance->Log(DEBUG,"epoll: WantWrite set on %d",eh->GetFd());
+	}
+}
+
 bool EPollEngine::DelFd(EventHandler* eh)
 {
 	ServerInstance->Log(DEBUG,"EPollEngine::DelFd(%d)",eh->GetFd());
@@ -118,8 +134,20 @@ int EPollEngine::DispatchEvents()
 	int i = epoll_wait(EngineHandle, events, MAX_DESCRIPTORS, 150);
 	for (int j = 0; j < i; j++)
 	{
-		ServerInstance->Log(DEBUG,"Handle %s event on fd %d",ref[events[j].data.fd]->Readable() ? "read" : "write", ref[events[j].data.fd]->GetFd());
-		ref[events[j].data.fd]->HandleEvent(ref[events[j].data.fd]->Readable() ? EVENT_READ : EVENT_WRITE);
+		ServerInstance->Log(DEBUG,"Handle %s event on fd %d",events[j].events & EPOLLOUT ? "write" : "read", events[j].data.fd);
+		if (events[j].events & EPOLLOUT)
+		{
+			ServerInstance->Log(DEBUG,"One shot, we should EPOLL_CTL_MOD here to set it read only.");
+			struct epoll_event ev;
+			ev.events = EPOLLIN;
+			ev.data.fd = events[j].data.fd;
+			int i = epoll_ctl(EngineHandle, EPOLL_CTL_MOD, events[j].data.fd, &ev);
+			if (i < 0)
+			{
+				ServerInstance->Log(DEBUG,"epoll: Could not reset fd %d!", events[j].data.fd);
+			}
+		}
+		ref[events[j].data.fd]->HandleEvent(events[j].events & EPOLLOUT ? EVENT_WRITE : EVENT_READ);
 	}
 
 	return i;
diff --git a/src/users.cpp b/src/users.cpp
index 3be7225d4..4b7e38f3c 100644
--- a/src/users.cpp
+++ b/src/users.cpp
@@ -639,11 +639,14 @@ void userrec::FlushWriteBuf()
 		}
 		if ((sendq.length()) && (this->fd != FD_MAGIC_NUMBER))
 		{
+			int old_sendq_length = sendq.length();
 			const char* tb = this->sendq.c_str();
 			int n_sent = write(this->fd,tb,this->sendq.length());
 			if (n_sent == -1)
 			{
-				if (errno != EAGAIN)
+				if (errno == EAGAIN)
+					this->ServerInstance->SE->WantWrite(this);
+				else
 					this->SetWriteError(strerror(errno));
 			}
 			else
@@ -654,6 +657,8 @@ void userrec::FlushWriteBuf()
 				// update the user's stats counters
 				this->bytes_out += n_sent;
 				this->cmds_out++;
+				if (n_sent != old_sendq_length)
+					this->ServerInstance->SE->WantWrite(this);
 			}
 		}
 	}
@@ -690,6 +695,7 @@ const char* userrec::GetWriteError()
 
 void userrec::Oper(const std::string &opertype)
 {
+	this->ServerInstance->SE->WantWrite(this);
 	try
 	{
 		this->modes[UM_OPERATOR] = 1;
@@ -2018,7 +2024,15 @@ void userrec::HandleEvent(EventType et)
 	/* WARNING: May delete this user! */
 	try
 	{
-		ServerInstance->ProcessUser(this);
+		switch (et)
+		{
+			case EVENT_READ:
+				ServerInstance->ProcessUser(this);
+			break;
+			case EVENT_WRITE:
+				this->FlushWriteBuf();
+			break;
+		}
 	}
 	catch (...)
 	{