From e04d8508ec97853ccc5d3f4992281e99e1c58e3c Mon Sep 17 00:00:00 2001 From: Vasil Dimov Date: Tue, 4 May 2021 18:37:19 +0200 Subject: [PATCH] net: use Sock::WaitMany() instead of CConnman::SocketEvents() Rename `GenerateSelectSet()` to `GenerateWaitSockets()` and adapt it to generate a wait data suitable for `Sock::WaitMany()`. Then call it from `CConnman::SocketHandler()` and feed the generated data to `Sock::WaitMany()`. This way `CConnman::SocketHandler()` can be unit tested because `Sock::WaitMany()` is mockable, so the usage of real sockets can be avoided. Resolves https://github.com/bitcoin/bitcoin/issues/21744 --- src/net.cpp | 152 +++++++--------------------------------------------- src/net.h | 17 +----- 2 files changed, 22 insertions(+), 147 deletions(-) diff --git a/src/net.cpp b/src/net.cpp index d79d8549112aab..57acf1d192e6fb 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -1327,10 +1327,12 @@ bool CConnman::InactivityCheck(const CNode& node) const return false; } -bool CConnman::GenerateSelectSet(const std::vector& nodes, std::set &recv_set, std::set &send_set, std::set &error_set) +Sock::WaitData CConnman::GenerateWaitSockets(const std::vector& nodes) { + Sock::WaitData what; + for (const ListenSocket& hListenSocket : vhListenSocket) { - recv_set.insert(hListenSocket.sock->Get()); + what.emplace(hListenSocket.sock, Sock::WaitEvents{Sock::RECV, 0}); } for (CNode* pnode : nodes) { @@ -1357,143 +1359,29 @@ bool CConnman::GenerateSelectSet(const std::vector& nodes, std::setm_sock->Get()); + Sock::Event requested{0}; if (select_send) { - send_set.insert(pnode->m_sock->Get()); - continue; - } - if (select_recv) { - recv_set.insert(pnode->m_sock->Get()); - } - } - - return !recv_set.empty() || !send_set.empty() || !error_set.empty(); -} - -#ifdef USE_POLL -void CConnman::SocketEvents(const std::vector& nodes, std::set &recv_set, std::set &send_set, std::set &error_set) -{ - std::set recv_select_set, send_select_set, error_select_set; - if (!GenerateSelectSet(nodes, recv_select_set, send_select_set, error_select_set)) { - interruptNet.sleep_for(std::chrono::milliseconds(SELECT_TIMEOUT_MILLISECONDS)); - return; - } - - std::unordered_map pollfds; - for (SOCKET socket_id : recv_select_set) { - pollfds[socket_id].fd = socket_id; - pollfds[socket_id].events |= POLLIN; - } - - for (SOCKET socket_id : send_select_set) { - pollfds[socket_id].fd = socket_id; - pollfds[socket_id].events |= POLLOUT; - } - - for (SOCKET socket_id : error_select_set) { - pollfds[socket_id].fd = socket_id; - // These flags are ignored, but we set them for clarity - pollfds[socket_id].events |= POLLERR|POLLHUP; - } - - std::vector vpollfds; - vpollfds.reserve(pollfds.size()); - for (auto it : pollfds) { - vpollfds.push_back(std::move(it.second)); - } - - if (poll(vpollfds.data(), vpollfds.size(), SELECT_TIMEOUT_MILLISECONDS) < 0) return; - - if (interruptNet) return; - - for (struct pollfd pollfd_entry : vpollfds) { - if (pollfd_entry.revents & POLLIN) recv_set.insert(pollfd_entry.fd); - if (pollfd_entry.revents & POLLOUT) send_set.insert(pollfd_entry.fd); - if (pollfd_entry.revents & (POLLERR|POLLHUP)) error_set.insert(pollfd_entry.fd); - } -} -#else -void CConnman::SocketEvents(const std::vector& nodes, std::set &recv_set, std::set &send_set, std::set &error_set) -{ - std::set recv_select_set, send_select_set, error_select_set; - if (!GenerateSelectSet(nodes, recv_select_set, send_select_set, error_select_set)) { - interruptNet.sleep_for(std::chrono::milliseconds(SELECT_TIMEOUT_MILLISECONDS)); - return; - } - - // - // Find which sockets have data to receive - // - struct timeval timeout; - timeout.tv_sec = 0; - timeout.tv_usec = SELECT_TIMEOUT_MILLISECONDS * 1000; // frequency to poll pnode->vSend - - fd_set fdsetRecv; - fd_set fdsetSend; - fd_set fdsetError; - FD_ZERO(&fdsetRecv); - FD_ZERO(&fdsetSend); - FD_ZERO(&fdsetError); - SOCKET hSocketMax = 0; - - for (SOCKET hSocket : recv_select_set) { - FD_SET(hSocket, &fdsetRecv); - hSocketMax = std::max(hSocketMax, hSocket); - } - - for (SOCKET hSocket : send_select_set) { - FD_SET(hSocket, &fdsetSend); - hSocketMax = std::max(hSocketMax, hSocket); - } - - for (SOCKET hSocket : error_select_set) { - FD_SET(hSocket, &fdsetError); - hSocketMax = std::max(hSocketMax, hSocket); - } - - int nSelect = select(hSocketMax + 1, &fdsetRecv, &fdsetSend, &fdsetError, &timeout); - - if (interruptNet) - return; - - if (nSelect == SOCKET_ERROR) - { - int nErr = WSAGetLastError(); - LogPrintf("socket select error %s\n", NetworkErrorString(nErr)); - for (unsigned int i = 0; i <= hSocketMax; i++) - FD_SET(i, &fdsetRecv); - FD_ZERO(&fdsetSend); - FD_ZERO(&fdsetError); - if (!interruptNet.sleep_for(std::chrono::milliseconds(SELECT_TIMEOUT_MILLISECONDS))) - return; - } - - for (SOCKET hSocket : recv_select_set) { - if (FD_ISSET(hSocket, &fdsetRecv)) { - recv_set.insert(hSocket); + requested = Sock::SEND; + } else if (select_recv) { + requested = Sock::RECV; } - } - for (SOCKET hSocket : send_select_set) { - if (FD_ISSET(hSocket, &fdsetSend)) { - send_set.insert(hSocket); - } + what.emplace(pnode->m_sock, Sock::WaitEvents{requested, 0}); } - for (SOCKET hSocket : error_select_set) { - if (FD_ISSET(hSocket, &fdsetError)) { - error_set.insert(hSocket); - } - } + return what; } -#endif void CConnman::SocketHandler() { const NodesSnapshot snap{*this, false}; - std::set recv_set, send_set, error_set; - SocketEvents(snap.Nodes(), recv_set, send_set, error_set); + const auto timeout = std::chrono::milliseconds(SELECT_TIMEOUT_MILLISECONDS); + + auto what = GenerateWaitSockets(snap.Nodes()); + if (what.empty() || !what.begin()->first->WaitMany(timeout, what)) { + interruptNet.sleep_for(timeout); + } if (interruptNet) return; @@ -1508,7 +1396,7 @@ void CConnman::SocketHandler() // for (const ListenSocket& hListenSocket : vhListenSocket) { - if (hListenSocket.sock->Get() != INVALID_SOCKET && recv_set.count(hListenSocket.sock->Get()) > 0) { + if (hListenSocket.sock->Get() != INVALID_SOCKET && what[hListenSocket.sock].occurred & Sock::RECV) { AcceptConnection(hListenSocket); } } @@ -1531,9 +1419,9 @@ void CConnman::SocketHandler() if (!pnode->m_sock) { continue; } - recvSet = recv_set.count(pnode->m_sock->Get()) > 0; - sendSet = send_set.count(pnode->m_sock->Get()) > 0; - errorSet = error_set.count(pnode->m_sock->Get()) > 0; + recvSet = what[pnode->m_sock].occurred & Sock::RECV; + sendSet = what[pnode->m_sock].occurred & Sock::SEND; + errorSet = what[pnode->m_sock].occurred & Sock::ERR; } if (recvSet || errorSet) { diff --git a/src/net.h b/src/net.h index 4f16ce65360746..da2094d1836e2d 100644 --- a/src/net.h +++ b/src/net.h @@ -1002,22 +1002,9 @@ class CConnman /** * Generate a collection of sockets that we want to be checked for readiness for IO. * @param[in] nodes Select from these nodes' sockets. - * @param[out] recv_set Sockets to check for read readiness. - * @param[out] send_set Sockets to check for write readiness. - * @param[out] error_set Sockets to check for errors. - * @return true if at least one socket is to be checked (the returned set is not empty) + * @return sockets to check for readiness */ - bool GenerateSelectSet(const std::vector& nodes, std::set &recv_set, std::set &send_set, std::set &error_set); - - /** - * Check which sockets are ready for IO. - * @param[in] nodes Select from these nodes' sockets. - * @param[out] recv_set Sockets which are ready for read. - * @param[out] send_set Sockets which are ready for write. - * @param[out] error_set Sockets which have errors. - * This calls `GenerateSelectSet()` to gather a list of sockets to check. - */ - void SocketEvents(const std::vector& nodes, std::set &recv_set, std::set &send_set, std::set &error_set); + Sock::WaitData GenerateWaitSockets(const std::vector& nodes); void SocketHandler(); void ThreadSocketHandler();