refactor: move wakeup pipe (de)registration to ETE

Author: kwvg

src/net.cpp

@@ -3385,31 +3385,9 @@ bool CConnman::Start(CDeterministicMNManager& dmnman, CMasternodeMetaMan& mn_met
         if (fcntl(wakeupPipe[1], F_SETFL, fFlags | O_NONBLOCK) == -1) {
             LogPrint(BCLog::NET, "fcntl for O_NONBLOCK on wakeupPipe failed\n");
         }
-#ifdef USE_KQUEUE
-        if (socketEventsMode == SocketEventsMode::KQueue) {
-            struct kevent event;
-            EV_SET(&event, wakeupPipe[0], EVFILT_READ, EV_ADD, 0, 0, nullptr);
-            int r = kevent(Assert(m_edge_trig_events)->m_fd, &event, 1, nullptr, 0, nullptr);
-            if (r != 0) {
-                LogPrint(BCLog::NET, "%s -- kevent(%d, %d, %d, ...) failed. error: %s\n", __func__,
-                         m_edge_trig_events->m_fd, EV_ADD, wakeupPipe[0], NetworkErrorString(WSAGetLastError()));
-                return false;
-            }
+        if (m_edge_trig_events && !m_edge_trig_events->RegisterPipe(wakeupPipe[0])) {
+            LogPrint(BCLog::NET, "EdgeTriggeredEvents::RegisterPipe() failed\n");
         }
-#endif
-#ifdef USE_EPOLL
-        if (socketEventsMode == SocketEventsMode::EPoll) {
-            epoll_event event;
-            event.events = EPOLLIN;
-            event.data.fd = wakeupPipe[0];
-            int r = epoll_ctl(Assert(m_edge_trig_events)->m_fd, EPOLL_CTL_ADD, wakeupPipe[0], &event);
-            if (r != 0) {
-                LogPrint(BCLog::NET, "%s -- epoll_ctl(%d, %d, %d, ...) failed. error: %s\n", __func__,
-                         m_edge_trig_events->m_fd, EPOLL_CTL_ADD, wakeupPipe[0], NetworkErrorString(WSAGetLastError()));
-                return false;
-            }
-        }
-#endif
     }
 #endif
 
@@ -3578,24 +3556,14 @@ void CConnman::StopNodes()
     semOutbound.reset();
     semAddnode.reset();
 
-#ifdef USE_KQUEUE
-    if (socketEventsMode == SocketEventsMode::KQueue && Assert(m_edge_trig_events)->m_fd != -1) {
+    if (m_edge_trig_events) {
 #ifdef USE_WAKEUP_PIPE
-        struct kevent event;
-        EV_SET(&event, wakeupPipe[0], EVFILT_READ, EV_DELETE, 0, 0, nullptr);
-        kevent(m_edge_trig_events->m_fd, &event, 1, nullptr, 0, nullptr);
-#endif
-        m_edge_trig_events.reset();
-    }
-#endif
-#ifdef USE_EPOLL
-    if (socketEventsMode == SocketEventsMode::EPoll && Assert(m_edge_trig_events)->m_fd != -1) {
-#ifdef USE_WAKEUP_PIPE
-        epoll_ctl(m_edge_trig_events->m_fd, EPOLL_CTL_DEL, wakeupPipe[0], nullptr);
+        if (!m_edge_trig_events->UnregisterPipe(wakeupPipe[0])) {
+            LogPrintf("EdgeTriggeredEvents::UnregisterPipe() failed\n");
+        }
 #endif
         m_edge_trig_events.reset();
     }
-#endif
 
 #ifdef USE_WAKEUP_PIPE
     if (wakeupPipe[0] != -1) close(wakeupPipe[0]);

src/util/edge.cpp

@@ -59,17 +59,17 @@ EdgeTriggeredEvents::~EdgeTriggeredEvents()
     }
 }
 
