net: use pointers to Sock instead of SOCKET in CConnman::SocketEvents()

This will make it easier to mock-test the surrounding code.

The pointed to `Sock` objects will not be destroyed while
`CConnman::SocketEvents()` is running because they are either pointers
to `CNode::m_sock` and we have incremented the refcount in
`CConnman::SocketHandler()` or pointers to `CConnman::vhListenSocket`
which is only destroyed at shutdown.

Author: vasild

src/net.cpp

@@ -1282,7 +1282,7 @@ bool CConnman::InactivityCheck(const CNode& node) const
 bool CConnman::GenerateSelectSet(const std::vector<CNode*>& nodes, Sockets& sockets)
 {
     for (const ListenSocket& hListenSocket : vhListenSocket) {
-        sockets.recv.insert(hListenSocket.sock->Get());
+        sockets.recv.insert(hListenSocket.sock.get());
     }
 
     for (CNode* pnode : nodes) {
@@ -1309,13 +1309,13 @@ bool CConnman::GenerateSelectSet(const std::vector<CNode*>& nodes, Sockets& sock
             continue;
         }
 
-        sockets.err.insert(pnode->m_sock->Get());
+        sockets.err.insert(pnode->m_sock.get());
         if (select_send) {
-            sockets.send.insert(pnode->m_sock->Get());
+            sockets.send.insert(pnode->m_sock.get());
             continue;
         }
         if (select_recv) {
-            sockets.recv.insert(pnode->m_sock->Get());
+            sockets.recv.insert(pnode->m_sock.get());
         }
     }
 
@@ -1331,42 +1331,52 @@ void CConnman::SocketEvents(const std::vector<CNode*>& nodes, Sockets& sockets)
         return;
     }
 
-    std::unordered_map<SOCKET, struct pollfd> pollfds;
-    for (SOCKET socket_id : sockets_to_check.recv) {
-        pollfds[socket_id].fd = socket_id;
-        pollfds[socket_id].events |= POLLIN;
-    }
+    // The same socket may be queued for more than one event (e.g. waiting for both recv
+    // and send - in both `sockets_to_check.recv` and `sockets_to_check.send`). We unique
+    // them into `by_socket`.
+    struct SockAndPollFd {
+        Sock* sock;
+        pollfd pfd;
+    };
+    std::unordered_map<SOCKET, SockAndPollFd> by_socket;
 
-    for (SOCKET socket_id : sockets_to_check.send) {
-        pollfds[socket_id].fd = socket_id;
-        pollfds[socket_id].events |= POLLOUT;
-    }
+    // Add the sockets from `from` to `by_socket` for the given events.
+    auto AddSocketsForEvents = [&by_socket](auto& from, short events) {
+        for (Sock* sock : from) {
+            const SOCKET fd = sock->Get();
+            by_socket[fd].sock = sock;
+            by_socket[fd].pfd.fd = fd;
+            by_socket[fd].pfd.events |= events;
+        }
+    };
 
-    for (SOCKET socket_id : sockets_to_check.err) {
-        pollfds[socket_id].fd = socket_id;
-        // These flags are ignored, but we set them for clarity
-        pollfds[socket_id].events |= POLLERR|POLLHUP;
-    }
+    AddSocketsForEvents(sockets_to_check.recv, POLLIN);
+    AddSocketsForEvents(sockets_to_check.send, POLLOUT);
+    // These flags are ignored by poll(2), but we set them for clarity.
+    AddSocketsForEvents(sockets_to_check.err, POLLERR | POLLHUP);
 
     std::vector<struct pollfd> vpollfds;
-    vpollfds.reserve(pollfds.size());
-    for (auto it : pollfds) {
-        vpollfds.push_back(std::move(it.second));
+    vpollfds.reserve(by_socket.size());
+    for (auto& [socket, sock_and_pollfd] : by_socket) {
+        vpollfds.push_back(sock_and_pollfd.pfd);
     }
 
     if (poll(vpollfds.data(), vpollfds.size(), SELECT_TIMEOUT_MILLISECONDS) < 0) return;
 
     if (interruptNet) return;
 
     for (struct pollfd pollfd_entry : vpollfds) {
+        auto pos = by_socket.find(pollfd_entry.fd);
+        assert(pos != by_socket.end());
+        Sock* sock = pos->second.sock;
         if (pollfd_entry.revents & POLLIN) {
-            sockets.recv.insert(pollfd_entry.fd);
+            sockets.recv.insert(sock);
         }
         if (pollfd_entry.revents & POLLOUT) {
-            sockets.send.insert(pollfd_entry.fd);
+            sockets.send.insert(sock);
         }
         if (pollfd_entry.revents & (POLLERR|POLLHUP)) {
-            sockets.err.insert(pollfd_entry.fd);
+            sockets.err.insert(sock);
         }
     }
 }
