Merge branch 'gh-698' of github.com:AntelopeIO/leap into gh-698

libraries/chain/include/eosio/chain/types.hpp

@@ -43,7 +43,7 @@
 
 #define _V(n, v)  fc::mutable_variant_object(n, v)
 
-namespace eosio { namespace chain {
+namespace eosio::chain {
    using                               std::map;
    using                               std::vector;
    using                               std::unordered_map;
@@ -396,7 +396,21 @@ namespace eosio { namespace chain {
    template<class... Ts> struct overloaded : Ts... { using Ts::operator()...; };
    template<class... Ts> overloaded(Ts...) -> overloaded<Ts...>;
 
-} }  // eosio::chain
+   // next_function is a function passed to an API (like send_transaction) and which is called at the end of
+   // the API processing on the main thread. The type T is a description of the API result that can be
+   // serialized as output.
+   // The function accepts a variant which can contain an exception_ptr (if an exception occured while
+   // processing the API) or the result T.
+   // The third option is a function which can be executed in a multithreaded context (likely on the
+   // http_plugin thread pool) and which completes the API processing and returns the result T.
+   // -------------------------------------------------------------------------------------------------------
+   template<typename T>
+   using next_function_variant = std::variant<fc::exception_ptr, T, std::function<T()>>;
+
+   template<typename T>
+   using next_function = std::function<void(const next_function_variant<T>&)>;
+
+}  // eosio::chain
 
 namespace chainbase {
    // chainbase::shared_cow_string

libraries/chain/include/eosio/chain/unapplied_transaction_queue.hpp

@@ -27,7 +27,7 @@ enum class trx_enum_type {
    incoming_p2p = 4 // incoming_end() needs to be updated if this changes
 };
 
-using next_func_t = std::function<void(const std::variant<fc::exception_ptr, transaction_trace_ptr>&)>;
+using next_func_t = next_function<transaction_trace_ptr>;
 
 struct unapplied_transaction {
    const transaction_metadata_ptr trx_meta;

plugins/chain_api_plugin/chain_api_plugin.cpp

@@ -62,6 +62,32 @@ parse_params<chain_apis::read_only::get_transaction_status_params, http_params_t
 
 #define CHAIN_RO_CALL_WITH_400(call_name, http_response_code, params_type) CALL_WITH_400(chain, ro_api, chain_apis::read_only, call_name, http_response_code, params_type)
 
+template<class API, class PARAMS_PARSER, class HANDLER>
+static api_entry make_api_entry(http_plugin& _http_plugin, API& api, const char* api_name,
+                                const char* call_name, PARAMS_PARSER params_parser, HANDLER handler) {
+   return api_entry(
+      std::string("/v1/") + api_name + "/" + call_name,
+      [&_http_plugin, api, api_name, call_name,
+       params_parser = std::move(params_parser), handler = std::move(handler)](string&&, string&& body, url_response_callback&& cb) {
+         auto deadline = api.start();
+         try {
+            auto start = fc::time_point::now();
+            auto params = params_parser(body);
+            FC_CHECK_DEADLINE(deadline);
+
+            // call first handler on main thread (likely because it accesses non thread-safe data)
+            // returns a thread-safe lambda that can be enqueued on the http thread pool to complete the request
+            auto completion_handler = handler(api, start, deadline, params, cb);
+            FC_CHECK_DEADLINE(deadline); // make sure remaining time is > 0
+
+            // execute thread-safe http_handler on _http_plugin's thread pool
+            _http_plugin.post_http_thread_pool(std::move(completion_handler));
+         } catch (...) {
+            http_plugin::handle_exception(api_name, call_name, body, cb);
+         }
+      });
+}
+
 void chain_api_plugin::plugin_startup() {
    ilog( "starting chain_api_plugin" );
    my.reset(new chain_api_plugin_impl(app().get_plugin<chain_plugin>().chain()));
@@ -128,41 +154,32 @@ void chain_api_plugin::plugin_startup() {
    }
 
    _http_plugin.add_api({
-      { std::string("/v1/chain/get_block"),
-        [ro_api, &_http_plugin, max_time=std::min(chain.get_abi_serializer_max_time(),max_response_time)]
-              ( string&&, string&& body, url_response_callback&& cb ) mutable {
-           auto deadline = ro_api.start();
-           try {
-              auto start = fc::time_point::now();
-              auto params = parse_params<chain_apis::read_only::get_raw_block_params, http_params_types::params_required>(body);
-              FC_CHECK_DEADLINE( deadline );
-              chain::signed_block_ptr block = ro_api.get_raw_block( params, deadline );
-
-              auto abi_cache = ro_api.get_block_serializers( block, max_time );
-              FC_CHECK_DEADLINE( deadline );
-
-              auto post_time = fc::time_point::now();
-              auto remaining_time = max_time - (post_time - start);
-              _http_plugin.post_http_thread_pool(
-                    [ro_api, cb, deadline, post_time, remaining_time, abi_cache{std::move(abi_cache)}, block{std::move( block )}]() mutable {
-                 try {
-                    auto new_deadline = deadline + (fc::time_point::now() - post_time);
-
-                    fc::variant result = ro_api.convert_block( block, std::move(abi_cache), remaining_time );
-
-                    cb( 200, new_deadline, std::move( result ) );
-                 } catch( ... ) {
-                    http_plugin::handle_exception( "chain", "get_block", "", cb );
-                 }
-              } );
-           } catch( ... ) {
-              http_plugin::handle_exception("chain", "get_block", body, cb);
-           }
-        }
-      }
-   }, appbase::exec_queue::read_only);
+         make_api_entry(
+            _http_plugin, ro_api, "chain", "get_block",
+            [](const string& body) {
+               return parse_params<chain_apis::read_only::get_raw_block_params, http_params_types::params_required>(body);
+            },
+            [max_response_time, &chain](auto& api, fc::time_point start, fc::time_point deadline, auto &params, const url_response_callback &cb) {
+
+               chain::signed_block_ptr block = api.get_raw_block(params, deadline);
+               auto max_time = std::min(chain.get_abi_serializer_max_time(), max_response_time);
+               auto abi_cache = api.get_block_serializers(block, max_time);
+               auto post_time = fc::time_point::now();
+               auto remaining_time = max_time - (post_time - start);
+
+               return [api, cb, deadline, post_time, remaining_time, abi_cache=std::move(abi_cache), block=std::move(block)]() mutable {
+                  try {
+                     auto new_deadline = deadline + (fc::time_point::now() - post_time);
+                     fc::variant result = api.convert_block(block, std::move(abi_cache), remaining_time);
+                     cb(200, new_deadline, std::move(result));
+                  } catch( ... ) {
+                     http_plugin::handle_exception("chain", "get_block", "", cb);
+                  }
+               };
+            })},
+      appbase::exec_queue::read_only);
 }
-
+   
 void chain_api_plugin::plugin_shutdown() {}
 
 }

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

@@ -8,13 +8,9 @@
 #include <eosio/chain/transaction_metadata.hpp>
 #include <eosio/chain/trace.hpp>
 
-namespace eosio { namespace chain { namespace plugin_interface {
+namespace eosio::chain::plugin_interface {
    using namespace eosio::chain;
    using namespace appbase;
-
-   template<typename T>
-   using next_function = std::function<void(const std::variant<fc::exception_ptr, T>&)>;
-
    struct chain_plugin_interface;
 
    namespace channels {
@@ -50,4 +46,4 @@ namespace eosio { namespace chain { namespace plugin_interface {
       }
    }
 
-} } }
+} // namespace eosio::chain::plugin_interface

plugins/chain_plugin/chain_plugin.cpp

@@ -2095,7 +2095,7 @@ void read_write::push_transaction(const read_write::push_transaction_params& par
       } EOS_RETHROW_EXCEPTIONS(chain::packed_transaction_type_exception, "Invalid packed transaction")
 
       app().get_method<incoming::methods::transaction_async>()(pretty_input, true, transaction_metadata::trx_type::input, false,
-            [this, next](const std::variant<fc::exception_ptr, transaction_trace_ptr>& result) -> void {
+            [this, next](const next_function_variant<transaction_trace_ptr>& result) -> void {
          if (std::holds_alternative<fc::exception_ptr>(result)) {
             next(std::get<fc::exception_ptr>(result));
          } else {
@@ -2169,13 +2169,15 @@ void read_write::push_transaction(const read_write::push_transaction_params& par
 }
 
 static void push_recurse(read_write* rw, int index, const std::shared_ptr<read_write::push_transactions_params>& params, const std::shared_ptr<read_write::push_transactions_results>& results, const next_function<read_write::push_transactions_results>& next) {
-   auto wrapped_next = [=](const std::variant<fc::exception_ptr, read_write::push_transaction_results>& result) {
+   auto wrapped_next = [=](const next_function_variant<read_write::push_transaction_results>& result) {
       if (std::holds_alternative<fc::exception_ptr>(result)) {
          const auto& e = std::get<fc::exception_ptr>(result);
          results->emplace_back( read_write::push_transaction_results{ transaction_id_type(), fc::mutable_variant_object( "error", e->to_detail_string() ) } );
-      } else {
+      } else if (std::holds_alternative<read_write::push_transaction_results>(result)) {
          const auto& r = std::get<read_write::push_transaction_results>(result);
          results->emplace_back( r );
+      } else {
+         assert(0);
       }
 
       size_t next_index = index + 1;
@@ -2214,12 +2216,11 @@ void read_write::send_transaction(const read_write::send_transaction_params& par
       } EOS_RETHROW_EXCEPTIONS(chain::packed_transaction_type_exception, "Invalid packed transaction")
 
       app().get_method<incoming::methods::transaction_async>()(pretty_input, true, transaction_metadata::trx_type::input, false,
-            [this, next](const std::variant<fc::exception_ptr, transaction_trace_ptr>& result) -> void {
+            [this, next](const next_function_variant<transaction_trace_ptr>& result) -> void {
          if (std::holds_alternative<fc::exception_ptr>(result)) {
             next(std::get<fc::exception_ptr>(result));
          } else {
             auto trx_trace_ptr = std::get<transaction_trace_ptr>(result);
-
             try {
                fc::variant output;
                try {
@@ -2257,7 +2258,7 @@ void read_write::send_transaction2(const read_write::send_transaction2_params& p
                   ("e", ptrx->expiration())("m", trx_retry->get_max_expiration_time()) );
 
       app().get_method<incoming::methods::transaction_async>()(ptrx, true, transaction_metadata::trx_type::input, static_cast<bool>(params.return_failure_trace),
-         [this, ptrx, next, retry, retry_num_blocks](const std::variant<fc::exception_ptr, transaction_trace_ptr>& result) -> void {
+         [this, ptrx, next, retry, retry_num_blocks](const next_function_variant<transaction_trace_ptr>& result) -> void {
             if( std::holds_alternative<fc::exception_ptr>( result ) ) {
                next( std::get<fc::exception_ptr>( result ) );
             } else {
@@ -2266,7 +2267,7 @@ void read_write::send_transaction2(const read_write::send_transaction2_params& p
                   if( retry && trx_retry.has_value() && !trx_trace_ptr->except) {
                      // will be ack'ed via next later
                      trx_retry->track_transaction( ptrx, retry_num_blocks,
-                        [ptrx, next](const std::variant<fc::exception_ptr, std::unique_ptr<fc::variant>>& result ) {
+                        [ptrx, next](const next_function_variant<std::unique_ptr<fc::variant>>& result ) {
                            if( std::holds_alternative<fc::exception_ptr>( result ) ) {
                               next( std::get<fc::exception_ptr>( result ) );
                            } else {
@@ -2578,7 +2579,7 @@ void read_only::send_transient_transaction(const Params& params, next_function<R
       } EOS_RETHROW_EXCEPTIONS(chain::packed_transaction_type_exception, "Invalid packed transaction")
 
       app().get_method<incoming::methods::transaction_async>()(pretty_input, true /* api_trx */, trx_type, true /* return_failure_trace */,
-          [this, next](const std::variant<fc::exception_ptr, transaction_trace_ptr>& result) -> void {
+          [this, next](const next_function_variant<transaction_trace_ptr>& result) -> void {
              if (std::holds_alternative<fc::exception_ptr>(result)) {
                    next(std::get<fc::exception_ptr>(result));
               } else {

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

@@ -718,7 +718,7 @@ class read_only {
 
 private:
    template<typename Params, typename Results>
-   void send_transient_transaction(const Params& params, next_function<Results> next, chain::transaction_metadata::trx_type trx_type) const;
+   void send_transient_transaction(const Params& params, eosio::chain::next_function<Results> next, chain::transaction_metadata::trx_type trx_type) const;
 };
 
 class read_write {

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

@@ -2,12 +2,10 @@
 #include <eosio/chain/types.hpp>
 #include <eosio/chain/block_state.hpp>
 #include <eosio/chain/trace.hpp>
+#include <eosio/chain/transaction_metadata.hpp>
 
 namespace eosio::chain_apis {
 
-template<typename T>
-using next_function = std::function<void(const std::variant<fc::exception_ptr, T>&)>;
-
 /**
  * This class manages the ephemeral indices and data that provide the transaction retry feature.
  * It is designed to be run on an API node, as it only tracks incoming API transactions.
@@ -51,7 +49,7 @@ class trx_retry_db {
     * @param next report result to user by calling next
     * @throws throw tx_resource_exhaustion if trx would exceeds max_mem_usage_size
     */
-   void track_transaction( chain::packed_transaction_ptr ptrx, std::optional<uint16_t> num_blocks, next_function<std::unique_ptr<fc::variant>> next );
+   void track_transaction( chain::packed_transaction_ptr ptrx, std::optional<uint16_t> num_blocks, eosio::chain::next_function<std::unique_ptr<fc::variant>> next );
 
    /**
     * Attach to chain applied_transaction signal