sysio::sync_manager Class Reference

Public Member Functions
	sync_manager (uint32_t span)
bool	syncing_with_peer () const
void	sync_reset_lib_num (const connection_ptr &conn, bool closing)
void	sync_reassign_fetch (const connection_ptr &c, go_away_reason reason)
void	rejected_block (const connection_ptr &c, uint32_t blk_num)
void	sync_recv_block (const connection_ptr &c, const block_id_type &blk_id, uint32_t blk_num, bool blk_applied)
void	sync_update_expected (const connection_ptr &c, const block_id_type &blk_id, uint32_t blk_num, bool blk_applied)
void	recv_handshake (const connection_ptr &c, const handshake_message &msg)
void	sync_recv_notice (const connection_ptr &c, const notice_message &msg)
std::unique_lock< std::mutex >	locked_sync_mutex ()
void	reset_last_requested_num (const std::unique_lock< std::mutex > &lock)

Static Public Member Functions
static void	send_handshakes ()

Detailed Description

Definition at line 119 of file net_plugin.cpp.

Constructor & Destructor Documentation

sync_manager()

sysio::sync_manager::sync_manager ( uint32_t span )

explicit

Definition at line 1537 of file net_plugin.cpp.

      :sync_known_lib_num( 0 )
      ,sync_last_requested_num( 0 )
      ,sync_next_expected_num( 1 )
      ,sync_req_span( req_span )
      ,sync_source()
      ,sync_state(in_sync)
   {
   }

Member Function Documentation

locked_sync_mutex()

std::unique_lock< std::mutex > sysio::sync_manager::locked_sync_mutex ( )

inline

Definition at line 157 of file net_plugin.cpp.

                                                          {
         return std::unique_lock<std::mutex>(sync_mtx);
      }

recv_handshake()

void sysio::sync_manager::recv_handshake	(	const connection_ptr &	c,
		const handshake_message &	msg )

