sysio::state_history_plugin_impl::session Struct Reference

Inheritance diagram for sysio::state_history_plugin_impl::session:

Collaboration diagram for sysio::state_history_plugin_impl::session:

Public Types
using	result_type = void

Public Member Functions
	session (std::shared_ptr< state_history_plugin_impl > plugin)
void	start (tcp::socket socket)
void	start_read ()
void	send (const char *s)
template<typename T >
void	send (T obj)
void	send ()
void	operator() (get_status_request_v0 &)
void	operator() (get_blocks_request_v0 &req)
void	operator() (get_blocks_ack_request_v0 &req)
void	send_update (get_blocks_result_v0 result, const block_state_ptr &block_state)
void	send_update (const block_state_ptr &block_state)
void	send_update (bool changed=false)
template<typename F >
void	catch_and_close (F f)
template<typename F >
void	callback (boost::system::error_code ec, const char *what, F f)
void	on_fail (boost::system::error_code ec, const char *what)
void	close ()

Public Attributes
std::shared_ptr< state_history_plugin_impl >	plugin
std::unique_ptr< ws::stream< tcp::socket > >	socket_stream
bool	sending = false
bool	sent_abi = false
std::vector< std::vector< char > >	send_queue
std::optional< get_blocks_request_v0 >	current_request
bool	need_to_send_update = false

Detailed Description

Definition at line 100 of file state_history_plugin.cpp.

Member Typedef Documentation

result_type

using sysio::state_history_plugin_impl::session::result_type = void

Definition at line 173 of file state_history_plugin.cpp.

Constructor & Destructor Documentation

session()

sysio::state_history_plugin_impl::session::session ( std::shared_ptr< state_history_plugin_impl > plugin )

inline

Definition at line 109 of file state_history_plugin.cpp.

          : plugin(std::move(plugin)) {}

Member Function Documentation

callback()

template<typename F >

void sysio::state_history_plugin_impl::session::callback ( boost::system::error_code ec, const char * what, F f )

inline

Definition at line 281 of file state_history_plugin.cpp.

                                                                     {
         app().post(priority::medium, [=]() {
            if (plugin->stopping)
               return;
            if (ec)
               return on_fail(ec, what);
            catch_and_close(f);
         });
      }

Here is the call graph for this function:

catch_and_close()

template<typename F >

void sysio::state_history_plugin_impl::session::catch_and_close ( F f )

inline

Definition at line 265 of file state_history_plugin.cpp.

                                {
         try {
            f();
         } catch (const fc::exception& e) {
            elog("${e}", ("e", e.to_detail_string()));
            close();
         } catch (const std::exception& e) {
            elog("${e}", ("e", e.what()));
            close();
         } catch (...) {
            elog("unknown exception");
            close();
         }
      }

Here is the call graph for this function:

Here is the caller graph for this function:

close()

void sysio::state_history_plugin_impl::session::close ( )

inline

Definition at line 304 of file state_history_plugin.cpp.

                   {
         socket_stream->next_layer().close();
         plugin->sessions.erase(this);
      }

Here is the caller graph for this function:

on_fail()

void sysio::state_history_plugin_impl::session::on_fail ( boost::system::error_code ec, const char * what )

inline

Definition at line 291 of file state_history_plugin.cpp.

                                                               {
         try {
            if (ec == boost::asio::error::eof) {
               dlog("${w}: ${m}", ("w", what)("m", ec.message()));
            } else {
               elog("${w}: ${m}", ("w", what)("m", ec.message()));
            }
            close();
         } catch (...) {
            elog("uncaught exception on close");
         }
      }

Here is the call graph for this function:

Here is the caller graph for this function:

operator()() [1/3]

void sysio::state_history_plugin_impl::session::operator() ( get_blocks_ack_request_v0 & req )

inline

Definition at line 204 of file state_history_plugin.cpp.

                                                      {
         if (!current_request)
            return;
         current_request->max_messages_in_flight += req.num_messages;
         send_update();
      }

Here is the call graph for this function:

operator()() [2/3]

void sysio::state_history_plugin_impl::session::operator() ( get_blocks_request_v0 & req )

inline

