Skip to content

Commit 6e68ccb

Browse files
vasildvasild
committed
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 bitcoin#21744
1 parent ae26346 commit 6e68ccb

File tree

2 files changed

+36
-180
lines changed

2 files changed

+36
-180
lines changed

‎src/net.cpp‎

Lines changed: 29 additions & 149 deletions
Original file line numberDiff line numberDiff line change
@@ -1395,13 +1395,12 @@ bool CConnman::InactivityCheck(const CNode& node) const
13951395
return false;
13961396
}
13971397

1398-
bool CConnman::GenerateSelectSet(const std::vector<CNode*>& nodes,
1399-
std::set<SOCKET>& recv_set,
1400-
std::set<SOCKET>& send_set,
1401-
std::set<SOCKET>& error_set)
1398+
Sock::EventsPerSock CConnman::GenerateWaitSockets(Span<CNode* const> nodes)
14021399
{
1400+
Sock::EventsPerSock events_per_sock;
1401+
14031402
for (const ListenSocket& hListenSocket : vhListenSocket) {
1404-
recv_set.insert(hListenSocket.sock->Get());
1403+
events_per_sock.emplace(hListenSocket.sock, Sock::Events{Sock::RECV});
14051404
}
14061405

14071406
for (CNode* pnode : nodes) {
@@ -1428,172 +1427,49 @@ bool CConnman::GenerateSelectSet(const std::vector<CNode*>& nodes,
14281427
continue;
14291428
}
14301429

1431-
error_set.insert(pnode->m_sock->Get());
1430+
Sock::Event requested{0};
14321431
if (select_send) {
1433-
send_set.insert(pnode->m_sock->Get());
1434-
continue;
1435-
}
1436-
if (select_recv) {
1437-
recv_set.insert(pnode->m_sock->Get());
1438-
}
1439-
}
1440-
1441-
return !recv_set.empty() || !send_set.empty() || !error_set.empty();
1442-
}
1443-
1444-
#ifdef USE_POLL
1445-
void CConnman::SocketEvents(const std::vector<CNode*>& nodes,
1446-
std::set<SOCKET>& recv_set,
1447-
std::set<SOCKET>& send_set,
1448-
std::set<SOCKET>& error_set)
1449-
{
1450-
std::set<SOCKET> recv_select_set, send_select_set, error_select_set;
1451-
if (!GenerateSelectSet(nodes, recv_select_set, send_select_set, error_select_set)) {
1452-
interruptNet.sleep_for(std::chrono::milliseconds(SELECT_TIMEOUT_MILLISECONDS));
1453-
return;
1454-
}
1455-
1456-
std::unordered_map<SOCKET, struct pollfd> pollfds;
1457-
for (SOCKET socket_id : recv_select_set) {
1458-
pollfds[socket_id].fd = socket_id;
1459-
pollfds[socket_id].events |= POLLIN;
1460-
}
1461-
1462-
for (SOCKET socket_id : send_select_set) {
1463-
pollfds[socket_id].fd = socket_id;
1464-
pollfds[socket_id].events |= POLLOUT;
1465-
}
1466-
1467-
for (SOCKET socket_id : error_select_set) {
1468-
pollfds[socket_id].fd = socket_id;
1469-
// These flags are ignored, but we set them for clarity
1470-
pollfds[socket_id].events |= POLLERR|POLLHUP;
1471-
}
1472-
1473-
std::vector<struct pollfd> vpollfds;
1474-
vpollfds.reserve(pollfds.size());
1475-
for (auto it : pollfds) {
1476-
vpollfds.push_back(std::move(it.second));
1477-
}
1478-
1479-
if (poll(vpollfds.data(), vpollfds.size(), SELECT_TIMEOUT_MILLISECONDS) < 0) return;
1480-
1481-
if (interruptNet) return;
1482-
1483-
for (struct pollfd pollfd_entry : vpollfds) {
1484-
if (pollfd_entry.revents & POLLIN) recv_set.insert(pollfd_entry.fd);
1485-
if (pollfd_entry.revents & POLLOUT) send_set.insert(pollfd_entry.fd);
1486-
if (pollfd_entry.revents & (POLLERR|POLLHUP)) error_set.insert(pollfd_entry.fd);
1487-
}
1488-
}
1489-
#else
1490-
void CConnman::SocketEvents(const std::vector<CNode*>& nodes,
1491-
std::set<SOCKET>& recv_set,
1492-
std::set<SOCKET>& send_set,
1493-
std::set<SOCKET>& error_set)
1494-
{
1495-
std::set<SOCKET> recv_select_set, send_select_set, error_select_set;
1496-
if (!GenerateSelectSet(nodes, recv_select_set, send_select_set, error_select_set)) {
1497-
interruptNet.sleep_for(std::chrono::milliseconds(SELECT_TIMEOUT_MILLISECONDS));
1498-
return;
1499-
}
1500-
1501-
//
1502-
// Find which sockets have data to receive
1503-
//
1504-
struct timeval timeout;
1505-
timeout.tv_sec = 0;
1506-
timeout.tv_usec = SELECT_TIMEOUT_MILLISECONDS * 1000; // frequency to poll pnode->vSend
1507-
1508-
fd_set fdsetRecv;
1509-
fd_set fdsetSend;
1510-
fd_set fdsetError;
1511-
FD_ZERO(&fdsetRecv);
1512-
FD_ZERO(&fdsetSend);
1513-
FD_ZERO(&fdsetError);
1514-
SOCKET hSocketMax = 0;
1515-
1516-
for (SOCKET hSocket : recv_select_set) {
1517-
FD_SET(hSocket, &fdsetRecv);
1518-
hSocketMax = std::max(hSocketMax, hSocket);
1519-
}
1520-
1521-
for (SOCKET hSocket : send_select_set) {
1522-
FD_SET(hSocket, &fdsetSend);
1523-
hSocketMax = std::max(hSocketMax, hSocket);
1524-
}
1525-
1526-
for (SOCKET hSocket : error_select_set) {
1527-
FD_SET(hSocket, &fdsetError);
1528-
hSocketMax = std::max(hSocketMax, hSocket);
1529-
}
1530-
1531-
int nSelect = select(hSocketMax + 1, &fdsetRecv, &fdsetSend, &fdsetError, &timeout);
1532-
1533-
if (interruptNet)
1534-
return;
1535-
1536-
if (nSelect == SOCKET_ERROR)
1537-
{
1538-
int nErr = WSAGetLastError();
1539-
LogPrintf("socket select error %s\n", NetworkErrorString(nErr));
1540-
for (unsigned int i = 0; i <= hSocketMax; i++)
1541-
FD_SET(i, &fdsetRecv);
1542-
FD_ZERO(&fdsetSend);
1543-
FD_ZERO(&fdsetError);
1544-
if (!interruptNet.sleep_for(std::chrono::milliseconds(SELECT_TIMEOUT_MILLISECONDS)))
1545-
return;
1546-
}
1547-
1548-
for (SOCKET hSocket : recv_select_set) {
1549-
if (FD_ISSET(hSocket, &fdsetRecv)) {
1550-
recv_set.insert(hSocket);
1432+
requested = Sock::SEND;
1433+
} else if (select_recv) {
1434+
requested = Sock::RECV;
15511435
}
1552-
}
15531436

1554-
for (SOCKET hSocket : send_select_set) {
1555-
if (FD_ISSET(hSocket, &fdsetSend)) {
1556-
send_set.insert(hSocket);
1557-
}
1437+
events_per_sock.emplace(pnode->m_sock, Sock::Events{requested});
15581438
}
15591439

1560-
for (SOCKET hSocket : error_select_set) {
1561-
if (FD_ISSET(hSocket, &fdsetError)) {
1562-
error_set.insert(hSocket);
1563-
}
1564-
}
1440+
return events_per_sock;
15651441
}
1566-
#endif
15671442