Definition at line 1774 of file net_plugin.cpp.

                                                                                            {
 
      if( c->is_transactions_only_connection() ) return;
 
      uint32_t lib_num = 0;
      uint32_t peer_lib = msg.last_irreversible_block_num;
      uint32_t head = 0;
      block_id_type head_id;
      std::tie( lib_num, std::ignore, head,
                std::ignore, std::ignore, head_id ) = my_impl->get_chain_info();
 
      sync_reset_lib_num(c, false);
 
      auto current_time_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
      int64_t network_latency_ns = current_time_ns - msg.time; // net latency in nanoseconds
      if( network_latency_ns < 0 ) {
         peer_wlog(c, "Peer sent a handshake with a timestamp skewed by at least ${t}ms", ("t", network_latency_ns/1000000));
         network_latency_ns = 0;
      }
      // number of blocks syncing node is behind from a peer node
      uint32_t nblk_behind_by_net_latency = static_cast<uint32_t>(network_latency_ns / block_interval_ns);
      // 2x for time it takes for message to reach back to peer node, +1 to compensate for integer division truncation
      uint32_t nblk_combined_latency = 2 * nblk_behind_by_net_latency + 1;
      // message in the log below is used in p2p_high_latency_test.py test
      peer_dlog(c, "Network latency is ${lat}ms, ${num} blocks discrepancy by network latency, ${tot_num} blocks discrepancy expected once message received",
                ("lat", network_latency_ns/1000000)("num", nblk_behind_by_net_latency)("tot_num", nblk_combined_latency));
 
      //--------------------------------
      // sync need checks; (lib == last irreversible block)
      //
      // 0. my head block id == peer head id means we are all caught up block wise
      // 1. my head block num < peer lib - start sync locally
      // 2. my lib > peer head num + nblk_combined_latency - send last_irr_catch_up notice if not the first generation
      //
      // 3  my head block num + nblk_combined_latency < peer head block num - update sync state and send a catchup request
      // 4  my head block num >= peer block num + nblk_combined_latency send a notice catchup if this is not the first generation
      //    4.1 if peer appears to be on a different fork ( our_id_for( msg.head_num ) != msg.head_id )
      //        then request peer's blocks
      //
      //-----------------------------
 
      if (head_id == msg.head_id) {
         peer_ilog( c, "handshake lib ${lib}, head ${head}, head id ${id}.. sync 0",
                    ("lib", msg.last_irreversible_block_num)("head", msg.head_num)("id", msg.head_id.str().substr(8,16)) );
         c->syncing = false;
         notice_message note;
         note.known_blocks.mode = none;
         note.known_trx.mode = catch_up;
         note.known_trx.pending = 0;
         c->enqueue( note );
         return;
      }
      if (head < peer_lib) {
         peer_ilog( c, "handshake lib ${lib}, head ${head}, head id ${id}.. sync 1",
                    ("lib", msg.last_irreversible_block_num)("head", msg.head_num)("id", msg.head_id.str().substr(8,16)) );
         c->syncing = false;
         if (c->sent_handshake_count > 0) {
            c->send_handshake();
         }
         return;
      }
      if (lib_num > msg.head_num + nblk_combined_latency) {
         peer_ilog( c, "handshake lib ${lib}, head ${head}, head id ${id}.. sync 2",
                    ("lib", msg.last_irreversible_block_num)("head", msg.head_num)("id", msg.head_id.str().substr(8,16)) );
         if (msg.generation > 1 || c->protocol_version > proto_base) {
            notice_message note;
            note.known_trx.pending = lib_num;
            note.known_trx.mode = last_irr_catch_up;
            note.known_blocks.mode = last_irr_catch_up;
            note.known_blocks.pending = head;
            c->enqueue( note );
         }
         c->syncing = true;
         return;
      }
 
      if (head + nblk_combined_latency < msg.head_num ) {
         peer_ilog( c, "handshake lib ${lib}, head ${head}, head id ${id}.. sync 3",
                    ("lib", msg.last_irreversible_block_num)("head", msg.head_num)("id", msg.head_id.str().substr(8,16)) );
         c->syncing = false;
         verify_catchup(c, msg.head_num, msg.head_id);
         return;
      } else if(head >= msg.head_num + nblk_combined_latency) {
         peer_ilog( c, "handshake lib ${lib}, head ${head}, head id ${id}.. sync 4",
                  ("lib", msg.last_irreversible_block_num)("head", msg.head_num)("id", msg.head_id.str().substr(8,16)) );
         if (msg.generation > 1 ||  c->protocol_version > proto_base) {
            notice_message note;
            note.known_trx.mode = none;
            note.known_blocks.mode = catch_up;
            note.known_blocks.pending = head;
            note.known_blocks.ids.push_back(head_id);
            c->enqueue( note );
         }
         c->syncing = false;
         app().post( priority::medium, [chain_plug = my_impl->chain_plug, c,
                                        msg_head_num = msg.head_num, msg_head_id = msg.head_id]() {
            bool on_fork = true;
            try {
               controller& cc = chain_plug->chain();
               on_fork = cc.get_block_id_for_num( msg_head_num ) != msg_head_id;
            } catch( ... ) {}
            if( on_fork ) {
               c->strand.post( [c]() {
                  request_message req;
                  req.req_blocks.mode = catch_up;
                  req.req_trx.mode = none;
                  c->enqueue( req );
               } );
            }
         } );
         return;
      } else {
         peer_dlog( c, "Block discrepancy is within network latency range.");
      }
   }

Here is the call graph for this function:

rejected_block()

void sysio::sync_manager::rejected_block	(	const connection_ptr &	c,
		uint32_t	blk_num )

Definition at line 1966 of file net_plugin.cpp.

                                                                                {
      c->block_status_monitor_.rejected();
      std::unique_lock<std::mutex> g( sync_mtx );
      reset_last_requested_num(g);
      if( c->block_status_monitor_.max_events_violated()) {
         peer_wlog( c, "block ${bn} not accepted, closing connection", ("bn", blk_num) );
         sync_source.reset();
         g.unlock();
         c->close();
      } else {
         g.unlock();
         c->send_handshake();
      }
   }

Here is the call graph for this function:

reset_last_requested_num()

void sysio::sync_manager::reset_last_requested_num ( const std::unique_lock< std::mutex > & lock )

inline

Definition at line 160 of file net_plugin.cpp.

                                                                                 {
         sync_last_requested_num = 0;
      }

