sysio::state_history_plugin_impl Struct Reference

Inheritance diagram for sysio::state_history_plugin_impl:

Collaboration diagram for sysio::state_history_plugin_impl:

Classes
struct	session

Public Member Functions
void	get_log_entry (state_history_log &log, uint32_t block_num, std::optional< bytes > &result)
void	get_block (uint32_t block_num, const block_state_ptr &block_state, std::optional< bytes > &result)
std::optional< chain::block_id_type >	get_block_id (uint32_t block_num)
void	listen ()
void	do_accept ()
void	on_applied_transaction (const transaction_trace_ptr &p, const packed_transaction_ptr &t)
void	on_accepted_block (const block_state_ptr &block_state)
void	on_block_start (uint32_t block_num)
void	clear_caches ()
void	store_traces (const block_state_ptr &block_state)
void	store_chain_state (const block_state_ptr &block_state)

Public Attributes
chain_plugin *	chain_plug = nullptr
std::optional< state_history_log >	trace_log
std::optional< state_history_log >	chain_state_log
bool	trace_debug_mode = false
bool	stopping = false
std::optional< scoped_connection >	applied_transaction_connection
std::optional< scoped_connection >	block_start_connection
std::optional< scoped_connection >	accepted_block_connection
string	endpoint_address = "0.0.0.0"
uint16_t	endpoint_port = 8080
std::unique_ptr< tcp::acceptor >	acceptor
state_history::trace_converter	trace_converter
std::map< session *, std::shared_ptr< session > >	sessions

Detailed Description

Definition at line 43 of file state_history_plugin.cpp.

Member Function Documentation

clear_caches()

void sysio::state_history_plugin_impl::clear_caches ( )

inline

Definition at line 387 of file state_history_plugin.cpp.

                       {
      trace_converter.cached_traces.clear();
      trace_converter.onblock_trace.reset();
   }

Here is the caller graph for this function:

do_accept()

void sysio::state_history_plugin_impl::do_accept ( )

inline

Definition at line 335 of file state_history_plugin.cpp.

                    {
      auto socket = std::make_shared<tcp::socket>(app().get_io_service());
      acceptor->async_accept(*socket, [self = shared_from_this(), socket, this](const boost::system::error_code& ec) {
         if (stopping)
            return;
         if (ec) {
            if (ec == boost::system::errc::too_many_files_open)
               catch_and_log([&] { do_accept(); });
            return;
         }
         catch_and_log([&] {
            auto s            = std::make_shared<session>(self);
            sessions[s.get()] = s;
            s->start(std::move(*socket));
         });
         catch_and_log([&] { do_accept(); });
      });
   }

Here is the call graph for this function:

Here is the caller graph for this function:

get_block()

void sysio::state_history_plugin_impl::get_block ( uint32_t block_num, const block_state_ptr & block_state, std::optional< bytes > & result )

inline

Definition at line 74 of file state_history_plugin.cpp.

                                                                                                      {
      chain::signed_block_ptr p;
      try {
         if( block_state && block_num == block_state->block_num ) {
            p = block_state->block;
         } else {
            p = chain_plug->chain().fetch_block_by_number( block_num );
         }
      } catch (...) {
         return;
      }
      if (p)
         result = fc::raw::pack(*p);
   }

Here is the call graph for this function:

get_block_id()

std::optional< chain::block_id_type > sysio::state_history_plugin_impl::get_block_id ( uint32_t block_num )

inline

Definition at line 89 of file state_history_plugin.cpp.

                                                                    {
      if (trace_log && block_num >= trace_log->begin_block() && block_num < trace_log->end_block())
         return trace_log->get_block_id(block_num);
      if (chain_state_log && block_num >= chain_state_log->begin_block() && block_num < chain_state_log->end_block())
         return chain_state_log->get_block_id(block_num);
      try {
         return chain_plug->chain().get_block_id_for_num(block_num);
      } catch (...) {}
      return {};
   }

Here is the call graph for this function:

get_log_entry()

void sysio::state_history_plugin_impl::get_log_entry ( state_history_log & log, uint32_t block_num, std::optional< bytes > & result )

