sysio::connection Class Reference

Inheritance diagram for sysio::connection:

Collaboration diagram for sysio::connection:

Public Member Functions
	connection (const string &endpoint)
	connection ()
	~connection ()=default
bool	start_session ()
bool	socket_is_open () const
const string &	peer_address () const
void	set_connection_type (const string &peer_addr)
bool	is_transactions_only_connection () const
bool	is_blocks_only_connection () const
void	set_heartbeat_timeout (std::chrono::milliseconds msec)
connection_status	get_status () const
bool	connected ()
bool	current ()
void	close (bool reconnect=true, bool shutdown=false)
bool	populate_handshake (handshake_message &hello)
bool	resolve_and_connect ()
void	connect (const std::shared_ptr< tcp::resolver > &resolver, tcp::resolver::results_type endpoints)
void	start_read_message ()
bool	process_next_message (uint32_t message_length)
	Process the next message from the pending message buffer.
void	send_handshake ()
void	blk_send_branch (const block_id_type &msg_head_id)
void	blk_send_branch_impl (uint32_t msg_head_num, uint32_t lib_num, uint32_t head_num)
void	blk_send (const block_id_type &blkid)
void	stop_send ()
void	enqueue (const net_message &msg)
void	enqueue_block (const signed_block_ptr &sb, bool to_sync_queue=false)
void	enqueue_buffer (const std::shared_ptr< std::vector< char > > &send_buffer, go_away_reason close_after_send, bool to_sync_queue=false)
void	cancel_sync (go_away_reason)
void	flush_queues ()
bool	enqueue_sync_block ()
void	request_sync_blocks (uint32_t start, uint32_t end)
void	cancel_wait ()
void	sync_wait ()
void	fetch_wait ()
void	sync_timeout (boost::system::error_code ec)
void	fetch_timeout (boost::system::error_code ec)
void	queue_write (const std::shared_ptr< vector< char > > &buff, std::function< void(boost::system::error_code, std::size_t)> callback, bool to_sync_queue=false)
void	do_queue_write ()
bool	is_valid (const handshake_message &msg) const
void	handle_message (const handshake_message &msg)
void	handle_message (const chain_size_message &msg)
void	handle_message (const go_away_message &msg)
void	handle_message (const notice_message &msg)
void	handle_message (const request_message &msg)
void	handle_message (const sync_request_message &msg)
void	handle_message (const signed_block &msg)=delete
void	handle_message (const block_id_type &id, signed_block_ptr msg)
void	handle_message (const packed_transaction &msg)=delete
void	handle_message (packed_transaction_ptr msg)
void	process_signed_block (const block_id_type &id, signed_block_ptr msg, block_state_ptr bsp)
fc::variant_object	get_logger_variant () const

Public Attributes
boost::asio::io_context::strand	strand
std::shared_ptr< tcp::socket >	socket
fc::message_buffer< 1024 *1024 >	pending_message_buffer
std::atomic< std::size_t >	outstanding_read_bytes {0}
queued_buffer	buffer_queue
fc::sha256	conn_node_id
string	short_conn_node_id
string	log_p2p_address
string	log_remote_endpoint_ip
string	log_remote_endpoint_port
string	local_endpoint_ip
string	local_endpoint_port
std::atomic< uint32_t >	trx_in_progress_size {0}
fc::time_point	last_dropped_trx_msg_time
const uint32_t	connection_id
int16_t	sent_handshake_count = 0
std::atomic< bool >	connecting {true}
std::atomic< bool >	syncing {false}
std::atomic< uint16_t >	protocol_version = 0
uint16_t	net_version = net_version_max
block_status_monitor	block_status_monitor_
std::atomic< uint16_t >	consecutive_immediate_connection_close = 0
std::mutex	response_expected_timer_mtx
boost::asio::steady_timer	response_expected_timer
std::atomic< go_away_reason >	no_retry {no_reason}
std::mutex	conn_mtx
std::optional< request_message >	last_req
handshake_message	last_handshake_recv
handshake_message	last_handshake_sent
block_id_type	fork_head
uint32_t	fork_head_num {0}
fc::time_point	last_close
string	remote_endpoint_ip
tstamp	latest_msg_time {0}
tstamp	hb_timeout {std::chrono::milliseconds{def_keepalive_interval}.count()}
tstamp	latest_blk_time {0}

Peer Timestamps
Time message handling
tstamp	org {0}
	originate timestamp
tstamp	rec {0}
	receive timestamp
tstamp	dst {0}
	destination timestamp
tstamp	xmt {0}
	transmit timestamp
void	check_heartbeat (tstamp current_time)
	Check heartbeat time and send Time_message.
void	send_time ()
	Populate and queue time_message.
void	send_time (const time_message &msg)
	Populate and queue time_message immediately using incoming time_message.
void	handle_message (const time_message &msg)
	Process time_message.
static tstamp	get_time ()
	Read system time and convert to a 64 bit integer.

Detailed Description

Definition at line 581 of file net_plugin.cpp.

Constructor & Destructor Documentation

connection() [1/2]

sysio::connection::connection ( const string & endpoint )

explicit

Definition at line 878 of file net_plugin.cpp.

      : peer_addr( endpoint ),
        strand( my_impl->thread_pool->get_executor() ),
        socket( new tcp::socket( my_impl->thread_pool->get_executor() ) ),
        log_p2p_address( endpoint ),
        connection_id( ++my_impl->current_connection_id ),
        response_expected_timer( my_impl->thread_pool->get_executor() ),
        last_handshake_recv(),
        last_handshake_sent()
   {
      fc_ilog( logger, "created connection ${c} to ${n}", ("c", connection_id)("n", endpoint) );
   }

connection() [2/2]

sysio::connection::connection

(

)

Definition at line 891 of file net_plugin.cpp.

      : peer_addr(),
        strand( my_impl->thread_pool->get_executor() ),
        socket( new tcp::socket( my_impl->thread_pool->get_executor() ) ),
        connection_id( ++my_impl->current_connection_id ),
        response_expected_timer( my_impl->thread_pool->get_executor() ),
        last_handshake_recv(),
        last_handshake_sent()
   {
      fc_dlog( logger, "new connection object created" );
   }

~connection()

sysio::connection::~connection ( )

default

Member Function Documentation

blk_send()

void sysio::connection::blk_send

(

const block_id_type &

blkid

)

Definition at line 1104 of file net_plugin.cpp.

                                                         {
      connection_wptr weak = shared_from_this();
      app().post( priority::medium, [blkid, weak{std::move(weak)}]() {
         connection_ptr c = weak.lock();
         if( !c ) return;
         try {
            controller& cc = my_impl->chain_plug->chain();
            signed_block_ptr b = cc.fetch_block_by_id( blkid );
            if( b ) {
               fc_dlog( logger, "fetch_block_by_id num ${n}, connection ${cid}",
                        ("n", b->block_num())("cid", c->connection_id) );
               my_impl->dispatcher->add_peer_block( blkid, c->connection_id );
               c->strand.post( [c, b{std::move(b)}]() {
                  c->enqueue_block( b );
               } );
            } else {
               fc_ilog( logger, "fetch block by id returned null, id ${id}, connection ${cid}",
                        ("id", blkid)("cid", c->connection_id) );
            }
         } catch( const assert_exception& ex ) {
            fc_elog( logger, "caught assert on fetch_block_by_id, ${ex}, id ${id}, connection ${cid}",
                     ("ex", ex.to_string())("id", blkid)("cid", c->connection_id) );
         } catch( ... ) {
            fc_elog( logger, "caught other exception fetching block id ${id}, connection ${cid}",
                     ("id", blkid)("cid", c->connection_id) );
         }
      });
   }

Here is the call graph for this function:

Here is the caller graph for this function:

blk_send_branch()

void sysio::connection::blk_send_branch

(

const block_id_type &

msg_head_id

)

