Skip to content

Commit

Permalink
net: use Sock::WaitMany() instead of CConnman::SocketEvents()
Browse files Browse the repository at this point in the history 
Summary:
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 bitcoin/bitcoin#21744

This concludes backport of [[bitcoin/bitcoin#24356 | core#24356]]
bitcoin/bitcoin@6e68ccb

Depends on D17133

Test Plan: `ninja all check-all`

Reviewers: #bitcoin_abc, Fabien

Reviewed By: #bitcoin_abc, Fabien

Differential Revision: https://reviews.bitcoinabc.org/D17134
  • Loading branch information
@vasild @PiRK
vasild authored and PiRK committed Nov 18, 2024
1 parent 9f4a3fa commit adc9c8e
Show file tree
Hide file tree
Showing 2 changed files with 42 additions and 198 deletions.
203 changes: 35 additions & 168 deletions src/net.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1567,12 +1567,11 @@ bool CConnman::InactivityCheck(const CNode &node) const {
return false;
}

bool CConnman::GenerateSelectSet(const std::vector<CNode *> &nodes,
std::set<SOCKET> &recv_set,
std::set<SOCKET> &send_set,
std::set<SOCKET> &error_set) {
Sock::EventsPerSock CConnman::GenerateWaitSockets(Span<CNode *const> nodes) {
Sock::EventsPerSock events_per_sock;

for (const ListenSocket &hListenSocket : vhListenSocket) {
recv_set.insert(hListenSocket.sock->Get());
events_per_sock.emplace(hListenSocket.sock, Sock::Events{Sock::RECV});
}

for (CNode *pnode : nodes) {
Expand Down Expand Up @@ -1600,187 +1599,50 @@ bool CConnman::GenerateSelectSet(const std::vector<CNode *> &nodes,
continue;
}

error_set.insert(pnode->m_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<CNode *> &nodes,
std::set<SOCKET> &recv_set,
std::set<SOCKET> &send_set,
std::set<SOCKET> &error_set) {
std::set<SOCKET> 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<SOCKET, struct pollfd> 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<struct pollfd> 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<CNode *> &nodes,
std::set<SOCKET> &recv_set,
std::set<SOCKET> &send_set,
std::set<SOCKET> &error_set) {
std::set<SOCKET> 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;
// frequency to poll pnode->vSend
timeout.tv_usec = SELECT_TIMEOUT_MILLISECONDS * 1000;

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);
}
events_per_sock.emplace(pnode->m_sock, Sock::Events{requested});
}

for (SOCKET hSocket : error_select_set) {
if (FD_ISSET(hSocket, &fdsetError)) {
error_set.insert(hSocket);
}
}
return events_per_sock;
}
#endif

void CConnman::SocketHandler() {
std::set<SOCKET> recv_set;
std::set<SOCKET> send_set;
std::set<SOCKET> error_set;
Sock::EventsPerSock events_per_sock;

{
const NodesSnapshot snap{*this, /*shuffle=*/false};

const auto timeout =
std::chrono::milliseconds(SELECT_TIMEOUT_MILLISECONDS);

// Check for the readiness of the already connected sockets and the
// listening sockets in one call ("readiness" as in poll(2) or
// select(2)). If none are ready, wait for a short while and return
// empty sets.
SocketEvents(snap.Nodes(), recv_set, send_set, error_set);
events_per_sock = GenerateWaitSockets(snap.Nodes());
if (events_per_sock.empty() ||
!events_per_sock.begin()->first->WaitMany(timeout,
events_per_sock)) {
interruptNet.sleep_for(timeout);
}

// Service (send/receive) each of the already connected nodes.
SocketHandlerConnected(snap.Nodes(), recv_set, send_set, error_set);
SocketHandlerConnected(snap.Nodes(), events_per_sock);
}

// Accept new connections from listening sockets.
SocketHandlerListening(recv_set);
SocketHandlerListening(events_per_sock);
}

void CConnman::SocketHandlerConnected(const std::vector<CNode *> &nodes,
const std::set<SOCKET> &recv_set,
const std::set<SOCKET> &send_set,
const std::set<SOCKET> &error_set) {
void CConnman::SocketHandlerConnected(
const std::vector<CNode *> &nodes,
const Sock::EventsPerSock &events_per_sock) {
for (CNode *pnode : nodes) {
if (interruptNet) {
return;
Expand All @@ -1797,9 +1659,12 @@ void CConnman::SocketHandlerConnected(const std::vector<CNode *> &nodes,
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;
const auto it = events_per_sock.find(pnode->m_sock);
if (it != events_per_sock.end()) {
recvSet = it->second.occurred & Sock::RECV;
sendSet = it->second.occurred & Sock::SEND;
errorSet = it->second.occurred & Sock::ERR;
}
}
if (recvSet || errorSet) {
// typical socket buffer is 8K-64K
Expand Down Expand Up @@ -1877,12 +1742,14 @@ void CConnman::SocketHandlerConnected(const std::vector<CNode *> &nodes,
}
}

void CConnman::SocketHandlerListening(const std::set<SOCKET> &recv_set) {
void CConnman::SocketHandlerListening(
const Sock::EventsPerSock &events_per_sock) {
for (const ListenSocket &listen_socket : vhListenSocket) {
if (interruptNet) {
return;
}
if (recv_set.count(listen_socket.sock->Get()) > 0) {
const auto it = events_per_sock.find(listen_socket.sock);
if (it != events_per_sock.end() && it->second.occurred & Sock::RECV) {
AcceptConnection(listen_socket);
}
}
Expand Down
37 changes: 7 additions & 30 deletions src/net.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1146,28 +1146,9 @@ class CConnman {
/**
* Generate a collection of sockets to check for IO readiness.
* @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<CNode *> &nodes,
std::set<SOCKET> &recv_set,
std::set<SOCKET> &send_set,
std::set<SOCKET> &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<CNode *> &nodes,
std::set<SOCKET> &recv_set, std::set<SOCKET> &send_set,
std::set<SOCKET> &error_set);
Sock::EventsPerSock GenerateWaitSockets(Span<CNode *const> nodes);

/**
* Check connected and listening sockets for IO readiness and process them
Expand All @@ -1178,22 +1159,18 @@ class CConnman {
/**
* Do the read/write for connected sockets that are ready for IO.
* @param[in] nodes Nodes to process. The socket of each node is checked
* against `recv_set`, `send_set` and `error_set`.
* @param[in] recv_set Sockets that are ready for read.
* @param[in] send_set Sockets that are ready for send.
* @param[in] error_set Sockets that have an exceptional condition (error).
* against `what`.
* @param[in] events_per_sock Sockets that are ready for IO.
*/
void SocketHandlerConnected(const std::vector<CNode *> &nodes,
const std::set<SOCKET> &recv_set,
const std::set<SOCKET> &send_set,
const std::set<SOCKET> &error_set)
const Sock::EventsPerSock &events_per_sock)
EXCLUSIVE_LOCKS_REQUIRED(!mutexMsgProc);

/**
* Accept incoming connections, one from each read-ready listening socket.
* @param[in] recv_set Sockets that are ready for read.
* @param[in] events_per_sock Sockets that are ready for IO.
*/
void SocketHandlerListening(const std::set<SOCKET> &recv_set);
void SocketHandlerListening(const Sock::EventsPerSock &events_per_sock);

void ThreadSocketHandler() EXCLUSIVE_LOCKS_REQUIRED(!mutexMsgProc);
void ThreadDNSAddressSeed()
Expand Down

0 comments on commit adc9c8e

Please sign in to comment.