15681443
void CConnman::SocketHandler()
15691444
{
15701445
AssertLockNotHeld(m_total_bytes_sent_mutex);
15711446

1572-
std::set<SOCKET> recv_set;
1573-
std::set<SOCKET> send_set;
1574-
std::set<SOCKET> error_set;
1447+
Sock::EventsPerSock events_per_sock;
15751448

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

1452+
const auto timeout = std::chrono::milliseconds(SELECT_TIMEOUT_MILLISECONDS);
1453+
15791454
// Check for the readiness of the already connected sockets and the
15801455
// listening sockets in one call ("readiness" as in poll(2) or
15811456
// select(2)). If none are ready, wait for a short while and return
15821457
// empty sets.
1583-
SocketEvents(snap.Nodes(), recv_set, send_set, error_set);
1458+
events_per_sock = GenerateWaitSockets(snap.Nodes());
1459+
if (events_per_sock.empty() || !events_per_sock.begin()->first->WaitMany(timeout, events_per_sock)) {
1460+
interruptNet.sleep_for(timeout);
1461+
}
15841462

15851463
// Service (send/receive) each of the already connected nodes.
1586-
SocketHandlerConnected(snap.Nodes(), recv_set, send_set, error_set);
1464+
SocketHandlerConnected(snap.Nodes(), events_per_sock);
15871465
}
15881466

15891467
// Accept new connections from listening sockets.
1590-
SocketHandlerListening(recv_set);
1468+
SocketHandlerListening(events_per_sock);
15911469
}
15921470

