sysio::net_plugin Class Reference

#include <net_plugin.hpp>

Inheritance diagram for sysio::net_plugin:

Collaboration diagram for sysio::net_plugin:

Public Member Functions
	net_plugin ()
virtual	~net_plugin ()
virtual void	set_program_options (options_description &cli, options_description &cfg) override
void	handle_sighup () override
void	plugin_initialize (const variables_map &options)
void	plugin_startup ()
void	plugin_shutdown ()
string	connect (const string &endpoint)
string	disconnect (const string &endpoint)
std::optional< connection_status >	status (const string &endpoint) const
vector< connection_status >	connections () const
Public Member Functions inherited from appbase::plugin< net_plugin >
	plugin ()
virtual	~plugin ()
virtual state	get_state () const override
virtual const std::string &	name () const override
virtual void	register_dependencies ()
virtual void	initialize (const variables_map &options) override
virtual void	startup () override
virtual void	shutdown () override
Public Member Functions inherited from appbase::abstract_plugin
virtual	~abstract_plugin ()

Additional Inherited Members
Public Types inherited from appbase::abstract_plugin
enum	state { registered , initialized , started , stopped }
Protected Member Functions inherited from appbase::plugin< net_plugin >
	plugin (const string &name)

Detailed Description

Definition at line 16 of file net_plugin.hpp.

Constructor & Destructor Documentation

net_plugin()

net_plugin::net_plugin

(

)

Definition at line 3539 of file net_plugin.cpp.

      :my( new net_plugin_impl ) {
      my_impl = my.get();
   }

~net_plugin()

net_plugin::~net_plugin ( )

virtual

Definition at line 3544 of file net_plugin.cpp.

                           {
   }

Member Function Documentation

connect()

string sysio::net_plugin::connect

(

const string &

host

)

Used to trigger a new connection from RPC API

Definition at line 3852 of file net_plugin.cpp.

                                                  {
      return my->connect( host );
   }

connections()

vector< connection_status > net_plugin::connections

(

)

const

Definition at line 3892 of file net_plugin.cpp.

                                                          {
      vector<connection_status> result;
      std::shared_lock<std::shared_mutex> g( my->connections_mtx );
      result.reserve( my->connections.size() );
      for( const auto& c : my->connections ) {
         result.push_back( c->get_status() );
      }
      return result;
   }

disconnect()

string net_plugin::disconnect

(

const string &

endpoint

)

Definition at line 3871 of file net_plugin.cpp.

                                                     {
      std::lock_guard<std::shared_mutex> g( my->connections_mtx );
      for( auto itr = my->connections.begin(); itr != my->connections.end(); ++itr ) {
         if( (*itr)->peer_address() == host ) {
            fc_ilog( logger, "disconnecting: ${cid}", ("cid", (*itr)->connection_id) );
            (*itr)->close();
            my->connections.erase(itr);
            return "connection removed";
         }
      }
      return "no known connection for host";
   }

handle_sighup()

void net_plugin::handle_sighup ( )

overridevirtual

Reimplemented from appbase::plugin< net_plugin >.

Definition at line 3801 of file net_plugin.cpp.

                                  {
      fc::logger::update( logger_name, logger );
   }

Here is the call graph for this function:

Here is the caller graph for this function:

plugin_initialize()

void net_plugin::plugin_initialize

(

const variables_map &

options

)

Definition at line 3596 of file net_plugin.cpp.

                                                                    {
      fc_ilog( logger, "Initialize net plugin" );
      try {
         peer_log_format = options.at( "peer-log-format" ).as<string>();
 
         my->sync_master.reset( new sync_manager( options.at( "sync-fetch-span" ).as<uint32_t>()));
 
         my->connector_period = std::chrono::seconds( options.at( "connection-cleanup-period" ).as<int>());
         my->max_cleanup_time_ms = options.at("max-cleanup-time-msec").as<int>();
         my->txn_exp_period = def_txn_expire_wait;
         my->p2p_dedup_cache_expire_time_us = fc::seconds( options.at( "p2p-dedup-cache-expire-time-sec" ).as<uint32_t>() );
         my->resp_expected_period = def_resp_expected_wait;
         my->max_client_count = options.at( "max-clients" ).as<int>();
         my->max_nodes_per_host = options.at( "p2p-max-nodes-per-host" ).as<int>();
         my->p2p_accept_transactions = options.at( "p2p-accept-transactions" ).as<bool>();
 
         my->use_socket_read_watermark = options.at( "use-socket-read-watermark" ).as<bool>();
         my->keepalive_interval = std::chrono::milliseconds( options.at( "p2p-keepalive-interval-ms" ).as<int>() );
         SYS_ASSERT( my->keepalive_interval.count() > 0, chain::plugin_config_exception,
                     "p2p-keepalive_interval-ms must be greater than 0" );
 
         if( options.count( "p2p-keepalive-interval-ms" )) {
            my->heartbeat_timeout = std::chrono::milliseconds( options.at( "p2p-keepalive-interval-ms" ).as<int>() * 2 );
         }
 
         if( options.count( "p2p-listen-endpoint" ) && options.at("p2p-listen-endpoint").as<string>().length()) {
            my->p2p_address = options.at( "p2p-listen-endpoint" ).as<string>();
            SYS_ASSERT( my->p2p_address.length() <= max_p2p_address_length, chain::plugin_config_exception,
                        "p2p-listen-endpoint too long, must be less than ${m}", ("m", max_p2p_address_length) );
         }
         if( options.count( "p2p-server-address" ) ) {
            my->p2p_server_address = options.at( "p2p-server-address" ).as<string>();
            SYS_ASSERT( my->p2p_server_address.length() <= max_p2p_address_length, chain::plugin_config_exception,
                        "p2p_server_address too long, must be less than ${m}", ("m", max_p2p_address_length) );
         }
 
         my->thread_pool_size = options.at( "net-threads" ).as<uint16_t>();
         SYS_ASSERT( my->thread_pool_size > 0, chain::plugin_config_exception,
                     "net-threads ${num} must be greater than 0", ("num", my->thread_pool_size) );
 
         if( options.count( "p2p-peer-address" )) {
            my->supplied_peers = options.at( "p2p-peer-address" ).as<vector<string> >();
         }
         if( options.count( "agent-name" )) {
            my->user_agent_name = options.at( "agent-name" ).as<string>();
            SYS_ASSERT( my->user_agent_name.length() <= max_handshake_str_length, chain::plugin_config_exception,
                        "agent-name too long, must be less than ${m}", ("m", max_handshake_str_length) );
         }
 
         if( options.count( "allowed-connection" )) {
            const std::vector<std::string> allowed_remotes = options["allowed-connection"].as<std::vector<std::string>>();
            for( const std::string& allowed_remote : allowed_remotes ) {
               if( allowed_remote == "any" )
                  my->allowed_connections |= net_plugin_impl::Any;
               else if( allowed_remote == "producers" )
                  my->allowed_connections |= net_plugin_impl::Producers;
               else if( allowed_remote == "specified" )
                  my->allowed_connections |= net_plugin_impl::Specified;
               else if( allowed_remote == "none" )
                  my->allowed_connections = net_plugin_impl::None;
            }
         }
 
         if( my->allowed_connections & net_plugin_impl::Specified )
            SYS_ASSERT( options.count( "peer-key" ),
                        plugin_config_exception,
                       "At least one peer-key must accompany 'allowed-connection=specified'" );
 
         if( options.count( "peer-key" )) {
            const std::vector<std::string> key_strings = options["peer-key"].as<std::vector<std::string>>();
            for( const std::string& key_string : key_strings ) {
               my->allowed_peers.push_back( dejsonify<chain::public_key_type>( key_string ));
            }
         }
 
         if( options.count( "peer-private-key" )) {
            const std::vector<std::string> key_id_to_wif_pair_strings = options["peer-private-key"].as<std::vector<std::string>>();
            for( const std::string& key_id_to_wif_pair_string : key_id_to_wif_pair_strings ) {
               auto key_id_to_wif_pair = dejsonify<std::pair<chain::public_key_type, std::string>>(
                     key_id_to_wif_pair_string );
               my->private_keys[key_id_to_wif_pair.first] = fc::crypto::private_key( key_id_to_wif_pair.second );
            }
         }
 
         my->chain_plug = app().find_plugin<chain_plugin>();
         SYS_ASSERT( my->chain_plug, chain::missing_chain_plugin_exception, ""  );
         my->chain_id = my->chain_plug->get_chain_id();
         fc::rand_pseudo_bytes( my->node_id.data(), my->node_id.data_size());
         const controller& cc = my->chain_plug->chain();
 
         if( cc.get_read_mode() == db_read_mode::IRREVERSIBLE || cc.get_read_mode() == db_read_mode::READ_ONLY ) {
            if( my->p2p_accept_transactions ) {
               my->p2p_accept_transactions = false;
               string m = cc.get_read_mode() == db_read_mode::IRREVERSIBLE ? "irreversible" : "read-only";
               wlog( "p2p-accept-transactions set to false due to read-mode: ${m}", ("m", m) );
            }
         }
         if( my->p2p_accept_transactions ) {
            my->chain_plug->enable_accept_transactions();
         }
 
      } FC_LOG_AND_RETHROW()
   }

Here is the call graph for this function:

plugin_shutdown()

void net_plugin::plugin_shutdown

(

)

Definition at line 3805 of file net_plugin.cpp.

                                    {
      try {
         fc_ilog( logger, "shutdown.." );
         my->in_shutdown = true;
         {
            std::lock_guard<std::mutex> g( my->connector_check_timer_mtx );
            if( my->connector_check_timer )
               my->connector_check_timer->cancel();
         }{
            std::lock_guard<std::mutex> g( my->expire_timer_mtx );
            if( my->expire_timer )
               my->expire_timer->cancel();
         }{
            std::lock_guard<std::mutex> g( my->keepalive_timer_mtx );
            if( my->keepalive_timer )
               my->keepalive_timer->cancel();
         }
 
         {
            fc_ilog( logger, "close ${s} connections", ("s", my->connections.size()) );
            std::lock_guard<std::shared_mutex> g( my->connections_mtx );
            for( auto& con : my->connections ) {
               fc_dlog( logger, "close: ${cid}", ("cid", con->connection_id) );
               con->close( false, true );
            }
            my->connections.clear();
         }
 
         if( my->thread_pool ) {
            my->thread_pool->stop();
         }
 
         if( my->acceptor ) {
            boost::system::error_code ec;
            my->acceptor->cancel( ec );
            my->acceptor->close( ec );
         }
 
         app().post( 0, [me = my](){} ); // keep my pointer alive until queue is drained
         fc_ilog( logger, "exit shutdown" );
      }
      FC_CAPTURE_AND_RETHROW()
   }

Here is the call graph for this function:

Here is the caller graph for this function:

plugin_startup()

void net_plugin::plugin_startup

(

)

