Connections and transport

3251
3252                // Receive messages
3253                bool fMoreNodeWork{m_msgproc->ProcessMessages(*pnode, flagInterruptMsgProc)};
3254                fMoreWork |= (fMoreNodeWork && !pnode->fPauseSend);
3255                if (flagInterruptMsgProc)
3256                    return;
3257                // Send messages
3258                m_msgproc->SendMessages(*pnode);
3259
3260                if (flagInterruptMsgProc)
3261                    return;
3262            }
3263        }
3264
3265        WAIT_LOCK(mutexMsgProc, lock);
3266        if (!fMoreWork) {
3267            condMsgProc.wait_until(lock, std::chrono::steady_clock::now() + std::chrono::milliseconds(100), [this]() EXCLUSIVE_LOCKS_REQUIRED(mutexMsgProc) { return fMsgProcWake; });
3268        }
3269        fMsgProcWake = false;
3270    }
3271}
3272
3273void CConnman::ThreadI2PAcceptIncoming()
3274{
3275    AssertLockNotHeld(m_nodes_mutex);
3276
3277    static constexpr auto err_wait_begin = 1s;
3278    static constexpr auto err_wait_cap = 5min;
3279    auto err_wait = err_wait_begin;
3280
3281    bool advertising_listen_addr = false;
3282    i2p::Connection conn;
3283
3284    auto SleepOnFailure = [&]() {
3285        m_interrupt_net->sleep_for(err_wait);
3286        if (err_wait < err_wait_cap) {
3287            err_wait += 1s;
3288        }
3289    };
3290
3291    while (!m_interrupt_net->interrupted()) {
3292
3293        if (!m_i2p_sam_session->Listen(conn)) {
3294            if (advertising_listen_addr && conn.me.IsValid()) {
3295                RemoveLocal(conn.me);
3296                advertising_listen_addr = false;
3297            }
3298            SleepOnFailure();
3299            continue;
3300        }
3301
3302        if (!advertising_listen_addr) {
3303            AddLocal(conn.me, LOCAL_MANUAL);
3304            advertising_listen_addr = true;
3305        }
3306
3307        if (!m_i2p_sam_session->Accept(conn)) {
3308            SleepOnFailure();
3309            continue;
3310        }
3311
3312        CreateNodeFromAcceptedSocket(std::move(conn.sock), NetPermissionFlags::None, conn.me, conn.peer);
3313
3314        err_wait = err_wait_begin;
3315    }
3316}
3317
3318void CConnman::ThreadPrivateBroadcast()
3319{
3320    AssertLockNotHeld(m_nodes_mutex);
3321    AssertLockNotHeld(m_unused_i2p_sessions_mutex);
3322
3323    size_t addrman_num_bad_addresses{0};
3324    while (!m_interrupt_net->interrupted()) {
3325
3326        if (!fNetworkActive) {
3327            m_interrupt_net->sleep_for(5s);
3328            continue;
3329        }
3330
3331        CountingSemaphoreGrant<> conn_max_grant{m_private_broadcast.m_sem_conn_max}; // Would block if too many are opened.
3332
3333        m_private_broadcast.NumToOpenWait();
3334
3335        if (m_interrupt_net->interrupted()) {
3336            break;
3337        }
3338
3339        std::optional<Proxy> proxy;
3340        const std::optional<Network> net{m_private_broadcast.PickNetwork(proxy)};
3341        if (!net.has_value()) {
3342            LogWarning("Unable to open -privatebroadcast connections: neither Tor nor I2P is reachable");
3343            m_interrupt_net->sleep_for(5s);
3344            continue;
3345        }
3346
3347        const auto [addr, _] = addrman.get().Select(/*new_only=*/false, {net.value()});
3348
3349        if (!addr.IsValid() || IsLocal(addr)) {
3350            ++addrman_num_bad_addresses;
3351            if (addrman_num_bad_addresses > 100) {
3352                LogDebug(BCLog::PRIVBROADCAST, "Connections needed but addrman keeps returning bad addresses, will retry");
3353                m_interrupt_net->sleep_for(500ms);
3354            }
3355            continue;
3356        }
3357        addrman_num_bad_addresses = 0;
3358
3359        auto target_str{addr.ToStringAddrPort()};
3360        if (proxy.has_value()) {
3361            target_str += " through the proxy at " + proxy->ToString();
3362        }
3363
3364        const bool use_v2transport(addr.nServices & GetLocalServices() & NODE_P2P_V2);
3365
3366        if (OpenNetworkConnection(addr,
3367                                  /*fCountFailure=*/true,
3368                                  std::move(conn_max_grant),
3369                                  /*pszDest=*/nullptr,
3370                                  ConnectionType::PRIVATE_BROADCAST,
3371                                  use_v2transport,
3372                                  proxy)) {
3373            const size_t remaining{m_private_broadcast.NumToOpenSub(1)};
3374            LogDebug(BCLog::PRIVBROADCAST, "Socket connected to %s; remaining connections to open: %d", target_str, remaining);
3375        } else {
3376            const size_t remaining{m_private_broadcast.NumToOpen()};
3377            if (remaining == 0) {
3378                LogDebug(BCLog::PRIVBROADCAST, "Failed to connect to %s, will not retry, no more connections needed", target_str);
3379            } else {
3380                LogDebug(BCLog::PRIVBROADCAST, "Failed to connect to %s, will retry to a different address; remaining connections to open: %d", target_str, remaining);
3381                m_interrupt_net->sleep_for(100ms); // Prevent busy loop if OpenNetworkConnection() fails fast repeatedly.
3382            }
3383        }
3384    }
3385}
3386
3387bool CConnman::BindListenPort(const CService& addrBind, bilingual_str& strError, NetPermissionFlags permissions)
3388{
3389    int nOne = 1;
3390
3391    // Create socket for listening for incoming connections
3392    struct sockaddr_storage sockaddr;
3393    socklen_t len = sizeof(sockaddr);
3394    if (!addrBind.GetSockAddr((struct sockaddr*)&sockaddr, &len))
3395    {
3396        strError = Untranslated(strprintf("Bind address family for %s not supported", addrBind.ToStringAddrPort()));
3397        LogError("%s\n", strError.original);
3398        return false;
3399    }
3400
3401    std::unique_ptr<Sock> sock = CreateSock(addrBind.GetSAFamily(), SOCK_STREAM, IPPROTO_TCP);
3402    if (!sock) {
3403        strError = Untranslated(strprintf("Couldn't open socket for incoming connections (socket returned error %s)", NetworkErrorString(WSAGetLastError())));
3404        LogError("%s\n", strError.original);
3405        return false;
3406    }
3407
3408    // Allow binding if the port is still in TIME_WAIT state after
3409    // the program was closed and restarted.
3410    if (sock->SetSockOpt(SOL_SOCKET, SO_REUSEADDR, &nOne, sizeof(int)) == SOCKET_ERROR) {
3411        strError = Untranslated(strprintf("Error setting SO_REUSEADDR on socket: %s, continuing anyway", NetworkErrorString(WSAGetLastError())));
3412        LogInfo("%s\n", strError.original);
3413    }
3414
3415    // some systems don't have IPV6_V6ONLY but are always v6only; others do have the option
3416    // and enable it by default or not. Try to enable it, if possible.
3417    if (addrBind.IsIPv6()) {
3418#ifdef IPV6_V6ONLY
3419        if (sock->SetSockOpt(IPPROTO_IPV6, IPV6_V6ONLY, &nOne, sizeof(int)) == SOCKET_ERROR) {
3420            strError = Untranslated(strprintf("Error setting IPV6_V6ONLY on socket: %s, continuing anyway", NetworkErrorString(WSAGetLastError())));
3421            LogInfo("%s\n", strError.original);
3422        }
3423#endif
3424#ifdef WIN32
3425        int nProtLevel = PROTECTION_LEVEL_UNRESTRICTED;
3426        if (sock->SetSockOpt(IPPROTO_IPV6, IPV6_PROTECTION_LEVEL, &nProtLevel, sizeof(int)) == SOCKET_ERROR) {
3427            strError = Untranslated(strprintf("Error setting IPV6_PROTECTION_LEVEL on socket: %s, continuing anyway", NetworkErrorString(WSAGetLastError())));
3428            LogInfo("%s\n", strError.original);
3429        }
3430#endif
3431    }
3432
3433    if (sock->Bind(reinterpret_cast<struct sockaddr*>(&sockaddr), len) == SOCKET_ERROR) {
3434        int nErr = WSAGetLastError();
3435        if (nErr == WSAEADDRINUSE)
3436            strError = strprintf(_("Unable to bind to %s on this computer. %s is probably already running."), addrBind.ToStringAddrPort(), CLIENT_NAME);
3437        else
3438            strError = strprintf(_("Unable to bind to %s on this computer (bind returned error %s)"), addrBind.ToStringAddrPort(), NetworkErrorString(nErr));
3439        LogError("%s\n", strError.original);
3440        return false;
3441    }
3442    LogInfo("Bound to %s\n", addrBind.ToStringAddrPort());
3443
3444    // Listen for incoming connections
3445    if (sock->Listen(SOMAXCONN) == SOCKET_ERROR)
3446    {
3447        strError = strprintf(_("Listening for incoming connections failed (listen returned error %s)"), NetworkErrorString(WSAGetLastError()));
3448        LogError("%s\n", strError.original);
3449        return false;
3450    }
3451
3452    vhListenSocket.emplace_back(std::move(sock), permissions);
3453    return true;
3454}
3455
3456void Discover()
3457{
3458    if (!fDiscover)
3459        return;
3460
3461    for (const CNetAddr &addr: GetLocalAddresses()) {
3462        if (AddLocal(addr, LOCAL_IF) && fLogIPs) {
3463            LogInfo("%s: %s\n", __func__, addr.ToStringAddr());
3464        }
3465    }
3466}
3467
3468void CConnman::SetNetworkActive(bool active)
3469{
3470    LogInfo("%s: %s\n", __func__, active);
3471
3472    if (fNetworkActive == active) {
3473        return;
3474    }
3475
3476    fNetworkActive = active;
3477
3478    if (m_client_interface) {
3479        m_client_interface->NotifyNetworkActiveChanged(fNetworkActive);
3480    }
3481}
3482
3483CConnman::CConnman(uint64_t nSeed0In,
3484                   uint64_t nSeed1In,
3485                   AddrMan& addrman_in,
3486                   const NetGroupManager& netgroupman,
3487                   const CChainParams& params,
3488                   bool network_active,
3489                   std::shared_ptr<CThreadInterrupt> interrupt_net)
3490    : addrman(addrman_in)
3491    , m_netgroupman{netgroupman}
3492    , nSeed0(nSeed0In)
3493    , nSeed1(nSeed1In)
3494    , m_interrupt_net{interrupt_net}
3495    , m_params(params)
3496{
3497    SetTryNewOutboundPeer(false);
3498
3499    Options connOptions;
3500    Init(connOptions);