-bool EdgeTriggeredEvents::AddSocket(SOCKET socket) const
+bool EdgeTriggeredEvents::RegisterEntity(int entity, std::string entity_name) const
 {
     assert(m_valid);
 
     if (m_mode == SocketEventsMode::EPoll) {
 #ifdef USE_EPOLL
         epoll_event event;
-        event.data.fd = socket;
+        event.data.fd = entity;
         event.events = EPOLLIN;
-        if (epoll_ctl(m_fd, EPOLL_CTL_ADD, socket, &event) != 0) {
-            LogPrintf("Failed to add socket to epoll fd (epoll_ctl returned error %s)\n",
+        if (epoll_ctl(m_fd, EPOLL_CTL_ADD, entity, &event) != 0) {
+            LogPrintf("Failed to add %s to epoll fd (epoll_ctl returned error %s)\n", entity_name,
                       NetworkErrorString(WSAGetLastError()));
             return false;
         }
@@ -79,9 +79,9 @@ bool EdgeTriggeredEvents::AddSocket(SOCKET socket) const
     } else if (m_mode == SocketEventsMode::KQueue) {
 #ifdef USE_KQUEUE
         struct kevent event;
-        EV_SET(&event, socket, EVFILT_READ, EV_ADD, 0, 0, nullptr);
+        EV_SET(&event, entity, EVFILT_READ, EV_ADD, 0, 0, nullptr);
         if (kevent(m_fd, &event, 1, nullptr, 0, nullptr) != 0) {
-            LogPrintf("Failed to add socket to kqueue fd (kevent returned error %s)\n",
+            LogPrintf("Failed to add %s to kqueue fd (kevent returned error %s)\n", entity_name,
                       NetworkErrorString(WSAGetLastError()));
             return false;
         }
@@ -94,14 +94,14 @@ bool EdgeTriggeredEvents::AddSocket(SOCKET socket) const
     return true;
 }
 
-bool EdgeTriggeredEvents::RemoveSocket(SOCKET socket) const
+bool EdgeTriggeredEvents::UnregisterEntity(int entity, std::string entity_name) const
 {
     assert(m_valid);
 
     if (m_mode == SocketEventsMode::EPoll) {
 #ifdef USE_EPOLL
-        if (epoll_ctl(m_fd, EPOLL_CTL_DEL, socket, nullptr) != 0) {
-            LogPrintf("Failed to remove socket from epoll fd (epoll_ctl returned error %s)\n",
+        if (epoll_ctl(m_fd, EPOLL_CTL_DEL, entity, nullptr) != 0) {
+            LogPrintf("Failed to remove %s from epoll fd (epoll_ctl returned error %s)\n", entity_name,
                       NetworkErrorString(WSAGetLastError()));
             return false;
         }
@@ -111,9 +111,9 @@ bool EdgeTriggeredEvents::RemoveSocket(SOCKET socket) const
     } else if (m_mode == SocketEventsMode::KQueue) {
 #ifdef USE_KQUEUE
         struct kevent event;
-        EV_SET(&event, socket, EVFILT_READ, EV_DELETE, 0, 0, nullptr);
+        EV_SET(&event, entity, EVFILT_READ, EV_DELETE, 0, 0, nullptr);
         if (kevent(m_fd, &event, 1, nullptr, 0, nullptr) != 0) {
-            LogPrintf("Failed to remove socket from kqueue fd (kevent returned error %s)\n",
+            LogPrintf("Failed to remove %s from kqueue fd (kevent returned error %s)\n", entity_name,
                       NetworkErrorString(WSAGetLastError()));
             return false;
         }
@@ -126,6 +126,38 @@ bool EdgeTriggeredEvents::RemoveSocket(SOCKET socket) const
     return true;
 }
 
+bool EdgeTriggeredEvents::AddSocket(SOCKET socket) const
+{
+    return RegisterEntity(socket, "socket");
+}
+
+bool EdgeTriggeredEvents::RemoveSocket(SOCKET socket) const
+{
+    return UnregisterEntity(socket, "socket");
+}
+
+bool EdgeTriggeredEvents::RegisterPipe(int wakeup_pipe)
+{
+    if (m_pipe_registered) {
+        LogPrintf("Pipe already registered, ignoring new registration request\n");
+        return false;
+    }
+    bool ret = RegisterEntity(wakeup_pipe, "wakeup pipe");
+    if (ret) m_pipe_registered = true;
+    return ret;
+}
+
+bool EdgeTriggeredEvents::UnregisterPipe(int wakeup_pipe)
+{
+    if (!m_pipe_registered) {
+        LogPrintf("No pipe currently registered to unregister, ignoring request\n");
+        return false;
+    }
+    bool ret = UnregisterEntity(wakeup_pipe, "wakeup pipe");
+    if (ret) m_pipe_registered = false;
+    return ret;
+}
+
 bool EdgeTriggeredEvents::RegisterEvents(SOCKET socket) const
 {
     assert(m_valid && socket != INVALID_SOCKET);

src/util/edge.h

@@ -8,6 +8,7 @@
 #include <compat.h>
 
 #include <cstdint>
+#include <string>
 
 enum class SocketEventsMode : int8_t;
 
@@ -21,23 +22,34 @@ class EdgeTriggeredEvents
     explicit EdgeTriggeredEvents(SocketEventsMode events_mode);
     ~EdgeTriggeredEvents();
 
+    bool IsValid() const { return m_valid; }
+
     /* Add socket to interest list */
     bool AddSocket(SOCKET socket) const;
     /* Remove socket from interest list */
     bool RemoveSocket(SOCKET socket) const;
 
-    bool IsValid() const { return m_valid; }
+    /* Register wakeup pipe with EdgeTriggeredEvents instance */
+    bool RegisterPipe(int wakeup_pipe);
+    /* Unregister wakeup pipe with EdgeTriggeredEvents instance */
+    bool UnregisterPipe(int wakeup_pipe);
 
     /* Register events for socket */
     bool RegisterEvents(SOCKET socket) const;
     /* Unregister events for socket */
     bool UnregisterEvents(SOCKET socket) const;
 
+private:
+    bool RegisterEntity(int entity, std::string entity_name) const;
+    bool UnregisterEntity(int entity, std::string entity_name) const;
+
 public:
     /* File descriptor used to interact with events mode */
     int m_fd{-1};
 
 private:
+    /* Flag set if pipe has been registered with instance */
+    bool m_pipe_registered{false};
     /* Instance validity flag set during construction */
     bool m_valid{false};
     /* Flag for storing selected socket events mode */