Definition at line 1029 of file net_plugin.cpp.

                                                                      {
      uint32_t head_num = 0;
      std::tie( std::ignore, std::ignore, head_num,
                std::ignore, std::ignore, std::ignore ) = my_impl->get_chain_info();
 
      peer_dlog(this, "head_num = ${h}",("h",head_num));
      if(head_num == 0) {
         notice_message note;
         note.known_blocks.mode = normal;
         note.known_blocks.pending = 0;
         enqueue(note);
         return;
      }
      std::unique_lock<std::mutex> g_conn( conn_mtx );
      if( last_handshake_recv.generation >= 1 ) {
         peer_dlog( this, "maybe truncating branch at = ${h}:${id}",
                    ("h", block_header::num_from_id(last_handshake_recv.head_id))("id", last_handshake_recv.head_id) );
      }
 
      block_id_type lib_id = last_handshake_recv.last_irreversible_block_id;
      g_conn.unlock();
      const auto lib_num = block_header::num_from_id(lib_id);
      if( lib_num == 0 ) return; // if last_irreversible_block_id is null (we have not received handshake or reset)
 
      app().post( priority::medium, [chain_plug = my_impl->chain_plug, c = shared_from_this(),
            lib_num, head_num, msg_head_id]() {
         auto msg_head_num = block_header::num_from_id(msg_head_id);
         bool on_fork = msg_head_num == 0;
         bool unknown_block = false;
         if( !on_fork ) {
            try {
               const controller& cc = chain_plug->chain();
               block_id_type my_id = cc.get_block_id_for_num( msg_head_num );
               on_fork = my_id != msg_head_id;
            } catch( const unknown_block_exception& ) {
               unknown_block = true;
            } catch( ... ) {
               on_fork = true;
            }
         }
         if( unknown_block ) {
            c->strand.post( [msg_head_num, c]() {
               peer_ilog( c, "Peer asked for unknown block ${mn}, sending: benign_other go away", ("mn", msg_head_num) );
               c->no_retry = benign_other;
               c->enqueue( go_away_message( benign_other ) );
            } );
         } else {
            if( on_fork ) msg_head_num = 0;
            // if peer on fork, start at their last lib, otherwise we can start at msg_head+1
            c->strand.post( [c, msg_head_num, lib_num, head_num]() {
               c->blk_send_branch_impl( msg_head_num, lib_num, head_num );
            } );
         }
      } );
   }

Here is the call graph for this function:

Here is the caller graph for this function:

blk_send_branch_impl()

void sysio::connection::blk_send_branch_impl	(	uint32_t	msg_head_num,
		uint32_t	lib_num,
		uint32_t	head_num )

Definition at line 1086 of file net_plugin.cpp.

                                                                                                     {
      if( !peer_requested ) {
         auto last = msg_head_num != 0 ? msg_head_num : lib_num;
         peer_requested = peer_sync_state( last+1, head_num, last );
      } else {
         auto last = msg_head_num != 0 ? msg_head_num : std::min( peer_requested->last, lib_num );
         uint32_t end   = std::max( peer_requested->end_block, head_num );
         peer_requested = peer_sync_state( last+1, end, last );
      }
      if( peer_requested->start_block <= peer_requested->end_block ) {
         peer_ilog( this, "enqueue ${s} - ${e}", ("s", peer_requested->start_block)("e", peer_requested->end_block) );
         enqueue_sync_block();
      } else {
         peer_ilog( this, "nothing to enqueue" );
         peer_requested.reset();
      }
   }

Here is the call graph for this function:

cancel_sync()

void sysio::connection::cancel_sync

(

go_away_reason

reason

)

Definition at line 1262 of file net_plugin.cpp.

                                                     {
      peer_dlog( this, "cancel sync reason = ${m}, write queue size ${o} bytes",
                 ("m", reason_str( reason ))("o", buffer_queue.write_queue_size()) );
      cancel_wait();
      flush_queues();
      switch (reason) {
      case validation :
      case fatal_other : {
         no_retry = reason;
         enqueue( go_away_message( reason ));
         break;
      }
      default:
         peer_ilog(this, "sending empty request but not calling sync wait");
         enqueue( ( sync_request_message ) {0,0} );
      }
   }

Here is the call graph for this function:

cancel_wait()

void sysio::connection::cancel_wait

(

)

Definition at line 1457 of file net_plugin.cpp.

                                {
      std::lock_guard<std::mutex> g( response_expected_timer_mtx );
      response_expected_timer.cancel();
   }

check_heartbeat()

void sysio::connection::check_heartbeat

(

tstamp

current_time

)

Definition at line 1156 of file net_plugin.cpp.

                                                         {
      if( latest_msg_time > 0 ) {
         if( current_time > latest_msg_time + hb_timeout ) {
            no_retry = benign_other;
            if( !peer_address().empty() ) {
               peer_wlog(this, "heartbeat timed out for peer address");
               close(true);
            } else {
               peer_wlog(this, "heartbeat timed out");
               close(false);
            }
            return;
         } else {
            const tstamp timeout = std::max(hb_timeout/2, 2*std::chrono::milliseconds(config::block_interval_ms).count());
            if ( current_time > latest_blk_time + timeout ) {
               send_handshake();
               return;
            }
         }
      }
 
      send_time();
   }

Here is the call graph for this function:

close()

void sysio::connection::close	(	bool	reconnect = true,
		bool	shutdown = false )

Parameters

reconnect	true if we should try and reconnect immediately after close
shutdown	true only if plugin is shutting down

Definition at line 985 of file net_plugin.cpp.

                                                         {
      strand.post( [self = shared_from_this(), reconnect, shutdown]() {
         connection::_close( self.get(), reconnect, shutdown );
      });
   }

Here is the caller graph for this function:

connect()

void sysio::connection::connect	(	const std::shared_ptr< tcp::resolver > &	resolver,
		tcp::resolver::results_type	endpoints )

Definition at line 2343 of file net_plugin.cpp.

                                                                                                           {
      connecting = true;
      pending_message_buffer.reset();
      buffer_queue.clear_out_queue();
      boost::asio::async_connect( *socket, endpoints,
         boost::asio::bind_executor( strand,
               [resolver, c = shared_from_this(), socket=socket]( const boost::system::error_code& err, const tcp::endpoint& endpoint ) {
            if( !err && socket->is_open() && socket == c->socket ) {
               if( c->start_session() ) {
                  c->send_handshake();
               }
            } else {
               fc_elog( logger, "connection failed to ${host}:${port} ${error}",
                        ("host", endpoint.address().to_string())("port", endpoint.port())( "error", err.message()));
               c->close( false );
            }
      } ) );
   }

Here is the call graph for this function:

connected()

bool sysio::connection::connected

(

)

Definition at line 973 of file net_plugin.cpp.

                              {
      return socket_is_open() && !connecting;
   }

Here is the call graph for this function:

Here is the caller graph for this function:

current()

bool sysio::connection::current

(

)

Definition at line 977 of file net_plugin.cpp.

                            {
      return (connected() && !syncing);
   }

Here is the call graph for this function:

do_queue_write()

void sysio::connection::do_queue_write

(

)

Definition at line 1212 of file net_plugin.cpp.

                                   {
      if( !buffer_queue.ready_to_send() )
         return;
      connection_ptr c(shared_from_this());
 
      std::vector<boost::asio::const_buffer> bufs;
      buffer_queue.fill_out_buffer( bufs );
 
      strand.post( [c{std::move(c)}, bufs{std::move(bufs)}]() {
         boost::asio::async_write( *c->socket, bufs,
            boost::asio::bind_executor( c->strand, [c, socket=c->socket]( boost::system::error_code ec, std::size_t w ) {
            try {
               c->buffer_queue.clear_out_queue();
               // May have closed connection and cleared buffer_queue
               if( !c->socket_is_open() || socket != c->socket ) {
                  peer_ilog( c, "async write socket ${r} before callback", ("r", c->socket_is_open() ? "changed" : "closed") );
                  c->close();
                  return;
               }
 
               if( ec ) {
                  if( ec.value() != boost::asio::error::eof ) {
                     peer_elog( c, "Error sending to peer: ${i}", ( "i", ec.message() ) );
                  } else {
                     peer_wlog( c, "connection closure detected on write" );
                  }
                  c->close();
                  return;
               }
 
               c->buffer_queue.out_callback( ec, w );
 
               c->enqueue_sync_block();
               c->do_queue_write();
            } catch ( const std::bad_alloc& ) {
              throw;
            } catch ( const boost::interprocess::bad_alloc& ) {
              throw;
            } catch( const fc::exception& ex ) {
               peer_elog( c, "fc::exception in do_queue_write: ${s}", ("s", ex.to_string()) );
            } catch( const std::exception& ex ) {
               peer_elog( c, "std::exception in do_queue_write: ${s}", ("s", ex.what()) );
            } catch( ... ) {
               peer_elog( c, "Unknown exception in do_queue_write" );
            }
         }));
      });
   }

Here is the call graph for this function:

Here is the caller graph for this function:

enqueue()

void sysio::connection::enqueue

(

const net_message &

msg

)

Definition at line 1414 of file net_plugin.cpp.

                                                  {
      verify_strand_in_this_thread( strand, __func__, __LINE__ );
      go_away_reason close_after_send = no_reason;
      if (std::holds_alternative<go_away_message>(m)) {
         close_after_send = std::get<go_away_message>(m).reason;
      }
 
      buffer_factory buff_factory;
      auto send_buffer = buff_factory.get_send_buffer( m );
      enqueue_buffer( send_buffer, close_after_send );
   }

Here is the call graph for this function:

Here is the caller graph for this function:

enqueue_block()

void sysio::connection::enqueue_block	(	const signed_block_ptr &	sb,
		bool	to_sync_queue = false )