Definition at line 191 of file state_history_plugin.cpp.

                                                  {
         for (auto& cp : req.have_positions) {
            if (req.start_block_num <= cp.block_num)
               continue;
            auto id = plugin->get_block_id(cp.block_num);
            if (!id || *id != cp.block_id)
               req.start_block_num = std::min(req.start_block_num, cp.block_num);
         }
         req.have_positions.clear();
         current_request = req;
         send_update(true);
      }

Here is the call graph for this function:

operator()() [3/3]

void sysio::state_history_plugin_impl::session::operator() ( get_status_request_v0 & )

inline

Definition at line 174 of file state_history_plugin.cpp.

                                              {
         auto&                chain = plugin->chain_plug->chain();
         get_status_result_v0 result;
         result.head              = {chain.head_block_num(), chain.head_block_id()};
         result.last_irreversible = {chain.last_irreversible_block_num(), chain.last_irreversible_block_id()};
         result.chain_id          = chain.get_chain_id();
         if (plugin->trace_log) {
            result.trace_begin_block = plugin->trace_log->begin_block();
            result.trace_end_block   = plugin->trace_log->end_block();
         }
         if (plugin->chain_state_log) {
            result.chain_state_begin_block = plugin->chain_state_log->begin_block();
            result.chain_state_end_block   = plugin->chain_state_log->end_block();
         }
         send(std::move(result));
      }

Here is the call graph for this function:

send() [1/3]

void sysio::state_history_plugin_impl::session::send ( )

inline

Definition at line 154 of file state_history_plugin.cpp.

                  {
         if (sending)
            return;
         if (send_queue.empty())
            return send_update();
         sending = true;
         socket_stream->binary(sent_abi);
         sent_abi = true;
         socket_stream->async_write(             //
             boost::asio::buffer(send_queue[0]), //
             [self = shared_from_this()](boost::system::error_code ec, size_t) {
                self->callback(ec, "async_write", [self] {
                   self->send_queue.erase(self->send_queue.begin());
                   self->sending = false;
                   self->send();
                });
             });
      }

Here is the call graph for this function:

Here is the caller graph for this function:

send() [2/3]

void sysio::state_history_plugin_impl::session::send ( const char * s )

inline

Definition at line 143 of file state_history_plugin.cpp.

                               {
         send_queue.push_back({s, s + strlen(s)});
         send();
      }

Here is the call graph for this function:

send() [3/3]

template<typename T >

void sysio::state_history_plugin_impl::session::send ( T obj )

inline

Definition at line 149 of file state_history_plugin.cpp.

                       {
         send_queue.push_back(fc::raw::pack(state_result{std::move(obj)}));
         send();
      }

Here is the call graph for this function:

send_update() [1/3]

void sysio::state_history_plugin_impl::session::send_update ( bool changed = false )

inline

Definition at line 252 of file state_history_plugin.cpp.

                                             {
         if (changed)
            need_to_send_update = true;
         if (!send_queue.empty() || !need_to_send_update || !current_request ||
             !current_request->max_messages_in_flight)
            return;
         auto&                chain = plugin->chain_plug->chain();
         get_blocks_result_v0 result;
         result.head = {chain.head_block_num(), chain.head_block_id()};
         send_update(std::move(result), {});
      }

Here is the call graph for this function:

send_update() [2/3]

void sysio::state_history_plugin_impl::session::send_update ( const block_state_ptr & block_state )

inline

Definition at line 243 of file state_history_plugin.cpp.

                                                           {
         need_to_send_update = true;
         if (!send_queue.empty() || !current_request || !current_request->max_messages_in_flight)
            return;
         get_blocks_result_v0 result;
         result.head = {block_state->block_num, block_state->id};
         send_update(std::move(result), block_state);
      }

Here is the call graph for this function:

send_update() [3/3]

void sysio::state_history_plugin_impl::session::send_update ( get_blocks_result_v0 result, const block_state_ptr & block_state )

inline