15931471
void CConnman::SocketHandlerConnected(const std::vector<CNode*>& nodes,
1594-
const std::set<SOCKET>& recv_set,
1595-
const std::set<SOCKET>& send_set,
1596-
const std::set<SOCKET>& error_set)
1472+
const Sock::EventsPerSock& events_per_sock)
15971473
{
15981474
AssertLockNotHeld(m_total_bytes_sent_mutex);
15991475

@@ -1612,9 +1488,12 @@ void CConnman::SocketHandlerConnected(const std::vector<CNode*>& nodes,
16121488
if (!pnode->m_sock) {
16131489
continue;
16141490
}
1615-
recvSet = recv_set.count(pnode->m_sock->Get()) > 0;
1616-
sendSet = send_set.count(pnode->m_sock->Get()) > 0;
1617-
errorSet = error_set.count(pnode->m_sock->Get()) > 0;
1491+
const auto it = events_per_sock.find(pnode->m_sock);
1492+
if (it != events_per_sock.end()) {
1493+
recvSet = it->second.occurred & Sock::RECV;
1494+
sendSet = it->second.occurred & Sock::SEND;
1495+
errorSet = it->second.occurred & Sock::ERR;
1496+
}
16181497
}
16191498
if (recvSet || errorSet)
16201499
{
@@ -1684,13 +1563,14 @@ void CConnman::SocketHandlerConnected(const std::vector<CNode*>& nodes,
16841563
}
16851564
}
16861565

1687-
void CConnman::SocketHandlerListening(const std::set<SOCKET>& recv_set)
1566+
void CConnman::SocketHandlerListening(const Sock::EventsPerSock& events_per_sock)
16881567
{
16891568
for (const ListenSocket& listen_socket : vhListenSocket) {
16901569
if (interruptNet) {
16911570
return;
16921571
}
1693-
if (recv_set.count(listen_socket.sock->Get()) > 0) {
1572+
const auto it = events_per_sock.find(listen_socket.sock);
1573+
if (it != events_per_sock.end() && it->second.occurred & Sock::RECV) {
16941574
AcceptConnection(listen_socket);
16951575
}
16961576
}

‎src/net.h‎

Lines changed: 7 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -980,28 +980,9 @@ class CConnman
980980
/**
981981
* Generate a collection of sockets to check for IO readiness.
982982
* @param[in] nodes Select from these nodes' sockets.
983-
* @param[out] recv_set Sockets to check for read readiness.
984-
* @param[out] send_set Sockets to check for write readiness.
985-
* @param[out] error_set Sockets to check for errors.
986-
* @return true if at least one socket is to be checked (the returned set is not empty)
983+
* @return sockets to check for readiness
987984
*/
988-
bool GenerateSelectSet(const std::vector<CNode*>& nodes,
989-
std::set<SOCKET>& recv_set,
990-
std::set<SOCKET>& send_set,
991-
std::set<SOCKET>& error_set);
992-
993-
/**
994-
* Check which sockets are ready for IO.
995-
* @param[in] nodes Select from these nodes' sockets.
996-
* @param[out] recv_set Sockets which are ready for read.
997-
* @param[out] send_set Sockets which are ready for write.
998-
* @param[out] error_set Sockets which have errors.
999-
* This calls `GenerateSelectSet()` to gather a list of sockets to check.
1000-
*/
1001-
void SocketEvents(const std::vector<CNode*>& nodes,
1002-
std::set<SOCKET>& recv_set,
1003-
std::set<SOCKET>& send_set,
1004-
std::set<SOCKET>& error_set);
985+
Sock::EventsPerSock GenerateWaitSockets(Span<CNode* const> nodes);
1005986

1006987
/**
1007988
* Check connected and listening sockets for IO readiness and process them accordingly.
@@ -1010,23 +991,18 @@ class CConnman
1010991

1011992
/**
1012993
* Do the read/write for connected sockets that are ready for IO.
1013-
* @param[in] nodes Nodes to process. The socket of each node is checked against
1014-
* `recv_set`, `send_set` and `error_set`.
1015-
* @param[in] recv_set Sockets that are ready for read.
1016-
* @param[in] send_set Sockets that are ready for send.
1017-
* @param[in] error_set Sockets that have an exceptional condition (error).
994+
* @param[in] nodes Nodes to process. The socket of each node is checked against `what`.
995+
* @param[in] events_per_sock Sockets that are ready for IO.
1018996
*/
1019997
void SocketHandlerConnected(const std::vector<CNode*>& nodes,
1020-
const std::set<SOCKET>& recv_set,
1021-
const std::set<SOCKET>& send_set,
1022-
const std::set<SOCKET>& error_set)
998+
const Sock::EventsPerSock& events_per_sock)
1023999
EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex, !mutexMsgProc);
10241000

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

10311007
void ThreadSocketHandler() EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex, !mutexMsgProc);
10321008
void ThreadDNSAddressSeed() EXCLUSIVE_LOCKS_REQUIRED(!m_addr_fetches_mutex, !m_nodes_mutex);

0 commit comments

Comments
 (0)