Definition at line 1427 of file net_plugin.cpp.

                                                                                {
      peer_dlog( this, "enqueue block ${num}", ("num", b->block_num()) );
      verify_strand_in_this_thread( strand, __func__, __LINE__ );
 
      block_buffer_factory buff_factory;
      auto sb = buff_factory.get_send_buffer( b );
      latest_blk_time = get_time();
      enqueue_buffer( sb, no_reason, to_sync_queue);
   }

Here is the call graph for this function:

enqueue_buffer()

void sysio::connection::enqueue_buffer	(	const std::shared_ptr< std::vector< char > > &	send_buffer,
		go_away_reason	close_after_send,
		bool	to_sync_queue = false )

Definition at line 1438 of file net_plugin.cpp.

   {
      connection_ptr self = shared_from_this();
      queue_write(send_buffer,
            [conn{std::move(self)}, close_after_send](boost::system::error_code ec, std::size_t ) {
                        if (ec) return;
                        if (close_after_send != no_reason) {
                           fc_ilog( logger, "sent a go away message: ${r}, closing connection ${cid}",
                                    ("r", reason_str(close_after_send))("cid", conn->connection_id) );
                           conn->close();
                           return;
                        }
                  },
                  to_sync_queue);
   }

Here is the call graph for this function:

enqueue_sync_block()

bool sysio::connection::enqueue_sync_block

(

)

Definition at line 1281 of file net_plugin.cpp.

                                       {
      if( !peer_requested ) {
         return false;
      } else {
         peer_dlog( this, "enqueue sync block ${num}", ("num", peer_requested->last + 1) );
      }
      uint32_t num = ++peer_requested->last;
      if(num == peer_requested->end_block) {
         peer_requested.reset();
         peer_dlog( this, "completing enqueue_sync_block ${num}", ("num", num) );
      }
      connection_wptr weak = shared_from_this();
      app().post( priority::medium, [num, weak{std::move(weak)}]() {
         connection_ptr c = weak.lock();
         if( !c ) return;
         controller& cc = my_impl->chain_plug->chain();
         signed_block_ptr sb;
         try {
            sb = cc.fetch_block_by_number( num );
         } FC_LOG_AND_DROP();
         if( sb ) {
            c->strand.post( [c, sb{std::move(sb)}]() {
               c->enqueue_block( sb, true );
            });
         } else {
            c->strand.post( [c, num]() {
               peer_ilog( c, "enqueue sync, unable to fetch block ${num}, sending benign_other go away", ("num", num) );
               c->peer_requested.reset(); // unable to provide requested blocks
               c->no_retry = benign_other;
               c->enqueue( go_away_message( benign_other ) );
            });
         }
      });
 
      return true;
   }

Here is the call graph for this function:

Here is the caller graph for this function:

fetch_timeout()

void sysio::connection::fetch_timeout

(

boost::system::error_code

ec

)

Definition at line 1495 of file net_plugin.cpp.

                                                            {
      if( !ec ) {
         my_impl->dispatcher->retry_fetch( shared_from_this() );
      } else if( ec != boost::asio::error::operation_aborted ) { // don't log on operation_aborted, called on destroy
         peer_elog( this, "setting timer for fetch request got error ${ec}", ("ec", ec.message() ) );
      }
   }

fetch_wait()

void sysio::connection::fetch_wait

(

)

Definition at line 1474 of file net_plugin.cpp.

                               {
      connection_ptr c( shared_from_this() );
      std::lock_guard<std::mutex> g( response_expected_timer_mtx );
      response_expected_timer.expires_from_now( my_impl->resp_expected_period );
      response_expected_timer.async_wait(
            boost::asio::bind_executor( c->strand, [c]( boost::system::error_code ec ) {
               c->fetch_timeout(ec);
            } ) );
   }

flush_queues()

void sysio::connection::flush_queues

(

)

Definition at line 981 of file net_plugin.cpp.

                                 {
      buffer_queue.clear_write_queue();
   }

Here is the call graph for this function:

Here is the caller graph for this function:

get_logger_variant()

fc::variant_object sysio::connection::get_logger_variant ( ) const

inline

Definition at line 790 of file net_plugin.cpp.

                                                  {
         fc::mutable_variant_object mvo;
         mvo( "_name", log_p2p_address)
            ( "_cid", connection_id )
            ( "_id", conn_node_id )
            ( "_sid", short_conn_node_id )
            ( "_ip", log_remote_endpoint_ip )
            ( "_port", log_remote_endpoint_port )
            ( "_lip", local_endpoint_ip )
            ( "_lport", local_endpoint_port );
         return mvo;
      }

get_status()

connection_status sysio::connection::get_status

(

)

const

Definition at line 943 of file net_plugin.cpp.

                                                 {
      connection_status stat;
      stat.peer = peer_addr;
      stat.connecting = connecting;
      stat.syncing = syncing;
      std::lock_guard<std::mutex> g( conn_mtx );
      stat.last_handshake = last_handshake_recv;
      return stat;
   }

get_time()

static tstamp sysio::connection::get_time ( )

inlinestatic

There are only two calls on this routine in the program. One when a packet arrives from the network and the other when a packet is placed on the send queue. Calls the kernel time of day routine and converts to a (at least) 64 bit integer.

Definition at line 729 of file net_plugin.cpp.

                               {
         return std::chrono::system_clock::now().time_since_epoch().count();
      }

Here is the caller graph for this function:

handle_message() [1/11]

void sysio::connection::handle_message	(	const block_id_type &	id,
		signed_block_ptr	msg )

Definition at line 3136 of file net_plugin.cpp.

                                                                                  {
      peer_dlog( this, "received signed_block ${num}, id ${id}", ("num", ptr->block_num())("id", id) );
 
      controller& cc = my_impl->chain_plug->chain();
      block_state_ptr bsp;
      bool exception = false;
      try {
         if( cc.fetch_block_state_by_id( id ) ) {
            my_impl->dispatcher->add_peer_block( id, connection_id );
            my_impl->sync_master->sync_recv_block( shared_from_this(), id, ptr->block_num(), false );
            return;
         }
         // this may return null if block is not immediately ready to be processed
         bsp = cc.create_block_state( id, ptr );
      } catch( const fc::exception& ex ) {
         exception = true;
         peer_elog(this, "bad block exception: #${n} ${id}...: ${m}",
                   ("n", ptr->block_num())("id", id.str().substr(8,16))("m",ex.to_string()));
      } catch( ... ) {
         exception = true;
         peer_elog(this, "bad block: #${n} ${id}...: unknown exception",
                   ("n", ptr->block_num())("id", id.str().substr(8,16)));
      }
      if( exception ) {
         my_impl->sync_master->rejected_block( shared_from_this(), ptr->block_num() );
         my_impl->dispatcher->rejected_block( id );
         return;
      }
 
      bool signal_producer = !!bsp; // ready to process immediately, so signal producer to interrupt start_block
      app().post(priority::medium, [ptr{std::move(ptr)}, bsp{std::move(bsp)}, id, c = shared_from_this()]() mutable {
         c->process_signed_block( id, std::move(ptr), std::move(bsp) );
      });
 
      if( signal_producer )
         my_impl->producer_plug->received_block();
   }

Here is the call graph for this function:

handle_message() [2/11]

void sysio::connection::handle_message

(

const chain_size_message &

msg

)

Definition at line 2759 of file net_plugin.cpp.

                                                                  {
      peer_dlog(this, "received chain_size_message");
   }

handle_message() [3/11]

void sysio::connection::handle_message

(

const go_away_message &

msg

)

Definition at line 2922 of file net_plugin.cpp.

                                                               {
      peer_wlog( this, "received go_away_message, reason = ${r}", ("r", reason_str( msg.reason )) );
 
      bool retry = no_retry == no_reason; // if no previous go away message
      no_retry = msg.reason;
      if( msg.reason == duplicate ) {
         conn_node_id = msg.node_id;
      }
      if( msg.reason == wrong_version ) {
         if( !retry ) no_retry = fatal_other; // only retry once on wrong version
      }
      else if ( msg.reason == benign_other ) {
         if ( retry ) peer_dlog( this, "received benign_other reason, retrying to connect");
      }
      else {
         retry = false;
      }
      flush_queues();
 
      close( retry ); // reconnect if wrong_version
   }

Here is the call graph for this function:

handle_message() [4/11]

void sysio::connection::handle_message

(

const handshake_message &

msg

)

