sysio::dispatch_manager Class Reference

Public Member Functions
	dispatch_manager (boost::asio::io_context &io_context)
void	bcast_transaction (const packed_transaction_ptr &trx)
void	rejected_transaction (const packed_transaction_ptr &trx, uint32_t head_blk_num)
void	bcast_block (const signed_block_ptr &b, const block_id_type &id)
void	rejected_block (const block_id_type &id)
void	recv_block (const connection_ptr &conn, const block_id_type &msg, uint32_t bnum)
void	expire_blocks (uint32_t bnum)
void	recv_notice (const connection_ptr &conn, const notice_message &msg, bool generated)
void	retry_fetch (const connection_ptr &conn)
bool	add_peer_block (const block_id_type &blkid, uint32_t connection_id)
bool	peer_has_block (const block_id_type &blkid, uint32_t connection_id) const
bool	have_block (const block_id_type &blkid) const
bool	add_peer_txn (const transaction_id_type id, const time_point_sec &trx_expires, uint32_t connection_id, const time_point_sec &now=time_point::now())
bool	have_txn (const transaction_id_type &tid) const
void	expire_txns ()

Public Attributes
boost::asio::io_context::strand	strand

Detailed Description

Definition at line 165 of file net_plugin.cpp.

Constructor & Destructor Documentation

dispatch_manager()

sysio::dispatch_manager::dispatch_manager ( boost::asio::io_context & io_context )

inlineexplicit

Definition at line 174 of file net_plugin.cpp.

      : strand( io_context ) {}

Member Function Documentation

add_peer_block()

bool sysio::dispatch_manager::add_peer_block	(	const block_id_type &	blkid,
		uint32_t	connection_id )

Definition at line 2060 of file net_plugin.cpp.

                                                                                            {
      std::lock_guard<std::mutex> g( blk_state_mtx );
      auto bptr = blk_state.get<by_id>().find( std::make_tuple( connection_id, std::ref( blkid )));
      bool added = (bptr == blk_state.end());
      if( added ) {
         blk_state.insert( {blkid, block_header::num_from_id( blkid ), connection_id, true} );
      } else if( !bptr->have_block ) {
         blk_state.modify( bptr, []( auto& pb ) {
            pb.have_block = true;
         });
      }
      return added;
   }

Here is the call graph for this function:

Here is the caller graph for this function:

add_peer_txn()

bool sysio::dispatch_manager::add_peer_txn	(	const transaction_id_type	id,
		const time_point_sec &	trx_expires,
		uint32_t	connection_id,
		const time_point_sec &	now = time_point::now() )

Definition at line 2091 of file net_plugin.cpp.

                                                                                            {
      std::lock_guard<std::mutex> g( local_txns_mtx );
      auto tptr = local_txns.get<by_id>().find( std::make_tuple( std::ref( id ), connection_id ) );
      bool added = (tptr == local_txns.end());
      if( added ) {
         // expire at either transaction expiration or configured max expire time whichever is less
         time_point_sec expires = now + my_impl->p2p_dedup_cache_expire_time_us;
         expires = std::min( trx_expires, expires );
         local_txns.insert( node_transaction_state{
            .id = id,
            .expires = expires,
            .connection_id = connection_id} );
      }
      return added;
   }

Here is the caller graph for this function:

bcast_block()

void sysio::dispatch_manager::bcast_block	(	const signed_block_ptr &	b,
		const block_id_type &	id )

Definition at line 2135 of file net_plugin.cpp.

                                                                                        {
      fc_dlog( logger, "bcast block ${b}", ("b", b->block_num()) );
 
      if( my_impl->sync_master->syncing_with_peer() ) return;
 
      block_buffer_factory buff_factory;
      const auto bnum = b->block_num();
      for_each_block_connection( [this, &id, &bnum, &b, &buff_factory]( auto& cp ) {
         fc_dlog( logger, "socket_is_open ${s}, connecting ${c}, syncing ${ss}, connection ${cid}",
                  ("s", cp->socket_is_open())("c", cp->connecting.load())("ss", cp->syncing.load())("cid", cp->connection_id) );
         if( !cp->current() ) return true;
         send_buffer_type sb = buff_factory.get_send_buffer( b );
 
         cp->strand.post( [this, cp, id, bnum, sb{std::move(sb)}]() {
            cp->latest_blk_time = cp->get_time();
            std::unique_lock<std::mutex> g_conn( cp->conn_mtx );
            bool has_block = cp->last_handshake_recv.last_irreversible_block_num >= bnum;
            g_conn.unlock();
            if( !has_block ) {
               if( !add_peer_block( id, cp->connection_id ) ) {
                  peer_dlog( cp, "not bcast block ${b}", ("b", bnum) );
                  return;
               }
               peer_dlog( cp, "bcast block ${b}", ("b", bnum) );
               cp->enqueue_buffer( sb, no_reason );
            }
         });
         return true;
      } );
   }

