<commit_before>/**
 * Copyright (c) 2011-2018 libbitcoin developers (see AUTHORS)
 *
 * This file is part of libbitcoin.
 *
 * libbitcoin is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License with
 * additional permissions to the one published by the Free Software
 * Foundation, either version 3 of the License, or (at your option)
 * any later version. For more information see LICENSE.
 *
 * 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/bitcoin/network/protocol.hpp>

#include <cstddef>
#include <cstdint>
#include <functional>
#include <memory>
#include <system_error>
#include <boost/date_time.hpp>
#include <boost/filesystem.hpp>
#include <boost/format.hpp>
#include <bitcoin/bitcoin/error.hpp>
#include <bitcoin/bitcoin/network/authority.hpp>
#include <bitcoin/bitcoin/network/hosts.hpp>
#include <bitcoin/bitcoin/network/handshake.hpp>
#include <bitcoin/bitcoin/network/seeder.hpp>
#include <bitcoin/bitcoin/utility/logger.hpp>
#include <bitcoin/bitcoin/utility/threadpool.hpp>

namespace libbitcoin {
namespace network {

using std::placeholders::_1;
using std::placeholders::_2;
using boost::filesystem::path;
using boost::format;
using boost::posix_time::time_duration;
using boost::posix_time::seconds;

// Based on http://bitcoinstats.com/network/dns-servers
#ifdef ENABLE_TESTNET
const hosts::authority_list protocol::default_seeds =
{
    { "testnet-seed.alexykot.me", 18333 },
    { "testnet-seed.bitcoin.petertodd.org", 18333 },
    { "testnet-seed.bluematt.me", 18333 },
    { "testnet-seed.bitcoin.schildbach.de", 18333 }
};
#else
const hosts::authority_list protocol::default_seeds =
{
    { "seed.bitnodes.io", 8333 },
    { "seed.bitcoinstats.com", 8333 },
    { "seed.bitcoin.sipa.be", 8333 },
    { "dnsseed.bluematt.me", 8333 },
    { "seed.bitcoin.jonasschnelli.ch", 8333 },
    { "dnsseed.bitcoin.dashjr.org", 8333 }

    // Previously also included:
    // bitseed.xf2.org:8333
    // archivum.info:8333
    // progressbar.sk:8333
    // faucet.bitcoin.st:8333
    // bitcoin.securepayment.cc:8333
};
#endif

const size_t protocol::default_max_outbound = 8;

// TODO: parameterize for config access.
const size_t watermark_connection_limit = 10;
const time_duration watermark_reset_interval = seconds(1);

protocol::protocol(threadpool& pool, hosts& peers, handshake& shake,
    network& net, const hosts::authority_list& seeds, uint16_t port,
    bool listen, size_t max_outbound)
  : strand_(pool),
    host_pool_(peers),
    handshake_(shake),
    network_(net),
    max_outbound_(max_outbound),
    watermark_timer_(pool.service()),
    watermark_count_(0),
    listen_port_(port),
    listen_enabled_(listen),
    channel_subscribe_(std::make_shared<channel_subscriber_type>(pool)),
    seeds_(seeds)
{
}

void protocol::start(completion_handler handle_complete)
{
    // Bootstrap from seeds if neccessary.
    bootstrap(
        strand_.wrap(&protocol::handle_bootstrap,
            this, _1, handle_complete));

    // Start handshake service. If it fails to start it's not a critical error.
    const auto service_started = [](const std::error_code& ec)
    {
        if (ec)
            log_error(LOG_PROTOCOL)
                << "Failed to start handshake service: " << ec.message();
    };

    handshake_.start(service_started);
}

void protocol::handle_bootstrap(const std::error_code& ec,
    completion_handler handle_complete)
{
    if (ec)
    {
        log_error(LOG_PROTOCOL) << "Failed to bootstrap: " << ec.message();
        handle_complete(ec);
        return;
    }

    handle_complete(error::success);
    run();
}

void protocol::stop(completion_handler handle_complete)
{
    host_pool_.save(
        strand_.wrap(&protocol::handle_save,
            this, _1, handle_complete));
}

void protocol::handle_save(const std::error_code& ec,
    completion_handler handle_complete)
{
    if (ec)
    {
        log_error(LOG_PROTOCOL) << "Failed to save hosts file: "
            << ec.message();
        handle_complete(ec);
        return;
    }

    channel_subscribe_->relay(error::service_stopped, nullptr);
    handle_complete(error::success);
}

void protocol::bootstrap(completion_handler handle_complete)
{
    host_pool_.load(
        strand_.wrap(&protocol::load_hosts,
            this, _1, handle_complete));
}

void protocol::load_hosts(const std::error_code& ec,
    completion_handler handle_complete)
{
    if (ec)
    {
        log_error(LOG_PROTOCOL)
            << "Failed to load hosts file: " << ec.message();
        handle_complete(ec);
        return;
    }

    host_pool_.fetch_count(
        strand_.wrap(&protocol::if_no_seeds,
            this, _1, _2, handle_complete));
}

void protocol::if_no_seeds(const std::error_code& ec, size_t hosts_count,
    completion_handler handle_complete)
{
    if (ec)
    {
        log_error(LOG_PROTOCOL)
            << "Failure checking existing hosts file: " << ec.message();
        handle_complete(ec);
        return;
    }

    if (hosts_count == 0)
    {
        seeder_ = std::make_shared<seeder>(this, seeds_, handle_complete);
        seeder_->start();
        return;
    }

    handle_complete(error::success); 
}

std::string protocol::state_to_string(connect_state state) const
{
    switch (state)
    {
        case connect_state::finding_peer:
            return "finding peer";
        case connect_state::connecting:
            return "connecting to peer";
        case connect_state::established:
            return "established connection";
        case connect_state::stopped:
            return "stopped";
    }

    // Unhandled state!
    BITCOIN_ASSERT(false);
    return "";
}

void protocol::modify_slot(slot_index slot, connect_state state)
{
    connect_states_[slot] = state;
    log_debug(LOG_PROTOCOL) << "Outbound connection on slot ["
        << slot << "] " << state_to_string(state) << ".";
}

void protocol::run()
{
    strand_.queue(&protocol::start_connecting, this);

    if (!listen_enabled_)
        return;

    network_.listen(listen_port_,
        strand_.wrap(&protocol::handle_listen,
            this, _1, _2));
}

void protocol::start_connecting()
{
    // Initialize the connection slots.
    BITCOIN_ASSERT(connections_.empty());
    BITCOIN_ASSERT(connect_states_.empty());
    connect_states_.resize(max_outbound_);

    for (size_t slot = 0; slot < connect_states_.size(); ++slot)
        modify_slot(slot, connect_state::stopped);

    // Start the main outbound connect loop.
    start_stopped_connects();
    start_watermark_reset_timer();
}

void protocol::start_stopped_connects()
{
    for (size_t slot = 0; slot < connect_states_.size(); ++slot)
        if (connect_states_[slot] == connect_state::stopped)
            try_connect_once(slot);
}

void protocol::try_connect_once(slot_index slot)
{
    ++watermark_count_;
    if (watermark_count_ > watermark_connection_limit)
        return;

    BITCOIN_ASSERT(slot <= connect_states_.size());
    BITCOIN_ASSERT(connect_states_[slot] == connect_state::stopped);

    // Begin connection flow: finding_peer -> connecting -> established.
    // Failures end with connect_state::stopped and loop back here again.
    modify_slot(slot, connect_state::finding_peer);
    host_pool_.fetch_address(
        strand_.wrap(&protocol::attempt_connect, 
            this, _1, _2, slot));
}

void protocol::start_watermark_reset_timer()
{
    // This timer just loops continuously at fixed intervals
    // resetting the watermark_count_ variable and starting stopped slots.
    const auto reset_watermark = [this](const boost::system::error_code& ec)
    {
        if (ec)
        {
            BITCOIN_ASSERT(ec == boost::asio::error::operation_aborted);
            return;
        }

        if (watermark_count_ > watermark_connection_limit)
            log_debug(LOG_PROTOCOL) << "Resuming connection attempts.";

        // Perform the reset, reallowing connection attempts.
        watermark_count_ = 0;
        start_stopped_connects();

        // Looping timer...
        start_watermark_reset_timer();
    };

    watermark_timer_.expires_from_now(watermark_reset_interval);
    watermark_timer_.async_wait(strand_.wrap(reset_watermark, _1));
}

template <typename ConnectionList>
bool already_connected(const network_address_type& address,
    const ConnectionList& connections)
{
    for (const auto& connection: connections)
        if (connection.address.ip == address.ip &&
            connection.address.port == address.port)
            return true;

    return false;
}

void protocol::attempt_connect(const std::error_code& ec,
    const network_address_type& address, slot_index slot)
{
    BITCOIN_ASSERT(connect_states_[slot] == connect_state::finding_peer);
    modify_slot(slot, connect_state::connecting);

    if (ec)
    {
        log_error(LOG_PROTOCOL)
            << "Failure randomly selecting a peer address for slot ["
            << slot << "] " << ec.message();
        return;
    }

    if (already_connected(address, connections_))
    {
        log_debug(LOG_PROTOCOL)
            << "Already connected to selected peer [" 
            << authority(address).to_string() << "]";

        // Retry another connection, still in same strand.
        modify_slot(slot, connect_state::stopped);
        try_connect_once(slot);
        return;
    }

    log_debug(LOG_PROTOCOL) << 
        "Connecting to peer [" << authority(address).to_string()
        << "] on slot [" << slot << "]";

    const authority peer(address);
    connect(handshake_, network_, peer.host, peer.port,
        strand_.wrap(&protocol::handle_connect,
            this, _1, _2, address, slot));
}

void protocol::handle_connect(const std::error_code& ec, channel_ptr node,
    const network_address_type& address, slot_index slot)
{
    BITCOIN_ASSERT(connect_states_[slot] == connect_state::connecting);

    if (ec)
    {
        log_warning(LOG_PROTOCOL) << "Failure connecting to peer ["
            << authority(address).to_string() << "] on slot [" << slot << "] "
            << ec.message();

        // Retry another connection, still in same strand.
        modify_slot(slot, connect_state::stopped);
        try_connect_once(slot);
        return;
    }

    modify_slot(slot, connect_state::established);
    BITCOIN_ASSERT(connections_.size() <= max_outbound_);
    connections_.push_back({address, node});

    log_info(LOG_PROTOCOL) << "Connected to peer ["
        << authority(address).to_string() << "] on slot [" << slot << "] ("
        << connections_.size() << " total)";

    // Remove channel from list of connections
    node->subscribe_stop(
        strand_.wrap(
            &protocol::outbound_channel_stopped,
                this, _1, node, slot));
<commit_msg>Use standard network call layout.<commit_after>/**
 * Copyright (c) 2011-2018 libbitcoin developers (see AUTHORS)
 *
 * This file is part of libbitcoin.
 *
 * libbitcoin is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License with
 * additional permissions to the one published by the Free Software
 * Foundation, either version 3 of the License, or (at your option)
 * any later version. For more information see LICENSE.
 *
 * 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/bitcoin/network/protocol.hpp>

#include <cstddef>
#include <cstdint>
#include <functional>
#include <memory>
#include <system_error>
#include <boost/date_time.hpp>
#include <boost/filesystem.hpp>
#include <boost/format.hpp>
#include <bitcoin/bitcoin/error.hpp>
#include <bitcoin/bitcoin/network/authority.hpp>
#include <bitcoin/bitcoin/network/hosts.hpp>
#include <bitcoin/bitcoin/network/handshake.hpp>
#include <bitcoin/bitcoin/network/seeder.hpp>
#include <bitcoin/bitcoin/utility/logger.hpp>
#include <bitcoin/bitcoin/utility/threadpool.hpp>

namespace libbitcoin {
namespace network {

using std::placeholders::_1;
using std::placeholders::_2;
using boost::filesystem::path;
using boost::format;
using boost::posix_time::time_duration;
using boost::posix_time::seconds;

// Based on http://bitcoinstats.com/network/dns-servers
#ifdef ENABLE_TESTNET
const hosts::authority_list protocol::default_seeds =
{
    { "testnet-seed.alexykot.me", 18333 },
    { "testnet-seed.bitcoin.petertodd.org", 18333 },
    { "testnet-seed.bluematt.me", 18333 },
    { "testnet-seed.bitcoin.schildbach.de", 18333 }
};
#else
const hosts::authority_list protocol::default_seeds =
{
    { "seed.bitnodes.io", 8333 },
    { "seed.bitcoinstats.com", 8333 },
    { "seed.bitcoin.sipa.be", 8333 },
    { "dnsseed.bluematt.me", 8333 },
    { "seed.bitcoin.jonasschnelli.ch", 8333 },
    { "dnsseed.bitcoin.dashjr.org", 8333 }

    // Previously also included:
    // bitseed.xf2.org:8333
    // archivum.info:8333
    // progressbar.sk:8333
    // faucet.bitcoin.st:8333
    // bitcoin.securepayment.cc:8333
};
#endif

const size_t protocol::default_max_outbound = 8;

// TODO: parameterize for config access.
const size_t watermark_connection_limit = 10;
const time_duration watermark_reset_interval = seconds(1);

protocol::protocol(threadpool& pool, hosts& peers, handshake& shake,
    network& net, const hosts::authority_list& seeds, uint16_t port,
    bool listen, size_t max_outbound)
  : strand_(pool),
    host_pool_(peers),
    handshake_(shake),
    network_(net),
    max_outbound_(max_outbound),
    watermark_timer_(pool.service()),
    watermark_count_(0),
    listen_port_(port),
    listen_enabled_(listen),
    channel_subscribe_(std::make_shared<channel_subscriber_type>(pool)),
    seeds_(seeds)
{
}

void protocol::start(completion_handler handle_complete)
{
    // Bootstrap from seeds if neccessary.
    bootstrap(
        strand_.wrap(&protocol::handle_bootstrap,
            this, _1, handle_complete));

    // Start handshake service. If it fails to start it's not a critical error.
    const auto service_started = [](const std::error_code& ec)
    {
        if (ec)
            log_error(LOG_PROTOCOL)
                << "Failed to start handshake service: " << ec.message();
    };

    handshake_.start(service_started);
}

void protocol::handle_bootstrap(const std::error_code& ec,
    completion_handler handle_complete)
{
    if (ec)
    {
        log_error(LOG_PROTOCOL) << "Failed to bootstrap: " << ec.message();
        handle_complete(ec);
        return;
    }

    handle_complete(error::success);
    run();
}

void protocol::stop(completion_handler handle_complete)
{
    host_pool_.save(
        strand_.wrap(&protocol::handle_save,
            this, _1, handle_complete));
}

void protocol::handle_save(const std::error_code& ec,
    completion_handler handle_complete)
{
    if (ec)
    {
        log_error(LOG_PROTOCOL) << "Failed to save hosts file: "
            << ec.message();
        handle_complete(ec);
        return;
    }

    channel_subscribe_->relay(error::service_stopped, nullptr);
    handle_complete(error::success);
}

void protocol::bootstrap(completion_handler handle_complete)
{
    host_pool_.load(
        strand_.wrap(&protocol::load_hosts,
            this, _1, handle_complete));
}

void protocol::load_hosts(const std::error_code& ec,
    completion_handler handle_complete)
{
    if (ec)
    {
        log_error(LOG_PROTOCOL)
            << "Failed to load hosts file: " << ec.message();
        handle_complete(ec);
        return;
    }

    host_pool_.fetch_count(
        strand_.wrap(&protocol::if_no_seeds,
            this, _1, _2, handle_complete));
}

void protocol::if_no_seeds(const std::error_code& ec, size_t hosts_count,
    completion_handler handle_complete)
{
    if (ec)
    {
        log_error(LOG_PROTOCOL)
            << "Failure checking existing hosts file: " << ec.message();
        handle_complete(ec);
        return;
    }

    if (hosts_count == 0)
    {
        seeder_ = std::make_shared<seeder>(this, seeds_, handle_complete);
        seeder_->start();
        return;
    }

    handle_complete(error::success); 
}

std::string protocol::state_to_string(connect_state state) const
{
    switch (state)
    {
        case connect_state::finding_peer:
            return "finding peer";
        case connect_state::connecting:
            return "connecting to peer";
        case connect_state::established:
            return "established connection";
        case connect_state::stopped:
            return "stopped";
    }

    // Unhandled state!
    BITCOIN_ASSERT(false);
    return "";
}

void protocol::modify_slot(slot_index slot, connect_state state)
{
    connect_states_[slot] = state;
    log_debug(LOG_PROTOCOL) << "Outbound connection on slot ["
        << slot << "] " << state_to_string(state) << ".";
}

void protocol::run()
{
    strand_.queue(&protocol::start_connecting, this);

    if (!listen_enabled_)
        return;

    network_.listen(listen_port_,
        strand_.wrap(&protocol::handle_listen,
            this, _1, _2));
}

void protocol::start_connecting()
{
    // Initialize the connection slots.
    BITCOIN_ASSERT(connections_.empty());
    BITCOIN_ASSERT(connect_states_.empty());
    connect_states_.resize(max_outbound_);

    for (size_t slot = 0; slot < connect_states_.size(); ++slot)
        modify_slot(slot, connect_state::stopped);

    // Start the main outbound connect loop.
    start_stopped_connects();
    start_watermark_reset_timer();
}

void protocol::start_stopped_connects()
{
    for (size_t slot = 0; slot < connect_states_.size(); ++slot)
        if (connect_states_[slot] == connect_state::stopped)
            try_connect_once(slot);
}

void protocol::try_connect_once(slot_index slot)
{
    ++watermark_count_;
    if (watermark_count_ > watermark_connection_limit)
        return;

    BITCOIN_ASSERT(slot <= connect_states_.size());
    BITCOIN_ASSERT(connect_states_[slot] == connect_state::stopped);

    // Begin connection flow: finding_peer -> connecting -> established.
    // Failures end with connect_state::stopped and loop back here again.
    modify_slot(slot, connect_state::finding_peer);
    host_pool_.fetch_address(
        strand_.wrap(&protocol::attempt_connect, 
            this, _1, _2, slot));
}

void protocol::start_watermark_reset_timer()
{
    // This timer just loops continuously at fixed intervals
    // resetting the watermark_count_ variable and starting stopped slots.
    const auto reset_watermark = [this](const boost::system::error_code& ec)
    {
        if (ec)
        {
            BITCOIN_ASSERT(ec == boost::asio::error::operation_aborted);
            return;
        }

        if (watermark_count_ > watermark_connection_limit)
            log_debug(LOG_PROTOCOL) << "Resuming connection attempts.";

        // Perform the reset, reallowing connection attempts.
        watermark_count_ = 0;
        start_stopped_connects();

        // Looping timer...
        start_watermark_reset_timer();
    };

    watermark_timer_.expires_from_now(watermark_reset_interval);
    watermark_timer_.async_wait(strand_.wrap(reset_watermark, _1));
}

template <typename ConnectionList>
bool already_connected(const network_address_type& address,
    const ConnectionList& connections)
{
    for (const auto& connection: connections)
        if (connection.address.ip == address.ip &&
            connection.address.port == address.port)
            return true;

    return false;
}

void protocol::attempt_connect(const std::error_code& ec,
    const network_address_type& address, slot_index slot)
{
    BITCOIN_ASSERT(connect_states_[slot] == connect_state::finding_peer);
    modify_slot(slot, connect_state::connecting);

    if (ec)
    {
        log_error(LOG_PROTOCOL)
            << "Failure randomly selecting a peer address for slot ["
            << slot << "] " << ec.message();
        return;
    }

    if (already_connected(address, connections_))
    {
        log_debug(LOG_PROTOCOL)
            << "Already connected to selected peer [" 
            << authority(address).to_string() << "]";

        // Retry another connection, still in same strand.
        modify_slot(slot, connect_state::stopped);
        try_connect_once(slot);
        return;
    }

    log_debug(LOG_PROTOCOL) << 
        "Connecting to peer [" << authority(address).to_string()
        << "] on slot [" << slot << "]";

    const authority peer(address);
    connect(handshake_, network_, peer.host, peer.port,
        strand_.wrap(&protocol::handle_connect,
            this, _1, _2, address, slot));
}

void protocol::handle_connect(const std::error_code& ec, channel_ptr node,
    const network_address_type& address, slot_index slot)
{
    BITCOIN_ASSERT(connect_states_[slot] == connect_state::connecting);

    if (ec)
    {
        log_warning(LOG_PROTOCOL) << "Failure connecting to peer ["
            << authority(address).to_string() << "] on slot [" << slot << "] "
            << ec.message();

        // Retry another connection, still in same strand.
        modify_slot(slot, connect_state::stopped);
        try_connect_once(slot);
        return;
    }

    modify_slot(slot, connect_state::established);
    BITCOIN_ASSERT(connections_.size() <= max_outbound_);
    connections_.push_back({address, node});

    log_info(LOG_PROTOCOL) << "Connected to peer ["
        << authority(address).to_string() << "] on slot [" << slot << "] ("
        << connections_.size() << " total)";

    // Remove channel from list of connections
    node->subscribe_stop(
        strand_.wrap(&protocol::outbound_channel_stopped,
            this, _1, node, slot));

{'fraction_non_alphanumeric': 0.11863495902189977, 'fraction_numerical': 0.00895696179855793, 'mean_word_length': 3.3166441136671176, 'pattern_counts': {'":': 0, '<': 215, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2099327', 'n_tokens_mistral': 6969, 'n_tokens_neox': 6759, 'n_words': 1493}