Connections and transport

4001    return nMaxOutboundLimit;
4002}
4003
4004std::chrono::seconds CConnman::GetMaxOutboundTimeframe() const
4005{
4006    return MAX_UPLOAD_TIMEFRAME;
4007}
4008
4009std::chrono::seconds CConnman::GetMaxOutboundTimeLeftInCycle() const
4010{
4011    AssertLockNotHeld(m_total_bytes_sent_mutex);
4012    LOCK(m_total_bytes_sent_mutex);
4013    return GetMaxOutboundTimeLeftInCycle_();
4014}
4015
4016std::chrono::seconds CConnman::GetMaxOutboundTimeLeftInCycle_() const
4017{
4018    AssertLockHeld(m_total_bytes_sent_mutex);
4019
4020    if (nMaxOutboundLimit == 0)
4021        return 0s;
4022
4023    if (nMaxOutboundCycleStartTime.count() == 0)
4024        return MAX_UPLOAD_TIMEFRAME;
4025
4026    const std::chrono::seconds cycleEndTime = nMaxOutboundCycleStartTime + MAX_UPLOAD_TIMEFRAME;
4027    const auto now = GetTime<std::chrono::seconds>();
4028    return (cycleEndTime < now) ? 0s : cycleEndTime - now;
4029}
4030
4031bool CConnman::OutboundTargetReached(bool historicalBlockServingLimit) const
4032{
4033    AssertLockNotHeld(m_total_bytes_sent_mutex);
4034    LOCK(m_total_bytes_sent_mutex);
4035    if (nMaxOutboundLimit == 0)
4036        return false;
4037
4038    if (historicalBlockServingLimit)
4039    {
4040        // keep a large enough buffer to at least relay each block once
4041        const std::chrono::seconds timeLeftInCycle = GetMaxOutboundTimeLeftInCycle_();
4042        const uint64_t buffer = timeLeftInCycle / std::chrono::minutes{10} * MAX_BLOCK_SERIALIZED_SIZE;
4043        if (buffer >= nMaxOutboundLimit || nMaxOutboundTotalBytesSentInCycle >= nMaxOutboundLimit - buffer)
4044            return true;
4045    }
4046    else if (nMaxOutboundTotalBytesSentInCycle >= nMaxOutboundLimit)
4047        return true;
4048
4049    return false;
4050}
4051
4052uint64_t CConnman::GetOutboundTargetBytesLeft() const
4053{
4054    AssertLockNotHeld(m_total_bytes_sent_mutex);
4055    LOCK(m_total_bytes_sent_mutex);
4056    if (nMaxOutboundLimit == 0)
4057        return 0;
4058
4059    return (nMaxOutboundTotalBytesSentInCycle >= nMaxOutboundLimit) ? 0 : nMaxOutboundLimit - nMaxOutboundTotalBytesSentInCycle;
4060}
4061
4062uint64_t CConnman::GetTotalBytesRecv() const
4063{
4064    return nTotalBytesRecv;
4065}
4066
4067uint64_t CConnman::GetTotalBytesSent() const
4068{
4069    AssertLockNotHeld(m_total_bytes_sent_mutex);
4070    LOCK(m_total_bytes_sent_mutex);
4071    return nTotalBytesSent;
4072}
4073
4074ServiceFlags CConnman::GetLocalServices() const
4075{
4076    return m_local_services;
4077}
4078
4079static std::unique_ptr<Transport> MakeTransport(NodeId id, bool use_v2transport, bool inbound) noexcept
4080{
4081    if (use_v2transport) {
4082        return std::make_unique<V2Transport>(id, /*initiating=*/!inbound);
4083    } else {
4084        return std::make_unique<V1Transport>(id);
4085    }
4086}
4087
4088CNode::CNode(NodeId idIn,
4089             std::shared_ptr<Sock> sock,
4090             const CAddress& addrIn,
4091             uint64_t nKeyedNetGroupIn,
4092             uint64_t nLocalHostNonceIn,
4093             const CService& addrBindIn,
4094             const std::string& addrNameIn,
4095             ConnectionType conn_type_in,
4096             bool inbound_onion,
4097             uint64_t network_key,
4098             CNodeOptions&& node_opts)
4099    : m_transport{MakeTransport(idIn, node_opts.use_v2transport, conn_type_in == ConnectionType::INBOUND)},
4100      m_permission_flags{node_opts.permission_flags},
4101      m_sock{sock},
4102      m_connected{NodeClock::now()},
4103      m_proxy_override{std::move(node_opts.proxy_override)},
4104      addr{addrIn},
4105      addrBind{addrBindIn},
4106      m_addr_name{addrNameIn.empty() ? addr.ToStringAddrPort() : addrNameIn},
4107      m_dest(addrNameIn),
4108      m_inbound_onion{inbound_onion},
4109      m_prefer_evict{node_opts.prefer_evict},
4110      nKeyedNetGroup{nKeyedNetGroupIn},
4111      m_network_key{network_key},
4112      m_conn_type{conn_type_in},
4113      id{idIn},
4114      nLocalHostNonce{nLocalHostNonceIn},
4115      m_recv_flood_size{node_opts.recv_flood_size},
4116      m_i2p_sam_session{std::move(node_opts.i2p_sam_session)}
4117{
4118    if (inbound_onion) assert(conn_type_in == ConnectionType::INBOUND);
4119
4120    for (const auto& msg : ALL_NET_MESSAGE_TYPES) {
4121        mapRecvBytesPerMsgType[msg] = 0;
4122    }
4123    mapRecvBytesPerMsgType[NET_MESSAGE_TYPE_OTHER] = 0;
4124
4125    if (fLogIPs) {
4126        LogDebug(BCLog::NET, "Added connection to %s peer=%d\n", m_addr_name, id);
4127    } else {
4128        LogDebug(BCLog::NET, "Added connection peer=%d\n", id);
4129    }
4130}
4131
4132void CNode::MarkReceivedMsgsForProcessing()
4133{
4134    AssertLockNotHeld(m_msg_process_queue_mutex);
4135
4136    size_t nSizeAdded = 0;
4137    for (const auto& msg : vRecvMsg) {
4138        // vRecvMsg contains only completed CNetMessage
4139        // the single possible partially deserialized message are held by TransportDeserializer
4140        nSizeAdded += msg.GetMemoryUsage();
4141    }
4142
4143    LOCK(m_msg_process_queue_mutex);
4144    m_msg_process_queue.splice(m_msg_process_queue.end(), vRecvMsg);
4145    m_msg_process_queue_size += nSizeAdded;
4146    fPauseRecv = m_msg_process_queue_size > m_recv_flood_size;
4147}
4148
4149std::optional<std::pair<CNetMessage, bool>> CNode::PollMessage()
4150{
4151    LOCK(m_msg_process_queue_mutex);
4152    if (m_msg_process_queue.empty()) return std::nullopt;
4153
4154    std::list<CNetMessage> msgs;
4155    // Just take one message
4156    msgs.splice(msgs.begin(), m_msg_process_queue, m_msg_process_queue.begin());
4157    m_msg_process_queue_size -= msgs.front().GetMemoryUsage();
4158    fPauseRecv = m_msg_process_queue_size > m_recv_flood_size;
4159
4160    return std::make_pair(std::move(msgs.front()), !m_msg_process_queue.empty());
4161}
4162
4163bool CConnman::NodeFullyConnected(const CNode* pnode)
4164{
4165    return pnode && pnode->fSuccessfullyConnected && !pnode->fDisconnect;
4166}
4167
4168/// Private broadcast connections only need to send certain message types.
4169/// Other messages are not needed and may degrade privacy.
4170static bool IsOutboundMessageAllowedInPrivateBroadcast(std::string_view type) noexcept
4171{
4172    return type == NetMsgType::VERSION ||
4173           type == NetMsgType::VERACK ||
4174           type == NetMsgType::INV ||
4175           type == NetMsgType::TX ||
4176           type == NetMsgType::PING;
4177}
4178
4179void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
4180{
4181    AssertLockNotHeld(m_total_bytes_sent_mutex);
4182
4183    if (pnode->IsPrivateBroadcastConn() && !IsOutboundMessageAllowedInPrivateBroadcast(msg.m_type)) {
4184        LogDebug(BCLog::PRIVBROADCAST, "Omitting send of message '%s', %s", msg.m_type, pnode->LogPeer());
4185        return;
4186    }
4187
4188    if (!m_private_broadcast.m_outbound_tor_ok_at_least_once.load() && !pnode->IsInboundConn() &&
4189        pnode->addr.IsTor() && msg.m_type == NetMsgType::VERACK) {
4190        // If we are sending the peer VERACK that means we successfully sent
4191        // and received another message to/from that peer (VERSION).
4192        m_private_broadcast.m_outbound_tor_ok_at_least_once.store(true);
4193    }
4194
4195    size_t nMessageSize = msg.data.size();
4196    LogDebug(BCLog::NET, "sending %s (%d bytes) peer=%d\n", msg.m_type, nMessageSize, pnode->GetId());
4197    if (m_capture_messages) {
4198        CaptureMessage(pnode->addr, msg.m_type, msg.data, /*is_incoming=*/false);
4199    }
4200
4201    TRACEPOINT(net, outbound_message,
4202        pnode->GetId(),
4203        pnode->m_addr_name.c_str(),
4204        pnode->ConnectionTypeAsString().c_str(),
4205        msg.m_type.c_str(),
4206        msg.data.size(),
4207        msg.data.data()
4208    );
4209
4210    size_t nBytesSent = 0;
4211    {
4212        LOCK(pnode->cs_vSend);
4213        // Check if the transport still has unsent bytes, and indicate to it that we're about to
4214        // give it a message to send.
4215        const auto& [to_send, more, _msg_type] =
4216            pnode->m_transport->GetBytesToSend(/*have_next_message=*/true);
4217        const bool queue_was_empty{to_send.empty() && pnode->vSendMsg.empty()};
4218
4219        // Update memory usage of send buffer.
4220        pnode->m_send_memusage += msg.GetMemoryUsage();
4221        if (pnode->m_send_memusage + pnode->m_transport->GetSendMemoryUsage() > nSendBufferMaxSize && !pnode->fPauseSend) {
4222            pnode->m_pause_send_since = NodeClock::now();
4223            pnode->fPauseSend = true;
4224        }
4225        // Move message to vSendMsg queue.
4226        pnode->vSendMsg.push_back(std::move(msg));
4227
4228        // If there was nothing to send before, and there is now (predicted by the "more" value
4229        // returned by the GetBytesToSend call above), attempt "optimistic write":
4230        // because the poll/select loop may pause for SELECT_TIMEOUT_MILLISECONDS before actually
4231        // doing a send, try sending from the calling thread if the queue was empty before.
4232        // With a V1Transport, more will always be true here, because adding a message always
4233        // results in sendable bytes there, but with V2Transport this is not the case (it may
4234        // still be in the handshake).
4235        if (queue_was_empty && more) {
4236            std::tie(nBytesSent, std::ignore) = SocketSendData(*pnode);
4237        }
4238    }
4239    if (nBytesSent) RecordBytesSent(nBytesSent);
4240}
4241
4242bool CConnman::ForNode(NodeId id, std::function<bool(CNode* pnode)> func)
4243{
4244    AssertLockNotHeld(m_nodes_mutex);
4245
4246    CNode* found = nullptr;
4247    LOCK(m_nodes_mutex);
4248    for (auto&& pnode : m_nodes) {
4249        if(pnode->GetId() == id) {
4250            found = pnode;