GH-1519 Remove hotstuff emit messages in controller. Simplify and opt…

…imize hotstuff message sending in net_plugin.

libraries/chain/controller.cpp

@@ -1974,22 +1974,6 @@ struct controller_impl {
       pending->push();
    }
 
-   void commit_hs_proposal_msg(hs_proposal_message_ptr msg){
-      emit( self.new_hs_proposal_message, msg );
-   }
-
-   void commit_hs_vote_msg(hs_vote_message_ptr msg){
-      emit( self.new_hs_vote_message, msg );
-   }
-
-   void commit_hs_new_view_msg(hs_new_view_message_ptr msg){
-      emit( self.new_hs_new_view_message, msg );
-   }
-
-   void commit_hs_new_block_msg(hs_new_block_message_ptr msg){
-      emit( self.new_hs_new_block_message, msg );
-   }
-
    /**
     *  This method is called from other threads. The controller_impl should outlive those threads.
     *  However, to avoid race conditions, it means that the behavior of this function should not change
@@ -2983,22 +2967,6 @@ void controller::commit_block() {
    my->commit_block(block_status::incomplete);
 }
 
-void controller::commit_hs_proposal_msg(hs_proposal_message_ptr msg) {
-   my->commit_hs_proposal_msg(msg);
-}
-
-void controller::commit_hs_vote_msg(hs_vote_message_ptr msg) {
-   my->commit_hs_vote_msg(msg);
-}
-
-void controller::commit_hs_new_view_msg(hs_new_view_message_ptr msg) {
-   my->commit_hs_new_view_msg(msg);
-}
-
-void controller::commit_hs_new_block_msg(hs_new_block_message_ptr msg) {
-   my->commit_hs_new_block_msg(msg);
-}
-
 deque<transaction_metadata_ptr> controller::abort_block() {
    return my->abort_block();
 }

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

@@ -21,15 +21,6 @@ namespace eosio { namespace vm { class wasm_allocator; }}
 
 namespace eosio { namespace chain {
 
-   struct hs_proposal_message;
-   struct hs_vote_message;
-   struct hs_new_view_message;
-   struct hs_new_block_message;
-   using hs_proposal_message_ptr = std::shared_ptr<hs_proposal_message>;
-   using hs_vote_message_ptr = std::shared_ptr<hs_vote_message>;
-   using hs_new_view_message_ptr = std::shared_ptr<hs_new_view_message>;
-   using hs_new_block_message_ptr = std::shared_ptr<hs_new_block_message>;
-
    class authorization_manager;
 
    namespace resource_limits {
@@ -176,12 +167,6 @@ namespace eosio { namespace chain {
          void sign_block( const signer_callback_type& signer_callback );
          void commit_block();
 
-         void commit_hs_proposal_msg(hs_proposal_message_ptr msg);
-         void commit_hs_vote_msg(hs_vote_message_ptr msg);
-
-         void commit_hs_new_view_msg(hs_new_view_message_ptr msg);
-         void commit_hs_new_block_msg(hs_new_block_message_ptr msg);
-
          // thread-safe
          std::future<block_state_ptr> create_block_state_future( const block_id_type& id, const signed_block_ptr& b );
          // thread-safe
@@ -352,20 +337,6 @@ namespace eosio { namespace chain {
          signal<void(const transaction_metadata_ptr&)> accepted_transaction;
          signal<void(std::tuple<const transaction_trace_ptr&, const packed_transaction_ptr&>)> applied_transaction;
          signal<void(const int&)>                      bad_alloc;
-         signal<void(const hs_proposal_message_ptr&)>    new_hs_proposal_message;
-         signal<void(const hs_vote_message_ptr&)> new_hs_vote_message;
-         signal<void(const hs_new_view_message_ptr&)>    new_hs_new_view_message;
-         signal<void(const hs_new_block_message_ptr&)> new_hs_new_block_message;
-
-         /*
-         signal<void()>                                  pre_apply_block;
-         signal<void()>                                  post_apply_block;
-         signal<void()>                                  abort_apply_block;
-         signal<void(const transaction_metadata_ptr&)>   pre_apply_transaction;
-         signal<void(const transaction_trace_ptr&)>      post_apply_transaction;
-         signal<void(const transaction_trace_ptr&)>  pre_apply_action;
-         signal<void(const transaction_trace_ptr&)>  post_apply_action;
-         */
 
          const apply_handler* find_apply_handler( account_name contract, scope_name scope, action_name act )const;
          wasm_interface_collection& get_wasm_interface();

libraries/hotstuff/chain_pacemaker.cpp