Definition at line 211 of file state_history_plugin.cpp.

                                                                                        {
         need_to_send_update = true;
         if (!send_queue.empty() || !current_request || !current_request->max_messages_in_flight)
            return;
         auto& chain              = plugin->chain_plug->chain();
         result.last_irreversible = {chain.last_irreversible_block_num(), chain.last_irreversible_block_id()};
         uint32_t current =
             current_request->irreversible_only ? result.last_irreversible.block_num : result.head.block_num;
         if (current_request->start_block_num <= current &&
             current_request->start_block_num < current_request->end_block_num) {
            auto block_id = plugin->get_block_id(current_request->start_block_num);
            if (block_id) {
               result.this_block  = block_position{current_request->start_block_num, *block_id};
               auto prev_block_id = plugin->get_block_id(current_request->start_block_num - 1);
               if (prev_block_id)
                  result.prev_block = block_position{current_request->start_block_num - 1, *prev_block_id};
               if (current_request->fetch_block) {
                  plugin->get_block( current_request->start_block_num, block_state, result.block );
               }
               if (current_request->fetch_traces && plugin->trace_log)
                  plugin->get_log_entry(*plugin->trace_log, current_request->start_block_num, result.traces);
               if (current_request->fetch_deltas && plugin->chain_state_log)
                  plugin->get_log_entry(*plugin->chain_state_log, current_request->start_block_num, result.deltas);
            }
            ++current_request->start_block_num;
         }
         send(std::move(result));
         --current_request->max_messages_in_flight;
         need_to_send_update = current_request->start_block_num <= current &&
                               current_request->start_block_num < current_request->end_block_num;
      }

Here is the call graph for this function:

Here is the caller graph for this function:

start()

void sysio::state_history_plugin_impl::session::start ( tcp::socket socket )

inline

Definition at line 112 of file state_history_plugin.cpp.

                                   {
         ilog("incoming connection");
         socket_stream = std::make_unique<ws::stream<tcp::socket>>(std::move(socket));
         socket_stream->binary(true);
         socket_stream->next_layer().set_option(boost::asio::ip::tcp::no_delay(true));
         socket_stream->next_layer().set_option(boost::asio::socket_base::send_buffer_size(1024 * 1024));
         socket_stream->next_layer().set_option(boost::asio::socket_base::receive_buffer_size(1024 * 1024));
         socket_stream->async_accept([self = shared_from_this()](boost::system::error_code ec) {
            self->callback(ec, "async_accept", [self] {
               self->start_read();
               self->send(state_history_plugin_abi);
            });
         });
      }

start_read()

void sysio::state_history_plugin_impl::session::start_read ( )

inline

Definition at line 127 of file state_history_plugin.cpp.

                        {
         auto in_buffer = std::make_shared<boost::beast::flat_buffer>();
         socket_stream->async_read(
             *in_buffer, [self = shared_from_this(), in_buffer](boost::system::error_code ec, size_t) {
                self->callback(ec, "async_read", [self, in_buffer] {
                   auto d = boost::asio::buffer_cast<char const*>(boost::beast::buffers_front(in_buffer->data()));
                   auto s = boost::asio::buffer_size(in_buffer->data());
                   fc::datastream<const char*> ds(d, s);
                   state_request               req;
                   fc::raw::unpack(ds, req);
                   std::visit(*self, req);
                   self->start_read();
                });
             });
      }

Here is the call graph for this function:

Member Data Documentation

current_request

std::optional<get_blocks_request_v0> sysio::state_history_plugin_impl::session::current_request

Definition at line 106 of file state_history_plugin.cpp.

need_to_send_update

bool sysio::state_history_plugin_impl::session::need_to_send_update = false

Definition at line 107 of file state_history_plugin.cpp.

plugin

std::shared_ptr<state_history_plugin_impl> sysio::state_history_plugin_impl::session::plugin

Definition at line 101 of file state_history_plugin.cpp.

send_queue

std::vector<std::vector<char> > sysio::state_history_plugin_impl::session::send_queue

Definition at line 105 of file state_history_plugin.cpp.

sending

bool sysio::state_history_plugin_impl::session::sending = false

Definition at line 103 of file state_history_plugin.cpp.

sent_abi

bool sysio::state_history_plugin_impl::session::sent_abi = false

Definition at line 104 of file state_history_plugin.cpp.

socket_stream

std::unique_ptr<ws::stream<tcp::socket> > sysio::state_history_plugin_impl::session::socket_stream

Definition at line 102 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