Here is the caller graph for this function:

send_handshakes()

void sysio::sync_manager::send_handshakes ( )

static

Definition at line 1711 of file net_plugin.cpp.

                                      {
      for_each_connection( []( auto& ci ) {
         if( ci->current() ) {
            ci->send_handshake();
         }
         return true;
      } );
   }

Here is the call graph for this function:

Here is the caller graph for this function:

sync_reassign_fetch()

void sysio::sync_manager::sync_reassign_fetch	(	const connection_ptr &	c,
		go_away_reason	reason )

Definition at line 1761 of file net_plugin.cpp.

                                                                                        {
      std::unique_lock<std::mutex> g( sync_mtx );
      peer_ilog( c, "reassign_fetch, our last req is ${cc}, next expected is ${ne}",
               ("cc", sync_last_requested_num)("ne", sync_next_expected_num) );
 
      if( c == sync_source ) {
         c->cancel_sync(reason);
         reset_last_requested_num(g);
         request_next_chunk( std::move(g) );
      }
   }

Here is the call graph for this function:

sync_recv_block()

void sysio::sync_manager::sync_recv_block	(	const connection_ptr &	c,
		const block_id_type &	blk_id,
		uint32_t	blk_num,
		bool	blk_applied )

Definition at line 1998 of file net_plugin.cpp.

                                                                                                                              {
      peer_dlog( c, "got block ${bn}", ("bn", blk_num) );
      if( app().is_quiting() ) {
         c->close( false, true );
         return;
      }
      c->block_status_monitor_.accepted();
      sync_update_expected( c, blk_id, blk_num, blk_applied );
      std::unique_lock<std::mutex> g_sync( sync_mtx );
      stages state = sync_state;
      peer_dlog( c, "state ${s}", ("s", stage_str( state )) );
      if( state == head_catchup ) {
         peer_dlog( c, "sync_manager in head_catchup state" );
         sync_source.reset();
         g_sync.unlock();
 
         block_id_type null_id;
         bool set_state_to_head_catchup = false;
         for_each_block_connection( [&null_id, blk_num, &blk_id, &c, &set_state_to_head_catchup]( const auto& cp ) {
            std::unique_lock<std::mutex> g_cp_conn( cp->conn_mtx );
            uint32_t fork_head_num = cp->fork_head_num;
            block_id_type fork_head_id = cp->fork_head;
            g_cp_conn.unlock();
            if( fork_head_id == null_id ) {
               // continue
            } else if( fork_head_num < blk_num || fork_head_id == blk_id ) {
               std::lock_guard<std::mutex> g_conn( c->conn_mtx );
               c->fork_head = null_id;
               c->fork_head_num = 0;
            } else {
               set_state_to_head_catchup = true;
            }
            return true;
         } );
 
         if( set_state_to_head_catchup ) {
            if( set_state( head_catchup ) ) {
                send_handshakes();
            }
         } else {
            set_state( in_sync );
            send_handshakes();
         }
      } else if( state == lib_catchup ) {
         if( blk_num >= sync_known_lib_num ) {
            peer_dlog( c, "All caught up with last known last irreversible block resending handshake" );
            set_state( in_sync );
            g_sync.unlock();
            send_handshakes();
         } else if( blk_num >= sync_last_requested_num ) {
            request_next_chunk( std::move( g_sync) );
         } else {
            g_sync.unlock();
            peer_dlog( c, "calling sync_wait" );
            c->sync_wait();
         }
      }
   }

Here is the call graph for this function:

sync_recv_notice()

void sysio::sync_manager::sync_recv_notice	(	const connection_ptr &	c,
		const notice_message &	msg )