Definition at line 2763 of file net_plugin.cpp.

                                                                 {
      peer_dlog( this, "received handshake_message" );
      if( !is_valid( msg ) ) {
         peer_elog( this, "bad handshake message");
         no_retry = go_away_reason::fatal_other;
         enqueue( go_away_message( fatal_other ) );
         return;
      }
      peer_dlog( this, "received handshake gen ${g}, lib ${lib}, head ${head}",
                 ("g", msg.generation)("lib", msg.last_irreversible_block_num)("head", msg.head_num) );
 
      std::unique_lock<std::mutex> g_conn( conn_mtx );
      last_handshake_recv = msg;
      g_conn.unlock();
 
      connecting = false;
      if (msg.generation == 1) {
         if( msg.node_id == my_impl->node_id) {
            peer_elog( this, "Self connection detected node_id ${id}. Closing connection", ("id", msg.node_id) );
            no_retry = go_away_reason::self;
            enqueue( go_away_message( go_away_reason::self ) );
            return;
         }
 
         log_p2p_address = msg.p2p_address;
         if( peer_address().empty() ) {
            set_connection_type( msg.p2p_address );
         }
 
         std::unique_lock<std::mutex> g_conn( conn_mtx );
         if( peer_address().empty() || last_handshake_recv.node_id == fc::sha256()) {
            auto c_time = last_handshake_sent.time;
            g_conn.unlock();
            peer_dlog( this, "checking for duplicate" );
            std::shared_lock<std::shared_mutex> g_cnts( my_impl->connections_mtx );
            for(const auto& check : my_impl->connections) {
               if(check.get() == this)
                  continue;
               std::unique_lock<std::mutex> g_check_conn( check->conn_mtx );
               fc_dlog( logger, "dup check: connected ${c}, ${l} =? ${r}",
                        ("c", check->connected())("l", check->last_handshake_recv.node_id)("r", msg.node_id) );
               if(check->connected() && check->last_handshake_recv.node_id == msg.node_id) {
                  if (net_version < proto_dup_goaway_resolution || msg.network_version < proto_dup_goaway_resolution) {
                     // It's possible that both peers could arrive here at relatively the same time, so
                     // we need to avoid the case where they would both tell a different connection to go away.
                     // Using the sum of the initial handshake times of the two connections, we will
                     // arbitrarily (but consistently between the two peers) keep one of them.
                  
                     auto check_time = check->last_handshake_sent.time + check->last_handshake_recv.time;
                     g_check_conn.unlock();
                     if (msg.time + c_time <= check_time)
                        continue;
                  } else if (net_version < proto_dup_node_id_goaway || msg.network_version < proto_dup_node_id_goaway) {
                     if (my_impl->p2p_address < msg.p2p_address) {
                        fc_dlog( logger, "my_impl->p2p_address '${lhs}' < msg.p2p_address '${rhs}'",
                                 ("lhs", my_impl->p2p_address)( "rhs", msg.p2p_address ) );
                        // only the connection from lower p2p_address to higher p2p_address will be considered as a duplicate,
                        // so there is no chance for both connections to be closed
                        continue;
                     }
                  } else if (my_impl->node_id < msg.node_id) {
                     fc_dlog( logger, "not duplicate, my_impl->node_id '${lhs}' < msg.node_id '${rhs}'",
                              ("lhs", my_impl->node_id)("rhs", msg.node_id) );
                     // only the connection from lower node_id to higher node_id will be considered as a duplicate,
                     // so there is no chance for both connections to be closed
                     continue; 
                  }
 
                  g_cnts.unlock();
                  peer_dlog( this, "sending go_away duplicate, msg.p2p_address: ${add}", ("add", msg.p2p_address) );
                  go_away_message gam(duplicate);
                  gam.node_id = conn_node_id;
                  enqueue(gam);
                  no_retry = duplicate;
                  return;
               }
            }
         } else {
            peer_dlog( this, "skipping duplicate check, addr == ${pa}, id = ${ni}",
                       ("pa", peer_address())( "ni", last_handshake_recv.node_id ) );
            g_conn.unlock();
         }
 
         if( msg.chain_id != my_impl->chain_id ) {
            peer_elog( this, "Peer on a different chain. Closing connection" );
            no_retry = go_away_reason::wrong_chain;
            enqueue( go_away_message(go_away_reason::wrong_chain) );
            return;
         }
         protocol_version = my_impl->to_protocol_version(msg.network_version);
         if( protocol_version != net_version ) {
            peer_ilog( this, "Local network version different: ${nv} Remote version: ${mnv}",
                       ("nv", net_version)("mnv", protocol_version.load()) );
         } else {
            peer_ilog( this, "Local network version: ${nv}", ("nv", net_version) );
         }
 
         conn_node_id = msg.node_id;
         short_conn_node_id = conn_node_id.str().substr( 0, 7 );
 
         if( !my_impl->authenticate_peer( msg ) ) {
            peer_elog( this, "Peer not authenticated.  Closing connection." );
            no_retry = go_away_reason::authentication;
            enqueue( go_away_message( go_away_reason::authentication ) );
            return;
         }
 
         uint32_t peer_lib = msg.last_irreversible_block_num;
         connection_wptr weak = shared_from_this();
         app().post( priority::medium, [peer_lib, chain_plug = my_impl->chain_plug, weak{std::move(weak)},
                                     msg_lib_id = msg.last_irreversible_block_id]() {
            connection_ptr c = weak.lock();
            if( !c ) return;
            controller& cc = chain_plug->chain();
            uint32_t lib_num = cc.last_irreversible_block_num();
 
            fc_dlog( logger, "handshake check for fork lib_num = ${ln}, peer_lib = ${pl}, connection ${cid}",
                     ("ln", lib_num)("pl", peer_lib)("cid", c->connection_id) );
 
            if( peer_lib <= lib_num && peer_lib > 0 ) {
               bool on_fork = false;
               try {
                  block_id_type peer_lib_id = cc.get_block_id_for_num( peer_lib );
                  on_fork = (msg_lib_id != peer_lib_id);
               } catch( const unknown_block_exception& ) {
                  // allow this for now, will be checked on sync
                  fc_dlog( logger, "peer last irreversible block ${pl} is unknown, connection ${cid}",
                           ("pl", peer_lib)("cid", c->connection_id) );
               } catch( ... ) {
                  fc_wlog( logger, "caught an exception getting block id for ${pl}, connection ${cid}",
                           ("pl", peer_lib)("cid", c->connection_id) );
                  on_fork = true;
               }
               if( on_fork ) {
                  c->strand.post( [c]() {
                     peer_elog( c, "Peer chain is forked, sending: forked go away" );
                     c->no_retry = go_away_reason::forked;
                     c->enqueue( go_away_message( go_away_reason::forked ) );
                  } );
               }
            }
         });
 
         // we don't support the 2.1 packed_transaction & signed_block, so tell 2.1 clients we are 2.0
         if( protocol_version >= proto_pruned_types && protocol_version < proto_wire_sysio_initial ) {
            sent_handshake_count = 0;
            net_version = proto_explicit_sync;
            send_handshake();
            return;
         }
 
         if( sent_handshake_count == 0 ) {
            send_handshake();
         }
      }
 
      my_impl->sync_master->recv_handshake( shared_from_this(), msg );
   }

Here is the call graph for this function:

handle_message() [5/11]

void sysio::connection::handle_message

(

const notice_message &

msg

)

Definition at line 2984 of file net_plugin.cpp.

                                                              {
      // peer tells us about one or more blocks or txns. When done syncing, forward on
      // notices of previously unknown blocks or txns,
      //
      peer_dlog( this, "received notice_message" );
      connecting = false;
      if( msg.known_blocks.ids.size() > 1 ) {
         peer_elog( this, "Invalid notice_message, known_blocks.ids.size ${s}, closing connection",
                    ("s", msg.known_blocks.ids.size()) );
         close( false );
         return;
      }
      if( msg.known_trx.mode != none ) {
         if( logger.is_enabled( fc::log_level::debug ) ) {
            const block_id_type& blkid = msg.known_blocks.ids.empty() ? block_id_type{} : msg.known_blocks.ids.back();
            peer_dlog( this, "this is a ${m} notice with ${n} pending blocks: ${num} ${id}...",
                       ("m", modes_str( msg.known_blocks.mode ))("n", msg.known_blocks.pending)
                       ("num", block_header::num_from_id( blkid ))("id", blkid.str().substr( 8, 16 )) );
         }
      }
      switch (msg.known_trx.mode) {
      case none:
         break;
      case last_irr_catch_up: {
         std::unique_lock<std::mutex> g_conn( conn_mtx );
         last_handshake_recv.head_num = msg.known_blocks.pending;
         g_conn.unlock();
         break;
      }
      case catch_up : {
         break;
      }
      case normal: {
         my_impl->dispatcher->recv_notice( shared_from_this(), msg, false );
      }
      }
 
      if( msg.known_blocks.mode != none ) {
         peer_dlog( this, "this is a ${m} notice with ${n} blocks",
                    ("m", modes_str( msg.known_blocks.mode ))( "n", msg.known_blocks.pending ) );
      }
      switch (msg.known_blocks.mode) {
      case none : {
         break;
      }
      case last_irr_catch_up:
      case catch_up: {
         my_impl->sync_master->sync_recv_notice( shared_from_this(), msg );
         break;
      }
      case normal : {
         my_impl->dispatcher->recv_notice( shared_from_this(), msg, false );
         break;
      }
      default: {
         peer_elog( this, "bad notice_message : invalid known_blocks.mode ${m}",
                    ("m", static_cast<uint32_t>(msg.known_blocks.mode)) );
      }
      }
   }

Here is the call graph for this function:

handle_message() [6/11]

void sysio::connection::handle_message ( const packed_transaction & msg )

delete

handle_message() [7/11]

void sysio::connection::handle_message

(

const request_message &

msg

)

Definition at line 3045 of file net_plugin.cpp.

                                                               {
      if( msg.req_blocks.ids.size() > 1 ) {
         peer_elog( this, "Invalid request_message, req_blocks.ids.size ${s}, closing",
                    ("s", msg.req_blocks.ids.size()) );
         close();
         return;
      }
 
      switch (msg.req_blocks.mode) {
      case catch_up :
         peer_dlog( this, "received request_message:catch_up" );
         blk_send_branch( msg.req_blocks.ids.empty() ? block_id_type() : msg.req_blocks.ids.back() );
         break;
      case normal :
         peer_dlog( this, "received request_message:normal" );
         if( !msg.req_blocks.ids.empty() ) {
            blk_send( msg.req_blocks.ids.back() );
         }
         break;
      default:;
      }
 
 
      switch (msg.req_trx.mode) {
      case catch_up :
         break;
      case none :
         if( msg.req_blocks.mode == none ) {
            stop_send();
         }
         // no break
      case normal :
         if( !msg.req_trx.ids.empty() ) {
            peer_elog( this, "Invalid request_message, req_trx.ids.size ${s}", ("s", msg.req_trx.ids.size()) );
            close();
            return;
         }
         break;
      default:;
      }
   }

Here is the call graph for this function:

handle_message() [8/11]

void sysio::connection::handle_message ( const signed_block & msg )

delete

handle_message() [9/11]

void sysio::connection::handle_message

(

const sync_request_message &

msg

)

Definition at line 3087 of file net_plugin.cpp.

                                                                    {
      peer_dlog( this, "peer requested ${start} to ${end}", ("start", msg.start_block)("end", msg.end_block) );
      if( msg.end_block == 0 ) {
         peer_requested.reset();
         flush_queues();
      } else {
         if (peer_requested) {
            // This happens when peer already requested some range and sync is still in progress
            // It could be higher in case of peer requested head catchup and current request is lib catchup
            // So to make sure peer will receive all requested blocks we assign end_block to highest value
            peer_requested->end_block = std::max(msg.end_block, peer_requested->end_block);
         }
         else {
            peer_requested = peer_sync_state( msg.start_block, msg.end_block, msg.start_block-1);
         }
         enqueue_sync_block();
      }
   }

Here is the call graph for this function:

handle_message() [10/11]

void sysio::connection::handle_message

(

const time_message &

msg

)

Calculate offset, delay and dispersion. Note carefully the implied processing. The first-order difference is done directly in 64-bit arithmetic, then the result is converted to floating double. All further processing is in floating-double arithmetic with rounding done by the hardware. This is necessary in order to avoid overflow and preserve precision.

Definition at line 2944 of file net_plugin.cpp.

                                                            {
      peer_ilog( this, "received time_message" );
 
      /* We've already lost however many microseconds it took to dispatch
       * the message, but it can't be helped.
       */
      msg.dst = get_time();
 
      // If the transmit timestamp is zero, the peer is horribly broken.
      if(msg.xmt == 0)
         return;                 /* invalid timestamp */
 
      if(msg.xmt == xmt)
         return;                 /* duplicate packet */
 
      xmt = msg.xmt;
      rec = msg.rec;
      dst = msg.dst;
 
      if( msg.org == 0 ) {
         send_time( msg );
         return;  // We don't have enough data to perform the calculation yet.
      }
 
      double offset = (double(rec - org) + double(msg.xmt - dst)) / 2;
      double NsecPerUsec{1000};
 
      if( logger.is_enabled( fc::log_level::all ) )
         logger.log( FC_LOG_MESSAGE( all, "Clock offset is ${o}ns (${us}us)",
                                     ("o", offset)( "us", offset / NsecPerUsec ) ) );
      org = 0;
      rec = 0;
 
      std::unique_lock<std::mutex> g_conn( conn_mtx );
      if( last_handshake_recv.generation == 0 ) {
         g_conn.unlock();
         send_handshake();
      }
   }

Here is the call graph for this function:

handle_message() [11/11]

void sysio::connection::handle_message

(

packed_transaction_ptr

msg

)

Definition at line 3110 of file net_plugin.cpp.

                                                               {
      const auto& tid = trx->id();
      peer_dlog( this, "received packed_transaction ${id}", ("id", tid) );
 
      trx_in_progress_size += calc_trx_size( trx );
      my_impl->chain_plug->accept_transaction( trx,
         [weak = weak_from_this(), trx](const std::variant<fc::exception_ptr, transaction_trace_ptr>& result) mutable {
         // next (this lambda) called from application thread
         if (std::holds_alternative<fc::exception_ptr>(result)) {
            fc_dlog( logger, "bad packed_transaction : ${m}", ("m", std::get<fc::exception_ptr>(result)->what()) );
         } else {
            const transaction_trace_ptr& trace = std::get<transaction_trace_ptr>(result);
            if( !trace->except ) {
               fc_dlog( logger, "chain accepted transaction, bcast ${id}", ("id", trace->id) );
            } else {
               fc_elog( logger, "bad packed_transaction : ${m}", ("m", trace->except->what()));
            }
         }
         connection_ptr conn = weak.lock();
         if( conn ) {
            conn->trx_in_progress_size -= calc_trx_size( trx );
         }
      });
   }

Here is the call graph for this function:

is_blocks_only_connection()

bool sysio::connection::is_blocks_only_connection ( ) const

inline

Definition at line 595 of file net_plugin.cpp.

{ return connection_type == blocks_only; }

Here is the caller graph for this function:

is_transactions_only_connection()

bool sysio::connection::is_transactions_only_connection ( ) const

inline

Definition at line 594 of file net_plugin.cpp.

{ return connection_type == transactions_only; }

Here is the caller graph for this function:

is_valid()

bool sysio::connection::is_valid

(

const handshake_message &

msg

)

const

Definition at line 2720 of file net_plugin.cpp.

                                                                 {
      // Do some basic validation of an incoming handshake_message, so things
      // that really aren't handshake messages can be quickly discarded without
      // affecting state.
      bool valid = true;
      if (msg.last_irreversible_block_num > msg.head_num) {
         peer_wlog( this, "Handshake message validation: last irreversible block (${i}) is greater than head block (${h})",
                  ("i", msg.last_irreversible_block_num)("h", msg.head_num) );
         valid = false;
      }
      if (msg.p2p_address.empty()) {
         peer_wlog( this, "Handshake message validation: p2p_address is null string" );
         valid = false;
      } else if( msg.p2p_address.length() > max_handshake_str_length ) {
         // see max_handshake_str_length comment in protocol.hpp
         peer_wlog( this, "Handshake message validation: p2p_address too large: ${p}",
                    ("p", msg.p2p_address.substr(0, max_handshake_str_length) + "...") );
         valid = false;
      }
      if (msg.os.empty()) {
         peer_wlog( this, "Handshake message validation: os field is null string" );
         valid = false;
      } else if( msg.os.length() > max_handshake_str_length ) {
         peer_wlog( this, "Handshake message validation: os field too large: ${p}",
                    ("p", msg.os.substr(0, max_handshake_str_length) + "...") );
         valid = false;
      }
      if( msg.agent.length() > max_handshake_str_length ) {
         peer_wlog( this, "Handshake message validation: agent field too large: ${p}",
                  ("p", msg.agent.substr(0, max_handshake_str_length) + "...") );
         valid = false;
      }
      if ((msg.sig != chain::signature_type() || msg.token != sha256()) && (msg.token != fc::sha256::hash(msg.time))) {
         peer_wlog( this, "Handshake message validation: token field invalid" );
         valid = false;
      }
      return valid;
   }

Here is the call graph for this function:

Here is the caller graph for this function:

peer_address()

const string & sysio::connection::peer_address ( ) const

inline

Definition at line 591 of file net_plugin.cpp.

{ return peer_addr; } // thread safe, const

Here is the caller graph for this function:

populate_handshake()

bool sysio::connection::populate_handshake

(

handshake_message &

hello

)

Definition at line 3504 of file net_plugin.cpp.

                                                                 {
      namespace sc = std::chrono;
      hello.network_version = net_version_base + net_version;
      uint32_t lib, head;
      std::tie( lib, std::ignore, head,
                hello.last_irreversible_block_id, std::ignore, hello.head_id ) = my_impl->get_chain_info();
      hello.last_irreversible_block_num = lib;
      hello.head_num = head;
      hello.chain_id = my_impl->chain_id;
      hello.node_id = my_impl->node_id;
      hello.key = my_impl->get_authentication_key();
      hello.time = sc::duration_cast<sc::nanoseconds>(sc::system_clock::now().time_since_epoch()).count();
      hello.token = fc::sha256::hash(hello.time);
      hello.sig = my_impl->sign_compact(hello.key, hello.token);
      // If we couldn't sign, don't send a token.
      if(hello.sig == chain::signature_type())
         hello.token = sha256();
      hello.p2p_address = my_impl->p2p_address;
      if( is_transactions_only_connection() ) hello.p2p_address += ":trx";
      if( is_blocks_only_connection() ) hello.p2p_address += ":blk";
      hello.p2p_address += " - " + hello.node_id.str().substr(0,7);
#if defined( __APPLE__ )
      hello.os = "osx";
#elif defined( __linux__ )
      hello.os = "linux";
#elif defined( _WIN32 )
      hello.os = "win32";
#else
      hello.os = "other";
#endif
      hello.agent = my_impl->user_agent_name;
 
      return true;
   }

Here is the call graph for this function:

process_next_message()

bool sysio::connection::process_next_message

(

uint32_t

message_length

)

Process the next message from the pending_message_buffer. message_length is the already determined length of the data part of the message that will handle the message. Returns true is successful. Returns false if an error was encountered unpacking or processing the message.

Definition at line 2560 of file net_plugin.cpp.

                                                                  {
      try {
         latest_msg_time = get_time();
 
         // if next message is a block we already have, exit early
         auto peek_ds = pending_message_buffer.create_peek_datastream();
         unsigned_int which{};
         fc::raw::unpack( peek_ds, which );
         if( which == signed_block_which ) {
            latest_blk_time = get_time();
            return process_next_block_message( message_length );
 
         } else if( which == packed_transaction_which ) {
            return process_next_trx_message( message_length );
 
         } else {
            auto ds = pending_message_buffer.create_datastream();
            net_message msg;
            fc::raw::unpack( ds, msg );
            msg_handler m( shared_from_this() );
            std::visit( m, msg );
         }
 
      } catch( const fc::exception& e ) {
         peer_elog( this, "Exception in handling message: ${s}", ("s", e.to_detail_string()) );
         close();
         return false;
      }
      return true;
   }

Here is the call graph for this function:

process_signed_block()

void sysio::connection::process_signed_block	(	const block_id_type &	id,
		signed_block_ptr	msg,
		block_state_ptr	bsp )

Definition at line 3175 of file net_plugin.cpp.

                                                                                                                 {
      controller& cc = my_impl->chain_plug->chain();
      uint32_t blk_num = msg->block_num();
      // use c in this method instead of this to highlight that all methods called on c-> must be thread safe
      connection_ptr c = shared_from_this();
 
      // if we have closed connection then stop processing
      if( !c->socket_is_open() )
         return;
 
      try {
         if( cc.fetch_block_by_id(blk_id) ) {
            c->strand.post( [sync_master = my_impl->sync_master.get(),
                             dispatcher = my_impl->dispatcher.get(), c, blk_id, blk_num]() {
               dispatcher->add_peer_block( blk_id, c->connection_id );
               sync_master->sync_recv_block( c, blk_id, blk_num, false );
            });
            return;
         }
      } catch(...) {
         // should this even be caught?
         fc_elog( logger, "Caught an unknown exception trying to recall block ID" );
      }
 
      fc::microseconds age( fc::time_point::now() - msg->timestamp);
      fc_dlog( logger, "received signed_block: #${n} block age in secs = ${age}, connection ${cid}",
               ("n", blk_num)("age", age.to_seconds())("cid", c->connection_id) );
 
      go_away_reason reason = fatal_other;
      try {
         bool accepted = my_impl->chain_plug->accept_block(msg, blk_id, bsp);
         my_impl->update_chain_info();
         if( !accepted ) return;
         reason = no_reason;
      } catch( const unlinkable_block_exception &ex) {
         fc_elog(logger, "unlinkable_block_exception connection ${cid}: #${n} ${id}...: ${m}",
                 ("cid", c->connection_id)("n", blk_num)("id", blk_id.str().substr(8,16))("m",ex.to_string()));
         reason = unlinkable;
      } catch( const block_validate_exception &ex ) {
         fc_elog(logger, "block_validate_exception connection ${cid}: #${n} ${id}...: ${m}",
                 ("cid", c->connection_id)("n", blk_num)("id", blk_id.str().substr(8,16))("m",ex.to_string()));
         reason = validation;
      } catch( const assert_exception &ex ) {
         fc_elog(logger, "block assert_exception connection ${cid}: #${n} ${id}...: ${m}",
                 ("cid", c->connection_id)("n", blk_num)("id", blk_id.str().substr(8,16))("m",ex.to_string()));
      } catch( const fc::exception &ex ) {
         fc_elog(logger, "bad block exception connection ${cid}: #${n} ${id}...: ${m}",
                 ("cid", c->connection_id)("n", blk_num)("id", blk_id.str().substr(8,16))("m",ex.to_string()));
      } catch( ... ) {
         fc_elog(logger, "bad block connection ${cid}: #${n} ${id}...: unknown exception",
                 ("cid", c->connection_id)("n", blk_num)("id", blk_id.str().substr(8,16)));
      }
 
      if( reason == no_reason ) {
         boost::asio::post( my_impl->thread_pool->get_executor(), [dispatcher = my_impl->dispatcher.get(), cid=c->connection_id, blk_id, msg]() {
            fc_dlog( logger, "accepted signed_block : #${n} ${id}...", ("n", msg->block_num())("id", blk_id.str().substr(8,16)) );
            dispatcher->add_peer_block( blk_id, cid );
         });
         c->strand.post( [sync_master = my_impl->sync_master.get(), dispatcher = my_impl->dispatcher.get(), c, blk_id, blk_num]() {
            dispatcher->recv_block( c, blk_id, blk_num );
            sync_master->sync_recv_block( c, blk_id, blk_num, true );
         });
      } else {
         c->strand.post( [sync_master = my_impl->sync_master.get(), dispatcher = my_impl->dispatcher.get(), c, blk_id, blk_num]() {
            sync_master->rejected_block( c, blk_num );
            dispatcher->rejected_block( blk_id );
         });
      }
   }

Here is the call graph for this function:

queue_write()

void sysio::connection::queue_write	(	const std::shared_ptr< vector< char > > &	buff,
		std::function< void(boost::system::error_code, std::size_t)>	callback,
		bool	to_sync_queue = false )

Definition at line 1200 of file net_plugin.cpp.

                                                    {
      if( !buffer_queue.add_write_queue( buff, callback, to_sync_queue )) {
         peer_wlog( this, "write_queue full ${s} bytes, giving up on connection", ("s", buffer_queue.write_queue_size()) );
         close();
         return;
      }
      do_queue_write();
   }

Here is the call graph for this function:

request_sync_blocks()

void sysio::connection::request_sync_blocks	(	uint32_t	start,
		uint32_t	end )

Definition at line 1504 of file net_plugin.cpp.

                                                                    {
      sync_request_message srm = {start,end};
      enqueue( net_message(srm) );
      sync_wait();
   }

resolve_and_connect()

bool sysio::connection::resolve_and_connect

(

)

Definition at line 2287 of file net_plugin.cpp.

                                        {
      switch ( no_retry ) {
         case no_reason:
         case wrong_version:
         case benign_other:
         case duplicate: // attempt reconnect in case connection has been dropped, should quickly disconnect if duplicate
            break;
         default:
            fc_dlog( logger, "Skipping connect due to go_away reason ${r}",("r", reason_str( no_retry )));
            return false;
      }
 
      string::size_type colon = peer_address().find(':');
      if (colon == std::string::npos || colon == 0) {
         fc_elog( logger, "Invalid peer address. must be \"host:port[:<blk>|<trx>]\": ${p}", ("p", peer_address()) );
         return false;
      }
 
      connection_ptr c = shared_from_this();
 
      if( consecutive_immediate_connection_close > def_max_consecutive_immediate_connection_close || no_retry == benign_other ) {
         auto connector_period_us = std::chrono::duration_cast<std::chrono::microseconds>( my_impl->connector_period );
         std::lock_guard<std::mutex> g( c->conn_mtx );
         if( last_close == fc::time_point() || last_close > fc::time_point::now() - fc::microseconds( connector_period_us.count() ) ) {
            return true; // true so doesn't remove from valid connections
         }
      }
 
      strand.post([c]() {
         string::size_type colon = c->peer_address().find(':');
         string::size_type colon2 = c->peer_address().find(':', colon + 1);
         string host = c->peer_address().substr( 0, colon );
         string port = c->peer_address().substr( colon + 1, colon2 == string::npos ? string::npos : colon2 - (colon + 1));
         c->set_connection_type( c->peer_address() );
 
         auto resolver = std::make_shared<tcp::resolver>( my_impl->thread_pool->get_executor() );
         connection_wptr weak_conn = c;
         // Note: need to add support for IPv6 too
         resolver->async_resolve( tcp::v4(), host, port, boost::asio::bind_executor( c->strand,
            [resolver, weak_conn, host, port]( const boost::system::error_code& err, tcp::resolver::results_type endpoints ) {
               auto c = weak_conn.lock();
               if( !c ) return;
               if( !err ) {
                  c->connect( resolver, endpoints );
               } else {
                  fc_elog( logger, "Unable to resolve ${host}:${port} ${error}",
                           ("host", host)("port", port)( "error", err.message() ) );
                  c->connecting = false;
                  ++c->consecutive_immediate_connection_close;
               }
         } ) );
      } );
      return true;
   }

