Merge pull request #619 from AntelopeIO/GH-529-optimize-block-processing

Optimize block processing - Part 1

libraries/chain/fork_database.cpp

@@ -94,7 +94,7 @@ namespace eosio::chain {
 
       void             open_impl( const char* desc, const std::filesystem::path& fork_db_file, fc::cfile_datastream& ds, validator_t& validator );
       void             close_impl( std::ofstream& out );
-      void             add_impl( const bsp_t& n, ignore_duplicate_t ignore_duplicate, bool validate, validator_t& validator );
+      bool             add_impl( const bsp_t& n, ignore_duplicate_t ignore_duplicate, bool validate, validator_t& validator );
       bool             is_valid() const;
 
       bsp_t            get_block_impl( const block_id_type& id, include_root_t include_root = include_root_t::no ) const;
@@ -240,7 +240,7 @@ namespace eosio::chain {
    }
 
    template <class BSP>
-   void fork_database_impl<BSP>::add_impl(const bsp_t& n, ignore_duplicate_t ignore_duplicate,
+   bool fork_database_impl<BSP>::add_impl(const bsp_t& n, ignore_duplicate_t ignore_duplicate,
                                           bool validate, validator_t& validator) {
       EOS_ASSERT( root, fork_database_exception, "root not yet set" );
       EOS_ASSERT( n, fork_database_exception, "attempt to add null block state" );
@@ -280,17 +280,17 @@ namespace eosio::chain {
       auto inserted = index.insert(n);
       EOS_ASSERT(ignore_duplicate == ignore_duplicate_t::yes || inserted.second, fork_database_exception,
                  "duplicate block added: ${id}", ("id", n->id()));
+
+      return inserted.second && n == head_impl(include_root_t::no);
    }
 
    template<class BSP>
-   void fork_database_t<BSP>::add( const bsp_t& n, ignore_duplicate_t ignore_duplicate ) {
+   bool fork_database_t<BSP>::add( const bsp_t& n, ignore_duplicate_t ignore_duplicate ) {
       std::lock_guard g( my->mtx );
-      my->add_impl( n, ignore_duplicate, false,
-                    []( block_timestamp_type timestamp,
-                        const flat_set<digest_type>& cur_features,
-                        const vector<digest_type>& new_features )
-                    {}
-      );
+      return my->add_impl(n, ignore_duplicate, false,
+                          [](block_timestamp_type         timestamp,
+                             const flat_set<digest_type>& cur_features,
+                             const vector<digest_type>&   new_features) {});
    }
 
    template<class BSP>

libraries/chain/include/eosio/chain/block_handle.hpp

@@ -6,7 +6,7 @@
 
 namespace eosio::chain {
 
-// Created via controller::create_block_handle(const block_id_type& id, const signed_block_ptr& b)
+// Created via controller::accept_block(const block_id_type& id, const signed_block_ptr& b)
 // Valid to request id and signed_block_ptr it was created from.
 struct block_handle {
 private:

libraries/chain/include/eosio/chain/controller.hpp

@@ -31,6 +31,46 @@ namespace eosio::vm { class wasm_allocator; }
 
 namespace eosio::chain {
 
+   struct speculative_block_metrics {
+      account_name block_producer{};
+      uint32_t     block_num             = 0;
+      int64_t      block_total_time_us   = 0;
+      int64_t      block_idle_us         = 0;
+      std::size_t  num_success_trx       = 0;
+      int64_t      success_trx_time_us   = 0;
+      std::size_t  num_fail_trx          = 0;
+      int64_t      fail_trx_time_us      = 0;
+      std::size_t  num_transient_trx     = 0;
+      int64_t      transient_trx_time_us = 0;
+      int64_t      block_other_time_us   = 0;
+   };
+
+   struct produced_block_metrics : public speculative_block_metrics {
+      std::size_t unapplied_transactions_total       = 0;
+      std::size_t subjective_bill_account_size_total = 0;
+      std::size_t scheduled_trxs_total               = 0;
+      std::size_t trxs_produced_total                = 0;
+      uint64_t    cpu_usage_us                       = 0;
+      int64_t     total_elapsed_time_us              = 0;
+      int64_t     total_time_us                      = 0;
+      uint64_t    net_usage_us                       = 0;
+
+      uint32_t last_irreversible = 0;
+      uint32_t head_block_num    = 0;
+   };
+
+   struct incoming_block_metrics {
+      std::size_t trxs_incoming_total   = 0;
+      uint64_t    cpu_usage_us          = 0;
+      int64_t     total_elapsed_time_us = 0;
+      int64_t     total_time_us         = 0;
+      uint64_t    net_usage_us          = 0;
+      int64_t     block_latency_us      = 0;
+
+      uint32_t last_irreversible = 0;
+      uint32_t head_block_num    = 0;
+   };
+
    using bls_pub_priv_key_map_t = std::map<std::string, std::string>;
    struct finalizer_policy;
 
@@ -182,41 +222,23 @@ namespace eosio::chain {
                                                            fc::time_point block_deadline, fc::microseconds max_transaction_time,
                                                            uint32_t billed_cpu_time_us, bool explicit_billed_cpu_time );
 
-         struct block_report {
-            size_t             total_net_usage = 0;
-            size_t             total_cpu_usage_us = 0;
-            fc::microseconds   total_elapsed_time{};
-            fc::microseconds   total_time{};
-         };
-
-         void assemble_and_complete_block( block_report& br, const signer_callback_type& signer_callback );
+         void assemble_and_complete_block( const signer_callback_type& signer_callback );
          void sign_block( const signer_callback_type& signer_callback );
-         void commit_block(block_report& br);
+         void commit_block();
          void testing_allow_voting(bool val);
          bool get_testing_allow_voting_flag();
          void set_async_voting(async_t val);
          void set_async_aggregation(async_t val);
-         void maybe_switch_forks(const forked_callback_t& cb, const trx_meta_cache_lookup& trx_lookup);
-
-         // thread-safe
-         std::future<block_handle> create_block_handle_future( const block_id_type& id, const signed_block_ptr& b );
-         // thread-safe
-         // returns empty optional if block b is not immediately ready to be processed
-         std::optional<block_handle> create_block_handle( const block_id_type& id, const signed_block_ptr& b ) const;
 
-         /**
-          * @param br returns statistics for block
-          * @param b block to push, created by create_block_handle
-          * @param cb calls cb with forked applied transactions for each forked block
-          * @param trx_lookup user provided lookup function for externally cached transaction_metadata
-          */
-         void push_block( block_report& br,
-                          const block_handle& b,
-                          const forked_callback_t& cb,
-                          const trx_meta_cache_lookup& trx_lookup );
+         /// Apply any blocks that are ready from the forkdb
+         void apply_blocks(const forked_callback_t& cb, const trx_meta_cache_lookup& trx_lookup);
 
-         /// Accept block into fork_database
-         void accept_block(const block_handle& b);
+         struct accepted_block_handle {
+            const bool is_new_best_head = false; // true if new best head
+            std::optional<block_handle> block;   // empty optional if block is unlinkable
+         };
+         // thread-safe
+         accepted_block_handle accept_block( const block_id_type& id, const signed_block_ptr& b ) const;
 
          boost::asio::io_context& get_thread_pool();
 
@@ -461,6 +483,9 @@ namespace eosio::chain {
       // is the bls key a registered finalizer key of this node, thread safe
       bool is_node_finalizer_key(const bls_public_key& key) const;
 
+      void register_update_produced_block_metrics(std::function<void(produced_block_metrics)>&&);
+      void register_update_speculative_block_metrics(std::function<void(speculative_block_metrics)>&&);
+      void register_update_incoming_block_metrics(std::function<void(incoming_block_metrics)>&&);
 
       private:
          const my_finalizers_t& get_node_finalizers() const;  // used for tests (purpose is inspecting fsi).

libraries/chain/include/eosio/chain/fork_database.hpp

@@ -67,8 +67,9 @@ namespace eosio::chain {
       /**
        *  Add block state to fork database.
        *  Must link to existing block in fork database or the root.
+       *  @return true if n becomes the new best head (and was not the best head before)
        */
-      void add( const bsp_t& next_block, ignore_duplicate_t ignore_duplicate );
+      bool add( const bsp_t& n, ignore_duplicate_t ignore_duplicate );
 
       void remove( const block_id_type& id );
 

libraries/testing/include/eosio/testing/tester.hpp

@@ -811,11 +811,10 @@ namespace eosio::testing {
       }
 
       void validate_push_block(const signed_block_ptr& sb) {
-         auto btf = validating_node->create_block_handle_future( sb->calculate_id(), sb );
-         controller::block_report br;
-         block_handle bh = btf.get();
-         validating_node->push_block( br, bh, {}, trx_meta_cache_lookup{} );
-         _check_for_vote_if_needed(*validating_node, bh);
+         auto [best_head, obh] = validating_node->accept_block( sb->calculate_id(), sb );
+         EOS_ASSERT(obh, unlinkable_block_exception, "block did not link ${b}", ("b", sb->calculate_id()));
+         validating_node->apply_blocks( {}, trx_meta_cache_lookup{} );
+         _check_for_vote_if_needed(*validating_node, *obh);
       }
 
       signed_block_ptr produce_empty_block( fc::microseconds skip_time = default_skip_time )override {

libraries/testing/tester.cpp

@@ -443,11 +443,11 @@ namespace eosio::testing {
 
    void base_tester::push_block(const signed_block_ptr& b) {
       auto block_id = b->calculate_id();
-      auto bhf = control->create_block_handle_future(block_id, b);
+      auto [best_fork, obh] = control->accept_block(block_id, b);
       unapplied_transactions.add_aborted( control->abort_block() );
-      controller::block_report br;
-      block_handle bh = bhf.get();
-      control->push_block( br, bh, [this]( const transaction_metadata_ptr& trx ) {
+      EOS_ASSERT(obh, unlinkable_block_exception, "block did not link ${b}", ("b", b->calculate_id()));
+      const block_handle& bh = *obh;
+      control->apply_blocks( [this]( const transaction_metadata_ptr& trx ) {
          unapplied_transactions.add_forked( trx );
       }, [this]( const transaction_id_type& id ) {
          return unapplied_transactions.get_trx( id );
@@ -564,8 +564,7 @@ namespace eosio::testing {
          }
       });
 
-      controller::block_report br;
-      control->assemble_and_complete_block( br, [&]( digest_type d ) {
+      control->assemble_and_complete_block( [&]( digest_type d ) {
          std::vector<signature_type> result;
          result.reserve(signing_keys.size());
          for (const auto& k: signing_keys)
@@ -574,7 +573,7 @@ namespace eosio::testing {
          return result;
       } );
 
-      control->commit_block(br);
+      control->commit_block();
 
       block_handle head = control->head();
 
@@ -1214,10 +1213,11 @@ namespace eosio::testing {
 
             auto block = a.control->fetch_block_by_number(i);
             if( block ) { //&& !b.control->is_known_block(block->id()) ) {
-               auto btf = b.control->create_block_handle_future( block->calculate_id(), block );
+               auto id = block->calculate_id();
+               auto [best_head, obh] = b.control->accept_block( id, block );
                b.control->abort_block();
-               controller::block_report br;
-               b.control->push_block(br, btf.get(), {}, trx_meta_cache_lookup{}); //, eosio::chain::validation_steps::created_block);
+               EOS_ASSERT(obh, unlinkable_block_exception, "block did not link ${b}", ("b", id));
+               b.control->apply_blocks({}, trx_meta_cache_lookup{});
             }
          }
       };

plugins/chain_interface/include/eosio/chain/plugin_interface.hpp

@@ -32,7 +32,7 @@ namespace eosio::chain::plugin_interface {
    namespace incoming {
       namespace methods {
          // synchronously push a block/trx to a single provider, block_state_legacy_ptr may be null
-         using block_sync            = method_decl<chain_plugin_interface, bool(const signed_block_ptr&, const block_id_type&, const std::optional<block_handle>&), first_provider_policy>;
+         using block_sync            = method_decl<chain_plugin_interface, bool(const signed_block_ptr&, const block_id_type&, const block_handle&), first_provider_policy>;
          using transaction_async     = method_decl<chain_plugin_interface, void(const packed_transaction_ptr&, bool, transaction_metadata::trx_type, bool, next_function<transaction_trace_ptr>), first_provider_policy>;
       }
    }

plugins/chain_plugin/chain_plugin.cpp

@@ -1208,8 +1208,8 @@ chain_apis::read_only chain_plugin::get_read_only_api(const fc::microseconds& ht
 }
 
 
-bool chain_plugin::accept_block(const signed_block_ptr& block, const block_id_type& id, const std::optional<block_handle>& obt ) {
-   return my->incoming_block_sync_method(block, id, obt);
+bool chain_plugin::accept_block(const signed_block_ptr& block, const block_id_type& id, const block_handle& bh ) {
+   return my->incoming_block_sync_method(block, id, bh);
 }
 
 void chain_plugin::accept_transaction(const chain::packed_transaction_ptr& trx, next_function<chain::transaction_trace_ptr> next) {
@@ -2116,7 +2116,10 @@ fc::variant read_only::get_block_header_state(const get_block_header_state_param
 void read_write::push_block(read_write::push_block_params&& params, next_function<read_write::push_block_results> next) {
    try {
       auto b = std::make_shared<signed_block>( std::move(params) );
-      app().get_method<incoming::methods::block_sync>()(b, b->calculate_id(), std::optional<block_handle>{});
+      block_id_type id = b->calculate_id();
+      auto [best_head, obh] = db.accept_block( id, b );
+      EOS_ASSERT(obh, unlinkable_block_exception, "block did not link ${b}", ("b", id));
+      app().get_method<incoming::methods::block_sync>()(b, id, *obh);
    } catch ( boost::interprocess::bad_alloc& ) {
       handle_db_exhaustion();
    } catch ( const std::bad_alloc& ) {

plugins/chain_plugin/include/eosio/chain_plugin/chain_plugin.hpp

@@ -998,7 +998,7 @@ class chain_plugin : public plugin<chain_plugin> {
    chain_apis::read_write get_read_write_api(const fc::microseconds& http_max_response_time);
    chain_apis::read_only get_read_only_api(const fc::microseconds& http_max_response_time) const;
 
-   bool accept_block( const chain::signed_block_ptr& block, const chain::block_id_type& id, const std::optional<chain::block_handle>& obt );
+   bool accept_block( const chain::signed_block_ptr& block, const chain::block_id_type& id, const chain::block_handle& bh );
    void accept_transaction(const chain::packed_transaction_ptr& trx, chain::plugin_interface::next_function<chain::transaction_trace_ptr> next);
 
    // Only call this after plugin_initialize()!