Definition at line 1937 of file net_plugin.cpp.

                                                                                          {
      peer_dlog( c, "sync_manager got ${m} block notice", ("m", modes_str( msg.known_blocks.mode )) );
      SYS_ASSERT( msg.known_blocks.mode == catch_up || msg.known_blocks.mode == last_irr_catch_up, plugin_exception,
                  "sync_recv_notice only called on catch_up" );
      if (msg.known_blocks.mode == catch_up) {
         if (msg.known_blocks.ids.size() == 0) {
            peer_elog( c, "got a catch up with ids size = 0" );
         } else {
            const block_id_type& id = msg.known_blocks.ids.back();
            peer_ilog( c, "notice_message, pending ${p}, blk_num ${n}, id ${id}...",
                     ("p", msg.known_blocks.pending)("n", block_header::num_from_id(id))("id",id.str().substr(8,16)) );
            if( !my_impl->dispatcher->have_block( id ) ) {
               verify_catchup( c, msg.known_blocks.pending, id );
            } else {
               // we already have the block, so update peer with our view of the world
               c->send_handshake();
            }
         }
      } else if (msg.known_blocks.mode == last_irr_catch_up) {
         {
            std::lock_guard<std::mutex> g_conn( c->conn_mtx );
            c->last_handshake_recv.last_irreversible_block_num = msg.known_trx.pending;
         }
         sync_reset_lib_num(c, false);
         start_sync(c, msg.known_trx.pending);
      }
   }

Here is the call graph for this function:

sync_reset_lib_num()

void sysio::sync_manager::sync_reset_lib_num	(	const connection_ptr &	conn,
		bool	closing )

Definition at line 1566 of file net_plugin.cpp.

                                                                              {
      std::unique_lock<std::mutex> g( sync_mtx );
      if( sync_state == in_sync ) {
         sync_source.reset();
      }
      if( !c ) return;
      if( !closing ) {
         std::lock_guard<std::mutex> g_conn( c->conn_mtx );
         if( c->last_handshake_recv.last_irreversible_block_num > sync_known_lib_num ) {
            sync_known_lib_num = c->last_handshake_recv.last_irreversible_block_num;
         }
      } else {
         // Closing connection, therefore its view of LIB can no longer be considered as we will no longer be connected.
         // Determine current LIB of remaining peers as our sync_known_lib_num.
         uint32_t highest_lib_num = 0;
         for_each_block_connection( [&highest_lib_num]( const auto& cc ) {
            std::lock_guard<std::mutex> g_conn( cc->conn_mtx );
            if( cc->current() && cc->last_handshake_recv.last_irreversible_block_num > highest_lib_num ) {
               highest_lib_num = cc->last_handshake_recv.last_irreversible_block_num;
            }
            return true;
         } );
         sync_known_lib_num = highest_lib_num;
 
         // if closing the connection we are currently syncing from, then reset our last requested and next expected.
         if( c == sync_source ) {
            reset_last_requested_num(g);
            // if starting to sync need to always start from lib as we might be on our own fork
            uint32_t lib_num = 0;
            std::tie( lib_num, std::ignore, std::ignore, std::ignore, std::ignore, std::ignore ) = my_impl->get_chain_info();
            sync_next_expected_num = lib_num + 1;
            request_next_chunk( std::move(g) );
         }
      }
   }

Here is the call graph for this function:

Here is the caller graph for this function:

sync_update_expected()

void sysio::sync_manager::sync_update_expected	(	const connection_ptr &	c,
		const block_id_type &	blk_id,
		uint32_t	blk_num,
		bool	blk_applied )

Definition at line 1982 of file net_plugin.cpp.

                                                                                                                                     {
      std::unique_lock<std::mutex> g_sync( sync_mtx );
      if( blk_num <= sync_last_requested_num ) {
         peer_dlog( c, "sync_last_requested_num: ${r}, sync_next_expected_num: ${e}, sync_known_lib_num: ${k}, sync_req_span: ${s}",
                  ("r", sync_last_requested_num)("e", sync_next_expected_num)("k", sync_known_lib_num)("s", sync_req_span) );
         if (blk_num != sync_next_expected_num && !blk_applied) {
            auto sync_next_expected = sync_next_expected_num;
            g_sync.unlock();
            peer_dlog( c, "expected block ${ne} but got ${bn}", ("ne", sync_next_expected)("bn", blk_num) );
            return;
         }
         sync_next_expected_num = blk_num + 1;
      }
   }

Here is the caller graph for this function:

syncing_with_peer()

bool sysio::sync_manager::syncing_with_peer ( ) const

inline

Definition at line 149 of file net_plugin.cpp.

{ return sync_state == lib_catchup; }

---

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

• plugins/net_plugin/net_plugin.cpp