Here is the call graph for this function:

send_handshake()

void sysio::connection::send_handshake

(

)

Definition at line 1137 of file net_plugin.cpp.

                                   {
      strand.post( [c = shared_from_this()]() {
         std::unique_lock<std::mutex> g_conn( c->conn_mtx );
         if( c->populate_handshake( c->last_handshake_sent ) ) {
            static_assert( std::is_same_v<decltype( c->sent_handshake_count ), int16_t>, "INT16_MAX based on int16_t" );
            if( c->sent_handshake_count == INT16_MAX ) c->sent_handshake_count = 1; // do not wrap
            c->last_handshake_sent.generation = ++c->sent_handshake_count;
            auto last_handshake_sent = c->last_handshake_sent;
            g_conn.unlock();
            peer_ilog( c, "Sending handshake generation ${g}, lib ${lib}, head ${head}, id ${id}",
                       ("g", last_handshake_sent.generation)
                       ("lib", last_handshake_sent.last_irreversible_block_num)
                       ("head", last_handshake_sent.head_num)("id", last_handshake_sent.head_id.str().substr(8,16)) );
            c->enqueue( last_handshake_sent );
         }
      });
   }

Here is the call graph for this function:

Here is the caller graph for this function:

send_time() [1/2]

void sysio::connection::send_time

(

)

Definition at line 1181 of file net_plugin.cpp.

                              {
      time_message xpkt;
      xpkt.org = rec;
      xpkt.rec = dst;
      xpkt.xmt = get_time();
      org = xpkt.xmt;
      enqueue(xpkt);
   }

Here is the call graph for this function:

Here is the caller graph for this function:

send_time() [2/2]

void sysio::connection::send_time

(

const time_message &

msg

)

Definition at line 1191 of file net_plugin.cpp.

                                                     {
      time_message xpkt;
      xpkt.org = msg.xmt;
      xpkt.rec = msg.dst;
      xpkt.xmt = get_time();
      enqueue(xpkt);
   }

Here is the call graph for this function:

set_connection_type()

void sysio::connection::set_connection_type

(

const string &

peer_addr

)

Definition at line 918 of file net_plugin.cpp.

                                                                {
      // host:port:[<trx>|<blk>]
      string::size_type colon = peer_add.find(':');
      string::size_type colon2 = peer_add.find(':', colon + 1);
      string::size_type end = colon2 == string::npos
            ? string::npos : peer_add.find_first_of( " :+=.,<>!$%^&(*)|-#@\t", colon2 + 1 ); // future proof by including most symbols without using regex
      string host = peer_add.substr( 0, colon );
      string port = peer_add.substr( colon + 1, colon2 == string::npos ? string::npos : colon2 - (colon + 1));
      string type = colon2 == string::npos ? "" : end == string::npos ?
            peer_add.substr( colon2 + 1 ) : peer_add.substr( colon2 + 1, end - (colon2 + 1) );
 
      if( type.empty() ) {
         fc_dlog( logger, "Setting connection ${c} type for: ${peer} to both transactions and blocks", ("c", connection_id)("peer", peer_add) );
         connection_type = both;
      } else if( type == "trx" ) {
         fc_dlog( logger, "Setting connection ${c} type for: ${peer} to transactions only", ("c", connection_id)("peer", peer_add) );
         connection_type = transactions_only;
      } else if( type == "blk" ) {
         fc_dlog( logger, "Setting connection ${c} type for: ${peer} to blocks only", ("c", connection_id)("peer", peer_add) );
         connection_type = blocks_only;
      } else {
         fc_wlog( logger, "Unknown connection ${c} type: ${t}, for ${peer}", ("c", connection_id)("t", type)("peer", peer_add) );
      }
   }

Here is the caller graph for this function:

set_heartbeat_timeout()

void sysio::connection::set_heartbeat_timeout ( std::chrono::milliseconds msec )

inline

Definition at line 596 of file net_plugin.cpp.

                                                             {
         std::chrono::system_clock::duration dur = msec;
         hb_timeout = dur.count();
      }

socket_is_open()

bool sysio::connection::socket_is_open ( ) const

inline

Definition at line 590 of file net_plugin.cpp.

{ return socket_open.load(); } // thread safe, atomic

Here is the caller graph for this function:

start_read_message()

void sysio::connection::start_read_message

(

)

Definition at line 2432 of file net_plugin.cpp.

                                       {
      try {
         std::size_t minimum_read =
               std::atomic_exchange<decltype(outstanding_read_bytes.load())>( &outstanding_read_bytes, 0 );
         minimum_read = minimum_read != 0 ? minimum_read : message_header_size;
 
         if (my_impl->use_socket_read_watermark) {
            const size_t max_socket_read_watermark = 4096;
            std::size_t socket_read_watermark = std::min<std::size_t>(minimum_read, max_socket_read_watermark);
            boost::asio::socket_base::receive_low_watermark read_watermark_opt(socket_read_watermark);
            boost::system::error_code ec;
            socket->set_option( read_watermark_opt, ec );
            if( ec ) {
               peer_elog( this, "unable to set read watermark: ${e1}", ("e1", ec.message()) );
            }
         }
 
         auto completion_handler = [minimum_read](boost::system::error_code ec, std::size_t bytes_transferred) -> std::size_t {
            if (ec || bytes_transferred >= minimum_read ) {
               return 0;
            } else {
               return minimum_read - bytes_transferred;
            }
         };
 
         uint32_t write_queue_size = buffer_queue.write_queue_size();
         if( write_queue_size > def_max_write_queue_size ) {
            peer_elog( this, "write queue full ${s} bytes, giving up on connection, closing", ("s", write_queue_size) );
            close( false );
            return;
         }
 
         boost::asio::async_read( *socket,
            pending_message_buffer.get_buffer_sequence_for_boost_async_read(), completion_handler,
            boost::asio::bind_executor( strand,
              [conn = shared_from_this(), socket=socket]( boost::system::error_code ec, std::size_t bytes_transferred ) {
               // may have closed connection and cleared pending_message_buffer
               if( !conn->socket_is_open() || socket != conn->socket ) return;
 
               bool close_connection = false;
               try {
                  if( !ec ) {
                     if (bytes_transferred > conn->pending_message_buffer.bytes_to_write()) {
                        peer_elog( conn, "async_read_some callback: bytes_transfered = ${bt}, buffer.bytes_to_write = ${btw}",
                                   ("bt",bytes_transferred)("btw",conn->pending_message_buffer.bytes_to_write()) );
                     }
                     SYS_ASSERT(bytes_transferred <= conn->pending_message_buffer.bytes_to_write(), plugin_exception, "");
                     conn->pending_message_buffer.advance_write_ptr(bytes_transferred);
                     while (conn->pending_message_buffer.bytes_to_read() > 0) {
                        uint32_t bytes_in_buffer = conn->pending_message_buffer.bytes_to_read();
 
                        if (bytes_in_buffer < message_header_size) {
                           conn->outstanding_read_bytes = message_header_size - bytes_in_buffer;
                           break;
                        } else {
                           uint32_t message_length;
                           auto index = conn->pending_message_buffer.read_index();
                           conn->pending_message_buffer.peek(&message_length, sizeof(message_length), index);
                           if(message_length > def_send_buffer_size*2 || message_length == 0) {
                              peer_elog( conn, "incoming message length unexpected (${i})", ("i", message_length) );
                              close_connection = true;
                              break;
                           }
 
                           auto total_message_bytes = message_length + message_header_size;
 
                           if (bytes_in_buffer >= total_message_bytes) {
                              conn->pending_message_buffer.advance_read_ptr(message_header_size);
                              conn->consecutive_immediate_connection_close = 0;
                              if (!conn->process_next_message(message_length)) {
                                 return;
                              }
                           } else {
                              auto outstanding_message_bytes = total_message_bytes - bytes_in_buffer;
                              auto available_buffer_bytes = conn->pending_message_buffer.bytes_to_write();
                              if (outstanding_message_bytes > available_buffer_bytes) {
                                 conn->pending_message_buffer.add_space( outstanding_message_bytes - available_buffer_bytes );
                              }
 
                              conn->outstanding_read_bytes = outstanding_message_bytes;
                              break;
                           }
                        }
                     }
                     if( !close_connection ) conn->start_read_message();
                  } else {
                     if (ec.value() != boost::asio::error::eof) {
                        peer_elog( conn, "Error reading message: ${m}", ( "m", ec.message() ) );
                     } else {
                        peer_ilog( conn, "Peer closed connection" );
                     }
                     close_connection = true;
                  }
               } 
               catch ( const std::bad_alloc& ) 
               {
                 throw;
               } 
               catch ( const boost::interprocess::bad_alloc& ) 
               {
                 throw;
               } 
               catch(const fc::exception &ex) 
               {
                  peer_elog( conn, "Exception in handling read data ${s}", ("s",ex.to_string()) );
                  close_connection = true;
               }
               catch(const std::exception &ex) {
                  peer_elog( conn, "Exception in handling read data: ${s}", ("s",ex.what()) );
                  close_connection = true;
               }
               catch (...) {
                  peer_elog( conn, "Undefined exception handling read data" );
                  close_connection = true;
               }
 
               if( close_connection ) {
                  peer_elog( conn, "Closing connection" );
                  conn->close();
               }
         }));
      } catch (...) {
         peer_elog( this, "Undefined exception in start_read_message, closing connection" );
         close();
      }
   }

