event/SocketMonitor: add `ready_flags`

src/event/Loop.cxx

@@ -87,20 +87,18 @@ EventLoop::Break() noexcept
 }
 
 bool
-EventLoop::Abandon(int _fd, SocketMonitor &m)  noexcept
+EventLoop::Abandon(int _fd)  noexcept
 {
 	assert(!IsAlive() || IsInside());
 
-	poll_result.Clear(&m);
 	return poll_group.Abandon(_fd);
 }
 
 bool
-EventLoop::RemoveFD(int _fd, SocketMonitor &m) noexcept
+EventLoop::RemoveFD(int _fd) noexcept
 {
 	assert(!IsAlive() || IsInside());
 
-	poll_result.Clear(&m);
 	return poll_group.Remove(_fd);
 }
 
@@ -230,8 +228,16 @@ EventLoop::Run() noexcept
 
 		/* wait for new event */
 
+		PollResult poll_result;
 		poll_group.ReadEvents(poll_result, ExportTimeoutMS(timeout));
 
+		ready_sockets.clear();
+		for (size_t i = 0; i < poll_result.GetSize(); ++i) {
+			auto &sm = *(SocketMonitor *)poll_result.GetObject(i);
+			sm.SetReadyFlags(poll_result.GetEvents(i));
+			ready_sockets.push_back(sm);
+		}
+
 		now = std::chrono::steady_clock::now();
 
 		{
@@ -240,19 +246,12 @@ EventLoop::Run() noexcept
 		}
 
 		/* invoke sockets */
-		for (size_t i = 0; i < poll_result.GetSize(); ++i) {
-			auto events = poll_result.GetEvents(i);
-			if (events != 0) {
-				if (quit)
-					break;
-
-				auto m = (SocketMonitor *)poll_result.GetObject(i);
-				m->Dispatch(events);
-			}
-		}
-
-		poll_result.Reset();
+		while (!ready_sockets.empty() && !quit) {
+			auto &sm = ready_sockets.front();
+			ready_sockets.pop_front();
 
+			sm.Dispatch();
+		}
 	} while (!quit);
 
 #ifndef NDEBUG

src/event/Loop.hxx

@@ -88,6 +88,19 @@ class EventLoop final : SocketMonitor
 				       boost::intrusive::constant_time_size<false>>;
 	DeferredList deferred;
 
+	using ReadySocketList =
+		boost::intrusive::list<SocketMonitor,
+				       boost::intrusive::member_hook<SocketMonitor,
+								     SocketMonitor::ReadyListHook,
+								     &SocketMonitor::ready_siblings>,
+				       boost::intrusive::constant_time_size<false>>;
+
+	/**
+	 * A linked list of #SocketMonitor instances which have a
+	 * non-zero "ready_flags" field, and need to be dispatched.
+	 */
+	ReadySocketList ready_sockets;
+
 #ifdef HAVE_URING
 	std::unique_ptr<Uring::Manager> uring;
 #endif
@@ -123,7 +136,6 @@ class EventLoop final : SocketMonitor
 #endif
 
 	PollGroup poll_group;
-	PollResult poll_result;
 
 	/**
 	 * A reference to the thread that is currently inside Run().
@@ -178,9 +190,9 @@ public:
 	 * has been closed.  This is like RemoveFD(), but does not
 	 * attempt to use #EPOLL_CTL_DEL.
 	 */
-	bool Abandon(int fd, SocketMonitor &m) noexcept;
+	bool Abandon(int fd) noexcept;
 
-	bool RemoveFD(int fd, SocketMonitor &m) noexcept;
+	bool RemoveFD(int fd) noexcept;
 
 	void AddIdle(IdleMonitor &i) noexcept;
 	void RemoveIdle(IdleMonitor &i) noexcept;

src/event/PollGroupEpoll.hxx

@@ -45,16 +45,6 @@ public:
 	void *GetObject(size_t i) const noexcept {
 		return events[i].data.ptr;
 	}
-
-	void Reset() noexcept {
-		n_events = 0;
-	}
-
-	void Clear(void *obj) noexcept {
-		for (size_t i = 0; i < n_events; ++i)
-			if (events[i].data.ptr == obj)
-				events[i].events = 0;
-	}
 };
 
 class PollGroupEpoll

src/event/PollResultGeneric.hxx

@@ -57,16 +57,6 @@ public:
 		return items[i].obj;
 	}
 
-	void Reset() noexcept {
-		items.clear();
-	}
-
-	void Clear(void *obj) noexcept {
-		for (auto i = items.begin(); i != items.end(); ++i)
-			if (i->obj == obj)
-				i->events = 0;
-	}
-
 	void Add(unsigned events, void *obj) noexcept {
 		items.emplace_back(events, obj);
 	}

src/event/SocketMonitor.cxx

@@ -28,9 +28,10 @@
 #endif
 
 void
-SocketMonitor::Dispatch(unsigned flags) noexcept
+SocketMonitor::Dispatch() noexcept
 {
-	flags &= GetScheduledFlags() | IMPLICIT_FLAGS;
+	const unsigned flags = std::exchange(ready_flags, 0) &
+		(GetScheduledFlags() | IMPLICIT_FLAGS);
 
 	if (flags != 0)
 		OnSocketReady(flags);
@@ -79,7 +80,7 @@ SocketMonitor::Schedule(unsigned flags) noexcept
 	if (scheduled_flags == 0)
 		success = loop.AddFD(fd.Get(), flags, *this);
 	else if (flags == 0)
-		success = loop.RemoveFD(fd.Get(), *this);
+		success = loop.RemoveFD(fd.Get());
 	else
 		success = loop.ModifyFD(fd.Get(), flags, *this);
 

src/event/SocketMonitor.hxx

@@ -23,6 +23,8 @@
 #include "PollGroup.hxx"
 #include "net/SocketDescriptor.hxx"
 
+#include <boost/intrusive/list_hook.hpp>
+
 #include <cassert>
 #include <cstddef>
 #include <type_traits>
@@ -43,14 +45,27 @@ class EventLoop;
  * as thread-safe.
  */
 class SocketMonitor {
+	friend class EventLoop;
+
+	using ReadyListHook = boost::intrusive::list_member_hook<boost::intrusive::link_mode<boost::intrusive::auto_unlink>>;
+	ReadyListHook ready_siblings;
+
 	SocketDescriptor fd = SocketDescriptor::Undefined();
 	EventLoop &loop;
 
 	/**
-	 * A bit mask of events that is currently registered in the EventLoop.
+	 * A bit mask of events that is currently registered in the
+	 * #EventLoop.
 	 */
 	unsigned scheduled_flags = 0;
 
+	/**
+	 * A bit mask of events which have been reported as "ready" by
+	 * epoll_wait().  If non-zero, then the #EventLoop will call
+	 * Dispatch() soon.
+	 */
+	unsigned ready_flags = 0;
+
 public:
 	static constexpr unsigned READ = PollGroup::READ;
 	static constexpr unsigned WRITE = PollGroup::WRITE;
@@ -103,6 +118,10 @@ public:
 		return scheduled_flags;
 	}
 
+	void SetReadyFlags(unsigned flags) noexcept {
+		ready_flags = flags;
+	}
+
 	/**
 	 * @return true on success, false on error (with errno set if
 	 * USE_EPOLL is defined)
@@ -136,7 +155,7 @@ protected:
 	virtual bool OnSocketReady(unsigned flags) noexcept = 0;
 
 public:
-	void Dispatch(unsigned flags) noexcept;
+	void Dispatch() noexcept;
 };
 
 #endif