41739478557

Committed 03 Nov 2026 11:28PM UTC coverage: 93.912% (-0.1%) from 94.038%

Build # 41739478557

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1004 from evoskuil/master

Define rpc body members w/o specialize, dispatch without cross-thread wakes.

Coverage Stats

209 of 219 new or added lines in 4 files covered. (95.43%)

134 existing lines in 39 files now uncovered.

14531 of 15473 relevant lines covered (93.91%)

2488.61 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.48

/src/net.cpp

/**
 * Copyright (c) 2011-2026 libbitcoin developers
 *
 * This file is part of libbitcoin.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
#include <bitcoin/network/net.hpp>

#include <bitcoin/network/async/async.hpp>
#include <bitcoin/network/channels/channels.hpp>
#include <bitcoin/network/config/config.hpp>
#include <bitcoin/network/define.hpp>
#include <bitcoin/network/log/log.hpp>
#include <bitcoin/network/messages/messages.hpp>
#include <bitcoin/network/net/net.hpp>
#include <bitcoin/network/protocols/protocols.hpp>
#include <bitcoin/network/sessions/sessions.hpp>
#include <bitcoin/network/settings.hpp>

namespace libbitcoin {
namespace network {

BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT)

using namespace system;
using namespace std::placeholders;
constexpr auto encryption = messages::peer::service::node_encrypted_transport;

// Seeding is not repeated within this many seeding timeouts of completion.
constexpr auto seed_backoff = 10;

net::net(const settings& settings, const logger& log,
    uint64_t required_services, uint64_t provided_services) NOEXCEPT
  : reporter(log),
    settings_(settings),
    privacy_(to_bool(bit_and(provided_services, to_value(encryption)))),
    threadpool_(settings.threads_()),
    strand_(threadpool_.service().get_executor()),
    hosts_(settings, log, required_services)
{
    ////LOG_LOG("Aplication log compiled..: ", news_defined);
    ////LOG_LOG("News log compiled........: ", news_defined);
    ////LOG_LOG("Session log compiled.....: ", session_defined);
    ////LOG_LOG("Protocol log compiled....: ", protocol_defined);
    ////LOG_LOG("ProXy log compiled.......: ", proxy_defined);
    ////LOG_LOG("Remote log compiled......: ", remote_defined);
    ////LOG_LOG("Fault log compiled.......: ", fault_defined);
    ////LOG_LOG("Quitting log compiled....: ", quitting_defined);
    ////LOG_LOG("Object log compiled......: ", objects_defined);
    ////LOG_LOG("Verbose log compiled.....: ", verbose_defined);
}

net::~net() NOEXCEPT
{
    // Weak references in threadpool closures safe as net joins threads here.
    net::close();
}

// I/O factories.
// ----------------------------------------------------------------------------

// server
acceptor::ptr net::create_service(socket::parameters&& params) NOEXCEPT
{
    return emplace_shared<acceptor>(log, strand(), service(),
        service_suspended_, std::move(params));
}

// inbound
acceptor::ptr net::create_acceptor(const socket::context& context) NOEXCEPT
{
    const auto& settings = network_settings();

    // bip324 (v2) inbound acceptance, v1 peers detected and passed through.
    const auto accept = privacy_ &&
        std::holds_alternative<std::monostate>(context) ?
            settings.peer_context() : context;

    socket::parameters params
    {
        .connect_timeout = settings.connect_timeout(settings.inbound),
        .maximum_request = settings.inbound.maximum_request,
        .minimum_buffer = settings.inbound.minimum_buffer,
        .maximum_buffer = settings.inbound.maximum_buffer,
        .context = accept
    };

    return emplace_shared<acceptor>(log, strand(), service(),
        accept_suspended_, std::move(params));
}

// inbound (sam)
acceptor_sam::ptr net::create_acceptor_sam() NOEXCEPT
{
    const auto& settings = network_settings();

    // bip324 (v2) inbound acceptance, v1 peers detected and passed through.
    socket::parameters params
    {
        .connect_timeout = settings.connect_timeout(settings.inbound),
        .maximum_request = settings.inbound.maximum_request,
        .minimum_buffer = settings.inbound.minimum_buffer,
        .maximum_buffer = settings.inbound.maximum_buffer,
        .context = privacy_ ? settings.peer_context() : socket::context{}
    };

    return emplace_shared<acceptor_sam>(log, strand(), service(),
        accept_suspended_, std::move(params), settings.inbound);
}

connector::ptr net::to_connector(const settings::socks5& socks,
    const settings::tcp_server& options,
    const steady_clock::duration& connect_timeout, size_t slot) NOEXCEPT
{
    socket::parameters params
    {
        .connect_timeout = connect_timeout,
        .maximum_request = options.maximum_request,
        .minimum_buffer = options.minimum_buffer,
        .maximum_buffer = options.maximum_buffer,
        .bind = options.binding(slot),
        .slot = slot
    };

    if (privacy_)
        params.context = network_settings().peer_context();

    if (socks.proxied())
        return emplace_shared<connector_socks>(log, strand(), service(),
            connect_suspended_, std::move(params), socks);

    // Above can handle both proxy and non-proxy, but this is more efficient.
    return emplace_shared<connector>(log, strand(), service(),
        connect_suspended_, std::move(params));
}

// outbound (general)
connector::ptr net::create_connector(const settings::socks5& socks,
    const settings::tcp_server& options, size_t slot) NOEXCEPT
{
    return to_connector(socks, options,
        network_settings().connect_timeout(options), slot);
}

// outbound (seed)
connector::ptr net::create_seed_connector(size_t slot) NOEXCEPT
{
    const auto& settings = network_settings();

    return to_connector(settings.outbound, settings.outbound,
        settings.outbound.seeding_timeout(), slot);
}

// outbound (manual)
connector::ptr net::create_manual_connector(size_t slot) NOEXCEPT
{
    const auto& settings = network_settings();

    return create_connector(settings.manual, settings.manual, slot);
}

// outbound (batch)
connectors_ptr net::create_connectors(size_t count, size_t slot) NOEXCEPT
{
    const auto& settings = network_settings();
    const auto connects = to_shared<connectors>();
    connects->reserve(count);

    for (size_t connect{}; connect < count; ++connect)
        connects->push_back(create_connector(settings.outbound,
            settings.outbound, slot));

    return connects;
}

// Start sequence.
// ----------------------------------------------------------------------------

void net::start(result_handler&& handler) NOEXCEPT
{
    // Threadpool is started on construct, stopped only by work starvation.
    io::post(strand_,
        std::bind(&net::do_start, this, std::move(handler)));
}

void net::do_start(const result_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());
    raise_descriptor_limit();
    manual_ = attach_manual_session();
    manual_->start(std::bind(&net::handle_start, this, _1, handler));
}

void net::handle_start(const code& ec, const result_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (ec)
    {
        handler(ec);
        return;
    }

    // Deserialize hosts from file.
    if (const auto error_code = start_hosts())
    {
        LOGF("Hosts file failed to deserialize, " << error_code.message());
        handler(error_code);
        return;
    }

    seed_ = attach_seed_session();
    seed_->start(move_copy(handler));
}

// Run sequence (seeding may be ongoing after its handler is invoked).
// ----------------------------------------------------------------------------

void net::run(result_handler&& handler) NOEXCEPT
{
    // Public methods can complete on caller thread.
    if (closed())
    {
        handler(error::service_stopped);
        return;
    }

    io::post(strand_,
        std::bind(&net::do_run, this, std::move(handler)));
}

void net::do_run(const result_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (closed())
    {
        handler(error::service_stopped);
        return;
    }

    // Start manual connections.
    for (const auto& peer: network_settings().manual.peers)
        do_connect(peer);

    // Start inbound connections.
    attach_inbound_session()->start(
        std::bind(&net::handle_run, this, _1, handler));
}

void net::handle_run(const code& ec, const result_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (ec)
    {
        handler(ec);
        return;
    }

    attach_outbound_session()->start(move_copy(handler));
}

// Shutdown sequence.
// ----------------------------------------------------------------------------

// Results in std::abort if called from a thread within the threadpool.
void net::close() NOEXCEPT
{
    closed_.store(true);
    io::post(strand_,
        std::bind(&net::do_close, this));

    // Blocks on join of all threadpool threads.
    if (!threadpool_.join())
    {
        BC_ASSERT_MSG(false, "failed to join threadpool");
        std::abort();
    }

    // Serialize hosts to file.
    if (const auto error_code = stop_hosts())
    {
        LOGF("Hosts file failed to serialize, " << error_code.message());
    }
}

void net::do_close() NOEXCEPT
{
    BC_ASSERT(stranded());

    // Release session references (also held by stop subscriber).
    if (manual_) manual_.reset();
    if (seed_) seed_.reset();

    // Notify and delete all stop subscribers (all sessions).
    stop_subscriber_.stop(error::service_stopped);
    service_subscriber_.stop(error::service_stopped);

    // Notify and delete subscribers to channel notifications.
    connect_subscriber_.stop_default(error::service_stopped);

    // Notify and delete subscribers to message broadcast notifications.
    broadcaster_.stop(error::service_stopped);

    // Stop threadpool keep-alive, all work must self-terminate to affect join.
    threadpool_.stop();
}

bool net::closed() const NOEXCEPT
{
    return closed_.load();
}

// Suspensions.
// ----------------------------------------------------------------------------
// Service connections are managed independently from P2P.

void net::suspend_services() NOEXCEPT
{
    service_suspended_.store(true);
}

void net::resume_services() NOEXCEPT
{
    service_suspended_.store(false);
}

void net::suspend(const code&) NOEXCEPT
{
    suspend_acceptors();
    suspend_connectors();
}

bool net::resume() NOEXCEPT
{
    resume_acceptors();
    resume_connectors();
    return true;
}

void net::stop_services(const code& ec) NOEXCEPT
{
    io::post(strand_,
        std::bind(&net::do_stop_services, this, ec));
}

void net::do_stop_services(const code& ec) NOEXCEPT
{
    BC_ASSERT(stranded());
    suspend_services();
    service_subscriber_.stop(ec);
}

bool net::suspended() const NOEXCEPT
{
    return connect_suspended_.load() || accept_suspended_.load();
}

void net::suspend_acceptors() NOEXCEPT
{
    accept_suspended_.store(true);
}

void net::resume_acceptors() NOEXCEPT
{
    accept_suspended_.store(false);
}

void net::suspend_connectors() NOEXCEPT
{
    connect_suspended_.store(true);
}

void net::resume_connectors() NOEXCEPT
{
    connect_suspended_.store(false);
}

// Properties.
// ----------------------------------------------------------------------------

const settings& net::network_settings() const NOEXCEPT
{
    return settings_;
}

io::context& net::service() NOEXCEPT
{
    return threadpool_.service();
}

io::strand& net::strand() NOEXCEPT
{
    return strand_;
}

bool net::stranded() const NOEXCEPT
{
    return strand_.running_in_this_thread();
}


// Subscriptions.
// ----------------------------------------------------------------------------
// Channel and network strands share same pool, and as long as a job is
// running in the pool, it will continue to accept work. Therefore handlers
// will not be orphaned during a stop as long as they remain in the pool.
// But when entering from outside the pool (such as subscribe) handler must be
// invoked when stopped as the handler will go uninvoked if the pool empties.

// public
void net::subscribe_connect(channel_notifier&& handler,
    channel_completer&& complete) NOEXCEPT
{
    // Public methods can complete on caller thread.
    if (closed())
    {
        complete(error::service_stopped, {});
        handler(error::service_stopped, {});
        return;
    }

    io::post(strand_,
        std::bind(&net::do_subscribe_connect,
            this, std::move(handler), std::move(complete)));
}

void net::do_subscribe_connect(const channel_notifier& handler,
    const channel_completer& complete) NOEXCEPT
{
    BC_ASSERT(stranded());

    const auto key = create_key();
    complete(connect_subscriber_.subscribe(move_copy(handler), key), key);
}

// protected
void net::notify_connect(const channel::ptr& channel) NOEXCEPT
{
    io::post(strand_,
        std::bind(&net::do_notify_connect, this, channel));
}

void net::do_notify_connect(const channel::ptr& channel) NOEXCEPT
{
    BC_ASSERT(stranded());
    connect_subscriber_.notify(error::success, channel);
}

void net::unsubscribe_connect(object_key key) NOEXCEPT
{
    io::post(strand_,
        std::bind(&net::do_unsubscribe_connect, this, key));
}

void net::do_unsubscribe_connect(object_key key) NOEXCEPT
{
    BC_ASSERT(stranded());
    connect_subscriber_.notify_one(key, error::desubscribed, nullptr);
}

// protected
void net::subscribe_close(stop_handler&& handler) NOEXCEPT
{
    BC_ASSERT(stranded());
    subscribe_close(std::move(handler), create_key());
}

// private
code net::subscribe_close(stop_handler&& handler, object_key key) NOEXCEPT
{
    BC_ASSERT(stranded());
    return stop_subscriber_.subscribe(std::move(handler), key);
}

code net::subscribe_service(stop_handler&& handler, object_key key) NOEXCEPT
{
    BC_ASSERT(stranded());
    return service_subscriber_.subscribe(std::move(handler), key);
}

// public
void net::subscribe_close(stop_handler&& handler,
    stop_completer&& complete) NOEXCEPT
{
    // Public methods can complete on caller thread.
    if (closed())
    {
        complete(error::service_stopped, {});
        handler(error::service_stopped);
        return;
    }

    io::post(strand_,
        std::bind(&net::do_subscribe_close,
            this, std::move(handler), std::move(complete)));
}

void net::do_subscribe_close(const stop_handler& handler,
    const stop_completer& complete) NOEXCEPT
{
    BC_ASSERT(stranded());
    
    const auto key = create_key();
    complete(subscribe_close(move_copy(handler), key), key);
}

void net::unsubscribe_close(object_key key) NOEXCEPT
{
    io::post(strand_,
        std::bind(&net::do_unsubscribe_close, this, key));
}

void net::do_unsubscribe_close(object_key key) NOEXCEPT
{
    BC_ASSERT(stranded());
    stop_subscriber_.notify_one(key, error::desubscribed);
}

// Broadcasts.
// ----------------------------------------------------------------------------

void net::demand_wire_tx(bool demand) NOEXCEPT
{
    if (demand)
        ++wire_tx_demand_;
    else
        --wire_tx_demand_;
}

void net::demand_wire_block(bool demand) NOEXCEPT
{
    if (demand)
        ++wire_block_demand_;
    else
        --wire_block_demand_;
}

bool net::wire_tx_demanded() const NOEXCEPT
{
    return is_nonzero(wire_tx_demand_.load());
}

bool net::wire_block_demanded() const NOEXCEPT
{
    return is_nonzero(wire_block_demand_.load());
}

// At one object/session/ns, this overflows in ~585 years (and handled).
net::object_key net::create_key() NOEXCEPT
{
    BC_ASSERT(stranded());

    if (is_zero(++keys_))
    {
        BC_ASSERT_MSG(false, "overflow");
        LOGF("Session object overflow.");
    }

    return keys_;
}

// P2P Manual connections.
// ----------------------------------------------------------------------------

void net::connect(const config::endpoint& endpoint) NOEXCEPT
{
    io::post(strand_,
        std::bind(&net::do_connect, this, endpoint));
}

void net::do_connect(const config::endpoint& endpoint) NOEXCEPT
{
    BC_ASSERT(stranded());
    if (manual_) manual_->connect(endpoint);
}

void net::connect(const config::endpoint& endpoint,
    channel_notifier&& handler) NOEXCEPT
{
    // Public methods can complete on caller thread.
    if (closed())
    {
        handler(error::service_stopped, {});
        return;
    }

    io::post(strand_,
        std::bind(&net::do_connect_handled, this, endpoint,
            std::move(handler)));
}

void net::do_connect_handled(const config::endpoint& endpoint,
    const channel_notifier& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (manual_)
        manual_->connect(endpoint, move_copy(handler));
    else
        handler(error::service_stopped, nullptr);
}

// P2P Administration.
// ----------------------------------------------------------------------------

void net::dump_addresses(address_handler&& handler) NOEXCEPT
{
    fetch(std::move(handler));
}

// P2P Properties.
// ----------------------------------------------------------------------------

size_t net::address_count() const NOEXCEPT
{
    return hosts_.count();
}

config::address_counts net::address_counts() const NOEXCEPT
{
    return hosts_.counts();
}

size_t net::reserved_count() const NOEXCEPT
{
    return hosts_.reserved();
}

void net::fetch_totals(totals_handler&& handler) NOEXCEPT
{
    if (closed())
    {
        handler(error::service_stopped, {});
        return;
    }

    io::post(strand_,
        std::bind(&net::do_fetch_totals, this, std::move(handler)));
}

void net::do_fetch_totals(const totals_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    handler(error::success,
    {
        peer_sent_.load(),
        peer_received_.load(),
        actives_,
    });
}

size_t net::channel_count() const NOEXCEPT
{
    return total_channel_count_;
}

size_t net::inbound_channel_count() const NOEXCEPT
{
    return inbound_channel_count_;
}

// P2P hosts collection.
// ----------------------------------------------------------------------------
// Protected, called from session (network strand) and channel (network pool).

// private
code net::start_hosts() NOEXCEPT
{
    return hosts_.start();
}

// private
code net::stop_hosts() NOEXCEPT
{
    return hosts_.stop();
}

void net::take(hosts::family family, address_item_handler&& handler) NOEXCEPT
{
    io::post(strand_,
        std::bind(&net::do_take, this, family, std::move(handler)));
}

void net::do_take(hosts::family family,
    const address_item_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());
    hosts_.take(family, move_copy(handler));
}

void net::restore(const address_item_cptr& address,
    result_handler&& handler) NOEXCEPT
{
    io::post(strand_,
        std::bind(&net::do_restore, this, address, std::move(handler)));
}

void net::do_restore(const address_item_cptr& address,
    const result_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());
    hosts_.restore(address, move_copy(handler));
}

void net::fetch(address_handler&& handler) NOEXCEPT
{
    io::post(strand_,
        std::bind(&net::do_fetch, this, std::move(handler)));
}

void net::do_fetch(const address_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    // Accelerate stop, since hosts keeps running until all threads closed.
    if (closed())
    {
        handler(error::service_stopped, {});
        return;
    }

    hosts_.fetch(move_copy(handler));
}

void net::save(const address_cptr& message, count_handler&& handler) NOEXCEPT
{
    io::post(strand_,
        std::bind(&net::do_save, this, message, std::move(handler)));
}

void net::do_save(const address_cptr& message,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    // Accelerate stop, since hosts keeps running until all threads closed.
    if (closed())
    {
        handler(error::service_stopped, {});
        return;
    }

    hosts_.save(message, move_copy(handler));
}

void net::seed() NOEXCEPT
{
    io::post(strand_,
        std::bind(&net::do_seed, this));
}

void net::do_seed() NOEXCEPT
{
    BC_ASSERT(stranded());

    // address_not_found may only mean that no pooled address suits the slot.
    const auto& outbound = network_settings().outbound;
    if (closed() || outbound.seeds.empty() ||
        address_count() >= outbound.minimum_address_count())
        return;

    BC_ASSERT(seed_);
    const auto completed = seed_->completed();
    const auto backoff = outbound.seeding_timeout() * seed_backoff;

    if (seed_->seeding() || (completed != steady_clock::time_point{} &&
        (steady_clock::now() - completed) < backoff))
        return;

    seed_ = attach_seed_session();
    seed_->start([](const code&) NOEXCEPT {});
}

// P2P self address.
// ----------------------------------------------------------------------------

// P2P loopback detection.
// ----------------------------------------------------------------------------

bool net::store_nonce(const channel_peer& channel) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (network_settings().inbound.enable_loopback || channel.inbound())
        return true;

    if (!nonces_.insert(channel.nonce()).second)
    {
        LOGF("Failed to store nonce for [" << channel.endpoint() << "].");
        return false;
    }

    return true;
}

bool net::unstore_nonce(const channel_peer& channel) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (network_settings().inbound.enable_loopback || channel.inbound())
        return true;

    if (!to_bool(nonces_.erase(channel.nonce())))
    {
        LOGF("Failed to unstore nonce for [" << channel.endpoint() << "].");
        return false;
    }

    return true;
}

bool net::is_loopback(const channel_peer& channel) const NOEXCEPT
{
    BC_ASSERT(stranded());

    if (network_settings().inbound.enable_loopback || !channel.inbound())
        return false;

    return to_bool(nonces_.count(channel.peer_version()->nonce));
}

// P2P channel counting with address deconfliction.
// ----------------------------------------------------------------------------

code net::count_channel(const channel_peer& channel) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (closed())
        return error::service_stopped;

    if (is_loopback(channel))
    {
        LOGS("Loopback detected from [" << channel.endpoint() << "].");
        return error::accept_failed;
    }

    if (channel.inbound() && is_zero(add1(inbound_channel_count_.load())))
    {
        LOGF("Overflow: inbound channel count.");
        return error::channel_overflow;
    }

    if (!channel.quiet() && is_zero(add1(total_channel_count_.load())))
    {
        LOGF("Overflow: total channel count.");
        return error::channel_overflow;
    }

    if (!hosts_.reserve(channel.endpoint()))
    {
        LOGS("Duplicate connection to [" << channel.endpoint() << "].");
        return error::address_in_use;
    }

    if (channel.inbound())
        ++inbound_channel_count_;

    if (!channel.quiet())
        ++total_channel_count_;

    actives_.push_back(channel.identifier());
    return error::success;
}

void net::uncount_channel(const channel_peer& channel) NOEXCEPT
{
    BC_ASSERT(stranded());

    hosts_.unreserve(channel.endpoint());
    peer_sent_ = ceilinged_add(peer_sent_.load(), channel.sent());
    peer_received_ = ceilinged_add(peer_received_.load(), channel.received());
    std::erase(actives_, channel.identifier());

    if (channel.inbound() && is_zero(inbound_channel_count_.load()))
    {
        LOGF("Underflow: inbound channel count.");
        return;
    }

    if (!channel.quiet() && is_zero(total_channel_count_.load()))
    {
        LOGF("Underflow: total channel count.");
        return;
    }

    if (channel.inbound())
        --inbound_channel_count_;

    if (!channel.quiet())
        --total_channel_count_;
}

// Specializations (protected).
// ----------------------------------------------------------------------------

session_seed::ptr net::attach_seed_session() NOEXCEPT
{
    BC_ASSERT(stranded());
    return attach<session_seed>(*this);
}

session_manual::ptr net::attach_manual_session() NOEXCEPT
{
    BC_ASSERT(stranded());
    return attach<session_manual>(*this);
}

session_inbound::ptr net::attach_inbound_session() NOEXCEPT
{
    BC_ASSERT(stranded());
    return attach<session_inbound>(*this);
}

session_outbound::ptr net::attach_outbound_session() NOEXCEPT
{
    BC_ASSERT(stranded());
    return attach<session_outbound>(*this);
}

BC_POP_WARNING()

} // namespace network
} // namespace libbitcoin

libbitcoin / libbitcoin-network / 41739478557

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous