store abi def in chainbase

Author: WLBF

CMakeLists.txt

@@ -2,7 +2,6 @@ file(GLOB HEADERS "include/eosio/elasticsearch_plugin/*.hpp")
 add_library( elasticsearch_plugin
              elasticsearch_plugin.cpp
              elastic_client.cpp
-             deserializer.cpp
              bulker.cpp
              ${HEADERS} )
 

README.md

@@ -142,4 +142,3 @@ Config Options for eosio::elasticsearch_plugin.
 ## TODO
 
 - [ ] Due to `libcurl` [100-continue feature](https://curl.haxx.se/mail/lib-2017-07/0013.html), consider replace [EOSLaoMao/elasticlient](https://github.com/EOSLaoMao/elasticlient) with other simple http client like [https://cpp-netlib.org/#](https://cpp-netlib.org/#)
-- [ ] Currently, the plugin will cache all abi information in the RAM. It's not a issue for now. Maybe it is better to embed some LRU cache library in the feature.

deserializer.cpp

@@ -29,7 +29,7 @@
 #include "elastic_client.hpp"
 #include "exceptions.hpp"
 #include "mappings.hpp"
-#include "deserializer.hpp"
+#include "serializer.hpp"
 #include "bulker.hpp"
 #include "ThreadPool/ThreadPool.h"
 
@@ -126,10 +126,6 @@ class elasticsearch_plugin_impl {
    bool store_transaction_traces = true;
    bool store_action_traces = true;
 
-   std::unique_ptr<elastic_client> es_client;
-   std::unique_ptr<deserializer> abi_deserializer;
-   std::unique_ptr<bulker_pool> bulk_pool;
-   std::unique_ptr<ThreadPool> thread_pool;
    size_t max_task_queue_size = 0;
    int task_queue_sleep_time = 0;
 
@@ -153,6 +149,11 @@ class elasticsearch_plugin_impl {
    boost::atomic<bool> startup{true};
    fc::optional<chain::chain_id_type> chain_id;
 
+   std::unique_ptr<elastic_client> es_client;
+   std::unique_ptr<serializer> serializer;
+   std::unique_ptr<bulker_pool> bulk_pool;
+   std::unique_ptr<ThreadPool> thread_pool;
+
    static const action_name newaccount;
    static const action_name setabi;
    static const action_name updateauth;
@@ -507,7 +508,7 @@ void elasticsearch_plugin_impl::upsert_account_setabi(
 {
    abi_def abi_def = fc::raw::unpack<chain::abi_def>( setabi.abi );
 
-   abi_deserializer->upsert_abi_cache( setabi.account, abi_def );
+   serializer->upsert_abi_cache( setabi.account, abi_def );
 
    param_doc("name", setabi.account.to_string());
    param_doc("abi", abi_def);
@@ -687,7 +688,7 @@ void elasticsearch_plugin_impl::_process_applied_transaction( chain::transaction
          for (auto& atrace : base_action_traces) {
             fc::mutable_variant_object action_traces_doc;
             chain::base_action_trace &base = atrace.get();
-            fc::from_variant( abi_deserializer->to_variant_with_abi( base ), action_traces_doc );
+            fc::from_variant( serializer->to_variant_with_abi( base ), action_traces_doc );
 
             fc::mutable_variant_object act_doc;
             fc::from_variant( action_traces_doc["act"], act_doc );
@@ -699,7 +700,7 @@ void elasticsearch_plugin_impl::_process_applied_transaction( chain::transaction
             fc::mutable_variant_object action_doc;
             action_doc("_index", action_traces_index);
             action_doc("_type", "_doc");
-            action_doc("_id", action_traces_doc["receipt"]["global_sequence"]);
+            action_doc("_id", base.receipt.global_sequence);
             action_doc("retry_on_conflict", 100);
 
             auto action = fc::json::to_string( fc::variant_object("index", action_doc) );
@@ -713,7 +714,7 @@ void elasticsearch_plugin_impl::_process_applied_transaction( chain::transaction
             // transaction trace index
 
             fc::mutable_variant_object trans_traces_doc;
-            fc::from_variant( abi_deserializer->to_variant_with_abi( *t ), trans_traces_doc );
+            fc::from_variant( serializer->to_variant_with_abi( *t ), trans_traces_doc );
             trans_traces_doc("createAt", now.count());
 
             fc::mutable_variant_object action_doc;
@@ -751,7 +752,7 @@ void elasticsearch_plugin_impl::_process_accepted_transaction( chain::transactio
          auto now = std::chrono::duration_cast<std::chrono::milliseconds>(
                std::chrono::microseconds{fc::time_point::now().time_since_epoch().count()} );
 
-         fc::from_variant( abi_deserializer->to_variant_with_abi( trx ), trans_doc );
+         fc::from_variant( serializer->to_variant_with_abi( trx ), trans_doc );
          trans_doc("trx_id", trx_id_str);
 
          fc::variant signing_keys;
@@ -794,7 +795,7 @@ void elasticsearch_plugin_impl::_process_accepted_block( chain::block_state_ptr
       [ bs{std::move(bs)}, this ]()
       {
          auto block_num = bs->block_num;
-         if( block_num % 1000 == 0 )
+         if( block_num % 10000 == 0 )
             ilog( "block_num: ${b}", ("b", block_num) );
 
          const auto block_id = bs->id;
@@ -856,7 +857,7 @@ void elasticsearch_plugin_impl::_process_accepted_block( chain::block_state_ptr
 
             params_doc("block_num", static_cast<int32_t>(block_num));
             params_doc("block_id", block_id_str);
-            params_doc("block", abi_deserializer->to_variant_with_abi( *bs->block ));
+            params_doc("block", serializer->to_variant_with_abi( *bs->block ));
             params_doc("irreversible", false);
             params_doc("createAt", now.count());
 
@@ -946,15 +947,14 @@ void elasticsearch_plugin_impl::_process_irreversible_block(chain::block_state_p
 
             block_doc("block_num", static_cast<int32_t>(block_num));
             block_doc("block_id", block_id_str);
-            block_doc("block", abi_deserializer->to_variant_with_abi( *bs->block ));
+            block_doc("block", serializer->to_variant_with_abi( *bs->block ));
             block_doc("irreversible", true);
             block_doc("validated", bs->validated);
             block_doc("createAt", now.count());
 
             doc("script", script_doc);
             doc("upsert", block_doc);
 
-
             fc::mutable_variant_object action_doc;
             action_doc("_index", blocks_index);
             action_doc("_type", "_doc");
@@ -1238,7 +1238,7 @@ void elasticsearch_plugin::plugin_initialize(const variables_map& options) {
             EOS_ASSERT(max_time > chain::config::default_abi_serializer_max_time_ms,
                        chain::plugin_config_exception, "--abi-serializer-max-time-ms required as default value not appropriate for parsing full blocks");
             fc::microseconds abi_serializer_max_time = app().get_plugin<chain_plugin>().get_abi_serializer_max_time();
-            my->abi_deserializer.reset( new deserializer( abi_serializer_max_time ));
+            my->serializer.reset( new serializer( app().data_dir() / "abi", abi_serializer_max_time ));
          }
 
          if( options.count( "elastic-queue-size" )) {

deserializer.hpp

@@ -0,0 +1,134 @@
+#pragma once
+#include <eosio/chain/multi_index_includes.hpp>
+#include <eosio/chain/database_utils.hpp>
+
+using namespace eosio;
+using namespace chainbase;
+
+struct by_account;
+
+class abi_cache: public chainbase::object<0, abi_cache> {
+   OBJECT_CTOR(abi_cache,(abi))
+
+   id_type                                id;
+   account_name                           account;
+   eosio::chain::shared_blob              abi;
+
+   void set_abi( const eosio::chain::abi_def& a ) {
+      abi.resize( fc::raw::pack_size( a ) );
+      fc::datastream<char*> ds( abi.data(), abi.size() );
+      fc::raw::pack( ds, a );
+   }
+};
+
+
+using abi_cache_index_t = chainbase::shared_multi_index_container<abi_cache,
+      indexed_by<
+            ordered_unique< member<abi_cache,abi_cache::id_type,&abi_cache::id> >,
+            ordered_unique< tag<by_account>,  member<abi_cache,account_name,&abi_cache::account> >
+      >
+>;
+
+CHAINBASE_SET_INDEX_TYPE( abi_cache, abi_cache_index_t )
+
+
+class serializer
+{
+public:
+   serializer(const bfs::path& dir, fc::microseconds abi_serializer_max_time)
+      :db(dir, database::read_write, 1024*1024*8), abi_serializer_max_time(abi_serializer_max_time)
+   {
+      db.add_index<abi_cache_index_t>();
+   }
+
+   optional<abi_serializer> get_abi_serializer( const account_name &name ) {
+      if( name.good() ) {
+         try {
+            const auto& a = db.get<abi_cache, by_account>(name);
+            abi_def abi;
+            if( abi_serializer::to_abi( a.abi, abi ))
+               return abi_def_to_serializer(name, abi);
+         } catch( std::out_of_range& e) {
+           // ignore missing abi excepton.
+         } FC_CAPTURE_AND_LOG((name))
+      }
+      return optional<abi_serializer>();
+   }
+
+   template<typename T>
+   fc::variant to_variant_with_abi( const T& obj ) {
+      fc::variant pretty_output;
+      abi_serializer::to_variant( obj, pretty_output,
+                                  [&]( account_name n ) { return get_abi_serializer( n ); },
+                                  abi_serializer_max_time );
+      return pretty_output;
+   }
+
+   void upsert_abi_cache( const account_name &name, const abi_def& abi ) {
+      if( name.good()) {
+         try {
+            auto* a = db.find<abi_cache, by_account>(name);
+            if ( a == nullptr ) {
+               db.create<abi_cache>( [&]( abi_cache& ca ) {
+                  ca.account = name;
+                  ca.set_abi(abi);
+               });
+            } else {
+               db.modify( *a, [&]( abi_cache& ca ) {
+                  ca.set_abi(abi);
+               });
+            }
+         } FC_CAPTURE_AND_LOG((name))
+      }
+   }
+
+private:
+
+   chainbase::database db;
+   fc::microseconds abi_serializer_max_time;
+
+   optional<abi_serializer> abi_def_to_serializer( const account_name &name, const abi_def& abi ) {
+      if( name.good()) {
+         try {
+            abi_serializer abis;
+
+            if( name == chain::config::system_account_name ) {
+               // redefine eosio setabi.abi from bytes to abi_def
+               // Done so that abi is stored as abi_def in elasticsearch instead of as bytes
+               abi_def abi_new = abi;
+               auto itr = std::find_if( abi_new.structs.begin(), abi_new.structs.end(),
+                                          []( const auto& s ) { return s.name == "setabi"; } );
+               if( itr != abi_new.structs.end() ) {
+                  auto itr2 = std::find_if( itr->fields.begin(), itr->fields.end(),
+                                             []( const auto& f ) { return f.name == "abi"; } );
+                  if( itr2 != itr->fields.end() ) {
+                     if( itr2->type == "bytes" ) {
+                        itr2->type = "abi_def";
+                        // unpack setabi.abi as abi_def instead of as bytes
+                        abis.add_specialized_unpack_pack( "abi_def",
+                              std::make_pair<abi_serializer::unpack_function, abi_serializer::pack_function>(
+                                    []( fc::datastream<const char*>& stream, bool is_array, bool is_optional ) -> fc::variant {
+                                       EOS_ASSERT( !is_array && !is_optional, chain::elasticsearch_exception, "unexpected abi_def");
+                                       chain::bytes temp;
+                                       fc::raw::unpack( stream, temp );
+                                       return fc::variant( fc::raw::unpack<abi_def>( temp ) );
+                                    },
+                                    []( const fc::variant& var, fc::datastream<char*>& ds, bool is_array, bool is_optional ) {
+                                       EOS_ASSERT( false, chain::elasticsearch_exception, "never called" );
+                                    }
+                              ) );
+                     }
+                  }
+               }
+               abis.set_abi( abi_new, abi_serializer_max_time );
+            } else {
+               abis.set_abi( abi, abi_serializer_max_time );
+            }
+
+            return abis;
+         } FC_CAPTURE_AND_LOG((name))
+      }
+      return optional<abi_serializer>();
+   }
+
+};