@@ -107,6 +107,23 @@ namespace eosio { namespace hotstuff {
    {
    }
 
+   void chain_pacemaker::register_bcast_functions(
+           std::function<void(const chain::hs_proposal_message&)> on_proposal_message,
+           std::function<void(const chain::hs_vote_message&)> on_vote_message,
+           std::function<void(const chain::hs_new_block_message&)> on_new_block_message,
+           std::function<void(const chain::hs_new_view_message&)> on_new_view_message
+   ) {
+      FC_ASSERT(on_proposal_message, "on_proposal_message must be provided");
+      FC_ASSERT(on_vote_message, "on_proposal_message must be provided");
+      FC_ASSERT(on_new_block_message, "on_proposal_message must be provided");
+      FC_ASSERT(on_new_view_message, "on_proposal_message must be provided");
+      std::lock_guard g( _hotstuff_global_mutex ); // not actually needed but doesn't hurt
+      bcast_proposal_message = std::move(on_proposal_message);
+      bcast_vote_message = std::move(on_vote_message);
+      bcast_new_block_message = std::move(on_new_block_message);
+      bcast_new_view_message = std::move(on_new_view_message);
+   }
+
    // Called internally by the chain_pacemaker to decide whether it should do something or not, based on feature activation.
    // Only methods called by the outside need to call this; methods called by qc_chain only don't need to check for enable().
    bool chain_pacemaker::enabled() const {
@@ -265,25 +282,22 @@ namespace eosio { namespace hotstuff {
    }
 
    void chain_pacemaker::send_hs_proposal_msg(const hs_proposal_message& msg, name id) {
-      hs_proposal_message_ptr msg_ptr = std::make_shared<hs_proposal_message>(msg);
-      _chain->commit_hs_proposal_msg(msg_ptr);
+      bcast_proposal_message(msg);
    }
 
    void chain_pacemaker::send_hs_vote_msg(const hs_vote_message& msg, name id) {
-      hs_vote_message_ptr msg_ptr = std::make_shared<hs_vote_message>(msg);
-      _chain->commit_hs_vote_msg(msg_ptr);
+      bcast_vote_message(msg);
    }
 
    void chain_pacemaker::send_hs_new_block_msg(const hs_new_block_message& msg, name id) {
-      hs_new_block_message_ptr msg_ptr = std::make_shared<hs_new_block_message>(msg);
-      _chain->commit_hs_new_block_msg(msg_ptr);
+      bcast_new_block_message(msg);
    }
 
    void chain_pacemaker::send_hs_new_view_msg(const hs_new_view_message& msg, name id) {
-      hs_new_view_message_ptr msg_ptr = std::make_shared<hs_new_view_message>(msg);
-      _chain->commit_hs_new_view_msg(msg_ptr);
+      bcast_new_view_message(msg);
    }
 
+   // called from net threads
    void chain_pacemaker::on_hs_proposal_msg(const hs_proposal_message& msg) {
       if (! enabled())
          return;
@@ -295,6 +309,7 @@ namespace eosio { namespace hotstuff {
       prof.core_out();
    }
 
+   // called from net threads
    void chain_pacemaker::on_hs_vote_msg(const hs_vote_message& msg) {
       if (! enabled())
          return;
@@ -306,6 +321,7 @@ namespace eosio { namespace hotstuff {
       prof.core_out();
    }
 
+   // called from net threads
    void chain_pacemaker::on_hs_new_block_msg(const hs_new_block_message& msg) {
       if (! enabled())
          return;
@@ -317,6 +333,7 @@ namespace eosio { namespace hotstuff {
       prof.core_out();
    }
 
+   // called from net threads
    void chain_pacemaker::on_hs_new_view_msg(const hs_new_view_message& msg) {
       if (! enabled())
          return;

libraries/hotstuff/include/eosio/hotstuff/chain_pacemaker.hpp

@@ -17,6 +17,12 @@ namespace eosio::hotstuff {
       //class-specific functions
 
       chain_pacemaker(controller* chain, std::set<account_name> my_producers, fc::logger& logger);
+      void register_bcast_functions(
+              std::function<void(const chain::hs_proposal_message&)> on_proposal_message,
+              std::function<void(const chain::hs_vote_message&)> on_vote_message,
+              std::function<void(const chain::hs_new_block_message&)> on_new_block_message,
+              std::function<void(const chain::hs_new_view_message&)> on_new_view_message
+      );
 
       void beat();
 
@@ -67,6 +73,10 @@ namespace eosio::hotstuff {
       chain::controller*      _chain = nullptr;
 
       qc_chain                _qc_chain;
+      std::function<void(const chain::hs_proposal_message&)>  bcast_proposal_message;
+      std::function<void(const chain::hs_vote_message&)>      bcast_vote_message;
+      std::function<void(const chain::hs_new_block_message&)> bcast_new_block_message;
+      std::function<void(const chain::hs_new_view_message&)>  bcast_new_view_message;
 
       uint32_t                _quorum_threshold = 15; //FIXME/TODO: calculate from schedule
       fc::logger&             _logger;

plugins/chain_plugin/chain_plugin.cpp

@@ -1119,6 +1119,20 @@ void chain_plugin::create_pacemaker(std::set<chain::account_name> my_producers)
    my->_chain_pacemaker.emplace(&chain(), std::move(my_producers), hotstuff_logger);
 }
 
+void chain_plugin::register_pacemaker_bcast_functions(
+        std::function<void(const chain::hs_proposal_message&)> on_proposal_message,
+        std::function<void(const chain::hs_vote_message&)> on_vote_message,
+        std::function<void(const chain::hs_new_block_message&)> on_new_block_message,
+        std::function<void(const chain::hs_new_view_message&)> on_new_view_message) {
+   EOS_ASSERT( my->_chain_pacemaker, plugin_config_exception, "chain_pacemaker not created" );
+   my->_chain_pacemaker->register_bcast_functions(
+           std::move(on_proposal_message),
+           std::move(on_vote_message),
+           std::move(on_new_block_message),
+           std::move(on_new_view_message));
+}
+
+
 void chain_plugin::plugin_initialize(const variables_map& options) {
    handle_sighup(); // Sets loggers
    my->plugin_initialize(options);
@@ -2651,7 +2665,7 @@ read_only::get_finalizer_state_results
 read_only::get_finalizer_state(const get_finalizer_state_params&, const fc::time_point& deadline ) const {
    get_finalizer_state_results results;
 
-   if ( chain_pacemaker ) {  // producer_plug is null when called from chain_plugin_tests.cpp and get_table_tests.cpp
+   if ( chain_pacemaker ) {  // is null when called from chain_plugin_tests.cpp and get_table_tests.cpp
       finalizer_state fs;
       chain_pacemaker->get_state( fs );
       results.chained_mode           = fs.chained_mode;
@@ -2665,8 +2679,9 @@ read_only::get_finalizer_state(const get_finalizer_state_params&, const fc::time
       results.high_qc                = fs.high_qc;
       results.current_qc             = fs.current_qc;
       results.schedule               = fs.schedule;
-      for (auto proposal: fs.proposals) {
-         chain::hs_proposal_message & p = proposal.second;
+      results.proposals.reserve( fs.proposals.size() );
+      for (const auto& proposal : fs.proposals) {
+         const chain::hs_proposal_message& p = proposal.second;
          results.proposals.push_back( hs_complete_proposal_message( p ) );
       }
    }
@@ -2675,18 +2690,22 @@ read_only::get_finalizer_state(const get_finalizer_state_params&, const fc::time
 
 } // namespace chain_apis
 
+// called from net threads
 void chain_plugin::notify_hs_vote_message( const hs_vote_message& msg ) {
    my->_chain_pacemaker->on_hs_vote_msg(msg);
 };
 
+// called from net threads
 void chain_plugin::notify_hs_proposal_message( const hs_proposal_message& msg ) {
    my->_chain_pacemaker->on_hs_proposal_msg(msg);
 };
 
+// called from net threads
 void chain_plugin::notify_hs_new_view_message( const hs_new_view_message& msg ) {
    my->_chain_pacemaker->on_hs_new_view_msg(msg);
 };
 
+// called from net threads
 void chain_plugin::notify_hs_new_block_message( const hs_new_block_message& msg ) {
    my->_chain_pacemaker->on_hs_new_block_msg(msg);
 };

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

@@ -843,7 +843,7 @@ class read_only : public api_base {
       uint8_t                    phase_counter   = 0;
       uint32_t                   block_height    = 0;
       uint64_t                   view_number     = 0;
-      explicit hs_complete_proposal_message( const chain::hs_proposal_message & p ) {
+      explicit hs_complete_proposal_message( const chain::hs_proposal_message& p ) {
          proposal_id    = p.proposal_id;
          block_id       = p.block_id;
          parent_id      = p.parent_id;
@@ -1032,6 +1032,12 @@ class chain_plugin : public plugin<chain_plugin> {
    const controller& chain() const;
 
    void create_pacemaker(std::set<chain::account_name> my_producers);
+   void register_pacemaker_bcast_functions(
+           std::function<void(const chain::hs_proposal_message&)> on_proposal_message,
+           std::function<void(const chain::hs_vote_message&)> on_vote_message,
+           std::function<void(const chain::hs_new_block_message&)> on_new_block_message,
+           std::function<void(const chain::hs_new_view_message&)> on_new_view_message
+   );
    void notify_hs_vote_message( const chain::hs_vote_message& msg );
    void notify_hs_proposal_message( const chain::hs_proposal_message& msg );
    void notify_hs_new_view_message( const chain::hs_new_view_message& msg );

plugins/net_plugin/include/eosio/net_plugin/net_plugin.hpp

@@ -3,6 +3,7 @@
 #include <eosio/chain/application.hpp>
 #include <eosio/net_plugin/protocol.hpp>
 #include <eosio/chain_plugin/chain_plugin.hpp>
+#include <eosio/producer_plugin/producer_plugin.hpp>
 
 namespace eosio {
    using namespace appbase;
@@ -26,7 +27,7 @@ namespace eosio {
         net_plugin();
         virtual ~net_plugin();
 
-        APPBASE_PLUGIN_REQUIRES((chain_plugin))
+        APPBASE_PLUGIN_REQUIRES((chain_plugin)(producer_plugin))
         virtual void set_program_options(options_description& cli, options_description& cfg) override;
         void handle_sighup() override;