Definition at line 3700 of file net_plugin.cpp.

                                   {
      handle_sighup();
      try {
 
      fc_ilog( logger, "my node_id is ${id}", ("id", my->node_id ));
 
      my->producer_plug = app().find_plugin<producer_plugin>();
 
      my->thread_pool.emplace( "net", my->thread_pool_size );
 
      my->dispatcher.reset( new dispatch_manager( my_impl->thread_pool->get_executor() ) );
 
      if( !my->p2p_accept_transactions && my->p2p_address.size() ) {
         fc_ilog( logger, "\n"
               "***********************************\n"
               "* p2p-accept-transactions = false *\n"
               "*    Transactions not forwarded   *\n"
               "***********************************\n" );
      }
 
      tcp::endpoint listen_endpoint;
      if( my->p2p_address.size() > 0 ) {
         auto host = my->p2p_address.substr( 0, my->p2p_address.find( ':' ));
         auto port = my->p2p_address.substr( host.size() + 1, my->p2p_address.size());
         tcp::resolver resolver( my->thread_pool->get_executor() );
         // Note: need to add support for IPv6 too?
         listen_endpoint = *resolver.resolve( tcp::v4(), host, port );
 
         my->acceptor.reset( new tcp::acceptor( my_impl->thread_pool->get_executor() ) );
 
         if( !my->p2p_server_address.empty() ) {
            my->p2p_address = my->p2p_server_address;
         } else {
            if( listen_endpoint.address().to_v4() == address_v4::any()) {
               boost::system::error_code ec;
               auto host = host_name( ec );
               if( ec.value() != boost::system::errc::success ) {
 
                  FC_THROW_EXCEPTION( fc::invalid_arg_exception,
                                      "Unable to retrieve host_name. ${msg}", ("msg", ec.message()));
 
               }
               auto port = my->p2p_address.substr( my->p2p_address.find( ':' ), my->p2p_address.size());
               my->p2p_address = host + port;
            }
         }
      }
 
      {
         chain::controller& cc = my->chain_plug->chain();
         cc.accepted_block.connect( [my = my]( const block_state_ptr& s ) {
            my->on_accepted_block( s );
         } );
         cc.pre_accepted_block.connect( [my = my]( const signed_block_ptr& s ) {
            my->on_pre_accepted_block( s );
         } );
         cc.irreversible_block.connect( [my = my]( const block_state_ptr& s ) {
            my->on_irreversible_block( s );
         } );
      }
 
      {
         std::lock_guard<std::mutex> g( my->keepalive_timer_mtx );
         my->keepalive_timer.reset( new boost::asio::steady_timer( my->thread_pool->get_executor() ) );
      }
 
      my->incoming_transaction_ack_subscription = app().get_channel<compat::channels::transaction_ack>().subscribe(
            std::bind(&net_plugin_impl::transaction_ack, my.get(), std::placeholders::_1));
 
      app().post(priority::highest, [my=my, listen_endpoint](){
         if( my->acceptor ) {
            try {
               my->acceptor->open(listen_endpoint.protocol());
               my->acceptor->set_option(tcp::acceptor::reuse_address(true));
               my->acceptor->bind(listen_endpoint);
               my->acceptor->listen();
            } catch (const std::exception& e) {
               elog( "net_plugin::plugin_startup failed to bind to port ${port}, ${what}",
                     ("port", listen_endpoint.port())("what", e.what()) );
               app().quit();
               return;
            }
            fc_ilog( logger, "starting listener, max clients is ${mc}",("mc",my->max_client_count) );
            my->start_listen_loop();
         }
 
         my->ticker();
         my->start_monitors();
         my->update_chain_info();
         for( const auto& seed_node : my->supplied_peers ) {
            my->connect( seed_node );
         }
      });
 
      } catch( ... ) {
         // always want plugin_shutdown even on exception
         plugin_shutdown();
         throw;
      }
   }

Here is the call graph for this function:

set_program_options()

void net_plugin::set_program_options ( options_description & cli, options_description & cfg )

overridevirtual

Implements appbase::abstract_plugin.