inline

Definition at line 57 of file state_history_plugin.cpp.

                                                                                              {
      if (block_num < log.begin_block() || block_num >= log.end_block())
         return;
      state_history_log_header header;
      auto&                    stream = log.get_entry(block_num, header);
      uint32_t                 s;
      // Compressed deltas now exceeds 4GB on one of the public chains. This length prefix
      // was intended to support adding additional fields in the future after the
      // packed deltas or packed traces. For now we're going to ignore on read.
      stream.read((char*)&s, sizeof(s));
      uint64_t s2 = header.payload_size - sizeof(s);
      bytes    compressed(s2);
      if (s2)
         stream.read(compressed.data(), s2);
      result = state_history::zlib_decompress(compressed);
   }

Here is the call graph for this function:

listen()

void sysio::state_history_plugin_impl::listen ( )

inline

Definition at line 311 of file state_history_plugin.cpp.

                 {
      boost::system::error_code ec;
 
      auto address  = boost::asio::ip::make_address(endpoint_address);
      auto endpoint = tcp::endpoint{address, endpoint_port};
      acceptor      = std::make_unique<tcp::acceptor>(app().get_io_service());
 
      auto check_ec = [&](const char* what) {
         if (!ec)
            return;
         elog("${w}: ${m}", ("w", what)("m", ec.message()));
         SYS_ASSERT(false, plugin_exception, "unable to open listen socket");
      };
 
      acceptor->open(endpoint.protocol(), ec);
      check_ec("open");
      acceptor->set_option(boost::asio::socket_base::reuse_address(true));
      acceptor->bind(endpoint, ec);
      check_ec("bind");
      acceptor->listen(boost::asio::socket_base::max_listen_connections, ec);
      check_ec("listen");
      do_accept();
   }

Here is the call graph for this function:

on_accepted_block()

void sysio::state_history_plugin_impl::on_accepted_block ( const block_state_ptr & block_state )

inline

Definition at line 359 of file state_history_plugin.cpp.

                                                              {
      try {
         store_traces(block_state);
         store_chain_state(block_state);
      } catch (const fc::exception& e) {
         elog("fc::exception: ${details}", ("details", e.to_detail_string()));
         // Both app().quit() and exception throwing are required. Without app().quit(),
         // the exception would be caught and drop before reaching main(). The exception is
         // to ensure the block won't be commited.
         appbase::app().quit();
         SYS_THROW(
             chain::state_history_write_exception,
             "State history encountered an Error which it cannot recover from.  Please resolve the error and relaunch "
             "the process");
      }
 
      for (auto& s : sessions) {
         auto& p = s.second;
         if (p) {
            if (p->current_request && block_state->block_num < p->current_request->start_block_num)
               p->current_request->start_block_num = block_state->block_num;
            p->send_update(block_state);
         }
      }
   }

Here is the call graph for this function:

on_applied_transaction()

void sysio::state_history_plugin_impl::on_applied_transaction ( const transaction_trace_ptr & p, const packed_transaction_ptr & t )

inline

Definition at line 354 of file state_history_plugin.cpp.

                                                                                                {
      if (trace_log)
         trace_converter.add_transaction(p, t);
   }

Here is the call graph for this function:

on_block_start()

void sysio::state_history_plugin_impl::on_block_start ( uint32_t block_num )

inline

Definition at line 385 of file state_history_plugin.cpp.

{ clear_caches(); }

Here is the call graph for this function:

store_chain_state()

void sysio::state_history_plugin_impl::store_chain_state ( const block_state_ptr & block_state )

inline