Here is the call graph for this function:

Here is the caller graph for this function:

start_session()

bool sysio::connection::start_session

(

)

Definition at line 954 of file net_plugin.cpp.

                                  {
      verify_strand_in_this_thread( strand, __func__, __LINE__ );
 
      update_endpoints();
      boost::asio::ip::tcp::no_delay nodelay( true );
      boost::system::error_code ec;
      socket->set_option( nodelay, ec );
      if( ec ) {
         peer_elog( this, "connection failed (set_option): ${e1}", ( "e1", ec.message() ) );
         close();
         return false;
      } else {
         peer_dlog( this, "connected" );
         socket_open = true;
         start_read_message();
         return true;
      }
   }

Here is the call graph for this function:

stop_send()

void sysio::connection::stop_send

(

)

Definition at line 1133 of file net_plugin.cpp.

                              {
      syncing = false;
   }

Here is the caller graph for this function:

sync_timeout()

void sysio::connection::sync_timeout

(

boost::system::error_code

ec

)

Definition at line 1485 of file net_plugin.cpp.

                                                           {
      if( !ec ) {
         my_impl->sync_master->sync_reassign_fetch( shared_from_this(), benign_other );
         close(true);
      } else if( ec != boost::asio::error::operation_aborted ) { // don't log on operation_aborted, called on destroy
         peer_elog( this, "setting timer for sync request got error ${ec}", ("ec", ec.message()) );
      }
   }

Here is the call graph for this function:

sync_wait()

void sysio::connection::sync_wait

(

)

Definition at line 1463 of file net_plugin.cpp.

                              {
      connection_ptr c(shared_from_this());
      std::lock_guard<std::mutex> g( response_expected_timer_mtx );
      response_expected_timer.expires_from_now( my_impl->resp_expected_period );
      response_expected_timer.async_wait(
            boost::asio::bind_executor( c->strand, [c]( boost::system::error_code ec ) {
               c->sync_timeout( ec );
            } ) );
   }

Member Data Documentation

block_status_monitor_

block_status_monitor sysio::connection::block_status_monitor_

Definition at line 645 of file net_plugin.cpp.

buffer_queue

queued_buffer sysio::connection::buffer_queue

Definition at line 626 of file net_plugin.cpp.

conn_mtx

std::mutex sysio::connection::conn_mtx

mutable

Definition at line 653 of file net_plugin.cpp.

conn_node_id

fc::sha256 sysio::connection::conn_node_id

Definition at line 628 of file net_plugin.cpp.

connecting

std::atomic<bool> sysio::connection::connecting {true}

Definition at line 640 of file net_plugin.cpp.

{true};

connection_id

const uint32_t sysio::connection::connection_id

Definition at line 638 of file net_plugin.cpp.

consecutive_immediate_connection_close

std::atomic<uint16_t> sysio::connection::consecutive_immediate_connection_close = 0

Definition at line 646 of file net_plugin.cpp.

dst

tstamp sysio::connection::dst {0}

Definition at line 671 of file net_plugin.cpp.

{0};          

fork_head

block_id_type sysio::connection::fork_head

Definition at line 657 of file net_plugin.cpp.

fork_head_num

uint32_t sysio::connection::fork_head_num {0}

Definition at line 658 of file net_plugin.cpp.

{0};

hb_timeout

tstamp sysio::connection::hb_timeout {std::chrono::milliseconds{def_keepalive_interval}.count()}

Definition at line 676 of file net_plugin.cpp.

{std::chrono::milliseconds{def_keepalive_interval}.count()};

last_close

fc::time_point sysio::connection::last_close

Definition at line 659 of file net_plugin.cpp.

last_dropped_trx_msg_time

fc::time_point sysio::connection::last_dropped_trx_msg_time

Definition at line 637 of file net_plugin.cpp.

last_handshake_recv

handshake_message sysio::connection::last_handshake_recv

Definition at line 655 of file net_plugin.cpp.

last_handshake_sent

handshake_message sysio::connection::last_handshake_sent

Definition at line 656 of file net_plugin.cpp.

last_req

std::optional<request_message> sysio::connection::last_req

Definition at line 654 of file net_plugin.cpp.

latest_blk_time

tstamp sysio::connection::latest_blk_time {0}

Definition at line 677 of file net_plugin.cpp.

{0};

latest_msg_time

tstamp sysio::connection::latest_msg_time {0}

Definition at line 675 of file net_plugin.cpp.

{0};

local_endpoint_ip

string sysio::connection::local_endpoint_ip

Definition at line 633 of file net_plugin.cpp.

local_endpoint_port

string sysio::connection::local_endpoint_port

Definition at line 634 of file net_plugin.cpp.

log_p2p_address

string sysio::connection::log_p2p_address

Definition at line 630 of file net_plugin.cpp.

log_remote_endpoint_ip

string sysio::connection::log_remote_endpoint_ip

Definition at line 631 of file net_plugin.cpp.

log_remote_endpoint_port

string sysio::connection::log_remote_endpoint_port

Definition at line 632 of file net_plugin.cpp.

net_version

uint16_t sysio::connection::net_version = net_version_max

Definition at line 644 of file net_plugin.cpp.

no_retry

std::atomic<go_away_reason> sysio::connection::no_retry {no_reason}

Definition at line 651 of file net_plugin.cpp.

{no_reason};

org

tstamp sysio::connection::org {0}

Definition at line 669 of file net_plugin.cpp.

{0};          

outstanding_read_bytes

std::atomic<std::size_t> sysio::connection::outstanding_read_bytes {0}

Definition at line 624 of file net_plugin.cpp.

{0}; // accessed only from strand threads

pending_message_buffer

fc::message_buffer<1024*1024> sysio::connection::pending_message_buffer

Definition at line 623 of file net_plugin.cpp.

protocol_version

std::atomic<uint16_t> sysio::connection::protocol_version = 0

Definition at line 643 of file net_plugin.cpp.

rec

tstamp sysio::connection::rec {0}

Definition at line 670 of file net_plugin.cpp.

{0};          

remote_endpoint_ip

string sysio::connection::remote_endpoint_ip

Definition at line 660 of file net_plugin.cpp.

response_expected_timer

boost::asio::steady_timer sysio::connection::response_expected_timer

Definition at line 649 of file net_plugin.cpp.

response_expected_timer_mtx

std::mutex sysio::connection::response_expected_timer_mtx

Definition at line 648 of file net_plugin.cpp.

sent_handshake_count

int16_t sysio::connection::sent_handshake_count = 0

Definition at line 639 of file net_plugin.cpp.

short_conn_node_id

string sysio::connection::short_conn_node_id

Definition at line 629 of file net_plugin.cpp.

socket

std::shared_ptr<tcp::socket> sysio::connection::socket

Definition at line 621 of file net_plugin.cpp.

strand

boost::asio::io_context::strand sysio::connection::strand

Definition at line 620 of file net_plugin.cpp.

syncing

std::atomic<bool> sysio::connection::syncing {false}

Definition at line 641 of file net_plugin.cpp.

{false};

trx_in_progress_size

std::atomic<uint32_t> sysio::connection::trx_in_progress_size {0}

Definition at line 636 of file net_plugin.cpp.

{0};

xmt

tstamp sysio::connection::xmt {0}

Definition at line 672 of file net_plugin.cpp.

{0};          

---

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

• plugins/net_plugin/net_plugin.cpp