Definition at line 3547 of file net_plugin.cpp.

   {
      cfg.add_options()
         ( "p2p-listen-endpoint", bpo::value<string>()->default_value( "0.0.0.0:9876" ), "The actual host:port used to listen for incoming p2p connections.")
         ( "p2p-server-address", bpo::value<string>(), "An externally accessible host:port for identifying this node. Defaults to p2p-listen-endpoint.")
         ( "p2p-peer-address", bpo::value< vector<string> >()->composing(),
           "The public endpoint of a peer node to connect to. Use multiple p2p-peer-address options as needed to compose a network.\n"
           "  Syntax: host:port[:<trx>|<blk>]\n"
           "  The optional 'trx' and 'blk' indicates to node that only transactions 'trx' or blocks 'blk' should be sent."
           "  Examples:\n"
           "    p2p.eos.io:9876\n"
           "    p2p.trx.eos.io:9876:trx\n"
           "    p2p.blk.eos.io:9876:blk\n")
         ( "p2p-max-nodes-per-host", bpo::value<int>()->default_value(def_max_nodes_per_host), "Maximum number of client nodes from any single IP address")
         ( "p2p-accept-transactions", bpo::value<bool>()->default_value(true), "Allow transactions received over p2p network to be evaluated and relayed if valid.")
         ( "agent-name", bpo::value<string>()->default_value("SYS Test Agent"), "The name supplied to identify this node amongst the peers.")
         ( "allowed-connection", bpo::value<vector<string>>()->multitoken()->default_value({"any"}, "any"), "Can be 'any' or 'producers' or 'specified' or 'none'. If 'specified', peer-key must be specified at least once. If only 'producers', peer-key is not required. 'producers' and 'specified' may be combined.")
         ( "peer-key", bpo::value<vector<string>>()->composing()->multitoken(), "Optional public key of peer allowed to connect.  May be used multiple times.")
         ( "peer-private-key", boost::program_options::value<vector<string>>()->composing()->multitoken(),
           "Tuple of [PublicKey, WIF private key] (may specify multiple times)")
         ( "max-clients", bpo::value<int>()->default_value(def_max_clients), "Maximum number of clients from which connections are accepted, use 0 for no limit")
         ( "connection-cleanup-period", bpo::value<int>()->default_value(def_conn_retry_wait), "number of seconds to wait before cleaning up dead connections")
         ( "max-cleanup-time-msec", bpo::value<int>()->default_value(10), "max connection cleanup time per cleanup call in milliseconds")
         ( "p2p-dedup-cache-expire-time-sec", bpo::value<uint32_t>()->default_value(10), "Maximum time to track transaction for duplicate optimization")
         ( "net-threads", bpo::value<uint16_t>()->default_value(my->thread_pool_size),
           "Number of worker threads in net_plugin thread pool" )
         ( "sync-fetch-span", bpo::value<uint32_t>()->default_value(def_sync_fetch_span), "number of blocks to retrieve in a chunk from any individual peer during synchronization")
         ( "use-socket-read-watermark", bpo::value<bool>()->default_value(false), "Enable experimental socket read watermark optimization")
         ( "peer-log-format", bpo::value<string>()->default_value( "[\"${_name}\" - ${_cid} ${_ip}:${_port}] " ),
           "The string used to format peers when logging messages about them.  Variables are escaped with ${<variable name>}.\n"
           "Available Variables:\n"
           "   _name  \tself-reported name\n\n"
           "   _cid   \tassigned connection id\n\n"
           "   _id    \tself-reported ID (64 hex characters)\n\n"
           "   _sid   \tfirst 8 characters of _peer.id\n\n"
           "   _ip    \tremote IP address of peer\n\n"
           "   _port  \tremote port number of peer\n\n"
           "   _lip   \tlocal IP address connected to peer\n\n"
           "   _lport \tlocal port number connected to peer\n\n")
         ( "p2p-keepalive-interval-ms", bpo::value<int>()->default_value(def_keepalive_interval), "peer heartbeat keepalive message interval in milliseconds")
 
        ;
   }

status()

std::optional< connection_status > net_plugin::status

(

const string &

endpoint

)

const

Definition at line 3884 of file net_plugin.cpp.

                                                                              {
      std::shared_lock<std::shared_mutex> g( my->connections_mtx );
      auto con = my->find_connection( host );
      if( con )
         return con->get_status();
      return std::optional<connection_status>();
   }

---

The documentation for this class was generated from the following files:

• plugins/net_plugin/include/sysio/net_plugin/net_plugin.hpp
• plugins/net_plugin/net_plugin.cpp