net: implement WaitMany variants for {epoll, kqueue}

Co-authored-by: UdjinM6 <UdjinM6@users.noreply.github.com>

Author: kwvg

src/i2p.cpp

@@ -160,7 +160,7 @@ bool Session::Accept(Connection& conn)
 
     while (!*m_interrupt) {
         Sock::Event occurred;
-        if (!conn.sock->Wait(MAX_WAIT_FOR_IO, Sock::RECV, SEM_LT_DEFAULT, &occurred)) {
+        if (!conn.sock->Wait(MAX_WAIT_FOR_IO, Sock::RECV, SocketEventsParams(), &occurred)) {
             errmsg = "wait on socket failed";
             break;
         }

src/net.cpp

@@ -56,14 +56,6 @@
 #include <ifaddrs.h>
 #endif
 
-#ifdef USE_EPOLL
-#include <sys/epoll.h>
-#endif
-
-#ifdef USE_KQUEUE
-#include <sys/event.h>
-#endif
-
 #include <algorithm>
 #include <array>
 #include <cstdint>
@@ -2375,79 +2367,6 @@ Sock::EventsPerSock CConnman::GenerateWaitSockets(Span<CNode* const> nodes)
     return events_per_sock;
 }
 
-#ifdef USE_KQUEUE
-void CConnman::SocketEventsKqueue(std::set<SOCKET>& recv_set,
-                                  std::set<SOCKET>& send_set,
-                                  std::set<SOCKET>& error_set,
-                                  bool only_poll)
-{
-    std::array<struct kevent, MAX_EVENTS> events{};
-    struct timespec timeout = MillisToTimespec(only_poll ? 0 : SELECT_TIMEOUT_MILLISECONDS);
-
-    int ret{-1};
-    ToggleWakeupPipe([&](){
-        ret = kevent(Assert(m_edge_trig_events)->GetFileDescriptor(), nullptr, 0, events.data(), events.size(), &timeout);
-    });
-    if (ret == SOCKET_ERROR) {
-        LogPrintf("kevent wait error\n");
-        return;
-    }
-
-    for (int i = 0; i < ret; i++) {
-        auto& event = events[i];
-        if ((event.flags & EV_ERROR) || (event.flags & EV_EOF)) {
-            error_set.insert((SOCKET)event.ident);
-            continue;
-        }
-
-        if (event.filter == EVFILT_READ) {
-            recv_set.insert((SOCKET)event.ident);
-        }
-
-        if (event.filter == EVFILT_WRITE) {
-            send_set.insert((SOCKET)event.ident);
-        }
-    }
-}
-#endif
-
-#ifdef USE_EPOLL
-void CConnman::SocketEventsEpoll(std::set<SOCKET>& recv_set,
-                                 std::set<SOCKET>& send_set,
-                                 std::set<SOCKET>& error_set,
-                                 bool only_poll)
-{
-    std::array<epoll_event, MAX_EVENTS> events{};
-
-    int ret{-1};
-    ToggleWakeupPipe([&](){
-        ret = epoll_wait(Assert(m_edge_trig_events)->GetFileDescriptor(), events.data(), events.size(),
-                         only_poll ? 0 : SELECT_TIMEOUT_MILLISECONDS);
-    });
-
-    if (ret == SOCKET_ERROR) {
-        LogPrintf("epoll_wait error\n");
-        return;
-    }
-
-    for (int i = 0; i < ret; i++) {
-        auto& e = events[i];
-        if((e.events & EPOLLERR) || (e.events & EPOLLHUP)) {
-            error_set.insert((SOCKET)e.data.fd);
-            continue;
-        }
-
-        if (e.events & EPOLLIN) {
-            recv_set.insert((SOCKET)e.data.fd);
-        }
-
-        if (e.events & EPOLLOUT) {
-            send_set.insert((SOCKET)e.data.fd);
-        }
-    }
-}
-#endif
-
 void CConnman::SocketHandler(CMasternodeSync& mn_sync)
 {
     AssertLockNotHeld(m_total_bytes_sent_mutex);
@@ -2480,7 +2399,7 @@ void CConnman::SocketHandler(CMasternodeSync& mn_sync)
         // select(2)). If none are ready, wait for a short while and return
         // empty sets.
         events_per_sock = GenerateWaitSockets(snap.Nodes());
-        if ((is_lt && events_per_sock.empty()) || !Sock::WaitManyInternal(timeout, events_per_sock, socketEventsMode)) {
+        if ((is_lt && events_per_sock.empty()) || !Sock::WaitManyInternal(timeout, events_per_sock, SocketEventsParams{socketEventsMode, GetModeFileDescriptor()})) {
             if (is_lt) {
                 interruptNet.sleep_for(std::chrono::milliseconds(SELECT_TIMEOUT_MILLISECONDS));
             }

src/net.h

@@ -1623,19 +1623,6 @@ friend class CNode;
      */
     Sock::EventsPerSock GenerateWaitSockets(Span<CNode* const> nodes);
 
-#ifdef USE_KQUEUE
-    void SocketEventsKqueue(std::set<SOCKET>& recv_set,
-                            std::set<SOCKET>& send_set,
-                            std::set<SOCKET>& error_set,
-                            bool only_poll);
-#endif
-#ifdef USE_EPOLL
-    void SocketEventsEpoll(std::set<SOCKET>& recv_set,
-                           std::set<SOCKET>& send_set,
-                           std::set<SOCKET>& error_set,
-                           bool only_poll);
-#endif
-
     /**
      * Check connected and listening sockets for IO readiness and process them accordingly.
      */
@@ -1867,6 +1854,14 @@ friend class CNode;
     std::unique_ptr<EdgeTriggeredEvents> m_edge_trig_events{nullptr};
     std::unique_ptr<WakeupPipe> m_wakeup_pipe{nullptr};
 
+    SOCKET GetModeFileDescriptor()
+    {
+        if (m_edge_trig_events) {
+            return static_cast<SOCKET>(m_edge_trig_events->GetFileDescriptor());
+        }
+        return INVALID_SOCKET;
+    }
+
     template <typename Callable>
     void ToggleWakeupPipe(Callable&& func)
     {

src/netbase.cpp

@@ -572,7 +572,7 @@ static bool ConnectToSocket(const Sock& sock, struct sockaddr* sockaddr, socklen
             // synchronously to check for successful connection with a timeout.
             const Sock::Event requested = Sock::RECV | Sock::SEND;
             Sock::Event occurred;
-            if (!sock.Wait(std::chrono::milliseconds{nConnectTimeout}, requested, SEM_LT_DEFAULT, &occurred)) {
+            if (!sock.Wait(std::chrono::milliseconds{nConnectTimeout}, requested, SocketEventsParams(), &occurred)) {
                 LogPrintf("wait for connect to %s failed: %s\n",
                           dest_str,
                           NetworkErrorString(WSAGetLastError()));

src/test/fuzz/util.cpp

@@ -270,7 +270,7 @@ bool FuzzedSock::IsSelectable(bool is_select) const
     return m_selectable;
 }
 
-bool FuzzedSock::Wait(std::chrono::milliseconds timeout, Event requested, SocketEventsMode event_mode, Event* occurred) const
+bool FuzzedSock::Wait(std::chrono::milliseconds timeout, Event requested, SocketEventsParams event_params, Event* occurred) const
 {
     constexpr std::array wait_errnos{
         EBADF,
@@ -287,7 +287,7 @@ bool FuzzedSock::Wait(std::chrono::milliseconds timeout, Event requested, Socket
     return true;
 }
 
-bool FuzzedSock::WaitMany(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock, SocketEventsMode event_mode) const
+bool FuzzedSock::WaitMany(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock, SocketEventsParams event_params) const
 {
     for (auto& [sock, events] : events_per_sock) {
         (void)sock;

src/test/fuzz/util.h

@@ -89,9 +89,9 @@ class FuzzedSock : public Sock
 
     bool IsSelectable(bool is_select) const override;
 
-    bool Wait(std::chrono::milliseconds timeout, Event requested, SocketEventsMode event_mode = SEM_LT_DEFAULT, Event* occurred = nullptr) const override;
+    bool Wait(std::chrono::milliseconds timeout, Event requested, SocketEventsParams event_params = SocketEventsParams(), Event* occurred = nullptr) const override;
 
-    bool WaitMany(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock, SocketEventsMode event_mode = SEM_LT_DEFAULT) const override;
+    bool WaitMany(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock, SocketEventsParams event_params = SocketEventsParams()) const override;
 
     bool IsConnected(std::string& errmsg) const override;
 };

src/test/util/net.h

@@ -202,7 +202,7 @@ class StaticContentsSock : public Sock
 
     bool Wait(std::chrono::milliseconds timeout,
               Event requested,
-              SocketEventsMode event_mode = SEM_LT_DEFAULT,
+              SocketEventsParams event_params = SocketEventsParams(),
               Event* occurred = nullptr) const override
     {
         if (occurred != nullptr) {
@@ -211,7 +211,9 @@ class StaticContentsSock : public Sock
         return true;
     }
 
-    bool WaitMany(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock, SocketEventsMode event_mode = SEM_LT_DEFAULT) const override
+    bool WaitMany(std::chrono::milliseconds timeout,
+                  EventsPerSock& events_per_sock,
+                  SocketEventsParams event_params = SocketEventsParams()) const override
     {
         for (auto& [sock, events] : events_per_sock) {
             (void)sock;

src/util/sock.cpp

@@ -15,6 +15,14 @@
 #include <stdexcept>
 #include <string>
 
+#ifdef USE_EPOLL
+#include <sys/epoll.h>
+#endif
+
+#ifdef USE_KQUEUE
+#include <sys/event.h>
+#endif
+
 #ifdef USE_POLL
 #include <poll.h>
 #endif
@@ -145,11 +153,16 @@ bool Sock::IsSelectable(bool is_select) const
     return IsSelectableSocket(m_socket, is_select);
 }
 
-bool Sock::Wait(std::chrono::milliseconds timeout, Event requested, SocketEventsMode event_mode, Event* occurred) const
+bool Sock::Wait(std::chrono::milliseconds timeout, Event requested, SocketEventsParams event_params, Event* occurred) const
 {
     EventsPerSock events_per_sock{std::make_pair(m_socket, Events{requested})};
 
-    if (!WaitMany(timeout, events_per_sock)) {
+    if (auto [sem, _] = event_params; sem != SocketEventsMode::Poll && sem != SocketEventsMode::Select) {
+        // We need to ensure we are only using a level-triggered mode because we are expecting
+        // a direct correlation between the events reported and the one socket we are querying
+        event_params = SocketEventsParams();
+    }
+    if (!WaitMany(timeout, events_per_sock, event_params)) {
         return false;
     }
 
@@ -160,16 +173,16 @@ bool Sock::Wait(std::chrono::milliseconds timeout, Event requested, SocketEvents
     return true;
 }
 
-bool Sock::WaitMany(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock, SocketEventsMode event_mode) const
+bool Sock::WaitMany(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock, SocketEventsParams event_params) const
 {
-    return WaitManyInternal(timeout, events_per_sock, event_mode);
+    return WaitManyInternal(timeout, events_per_sock, event_params);
 }
 
-bool Sock::WaitManyInternal(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock, SocketEventsMode event_mode)
+bool Sock::WaitManyInternal(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock, SocketEventsParams event_params)
 {
     std::string debug_str;
 
-    switch (event_mode)
+    switch (event_params.m_event_mode)
     {
         case SocketEventsMode::Poll:
 #ifdef USE_POLL
@@ -181,11 +194,21 @@ bool Sock::WaitManyInternal(std::chrono::milliseconds timeout, EventsPerSock& ev
         case SocketEventsMode::Select:
             return WaitManySelect(timeout, events_per_sock);
         case SocketEventsMode::EPoll:
-            debug_str += "Sock::Wait -- Unimplemented for epoll, falling back on ";
+#ifdef USE_EPOLL
+            assert(event_params.m_event_fd != INVALID_SOCKET);
+            return WaitManyEPoll(timeout, events_per_sock, event_params.m_event_fd);
+#else
+            debug_str += "Sock::Wait -- Support for epoll not compiled in, falling back on ";
             break;
+#endif /* USE_EPOLL */
         case SocketEventsMode::KQueue:
-            debug_str += "Sock::Wait -- Unimplemented for kqueue, falling back on ";
+#ifdef USE_KQUEUE
+            assert(event_params.m_event_fd != INVALID_SOCKET);
+            return WaitManyKQueue(timeout, events_per_sock, event_params.m_event_fd);
+#else
+            debug_str += "Sock::Wait -- Support for kqueue not compiled in, falling back on ";
             break;
+#endif /* USE_KQUEUE */
         default:
             assert(false);
     }
@@ -202,6 +225,75 @@ bool Sock::WaitManyInternal(std::chrono::milliseconds timeout, EventsPerSock& ev
 #endif /* USE_POLL */
 }
 
+#ifdef USE_EPOLL
+bool Sock::WaitManyEPoll(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock, SOCKET epoll_fd)
+{
+    std::array<epoll_event, MAX_EVENTS> events{};
+
+    int ret = epoll_wait(epoll_fd, events.data(), events.size(), count_milliseconds(timeout));
+    if (ret == SOCKET_ERROR) {
+        return false;
+    }
+
+    // Events reported do not correspond to sockets requested in edge-triggered modes, we will clear the
+    // entire map before populating it with our events data.
+    events_per_sock.clear();
+
+    for (int idx = 0; idx < ret; idx++) {
+        auto& ev = events[idx];
+        Event occurred = 0;
+        if (ev.events & (EPOLLERR | EPOLLHUP)) {
+            occurred |= ERR;
+        } else {
+            if (ev.events & EPOLLIN) {
+                occurred |= RECV;
+            }
+            if (ev.events & EPOLLOUT) {
+                occurred |= SEND;
+            }
+        }
+        events_per_sock.emplace(static_cast<SOCKET>(ev.data.fd), Sock::Events{/*req=*/RECV | SEND, occurred});
+    }
+
+    return true;
+}
+#endif /* USE_EPOLL */
+
+#ifdef USE_KQUEUE
+bool Sock::WaitManyKQueue(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock, SOCKET kqueue_fd)
+{
+    std::array<struct kevent, MAX_EVENTS> events{};
+    struct timespec ts = MillisToTimespec(timeout);
+
+    int ret = kevent(kqueue_fd, nullptr, 0, events.data(), events.size(), &ts);
+    if (ret == SOCKET_ERROR) {
+        return false;
+    }
+
+    // Events reported do not correspond to sockets requested in edge-triggered modes, we will clear the
+    // entire map before populating it with our events data.
+    events_per_sock.clear();
+
+    for (int idx = 0; idx < ret; idx++) {
+        auto& ev = events[idx];
+        Event occurred = 0;
+        if (ev.flags & (EV_ERROR | EV_EOF)) {
+            occurred |= ERR;
+        } else {
+            if (ev.filter == EVFILT_READ) {
+                occurred |= RECV;
+            }
+            if (ev.filter == EVFILT_WRITE) {
+                occurred |= SEND;
+            }
+        }
+        events_per_sock.emplace(static_cast<SOCKET>(ev.ident), Sock::Events{/*req=*/RECV | SEND, occurred});
+    }
+
+    return true;
+}
+#endif /* USE_KQUEUE */
+
 #ifdef USE_POLL
 bool Sock::WaitManyPoll(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock)
 {