Connections and transport

2001            }
2002        }
2003    }
2004    {
2005        // Delete disconnected nodes
2006        std::list<CNode*> nodes_disconnected_copy = m_nodes_disconnected;
2007        for (CNode* pnode : nodes_disconnected_copy)
2008        {
2009            // Destroy the object only after other threads have stopped using it.
2010            if (pnode->GetRefCount() <= 0) {
2011                m_nodes_disconnected.remove(pnode);
2012                DeleteNode(pnode);
2013            }
2014        }
2015    }
2016    {
2017        // Move entries from reconnections_to_add to m_reconnections.
2018        LOCK(m_reconnections_mutex);
2019        m_reconnections.splice(m_reconnections.end(), std::move(reconnections_to_add));
2020    }
2021}
2022
2023void CConnman::NotifyNumConnectionsChanged()
2024{
2025    AssertLockNotHeld(m_nodes_mutex);
2026
2027    size_t nodes_size;
2028    {
2029        LOCK(m_nodes_mutex);
2030        nodes_size = m_nodes.size();
2031    }
2032    if(nodes_size != nPrevNodeCount) {
2033        nPrevNodeCount = nodes_size;
2034        if (m_client_interface) {
2035            m_client_interface->NotifyNumConnectionsChanged(nodes_size);
2036        }
2037    }
2038}
2039
2040bool CConnman::ShouldRunInactivityChecks(const CNode& node, NodeClock::time_point now) const
2041{
2042    return node.m_connected + m_peer_connect_timeout < now;
2043}
2044
2045bool CConnman::InactivityCheck(const CNode& node, NodeClock::time_point now) const
2046{
2047    // Tests that see disconnects after using mocktime can start nodes with a
2048    // large timeout. For example, -peertimeout=999999999.
2049    const auto last_send{node.m_last_send.load()};
2050    const auto last_recv{node.m_last_recv.load()};
2051
2052    if (!ShouldRunInactivityChecks(node, now)) return false;
2053
2054    bool has_received{last_recv > NodeClock::epoch};
2055    bool has_sent{last_send > NodeClock::epoch};
2056
2057    if (!has_received || !has_sent) {
2058        std::string has_never;
2059        if (!has_received) has_never += ", never received from peer";
2060        if (!has_sent) has_never += ", never sent to peer";
2061        LogDebug(BCLog::NET,
2062            "socket no message in first %i seconds%s, %s",
2063            count_seconds(m_peer_connect_timeout),
2064            has_never,
2065            node.DisconnectMsg()
2066        );
2067        return true;
2068    }
2069
2070    if (now > last_send + TIMEOUT_INTERVAL) {
2071        LogDebug(BCLog::NET,
2072            "socket sending timeout: %is, %s", Ticks<std::chrono::seconds>(now - last_send),
2073            node.DisconnectMsg()
2074        );
2075        return true;
2076    }
2077
2078    if (now > last_recv + TIMEOUT_INTERVAL) {
2079        LogDebug(BCLog::NET,
2080            "socket receive timeout: %is, %s", Ticks<std::chrono::seconds>(now - last_recv),
2081            node.DisconnectMsg()
2082        );
2083        return true;
2084    }
2085
2086    if (node.fPauseSend) {
2087        const auto last_pause{node.m_pause_send_since.load()};
2088        if (now > last_pause + TIMEOUT_INTERVAL) {
2089            LogDebug(BCLog::NET,
2090                "send buffer full for %is, %s", Ticks<std::chrono::seconds>(now - last_pause),
2091                node.DisconnectMsg()
2092            );
2093            return true;
2094        }
2095    }
2096
2097    if (!node.fSuccessfullyConnected) {
2098        if (node.m_transport->GetInfo().transport_type == TransportProtocolType::DETECTING) {
2099            LogDebug(BCLog::NET, "V2 handshake timeout, %s", node.DisconnectMsg());
2100        } else {
2101            LogDebug(BCLog::NET, "version handshake timeout, %s", node.DisconnectMsg());
2102        }
2103        return true;
2104    }
2105
2106    return false;
2107}
2108
2109Sock::EventsPerSock CConnman::GenerateWaitSockets(std::span<CNode* const> nodes)
2110{
2111    Sock::EventsPerSock events_per_sock;
2112
2113    for (const ListenSocket& hListenSocket : vhListenSocket) {
2114        events_per_sock.emplace(hListenSocket.sock, Sock::Events{Sock::RecvEvent});
2115    }
2116
2117    for (CNode* pnode : nodes) {
2118        bool select_recv = !pnode->fPauseRecv;
2119        bool select_send;
2120        {
2121            LOCK(pnode->cs_vSend);
2122            // Sending is possible if either there are bytes to send right now, or if there will be
2123            // once a potential message from vSendMsg is handed to the transport. GetBytesToSend
2124            // determines both of these in a single call.
2125            const auto& [to_send, more, _msg_type] = pnode->m_transport->GetBytesToSend(!pnode->vSendMsg.empty());
2126            select_send = !to_send.empty() || more;
2127        }
2128        if (!select_recv && !select_send) continue;
2129
2130        LOCK(pnode->m_sock_mutex);
2131        if (pnode->m_sock) {
2132            Sock::Event event = (select_send ? Sock::SendEvent : 0) | (select_recv ? Sock::RecvEvent : 0);
2133            events_per_sock.emplace(pnode->m_sock, Sock::Events{event});
2134        }
2135    }
2136
2137    return events_per_sock;
2138}
2139
2140void CConnman::SocketHandler()
2141{
2142    AssertLockNotHeld(m_nodes_mutex);
2143    AssertLockNotHeld(m_total_bytes_sent_mutex);
2144
2145    Sock::EventsPerSock events_per_sock;
2146
2147    {
2148        const NodesSnapshot snap{*this, /*shuffle=*/false};
2149
2150        const auto timeout = std::chrono::milliseconds(SELECT_TIMEOUT_MILLISECONDS);
2151
2152        // Check for the readiness of the already connected sockets and the
2153        // listening sockets in one call ("readiness" as in poll(2) or
2154        // select(2)). If none are ready, wait for a short while and return
2155        // empty sets.
2156        events_per_sock = GenerateWaitSockets(snap.Nodes());
2157        if (events_per_sock.empty() || !events_per_sock.begin()->first->WaitMany(timeout, events_per_sock)) {
2158            m_interrupt_net->sleep_for(timeout);
2159        }
2160
2161        // Service (send/receive) each of the already connected nodes.
2162        SocketHandlerConnected(snap.Nodes(), events_per_sock);
2163    }
2164
2165    // Accept new connections from listening sockets.
2166    SocketHandlerListening(events_per_sock);
2167}
2168
2169void CConnman::SocketHandlerConnected(const std::vector<CNode*>& nodes,
2170                                      const Sock::EventsPerSock& events_per_sock)
2171{
2172    AssertLockNotHeld(m_total_bytes_sent_mutex);
2173
2174    const auto now{NodeClock::now()};
2175
2176    for (CNode* pnode : nodes) {
2177        if (m_interrupt_net->interrupted()) {
2178            return;
2179        }
2180
2181        //
2182        // Receive
2183        //
2184        bool recvSet = false;
2185        bool sendSet = false;
2186        bool errorSet = false;
2187        {
2188            LOCK(pnode->m_sock_mutex);
2189            if (!pnode->m_sock) {
2190                continue;
2191            }
2192            const auto it = events_per_sock.find(pnode->m_sock);
2193            if (it != events_per_sock.end()) {
2194                recvSet = it->second.occurred & Sock::RecvEvent;
2195                sendSet = it->second.occurred & Sock::SendEvent;
2196                errorSet = it->second.occurred & Sock::ErrorEvent;
2197            }
2198        }
2199
2200        if (sendSet) {
2201            // Send data
2202            auto [bytes_sent, data_left] = WITH_LOCK(pnode->cs_vSend, return SocketSendData(*pnode));
2203            if (bytes_sent) {
2204                RecordBytesSent(bytes_sent);
2205
2206                // If both receiving and (non-optimistic) sending were possible, we first attempt
2207                // sending. If that succeeds, but does not fully drain the send queue, do not
2208                // attempt to receive. This avoids needlessly queueing data if the remote peer
2209                // is slow at receiving data, by means of TCP flow control. We only do this when
2210                // sending actually succeeded to make sure progress is always made; otherwise a
2211                // deadlock would be possible when both sides have data to send, but neither is
2212                // receiving.
2213                if (data_left) recvSet = false;
2214            }
2215        }
2216
2217        if (recvSet || errorSet)
2218        {
2219            // typical socket buffer is 8K-64K
2220            uint8_t pchBuf[0x10000];
2221            int nBytes = 0;
2222            {
2223                LOCK(pnode->m_sock_mutex);
2224                if (!pnode->m_sock) {
2225                    continue;
2226                }
2227                nBytes = pnode->m_sock->Recv(pchBuf, sizeof(pchBuf), MSG_DONTWAIT);
2228            }
2229            if (nBytes > 0)
2230            {
2231                bool notify = false;
2232                if (!pnode->ReceiveMsgBytes({pchBuf, (size_t)nBytes}, notify)) {
2233                    LogDebug(BCLog::NET,
2234                        "receiving message bytes failed, %s",
2235                        pnode->DisconnectMsg()
2236                    );
2237                    pnode->CloseSocketDisconnect();
2238                }
2239                RecordBytesRecv(nBytes);
2240                if (notify) {
2241                    pnode->MarkReceivedMsgsForProcessing();
2242                    WakeMessageHandler();
2243                }
2244            }
2245            else if (nBytes == 0)
2246            {
2247                // socket closed gracefully
2248                if (!pnode->fDisconnect) {
2249                    LogDebug(BCLog::NET, "socket closed, %s", pnode->DisconnectMsg());
2250                }