Here is the call graph for this function:

bcast_transaction()

void sysio::dispatch_manager::bcast_transaction

(

const packed_transaction_ptr &

trx

)

Definition at line 2187 of file net_plugin.cpp.

                                                                             {
      trx_buffer_factory buff_factory;
      const auto now = fc::time_point::now();
      for_each_connection( [this, &trx, &now, &buff_factory]( auto& cp ) {
         if( cp->is_blocks_only_connection() || !cp->current() ) {
            return true;
         }
         if( !add_peer_txn(trx->id(), trx->expiration(), cp->connection_id, now) ) {
            return true;
         }
 
         send_buffer_type sb = buff_factory.get_send_buffer( trx );
         fc_dlog( logger, "sending trx: ${id}, to connection ${cid}", ("id", trx->id())("cid", cp->connection_id) );
         cp->strand.post( [cp, sb{std::move(sb)}]() {
            cp->enqueue_buffer( sb, no_reason );
         } );
         return true;
      } );
   }

Here is the call graph for this function:

expire_blocks()

void sysio::dispatch_manager::expire_blocks

(

uint32_t

bnum

)

Definition at line 2128 of file net_plugin.cpp.

                                                          {
      std::lock_guard<std::mutex> g(blk_state_mtx);
      auto& stale_blk = blk_state.get<by_block_num>();
      stale_blk.erase( stale_blk.lower_bound(1), stale_blk.upper_bound(lib_num) );
   }

expire_txns()

void sysio::dispatch_manager::expire_txns

(

)

Definition at line 2114 of file net_plugin.cpp.

                                      {
      size_t start_size = 0, end_size = 0;
 
      std::unique_lock<std::mutex> g( local_txns_mtx );
      start_size = local_txns.size();
      auto& old = local_txns.get<by_expiry>();
      auto ex_lo = old.lower_bound( fc::time_point_sec( 0 ) );
      auto ex_up = old.upper_bound( time_point::now() );
      old.erase( ex_lo, ex_up );
      g.unlock();
 
      fc_dlog( logger, "expire_local_txns size ${s} removed ${r}", ("s", start_size)( "r", start_size - end_size ) );
   }

Here is the call graph for this function:

have_block()

bool sysio::dispatch_manager::have_block

(

const block_id_type &

blkid

)

const

Definition at line 2080 of file net_plugin.cpp.

                                                                       {
      std::lock_guard<std::mutex> g(blk_state_mtx);
      // by_peer_block_id sorts have_block by greater so have_block == true will be the first one found
      const auto& index = blk_state.get<by_peer_block_id>();
      auto blk_itr = index.find( blkid );
      if( blk_itr != index.end() ) {
         return blk_itr->have_block;
      }
      return false;
   }

have_txn()

bool sysio::dispatch_manager::have_txn

(

const transaction_id_type &

tid

)

const

Definition at line 2108 of file net_plugin.cpp.

                                                                         {
      std::lock_guard<std::mutex> g( local_txns_mtx );
      const auto tptr = local_txns.get<by_id>().find( tid );
      return tptr != local_txns.end();
   }

peer_has_block()

bool sysio::dispatch_manager::peer_has_block	(	const block_id_type &	blkid,
		uint32_t	connection_id ) const

Definition at line 2074 of file net_plugin.cpp.

                                                                                                   {
      std::lock_guard<std::mutex> g(blk_state_mtx);
      const auto blk_itr = blk_state.get<by_id>().find( std::make_tuple( connection_id, std::ref( blkid )));
      return blk_itr != blk_state.end();
   }

Here is the caller graph for this function:

recv_block()