@@ -1394,20 +1404,18 @@ void CConnman::SocketEvents(const std::vector<CNode*>& nodes, Sockets& sockets)
     FD_ZERO(&fdsetError);
     SOCKET hSocketMax = 0;
 
-    for (SOCKET hSocket : sockets_to_check.recv) {
-        FD_SET(hSocket, &fdsetRecv);
-        hSocketMax = std::max(hSocketMax, hSocket);
-    }
-
-    for (SOCKET hSocket : sockets_to_check.send) {
-        FD_SET(hSocket, &fdsetSend);
-        hSocketMax = std::max(hSocketMax, hSocket);
-    }
+    // Add the sockets from `from` to the given fd set.
+    auto AddSocketsForEvents = [&hSocketMax](auto& from, fd_set& set) {
+        for (Sock* sock : from) {
+            const SOCKET fd = sock->Get();
+            FD_SET(fd, &set);
+            hSocketMax = std::max(hSocketMax, fd);
+        }
+    };
 
-    for (SOCKET hSocket : sockets_to_check.err) {
-        FD_SET(hSocket, &fdsetError);
-        hSocketMax = std::max(hSocketMax, hSocket);
-    }
+    AddSocketsForEvents(sockets_to_check.recv, fdsetRecv);
+    AddSocketsForEvents(sockets_to_check.send, fdsetSend);
+    AddSocketsForEvents(sockets_to_check.err, fdsetError);
 
     int nSelect = select(hSocketMax + 1, &fdsetRecv, &fdsetSend, &fdsetError, &timeout);
 
@@ -1426,21 +1434,21 @@ void CConnman::SocketEvents(const std::vector<CNode*>& nodes, Sockets& sockets)
             return;
     }
 
-    for (SOCKET hSocket : sockets_to_check.recv) {
-        if (FD_ISSET(hSocket, &fdsetRecv)) {
-            sockets.recv.insert(hSocket);
+    for (Sock* sock : sockets_to_check.recv) {
+        if (FD_ISSET(sock->Get(), &fdsetRecv)) {
+            sockets.recv.insert(sock);
         }
     }
 
-    for (SOCKET hSocket : sockets_to_check.send) {
-        if (FD_ISSET(hSocket, &fdsetSend)) {
-            sockets.send.insert(hSocket);
+    for (Sock* sock : sockets_to_check.send) {
+        if (FD_ISSET(sock->Get(), &fdsetSend)) {
+            sockets.send.insert(sock);
         }
     }
 
-    for (SOCKET hSocket : sockets_to_check.err) {
-        if (FD_ISSET(hSocket, &fdsetError)) {
-            sockets.err.insert(hSocket);
+    for (Sock* sock : sockets_to_check.err) {
+        if (FD_ISSET(sock->Get(), &fdsetError)) {
+            sockets.err.insert(sock);
         }
     }
 }
@@ -1460,8 +1468,7 @@ void CConnman::SocketHandler()
     //
     for (const ListenSocket& hListenSocket : vhListenSocket)
     {
-        if (hListenSocket.sock->IsValid() && sockets.recv.count(hListenSocket.sock->Get()) > 0)
-        {
+        if (hListenSocket.sock->IsValid() && sockets.recv.count(hListenSocket.sock.get()) > 0) {
             AcceptConnection(hListenSocket);
         }
     }
@@ -1484,9 +1491,9 @@ void CConnman::SocketHandler()
             if (!pnode->m_sock->IsValid()) {
                 continue;
             }
-            recvSet = sockets.recv.count(pnode->m_sock->Get()) > 0;
-            sendSet = sockets.send.count(pnode->m_sock->Get()) > 0;
-            errorSet = sockets.err.count(pnode->m_sock->Get()) > 0;
+            recvSet = sockets.recv.count(pnode->m_sock.get()) > 0;
+            sendSet = sockets.send.count(pnode->m_sock.get()) > 0;
+            errorSet = sockets.err.count(pnode->m_sock.get()) > 0;
         }
         if (recvSet || errorSet)
         {

src/net.h

@@ -1080,9 +1080,9 @@ class CConnman
      * Used to check which of a set of sockets is ready for IO.
      */
     struct Sockets {
-        std::unordered_set<SOCKET> recv;
-        std::unordered_set<SOCKET> send;
-        std::unordered_set<SOCKET> err;
+        std::unordered_set<Sock*> recv;
+        std::unordered_set<Sock*> send;
+        std::unordered_set<Sock*> err;
     };
 
     /**