Definition at line 411 of file state_history_plugin.cpp.

                                                              {
      if (!chain_state_log)
         return;
      bool fresh = chain_state_log->begin_block() == chain_state_log->end_block();
      if (fresh)
         ilog("Placing initial state in block ${n}", ("n", block_state->block->block_num()));
 
      std::vector<table_delta> deltas     = state_history::create_deltas(chain_plug->chain().db(), fresh);
      auto                     deltas_bin = state_history::zlib_compress_bytes(fc::raw::pack(deltas));
      state_history_log_header header{.magic        = ship_magic(ship_current_version, 0),
                                      .block_id     = block_state->id,
                                      .payload_size = sizeof(uint32_t) + deltas_bin.size()};
      chain_state_log->write_entry(header, block_state->block->previous, [&](auto& stream) {
         // Compressed deltas now exceeds 4GB on one of the public chains. This length prefix
         // was intended to support adding additional fields in the future after the
         // packed deltas. For now we're going to ignore on read. The 0 is an attempt to signal
         // old versions that something's not quite right.
         uint32_t s = (uint32_t)deltas_bin.size();
         if (s != deltas_bin.size())
            s = 0;
         stream.write((char*)&s, sizeof(s));
         if (!deltas_bin.empty())
            stream.write(deltas_bin.data(), deltas_bin.size());
      });
   } // store_chain_state

Here is the call graph for this function:

Here is the caller graph for this function:

store_traces()

void sysio::state_history_plugin_impl::store_traces ( const block_state_ptr & block_state )

inline

Definition at line 392 of file state_history_plugin.cpp.

                                                         {
      if (!trace_log)
         return;
      auto traces_bin = state_history::zlib_compress_bytes(
          trace_converter.pack(chain_plug->chain().db(), trace_debug_mode, block_state));
 
      SYS_ASSERT(traces_bin.size() == (uint32_t)traces_bin.size(), plugin_exception, "traces is too big");
 
      state_history_log_header header{.magic        = ship_magic(ship_current_version, 0),
                                      .block_id     = block_state->id,
                                      .payload_size = sizeof(uint32_t) + traces_bin.size()};
      trace_log->write_entry(header, block_state->block->previous, [&](auto& stream) {
         uint32_t s = (uint32_t)traces_bin.size();
         stream.write((char*)&s, sizeof(s));
         if (!traces_bin.empty())
            stream.write(traces_bin.data(), traces_bin.size());
      });
   }

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

accepted_block_connection

std::optional<scoped_connection> sysio::state_history_plugin_impl::accepted_block_connection

Definition at line 51 of file state_history_plugin.cpp.

acceptor

std::unique_ptr<tcp::acceptor> sysio::state_history_plugin_impl::acceptor

Definition at line 54 of file state_history_plugin.cpp.

applied_transaction_connection

std::optional<scoped_connection> sysio::state_history_plugin_impl::applied_transaction_connection

Definition at line 49 of file state_history_plugin.cpp.

block_start_connection

std::optional<scoped_connection> sysio::state_history_plugin_impl::block_start_connection

Definition at line 50 of file state_history_plugin.cpp.

chain_plug

chain_plugin* sysio::state_history_plugin_impl::chain_plug = nullptr

Definition at line 44 of file state_history_plugin.cpp.

chain_state_log

std::optional<state_history_log> sysio::state_history_plugin_impl::chain_state_log

Definition at line 46 of file state_history_plugin.cpp.

endpoint_address

string sysio::state_history_plugin_impl::endpoint_address = "0.0.0.0"

Definition at line 52 of file state_history_plugin.cpp.

endpoint_port

uint16_t sysio::state_history_plugin_impl::endpoint_port = 8080

Definition at line 53 of file state_history_plugin.cpp.

sessions

std::map<session*, std::shared_ptr<session> > sysio::state_history_plugin_impl::sessions

Definition at line 309 of file state_history_plugin.cpp.

stopping

bool sysio::state_history_plugin_impl::stopping = false

Definition at line 48 of file state_history_plugin.cpp.

trace_converter

state_history::trace_converter sysio::state_history_plugin_impl::trace_converter

Definition at line 55 of file state_history_plugin.cpp.

trace_debug_mode

bool sysio::state_history_plugin_impl::trace_debug_mode = false

Definition at line 47 of file state_history_plugin.cpp.

trace_log

std::optional<state_history_log> sysio::state_history_plugin_impl::trace_log

Definition at line 45 of file state_history_plugin.cpp.

---

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

• plugins/state_history_plugin/state_history_plugin.cpp