void sysio::dispatch_manager::recv_block	(	const connection_ptr &	conn,
		const block_id_type &	msg,
		uint32_t	bnum )

Definition at line 2167 of file net_plugin.cpp.

                                                                                                    {
      std::unique_lock<std::mutex> g( c->conn_mtx );
      if (c &&
          c->last_req &&
          c->last_req->req_blocks.mode != none &&
          !c->last_req->req_blocks.ids.empty() &&
          c->last_req->req_blocks.ids.back() == id) {
         peer_dlog( c, "resetting last_req" );
         c->last_req.reset();
      }
      g.unlock();
 
      peer_dlog(c, "canceling wait");
      c->cancel_wait();
   }

recv_notice()

void sysio::dispatch_manager::recv_notice	(	const connection_ptr &	conn,
		const notice_message &	msg,
		bool	generated )

Definition at line 2213 of file net_plugin.cpp.

                                                                                                        {
      if (msg.known_trx.mode == normal) {
      } else if (msg.known_trx.mode != none) {
         peer_elog( c, "passed a notice_message with something other than a normal on none known_trx" );
         return;
      }
      if (msg.known_blocks.mode == normal) {
         // known_blocks.ids is never > 1
         if( !msg.known_blocks.ids.empty() ) {
            if( msg.known_blocks.pending == 1 ) { // block id notify of 2.0.0, ignore
               return;
            }
         }
      } else if (msg.known_blocks.mode != none) {
         peer_elog( c, "passed a notice_message with something other than a normal on none known_blocks" );
         return;
      }
   }

rejected_block()

void sysio::dispatch_manager::rejected_block

(

const block_id_type &

id

)

Definition at line 2183 of file net_plugin.cpp.

                                                                {
      fc_dlog( logger, "rejected block ${id}", ("id", id) );
   }

rejected_transaction()

void sysio::dispatch_manager::rejected_transaction	(	const packed_transaction_ptr &	trx,
		uint32_t	head_blk_num )

Definition at line 2207 of file net_plugin.cpp.

                                                                                                       {
      fc_dlog( logger, "not sending rejected transaction ${tid}", ("tid", trx->id()) );
      // keep rejected transaction around for awhile so we don't broadcast it, don't remove from local_txns
   }

retry_fetch()

void sysio::dispatch_manager::retry_fetch

(

const connection_ptr &

conn

)

Definition at line 2233 of file net_plugin.cpp.

                                                             {
      peer_dlog( c, "retry fetch" );
      request_message last_req;
      block_id_type bid;
      {
         std::lock_guard<std::mutex> g_c_conn( c->conn_mtx );
         if( !c->last_req ) {
            return;
         }
         peer_wlog( c, "failed to fetch from peer" );
         if( c->last_req->req_blocks.mode == normal && !c->last_req->req_blocks.ids.empty() ) {
            bid = c->last_req->req_blocks.ids.back();
         } else {
            peer_wlog( c, "no retry, block mpde = ${b} trx mode = ${t}",
                       ("b", modes_str( c->last_req->req_blocks.mode ))( "t", modes_str( c->last_req->req_trx.mode ) ) );
            return;
         }
         last_req = *c->last_req;
      }
      for_each_block_connection( [this, &c, &last_req, &bid]( auto& conn ) {
         if( conn == c )
            return true;
 
         {
            std::lock_guard<std::mutex> guard( conn->conn_mtx );
            if( conn->last_req ) {
               return true;
            }
         }
 
         bool sendit = peer_has_block( bid, conn->connection_id );
         if( sendit ) {
            conn->strand.post( [conn, last_req{std::move(last_req)}]() {
               conn->enqueue( last_req );
               conn->fetch_wait();
               std::lock_guard<std::mutex> g_conn_conn( conn->conn_mtx );
               conn->last_req = last_req;
            } );
            return false;
         }
         return true;
      } );
 
      // at this point no other peer has it, re-request or do nothing?
      peer_wlog( c, "no peer has last_req" );
      if( c->connected() ) {
         c->enqueue( last_req );
         c->fetch_wait();
      }
   }

Here is the call graph for this function:

Member Data Documentation

strand

boost::asio::io_context::strand sysio::dispatch_manager::strand

Definition at line 172 of file net_plugin.cpp.

---

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

• plugins/net_plugin/net_plugin.cpp