abstract_tcp_server2.inl

Go to the documentation of this file.

// Copyright (c) 2006-2013, Andrey N. Sabelnikov, www.sabelnikov.net
// All rights reserved.
// 
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
// * Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
// * Redistributions in binary form must reproduce the above copyright
// notice, this list of conditions and the following disclaimer in the
// documentation and/or other materials provided with the distribution.
// * Neither the name of the Andrey N. Sabelnikov nor the
// names of its contributors may be used to endorse or promote products
// derived from this software without specific prior written permission.
// 
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER  BE LIABLE FOR ANY
// DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
// (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
// LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
// ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
// 
 
 
#include <boost/asio/post.hpp>
#include <boost/foreach.hpp>
#include <boost/uuid/random_generator.hpp>
#include <boost/uuid/uuid_io.hpp>
#include <boost/chrono.hpp>
#include <boost/utility/value_init.hpp>
#include <boost/asio/bind_executor.hpp>
#include <boost/asio/deadline_timer.hpp>
#include <boost/date_time/posix_time/posix_time.hpp> // TODO
#include <boost/thread/condition_variable.hpp> // TODO
#include <boost/make_shared.hpp>
#include <boost/thread.hpp>
#include "warnings.h"
#include "string_tools_lexical.h"
#include "misc_language.h"
 
#include <sstream>
#include <iomanip>
#include <algorithm>
#include <functional>
#include <random>
 
#undef MONERO_DEFAULT_LOG_CATEGORY
#define MONERO_DEFAULT_LOG_CATEGORY "net"
 
#define AGGRESSIVE_TIMEOUT_THRESHOLD 120 // sockets
#define NEW_CONNECTION_TIMEOUT_LOCAL 1200000 // 2 minutes
#define NEW_CONNECTION_TIMEOUT_REMOTE 10000 // 10 seconds
#define DEFAULT_TIMEOUT_MS_LOCAL 1800000 // 30 minutes
#define DEFAULT_TIMEOUT_MS_REMOTE 300000 // 5 minutes
#define TIMEOUT_EXTRA_MS_PER_BYTE 0.2
 
 
namespace epee
{
namespace net_utils
{
  template<typename T>
  T& check_and_get(std::shared_ptr<T>& ptr)
  {
    CHECK_AND_ASSERT_THROW_MES(bool(ptr), "shared_state cannot be null");
    return *ptr;
  }
 
  /************************************************************************/
  /*                                                                      */
  /************************************************************************/
  template<typename T>
  unsigned int connection<T>::host_count(int delta)
  {
    static std::mutex hosts_mutex;
    std::lock_guard<std::mutex> guard(hosts_mutex);
    static std::map<std::string, unsigned int> hosts;
    unsigned int &val = hosts[m_host];
    if (delta > 0)
      MTRACE("New connection from host " << m_host << ": " << val);
    else if (delta < 0)
      MTRACE("Closed connection from host " << m_host << ": " << val);
    CHECK_AND_ASSERT_THROW_MES(delta >= 0 || val >= (unsigned)-delta, "Count would go negative");
    CHECK_AND_ASSERT_THROW_MES(delta <= 0 || val <= std::numeric_limits<unsigned int>::max() - (unsigned)delta, "Count would wrap");
    val += delta;
    return val;
  }
 
  template<typename T>
  typename connection<T>::duration_t connection<T>::get_default_timeout()
  {
    unsigned count{};
    try { count = host_count(); } catch (...) {}
    const unsigned shift = (
      connection_basic::get_state().sock_count > AGGRESSIVE_TIMEOUT_THRESHOLD ?
      std::min(std::max(count, 1u) - 1, 8u) :
      0
    );
    return (
      m_local ?
      std::chrono::milliseconds(DEFAULT_TIMEOUT_MS_LOCAL >> shift) :
      std::chrono::milliseconds(DEFAULT_TIMEOUT_MS_REMOTE >> shift)
    );
  }
 
  template<typename T>
  typename connection<T>::duration_t connection<T>::get_timeout_from_bytes_read(size_t bytes) const
  {
    return std::chrono::duration_cast<connection<T>::duration_t>(
      std::chrono::duration<double, std::chrono::milliseconds::period>(
        bytes * TIMEOUT_EXTRA_MS_PER_BYTE
      )
    );
  }
 
  template<typename T>
  void connection<T>::state_status_check()
  {
    switch (m_state.status)
    {
      case status_t::RUNNING:
        interrupt();
        break;
      case status_t::INTERRUPTED:
        on_interrupted();
        break;
      case status_t::TERMINATING:
        on_terminating();
        break;
      default:
        break;
    }
  }
 
  template<typename T>
  void connection<T>::start_timer(duration_t duration, bool add)
  {
    if (m_state.timers.general.wait_expire) {
      m_state.timers.general.cancel_expire = true;
      m_state.timers.general.reset_expire = true;
      m_timers.general.expires_after(
        std::min(
          duration + (add ? (m_timers.general.expiry() - std::chrono::steady_clock::now()) : duration_t{}),
          get_default_timeout()
        )
      );
    }
    else {
      m_timers.general.expires_after(
        std::min(
          duration + (add ? (m_timers.general.expiry() - std::chrono::steady_clock::now()) : duration_t{}),
          get_default_timeout()
        )
      );
      async_wait_timer();
    }
  }
 
  template<typename T>
  void connection<T>::async_wait_timer()
  {
    if (m_state.timers.general.wait_expire)
      return;
    m_state.timers.general.wait_expire = true;
    auto self = connection<T>::shared_from_this();
    auto on_wait = [this, self] {
      std::lock_guard<std::mutex> guard(m_state.lock);
      m_state.timers.general.wait_expire = false;
      if (m_state.timers.general.cancel_expire) {
        m_state.timers.general.cancel_expire = false;
        if (m_state.timers.general.reset_expire) {
          m_state.timers.general.reset_expire = false;
          async_wait_timer();
        }
        else if (m_state.status == status_t::INTERRUPTED)
          on_interrupted();
        else if (m_state.status == status_t::TERMINATING)
          on_terminating();
      }
      else if (m_state.status == status_t::RUNNING)
        interrupt();
      else if (m_state.status == status_t::INTERRUPTED)
        terminate();
    };
    m_timers.general.async_wait([this, self, on_wait](const ec_t & ec){
      boost::asio::post(m_strand, on_wait);
    });
  }
 
  template<typename T>
  void connection<T>::cancel_timer()
  {
    if (!m_state.timers.general.wait_expire)
      return;
    m_state.timers.general.cancel_expire = true;
    m_state.timers.general.reset_expire = false;
    m_timers.general.cancel();
  }
 
  template<typename T>
  void connection<T>::start_handshake()
  {
    if (m_state.socket.wait_handshake)
      return;
    static_assert(
      epee::net_utils::get_ssl_magic_size() <= sizeof(m_state.data.read.buffer),
      ""
    );
    auto self = connection<T>::shared_from_this();
    if (!m_state.ssl.forced && !m_state.ssl.detected) {
      m_state.socket.wait_read = true;
      boost::asio::async_read(
        connection_basic::socket_.next_layer(),
        boost::asio::buffer(
          m_state.data.read.buffer.data(),
          m_state.data.read.buffer.size()
        ),
        boost::asio::transfer_exactly(epee::net_utils::get_ssl_magic_size()),
        boost::asio::bind_executor(
          m_strand,
          [this, self](const ec_t &ec, size_t bytes_transferred){
            std::lock_guard<std::mutex> guard(m_state.lock);
            m_state.socket.wait_read = false;
            if (m_state.socket.cancel_read) {
              m_state.socket.cancel_read = false;
              state_status_check();
            }
            else if (ec.value()) {
              terminate();
            }
            else if (
              !epee::net_utils::is_ssl(
                static_cast<const unsigned char *>(
                  m_state.data.read.buffer.data()
                ),
                bytes_transferred
              )
            ) {
              m_state.ssl.enabled = false;
              handle_read(bytes_transferred);
            }
            else {
              m_state.ssl.detected = true;
              start_handshake();
            }
          }
        )
      );
      return;
    }
 
    m_state.socket.wait_handshake = true;
    auto on_handshake = [this, self](const ec_t &ec, size_t bytes_transferred){
      std::lock_guard<std::mutex> guard(m_state.lock);
      m_state.socket.wait_handshake = false;
      if (m_state.socket.cancel_handshake) {
        m_state.socket.cancel_handshake = false;
        state_status_check();
      }
      else if (ec.value()) {
        ec_t ec;
        connection_basic::socket_.next_layer().shutdown(
          socket_t::shutdown_both,
          ec
        );
        connection_basic::socket_.next_layer().close(ec);
        m_state.socket.connected = false;
        interrupt();
      }
      else {
        m_state.ssl.handshaked = true;
        start_write();
        start_read();
      }
    };
    const auto handshake = handshake_t::server;
    static_cast<shared_state&>(
      connection_basic::get_state()
    ).ssl_options().configure(connection_basic::socket_, handshake);
    boost::asio::post(
      m_strand,
      [this, self, on_handshake]{
        connection_basic::socket_.async_handshake(
          handshake,
          boost::asio::buffer(
            m_state.data.read.buffer.data(),
            m_state.ssl.forced ? 0 :
            epee::net_utils::get_ssl_magic_size()
          ),
          boost::asio::bind_executor(m_strand, on_handshake)
        );
      }
    );
  }
 
  template<typename T>
  void connection<T>::start_read()
  {
    if (m_state.timers.throttle.in.wait_expire || m_state.socket.wait_read ||
      m_state.socket.handle_read || m_state.socket.shutdown_read
    ) {
      return;
    }
    auto self = connection<T>::shared_from_this();
    if (speed_limit_is_enabled()) {
      auto calc_duration = []{
        CRITICAL_REGION_LOCAL(
          network_throttle_manager_t::m_lock_get_global_throttle_in
        );
        return std::chrono::duration_cast<connection<T>::duration_t>(
            std::chrono::duration<double, std::chrono::seconds::period>(
              std::min(
                network_throttle_manager_t::get_global_throttle_in(
                ).get_sleep_time_after_tick(1),
                1.0
              )
            )
        );
      };
      const auto duration = calc_duration();
      if (duration > duration_t{}) {
        m_timers.throttle.in.expires_after(duration);
        m_state.timers.throttle.in.wait_expire = true;
        auto on_wait = [this, self](const ec_t &ec){
          std::lock_guard<std::mutex> guard(m_state.lock);
          m_state.timers.throttle.in.wait_expire = false;
          if (m_state.timers.throttle.in.cancel_expire) {
            m_state.timers.throttle.in.cancel_expire = false;
            state_status_check();
          }
          else if (ec.value())
            interrupt();
        };
        m_timers.throttle.in.async_wait([this, self, on_wait](const ec_t &ec){
          std::lock_guard<std::mutex> guard(m_state.lock);
          const bool error_status = m_state.timers.throttle.in.cancel_expire || ec.value();
          if (error_status)
            boost::asio::post(m_strand, std::bind(on_wait, ec));
          else {
            m_state.timers.throttle.in.wait_expire = false;
            start_read();
          }
        });
        return;
      }
    }
    m_state.socket.wait_read = true;
    auto on_read = [this, self](const ec_t &ec, size_t bytes_transferred){
      std::lock_guard<std::mutex> guard(m_state.lock);
      m_state.socket.wait_read = false;
      if (m_state.socket.cancel_read) {
        m_state.socket.cancel_read = false;
        state_status_check();
      }
      else if (ec.value())
        terminate();
      else {
        {
          m_state.stat.in.throttle.handle_trafic_exact(bytes_transferred);
          const auto speed = m_state.stat.in.throttle.get_current_speed();
          m_conn_context.m_current_speed_down = speed;
          m_conn_context.m_max_speed_down = std::max(
            m_conn_context.m_max_speed_down,
            speed
          );
          if (speed_limit_is_enabled()) {
            CRITICAL_REGION_LOCAL(
              network_throttle_manager_t::m_lock_get_global_throttle_in
            );
            network_throttle_manager_t::get_global_throttle_in(
            ).handle_trafic_exact(bytes_transferred);
          }
          connection_basic::logger_handle_net_read(bytes_transferred);
          m_conn_context.m_last_recv = time(NULL);
          m_conn_context.m_recv_cnt += bytes_transferred;
          start_timer(get_timeout_from_bytes_read(bytes_transferred), true);
        }
        handle_read(bytes_transferred);
      }
    };
    if (!m_state.ssl.enabled)
      connection_basic::socket_.next_layer().async_read_some(
        boost::asio::buffer(
          m_state.data.read.buffer.data(),
          m_state.data.read.buffer.size()
        ),
        boost::asio::bind_executor(m_strand, on_read)
      );
    else
      boost::asio::post(
        m_strand,
        [this, self, on_read]{
          connection_basic::socket_.async_read_some(
            boost::asio::buffer(
              m_state.data.read.buffer.data(),
              m_state.data.read.buffer.size()
            ),
            boost::asio::bind_executor(m_strand, on_read)
          );
        }
      );
  }
 
  template<typename T>
  void connection<T>::handle_read(size_t bytes_transferred)
  {
    // Post handle_recv to a separate `strand_`, distinct from `m_strand`
    // which is listening for reads/writes. This avoids a circular dep.
    // handle_recv can queue many writes, and `m_strand` will process those
    // writes until the connection terminates without deadlocking waiting
    // for handle_recv.
    m_state.socket.handle_read = true;
    auto self = connection<T>::shared_from_this();
    boost::asio::post(
      connection_basic::strand_,
      [this, self, bytes_transferred]{
        bool success = false;
        TRY_ENTRY();
        success = m_handler.handle_recv(
          reinterpret_cast<char *>(m_state.data.read.buffer.data()),
          bytes_transferred
        );
        CATCH_ENTRY_SWALLOW_EX("m_handler.handle_recv");
        std::lock_guard<std::mutex> guard(m_state.lock);
        const bool error_status = m_state.status == status_t::INTERRUPTED
            || m_state.status == status_t::TERMINATING
            || !success;
        if (!error_status) {
          m_state.socket.handle_read = false;
          start_read();
          return;
        }
        boost::asio::post(
          m_strand,
          [this, self, success]{
            // expect error_status == true
            std::lock_guard<std::mutex> guard(m_state.lock);
            m_state.socket.handle_read = false;
            if (m_state.status == status_t::INTERRUPTED)
              on_interrupted();
            else if (m_state.status == status_t::TERMINATING)
              on_terminating();
            else if (!success) {
              ec_t ec;
              if (m_state.socket.wait_write) {
                // Allow the already queued writes time to finish, but no more new reads
                connection_basic::socket_.next_layer().shutdown(
                  socket_t::shutdown_receive,
                  ec
                );
                m_state.socket.shutdown_read = true;
              }
              if (!m_state.socket.wait_write || ec.value()) {
                interrupt();
              }
            }
          }
        );
      }
    );
  }
 
  template<typename T>
  void connection<T>::start_write()
  {
    if (m_state.timers.throttle.out.wait_expire || m_state.socket.wait_write ||
      m_state.data.write.queue.empty() ||
      (m_state.ssl.enabled && !m_state.ssl.handshaked)
    ) {
      return;
    }
    auto self = connection<T>::shared_from_this();
    if (speed_limit_is_enabled()) {
      auto calc_duration = [this]{
        CRITICAL_REGION_LOCAL(
          network_throttle_manager_t::m_lock_get_global_throttle_out
        );
        return std::chrono::duration_cast<connection<T>::duration_t>(
            std::chrono::duration<double, std::chrono::seconds::period>(
              std::min(
                network_throttle_manager_t::get_global_throttle_out(
                ).get_sleep_time_after_tick(
                  m_state.data.write.queue.back().size()
                ),
                1.0
              )
            )
        );
      };
      const auto duration = calc_duration();
      if (duration > duration_t{}) {
        m_timers.throttle.out.expires_after(duration);
        m_state.timers.throttle.out.wait_expire = true;
        auto on_wait = [this, self](const ec_t &ec){
          std::lock_guard<std::mutex> guard(m_state.lock);
          m_state.timers.throttle.out.wait_expire = false;
          if (m_state.timers.throttle.out.cancel_expire) {
            m_state.timers.throttle.out.cancel_expire = false;
            state_status_check();
          }
          else if (ec.value())
            interrupt();
        };
        m_timers.throttle.out.async_wait([this, self, on_wait](const ec_t &ec){
          std::lock_guard<std::mutex> guard(m_state.lock);
          const bool error_status = m_state.timers.throttle.out.cancel_expire || ec.value();
          if (error_status)
            boost::asio::post(m_strand, std::bind(on_wait, ec));
          else {
            m_state.timers.throttle.out.wait_expire = false;
            start_write();
          }
        });
        return;
      }
    }
 
    m_state.socket.wait_write = true;
    auto on_write = [this, self](const ec_t &ec, size_t bytes_transferred){
      std::lock_guard<std::mutex> guard(m_state.lock);
      m_state.socket.wait_write = false;
      if (m_state.socket.cancel_write) {
        m_state.socket.cancel_write = false;
        m_state.data.write.queue.clear();
        m_state.data.write.total_bytes = 0;
        state_status_check();
      }
      else if (ec.value()) {
        m_state.data.write.queue.clear();
        m_state.data.write.total_bytes = 0;
        interrupt();
      }
      else {
        {
          m_state.stat.out.throttle.handle_trafic_exact(bytes_transferred);
          const auto speed = m_state.stat.out.throttle.get_current_speed();
          m_conn_context.m_current_speed_up = speed;
          m_conn_context.m_max_speed_down = std::max(
            m_conn_context.m_max_speed_down,
            speed
          );
          if (speed_limit_is_enabled()) {
            CRITICAL_REGION_LOCAL(
              network_throttle_manager_t::m_lock_get_global_throttle_out
            );
            network_throttle_manager_t::get_global_throttle_out(
            ).handle_trafic_exact(bytes_transferred);
          }
          connection_basic::logger_handle_net_write(bytes_transferred);
          m_conn_context.m_last_send = time(NULL);
          m_conn_context.m_send_cnt += bytes_transferred;
 
          start_timer(get_default_timeout(), true);
        }
        const std::size_t byte_count = m_state.data.write.queue.back().size();
        assert(bytes_transferred == byte_count);
        m_state.data.write.queue.pop_back();
        m_state.data.write.total_bytes -=
          std::min(m_state.data.write.total_bytes, byte_count);
        m_state.condition.notify_all();
        if (m_state.data.write.queue.empty() && m_state.socket.shutdown_read) {
          // All writes have been sent and reads shutdown already, connection can be closed
          interrupt();
        } else {
          start_write();
        }
      }
    };
    if (!m_state.ssl.enabled)
      boost::asio::async_write(
        connection_basic::socket_.next_layer(),
        boost::asio::buffer(
          m_state.data.write.queue.back().data(),
          m_state.data.write.queue.back().size()
        ),
        boost::asio::bind_executor(m_strand, on_write)
      );
    else
      boost::asio::post(
        m_strand,
        [this, self, on_write]{
          boost::asio::async_write(
            connection_basic::socket_,
            boost::asio::buffer(
              m_state.data.write.queue.back().data(),
              m_state.data.write.queue.back().size()
            ),
            boost::asio::bind_executor(m_strand, on_write)
          );
        }
      );
  }
 
  template<typename T>
  void connection<T>::start_shutdown()
  {
    if (m_state.socket.wait_shutdown)
      return;
    auto self = connection<T>::shared_from_this();
    m_state.socket.wait_shutdown = true;
    auto on_shutdown = [this, self](const ec_t &ec){
      std::lock_guard<std::mutex> guard(m_state.lock);
      m_state.socket.wait_shutdown = false;
      if (m_state.socket.cancel_shutdown) {
        m_state.socket.cancel_shutdown = false;
        switch (m_state.status)
        {
          case status_t::RUNNING:
            interrupt();
            break;
          case status_t::INTERRUPTED:
            terminate();
            break;
          case status_t::TERMINATING:
            on_terminating();
            break;
          default:
            break;
        }
      }
      else {
        terminate();
      }
    };
    boost::asio::post(
      m_strand,
      [this, self, on_shutdown]{
        connection_basic::socket_.async_shutdown(
          boost::asio::bind_executor(m_strand, on_shutdown)
        );
      }
    );
    start_timer(get_default_timeout());
  }
 
  template<typename T>
  void connection<T>::cancel_socket()
  {
    bool wait_socket = false;
    if (m_state.socket.wait_handshake)
      wait_socket = m_state.socket.cancel_handshake = true;
    if (m_state.timers.throttle.in.wait_expire) {
      m_state.timers.throttle.in.cancel_expire = true;
      m_timers.throttle.in.cancel();
    }
    if (m_state.socket.wait_read)
      wait_socket = m_state.socket.cancel_read = true;
    if (m_state.timers.throttle.out.wait_expire) {
      m_state.timers.throttle.out.cancel_expire = true;
      m_timers.throttle.out.cancel();
    }
    if (m_state.socket.wait_write)
      wait_socket = m_state.socket.cancel_write = true;
    if (m_state.socket.wait_shutdown)
      wait_socket = m_state.socket.cancel_shutdown = true;
    if (wait_socket) {
      ec_t ec;
      connection_basic::socket_.next_layer().cancel(ec);
    }
  }
 
  template<typename T>
  void connection<T>::cancel_handler()
  {
    if (m_state.protocol.released || m_state.protocol.wait_release)
      return;
    m_state.protocol.wait_release = true;
    m_state.lock.unlock();
    m_handler.release_protocol();
    m_state.lock.lock();
    m_state.protocol.wait_release = false;
    m_state.protocol.released = true;
    if (m_state.status == status_t::INTERRUPTED)
      on_interrupted();
    else if (m_state.status == status_t::TERMINATING)
      on_terminating();
  }
 
  template<typename T>
  void connection<T>::interrupt()
  {
    if (m_state.status != status_t::RUNNING)
      return;
    m_state.status = status_t::INTERRUPTED;
    cancel_timer();
    cancel_socket();
    on_interrupted();
    m_state.condition.notify_all();
    cancel_handler();
  }
 
  template<typename T>
  void connection<T>::on_interrupted()
  {
    assert(m_state.status == status_t::INTERRUPTED);
    if (m_state.timers.general.wait_expire)
      return;
    if (m_state.socket.wait_handshake)
      return;
    if (m_state.timers.throttle.in.wait_expire)
      return;
    if (m_state.socket.wait_read)
      return;
    if (m_state.socket.handle_read)
      return;
    if (m_state.timers.throttle.out.wait_expire)
      return;
    // \NOTE See on_terminating() comments
    //if (m_state.socket.wait_write)
    // return;
    if (m_state.socket.wait_shutdown)
      return;
    if (m_state.protocol.wait_init)
      return;
    if (m_state.protocol.wait_callback)
      return;
    if (m_state.protocol.wait_release)
      return;
    if (m_state.socket.connected) {
      if (!m_state.ssl.enabled) {
        ec_t ec;
        connection_basic::socket_.next_layer().shutdown(
          socket_t::shutdown_both,
          ec
        );
        connection_basic::socket_.next_layer().close(ec);
        m_state.socket.connected = false;
        m_state.status = status_t::WASTED;
      }
      else
        start_shutdown();
    }
    else
      m_state.status = status_t::WASTED;
    m_state.condition.notify_all();
  }
 
  template<typename T>
  void connection<T>::terminate()
  {
    if (m_state.status != status_t::RUNNING &&
      m_state.status != status_t::INTERRUPTED
    )
      return;
    m_state.status = status_t::TERMINATING;
    cancel_timer();
    cancel_socket();
    on_terminating();
    m_state.condition.notify_all();
    cancel_handler();
  }
 
  template<typename T>
  void connection<T>::on_terminating()
  {
    assert(m_state.status == status_t::TERMINATING);
    if (m_state.timers.general.wait_expire)
      return;
    if (m_state.socket.wait_handshake)
      return;
    if (m_state.timers.throttle.in.wait_expire)
      return;
    if (m_state.socket.wait_read)
      return;
    if (m_state.socket.handle_read)
      return;
    if (m_state.timers.throttle.out.wait_expire)
      return;
    // Writes cannot be canceled due to `async_write` being a "composed"
    // handler. ASIO has new cancellation routines, not available in 1.66, to
    // handle this situation. The problem is that if cancel is called after an
    // intermediate handler is queued, the op will not check the cancel flag in
    // our code, and will instead queue up another write.
    //if (m_state.socket.wait_write)
    //  return;
    if (m_state.socket.wait_shutdown)
      return;
    if (m_state.protocol.wait_init)
      return;
    if (m_state.protocol.wait_callback)
      return;
    if (m_state.protocol.wait_release)
      return;
    if (m_state.socket.connected) {
      ec_t ec;
      connection_basic::socket_.next_layer().shutdown(
        socket_t::shutdown_both,
        ec
      );
      connection_basic::socket_.next_layer().close(ec);
      m_state.socket.connected = false;
    }
    m_state.status = status_t::WASTED;
    m_state.condition.notify_all();
  }
 
  template<typename T>
  void connection<T>::terminate_async()
  {
    // synchronize with intermediate writes on `m_strand`
    auto self = connection<T>::shared_from_this();
    boost::asio::post(m_strand, [this, self] {
      std::lock_guard<std::mutex> guard(m_state.lock);
      terminate();
    });
  }
 
  template<typename T>
  bool connection<T>::send(epee::byte_slice message)
  {
    std::lock_guard<std::mutex> guard(m_state.lock);
    if (m_state.status != status_t::RUNNING || m_state.socket.wait_handshake)
      return false;
    if (std::numeric_limits<std::size_t>::max() - m_state.data.write.total_bytes < message.size())
      return false;
 
    // Wait for the write queue to fall below the max. If it doesn't after a
    // randomized delay, drop the connection. P2P senders fail fast instead of
    // parking an io_context worker thread here.
    auto wait_consume = [this] {
      auto random_delay = []{
        using engine = std::mt19937;
        std::random_device dev;
        std::seed_seq::result_type rand[
          engine::state_size // Use complete bit space
        ]{};
        std::generate_n(rand, engine::state_size, std::ref(dev));
        std::seed_seq seed(rand, rand + engine::state_size);
        engine rng(seed);
        return std::chrono::milliseconds(
          std::uniform_int_distribution<>(5000, 6000)(rng)
        );
      };
 
      // The bytes check intentionally does not include incoming message size.
      // This allows for a soft overflow; a single http response will never fail
      // this check, but multiple responses could. Clients can avoid this case
      // by reading the entire response before making another request. P2P
      // should never hit the MAX_BYTES check (when using default values).
      if (m_state.data.write.queue.size() <= ABSTRACT_SERVER_SEND_QUE_MAX_COUNT &&
          m_state.data.write.total_bytes <= static_cast<shared_state&>(connection_basic::get_state()).response_soft_limit)
        return true;
 
      if (m_connection_type == e_connection_type_P2P) {
        MWARNING("Connection " << m_conn_context.m_connection_id << " tripped write limit, terminating");
        terminate_async();
        return false;
      }
      m_state.data.write.wait_consume = true;
      bool success = m_state.condition.wait_for(
        m_state.lock,
        random_delay(),
        [this]{
          return (
            m_state.status != status_t::RUNNING ||
            (
              m_state.data.write.queue.size() <=
                ABSTRACT_SERVER_SEND_QUE_MAX_COUNT &&
              m_state.data.write.total_bytes <=
                static_cast<shared_state&>(connection_basic::get_state()).response_soft_limit
            )
          );
        }
      );
      m_state.data.write.wait_consume = false;
      if (!success) {
        terminate_async();
        return false;
      }
      else
        return m_state.status == status_t::RUNNING;
    };
    auto wait_sender = [this] {
      m_state.condition.wait(
        m_state.lock,
        [this] {
          return (
            m_state.status != status_t::RUNNING ||
            !m_state.data.write.wait_consume
          );
        }
      );
      return m_state.status == status_t::RUNNING;
    };
    if (!wait_sender())
      return false;
    /* CHUNK_SIZE indirectly caps outgoing to 128 * 1024 * 1000
     (ABSTRACT_SERVER_SEND_QUE_MAX_COUNT). The "soft" limit total is currently
     100 MiB (ABSTRACT_SERVER_SEND_QUE_MAX_BYTES_DEFAULT). These values will
     need to be re-visited alongside block limit increases. */
    constexpr size_t CHUNK_SIZE = 128 * 1024;
    if (m_connection_type == e_connection_type_RPC ||
      message.size() <= 2 * CHUNK_SIZE
    ) {
      if (!wait_consume())
        return false;
      const std::size_t byte_count = message.size();
      m_state.data.write.queue.emplace_front(std::move(message));
      m_state.data.write.total_bytes += byte_count;
      start_write();
    }
    else {
      std::size_t soft_limit = 0;
      const epee::misc_utils::auto_scope_leave_caller scope_exit_handler =
        epee::misc_utils::create_scope_leave_handler([&soft_limit, this] {
          m_state.data.write.total_bytes += soft_limit;
        });
 
      while (!message.empty()) {
        if (!wait_consume())
          return false;
        m_state.data.write.queue.emplace_front(
          message.take_slice(CHUNK_SIZE)
        );
        soft_limit += m_state.data.write.queue.front().size();
        start_write();
      }
    }
    m_state.condition.notify_all();
    return true;
  }
 
  template<typename T>
  bool connection<T>::start_internal(
    bool is_income,
    bool is_multithreaded,
    boost::optional<network_address> real_remote
  )
  {
    std::unique_lock<std::mutex> guard(m_state.lock);
    if (m_state.status != status_t::TERMINATED)
      return false;
    if (!real_remote) {
      ec_t ec;
      auto endpoint = connection_basic::socket_.next_layer().remote_endpoint(
        ec
      );
      if (ec.value())
        return false;
      real_remote = (
        endpoint.address().is_v6() ?
        network_address{
          ipv6_network_address{endpoint.address().to_v6(), endpoint.port()}
        } :
        network_address{
          ipv4_network_address{
            uint32_t{
              boost::asio::detail::socket_ops::host_to_network_long(
                endpoint.address().to_v4().to_uint()
              )
            },
            endpoint.port()
          }
        }
      );
    }
    auto *filter = static_cast<shared_state&>(
      connection_basic::get_state()
    ).pfilter;
    if (filter && !filter->is_remote_host_allowed(*real_remote))
      return false;
 
    auto *limit = static_cast<shared_state&>(
      connection_basic::get_state()
    ).plimit;
    if (is_income && limit && limit->is_host_limit(*real_remote))
      return false;
 
    ec_t ec;
    #if !defined(_WIN32) || !defined(__i686)
    if (real_remote->get_type_id() == ipv4_network_address::get_type_id()) {
      connection_basic::socket_.next_layer().set_option(
        boost::asio::detail::socket_option::integer<IPPROTO_IP, IP_TOS>{
          connection_basic::get_tos_flag()
        },
        ec
      );
      if (ec.value())
        return false;
    }
    #endif
    connection_basic::socket_.next_layer().set_option(
      boost::asio::ip::tcp::no_delay{false},
      ec
    );
    if (ec.value())
      return false;
    connection_basic::m_is_multithreaded = is_multithreaded;
    m_conn_context.set_details(
      boost::uuids::random_generator()(),
      *real_remote,
      is_income,
      connection_basic::m_ssl_support == ssl_support_t::e_ssl_support_enabled
    );
    m_host = real_remote->host_str();
    try { host_count(1); } catch(...) { /* ignore */ }
    m_local = real_remote->is_loopback() || real_remote->is_local();
    m_state.ssl.enabled = (
      connection_basic::m_ssl_support != ssl_support_t::e_ssl_support_disabled
    );
    m_state.ssl.forced = (
      connection_basic::m_ssl_support == ssl_support_t::e_ssl_support_enabled
    );
    m_state.socket.connected = true;
    m_state.status = status_t::RUNNING;
    start_timer(
      std::chrono::milliseconds(
        m_local ? NEW_CONNECTION_TIMEOUT_LOCAL : NEW_CONNECTION_TIMEOUT_REMOTE
      )
    );
    m_state.protocol.wait_init = true;
    guard.unlock();
    m_handler.after_init_connection();
    guard.lock();
    m_state.protocol.wait_init = false;
    m_state.protocol.initialized = true;
    if (m_state.status == status_t::INTERRUPTED)
      on_interrupted();
    else if (m_state.status == status_t::TERMINATING)
      on_terminating();
    else if (!is_income || !m_state.ssl.enabled)
      start_read();
    else
      start_handshake();
    return true;
  }
 
  template<typename T>
  connection<T>::connection(
    io_context_t &io_context,
    std::shared_ptr<shared_state> shared_state,
    t_connection_type connection_type,
    ssl_support_t ssl_support,
    t_connection_context&& initial
  ):
    connection(
      io_context,
      socket_t{io_context},
      std::move(shared_state),
      connection_type,
      ssl_support,
      std::move(initial)
    )
  {
  }
 
  template<typename T>
  connection<T>::connection(
    io_context_t &io_context,
    socket_t &&socket,
    std::shared_ptr<shared_state> shared_state,
    t_connection_type connection_type,
    ssl_support_t ssl_support,
    t_connection_context&& initial
  ):
    connection_basic(io_context, std::move(socket), shared_state, ssl_support),
    m_handler(this, *shared_state, m_conn_context),
    m_connection_type(connection_type),
    m_io_context{io_context},
    m_conn_context(std::move(initial)),
    m_strand{m_io_context},
    m_timers{m_io_context}
  {
  }
 
  template<typename T>
  connection<T>::~connection() noexcept(false)
  {
    std::lock_guard<std::mutex> guard(m_state.lock);
    assert(m_state.status == status_t::TERMINATED ||
      m_state.status == status_t::WASTED ||
      m_io_context.stopped()
    );
    if (m_state.status != status_t::WASTED)
      return;
    try { host_count(-1); } catch (...) { /* ignore */ }
  }
 
  template<typename T>
  bool connection<T>::start(
    bool is_income,
    bool is_multithreaded
  )
  {
    return start_internal(is_income, is_multithreaded, {});
  }
 
  template<typename T>
  bool connection<T>::start(
    bool is_income,
    bool is_multithreaded,
    network_address real_remote
  )
  {
    return start_internal(is_income, is_multithreaded, real_remote);
  }
 
  template<typename T>
  void connection<T>::save_dbg_log()
  {
    std::lock_guard<std::mutex> guard(m_state.lock);
    std::string address;
    std::string port;
    ec_t ec;
    auto endpoint = connection_basic::socket().remote_endpoint(ec);
    if (ec.value()) {
      address = "<not connected>";
      port = "<not connected>";
    }
    else {
      address = endpoint.address().to_string();
      port = std::to_string(endpoint.port());
    }
    MDEBUG(
      " connection type " << std::to_string(m_connection_type) <<
      " " << connection_basic::socket().local_endpoint().address().to_string() <<
      ":" << connection_basic::socket().local_endpoint().port() <<
      " <--> " << m_conn_context.m_remote_address.str() <<
      " (via " << address << ":" << port << ")"
    );
  }
 
  template<typename T>
  bool connection<T>::speed_limit_is_enabled() const
  {
    return m_connection_type == e_connection_type_P2P;
  }
 
  template<typename T>
  bool connection<T>::cancel(const bool wait_for_shutdown)
  {
    return close(wait_for_shutdown);
  }
 
  template<typename T>
  bool connection<T>::do_send(byte_slice message)
  {
    return send(std::move(message));
  }
 
  template<typename T>
  bool connection<T>::send_done()
  {
    return true;
  }
 
  template<typename T>
  bool connection<T>::close(const bool wait_for_shutdown)
  {
    std::lock_guard<std::mutex> guard(m_state.lock);
    if (m_state.status == status_t::TERMINATED || m_state.status == status_t::WASTED)
      return true;
    if (!wait_for_shutdown && m_state.status != status_t::RUNNING)
      return false;
    terminate_async();
 
    // Sometimes we do *not* want to wait for the connection to shut down because for example handle_recv might try to
    // close the connection when handling a request. But handle_recv can't complete the shutdown sequence because
    // handle_read is set to true. So, in that case we call terminate_async and return here.
    if (!wait_for_shutdown)
      return true;
 
    // Sometimes we *do* want to wait for the connection to shut down for example when stopping the server. When
    // stopping the server, we don't want the io_context to stop before the shutdown sequence completes, since we
    // execute terminate inside m_strand. So we wait for the connection's shutdown sequence to complete before stopping
    // the io_context.
    MDEBUG("Waiting for connection " << m_conn_context.m_connection_id << " to shutdown, current state: " << m_state.status);
    const bool shutdown = m_state.condition.wait_for(
      m_state.lock,
      std::chrono::seconds(5),
      [this]{
        return (
          m_state.status == status_t::TERMINATED || m_state.status == status_t::WASTED
        );
      }
    );
    if (shutdown)
      MDEBUG("Shut down connection " << m_conn_context.m_connection_id);
    else
      MERROR("Connection " << m_conn_context.m_connection_id << " did not shut down");
 
    return shutdown;
  }
 
  template<typename T>
  bool connection<T>::call_run_once_service_io()
  {
    if(connection_basic::m_is_multithreaded) {
      if (!m_io_context.poll_one())
        misc_utils::sleep_no_w(1);
    }
    else {
      if (!m_io_context.run_one())
        return false;
    }
    return true;
  }
 
  template<typename T>
  bool connection<T>::request_callback()
  {
    std::lock_guard<std::mutex> guard(m_state.lock);
    if (m_state.status != status_t::RUNNING)
      return false;
    auto self = connection<T>::shared_from_this();
    ++m_state.protocol.wait_callback;
    boost::asio::post(connection_basic::strand_, [this, self]{
      TRY_ENTRY();
      m_handler.handle_qued_callback();
      CATCH_ENTRY_SWALLOW_EX("m_handler.handle_qued_callback");
      std::lock_guard<std::mutex> guard(m_state.lock);
      --m_state.protocol.wait_callback;
      if (m_state.status == status_t::INTERRUPTED)
        on_interrupted();
      else if (m_state.status == status_t::TERMINATING)
        on_terminating();
    });
    return true;
  }
 
  template<typename T>
  typename connection<T>::io_context_t &connection<T>::get_io_context()
  {
    return m_io_context;
  }
 
  template<typename T>
  bool connection<T>::add_ref()
  {
    auto self = connection<T>::weak_from_this().lock();
    if (!self)
      return false;
    std::lock_guard<std::mutex> guard(m_state.lock);
    this->self = std::move(self);
    ++m_state.protocol.reference_counter;
    return true;
  }
 
  template<typename T>
  bool connection<T>::release()
  {
    connection_ptr self;
    std::lock_guard<std::mutex> guard(m_state.lock);
    if (!(--m_state.protocol.reference_counter))
      self = std::move(this->self);
    return true;
  }
 
  template<typename T>
  void connection<T>::setRpcStation()
  {
    std::lock_guard<std::mutex> guard(m_state.lock);
    m_connection_type = e_connection_type_RPC;
  }
 
  template<class t_protocol_handler>
  boosted_tcp_server<t_protocol_handler>::boosted_tcp_server( t_connection_type connection_type ) :
    m_state(std::make_shared<typename connection<t_protocol_handler>::shared_state>()),
    m_io_context_local_instance(new worker()),
    io_context_(m_io_context_local_instance->io_context),
    acceptor_(io_context_),
    acceptor_ipv6(io_context_),
    default_remote(),
    m_stop_signal_sent(false), m_port(0), 
    m_threads_count(0),
    m_thread_index(0),
        m_connection_type( connection_type ),
    new_connection_(),
    new_connection_ipv6()
  {
    create_server_type_map();
    m_thread_name_prefix = "NET";
  }
 
  template<class t_protocol_handler>
  boosted_tcp_server<t_protocol_handler>::boosted_tcp_server(boost::asio::io_context& extarnal_io_context, t_connection_type connection_type) :
    m_state(std::make_shared<typename connection<t_protocol_handler>::shared_state>()),
    io_context_(extarnal_io_context),
    acceptor_(io_context_),
    acceptor_ipv6(io_context_),
    default_remote(),
    m_stop_signal_sent(false), m_port(0),
    m_threads_count(0),
    m_thread_index(0),
        m_connection_type(connection_type),
    new_connection_(),
    new_connection_ipv6()
  {
    create_server_type_map();
    m_thread_name_prefix = "NET";
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  boosted_tcp_server<t_protocol_handler>::~boosted_tcp_server()
  {
    send_stop_signal();
    timed_wait_server_stop(10000);
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  void boosted_tcp_server<t_protocol_handler>::create_server_type_map() 
  {
        server_type_map["NET"] = e_connection_type_NET;
        server_type_map["RPC"] = e_connection_type_RPC;
        server_type_map["P2P"] = e_connection_type_P2P;
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
    bool boosted_tcp_server<t_protocol_handler>::init_server(uint32_t port,  const std::string& address,
    uint32_t port_ipv6, const std::string& address_ipv6, bool use_ipv6, bool require_ipv4,
    ssl_options_t ssl_options)
  {
    TRY_ENTRY();
    m_stop_signal_sent = false;
    m_port = port;
    m_port_ipv6 = port_ipv6;
    m_address = address;
    m_address_ipv6 = address_ipv6;
    m_use_ipv6 = use_ipv6;
    m_require_ipv4 = require_ipv4;
 
    if (ssl_options)
      m_state->configure_ssl(std::move(ssl_options));
 
    std::string ipv4_failed = "";
    std::string ipv6_failed = "";
 
    boost::asio::ip::tcp::resolver resolver(io_context_);
 
    try
    {
      const auto results = resolver.resolve(
        address, boost::lexical_cast<std::string>(port), boost::asio::ip::tcp::resolver::canonical_name
      );
      acceptor_.open(results.begin()->endpoint().protocol());
#if !defined(_WIN32)
      acceptor_.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
#endif
      acceptor_.bind(*results.begin());
      acceptor_.listen();
      boost::asio::ip::tcp::endpoint binded_endpoint = acceptor_.local_endpoint();
      m_port = binded_endpoint.port();
      MDEBUG("start accept (IPv4)");
      new_connection_.reset(new connection<t_protocol_handler>(io_context_, m_state, m_connection_type, m_state->ssl_options().support));
      acceptor_.async_accept(new_connection_->socket(),
            boost::bind(&boosted_tcp_server<t_protocol_handler>::handle_accept_ipv4, this,
              boost::asio::placeholders::error));
    }
    catch (const std::exception &e)
    {
      ipv4_failed = e.what();
    }
 
    if (ipv4_failed != "")
    {
      MERROR("Failed to bind IPv4: " << ipv4_failed);
      if (require_ipv4)
      {
        throw std::runtime_error("Failed to bind IPv4 (set to required)");
      }
    }
 
    if (use_ipv6)
    {
      try
      {
        if (port_ipv6 == 0) port_ipv6 = port; // default arg means bind to same port as ipv4
 
        const auto results = resolver.resolve(
          address_ipv6, boost::lexical_cast<std::string>(port_ipv6), boost::asio::ip::tcp::resolver::canonical_name
        );
 
        acceptor_ipv6.open(results.begin()->endpoint().protocol());
#if !defined(_WIN32)
        acceptor_ipv6.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
#endif
        acceptor_ipv6.set_option(boost::asio::ip::v6_only(true));
        acceptor_ipv6.bind(*results.begin());
        acceptor_ipv6.listen();
        boost::asio::ip::tcp::endpoint binded_endpoint = acceptor_ipv6.local_endpoint();
        m_port_ipv6 = binded_endpoint.port();
        MDEBUG("start accept (IPv6)");
        new_connection_ipv6.reset(new connection<t_protocol_handler>(io_context_, m_state, m_connection_type, m_state->ssl_options().support));
        acceptor_ipv6.async_accept(new_connection_ipv6->socket(),
          boost::bind(&boosted_tcp_server<t_protocol_handler>::handle_accept_ipv6, this,
          boost::asio::placeholders::error));
      }
      catch (const std::exception &e)
      {
        ipv6_failed = e.what();
      }
    }
 
      if (use_ipv6 && ipv6_failed != "")
      {
        MERROR("Failed to bind IPv6: " << ipv6_failed);
        if (ipv4_failed != "")
        {
          throw std::runtime_error("Failed to bind IPv4 and IPv6");
        }
      }
 
    return true;
    }
    catch (const std::exception &e)
    {
      MFATAL("Error starting server: " << e.what());
      return false;
    }
    catch (...)
    {
      MFATAL("Error starting server");
      return false;
    }
  }
  //-----------------------------------------------------------------------------
  template<class t_protocol_handler>
  bool boosted_tcp_server<t_protocol_handler>::init_server(const std::string port,  const std::string& address,
      const std::string port_ipv6, const std::string address_ipv6, bool use_ipv6, bool require_ipv4,
      ssl_options_t ssl_options)
  {
    uint32_t p = 0;
    uint32_t p_ipv6 = 0;
 
    if (port.size() && !string_tools::get_xtype_from_string(p, port)) {
      MERROR("Failed to convert port no = " << port);
      return false;
    }
 
    if (port_ipv6.size() && !string_tools::get_xtype_from_string(p_ipv6, port_ipv6)) {
      MERROR("Failed to convert port no = " << port_ipv6);
      return false;
    }
    return this->init_server(p, address, p_ipv6, address_ipv6, use_ipv6, require_ipv4, std::move(ssl_options));
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  bool boosted_tcp_server<t_protocol_handler>::worker_thread()
  {
    TRY_ENTRY();
    const uint32_t local_thr_index = m_thread_index++; // atomically increment, getting value before increment
    std::string thread_name = std::string("[") + m_thread_name_prefix;
    thread_name += boost::to_string(local_thr_index) + "]";
    MLOG_SET_THREAD_NAME(thread_name);
    //   _fact("Thread name: " << m_thread_name_prefix);
    while(!m_stop_signal_sent)
    {
      try
      {
        io_context_.run();
        return true;
      }
      catch(const std::exception& ex)
      {
        _erro("Exception at server worker thread, what=" << ex.what());
      }
      catch(...)
      {
        _erro("Exception at server worker thread, unknown execption");
      }
    }
    //_info("Worker thread finished");
    return true;
    CATCH_ENTRY_L0("boosted_tcp_server<t_protocol_handler>::worker_thread", false);
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  void boosted_tcp_server<t_protocol_handler>::set_threads_prefix(const std::string& prefix_name)
  {
    m_thread_name_prefix = prefix_name;
        auto it = server_type_map.find(m_thread_name_prefix);
        if (it==server_type_map.end()) throw std::runtime_error("Unknown prefix/server type:" + std::string(prefix_name));
    auto connection_type = it->second; // the value of type
    MINFO("Set server type to: " << connection_type << " from name: " << m_thread_name_prefix << ", prefix_name = " << prefix_name);
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  void boosted_tcp_server<t_protocol_handler>::set_connection_filter(i_connection_filter* pfilter)
  {
    assert(m_state != nullptr); // always set in constructor
    m_state->pfilter = pfilter;
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  void boosted_tcp_server<t_protocol_handler>::set_connection_limit(i_connection_limit* plimit)
  {
    assert(m_state != nullptr); // always set in constructor
    m_state->plimit = plimit;
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  void boosted_tcp_server<t_protocol_handler>::set_response_soft_limit(const std::size_t limit)
  {
    assert(m_state != nullptr); // always set in constructor
    m_state->response_soft_limit = limit;
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  bool boosted_tcp_server<t_protocol_handler>::run_server(size_t threads_count, bool wait, const boost::thread::attributes& attrs)
  {
    TRY_ENTRY();
    m_threads_count = threads_count;
    m_main_thread_id = boost::this_thread::get_id();
    MLOG_SET_THREAD_NAME("[SRV_MAIN]");
    while(!m_stop_signal_sent)
    {
 
      // Create a pool of threads to run all of the io_contexts.
      CRITICAL_REGION_BEGIN(m_threads_lock);
      for (std::size_t i = 0; i < threads_count; ++i)
      {
        boost::shared_ptr<boost::thread> thread(new boost::thread(
          attrs, boost::bind(&boosted_tcp_server<t_protocol_handler>::worker_thread, this)));
          _note("Run server thread name: " << m_thread_name_prefix);
        m_threads.push_back(thread);
      }
      CRITICAL_REGION_END();
      // Wait for all threads in the pool to exit.
      if (wait)
      {
        _fact("JOINING all threads");
        for (std::size_t i = 0; i < m_threads.size(); ++i) {
            m_threads[i]->join();
         }
         _fact("JOINING all threads - almost");
        m_threads.clear();
        _fact("JOINING all threads - DONE");
 
      } 
      else {
        _dbg1("Reiniting OK.");
        return true;
      }
 
      if(wait && !m_stop_signal_sent)
      {
        //some problems with the listening socket ?..
        _dbg1("Net service stopped without stop request, restarting...");
        if(!this->init_server(m_port, m_address, m_port_ipv6, m_address_ipv6, m_use_ipv6, m_require_ipv4))
        {
          _dbg1("Reiniting service failed, exit.");
          return false;
        }else
        {
          _dbg1("Reiniting OK.");
        }
      }
    }
    return true;
    CATCH_ENTRY_L0("boosted_tcp_server<t_protocol_handler>::run_server", false);
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  bool boosted_tcp_server<t_protocol_handler>::is_thread_worker()
  {
    TRY_ENTRY();
    CRITICAL_REGION_LOCAL(m_threads_lock);
    BOOST_FOREACH(boost::shared_ptr<boost::thread>& thp,  m_threads)
    {
      if(thp->get_id() == boost::this_thread::get_id())
        return true;
    }
    if(m_threads_count == 1 && boost::this_thread::get_id() == m_main_thread_id)
      return true;
    return false;
    CATCH_ENTRY_L0("boosted_tcp_server<t_protocol_handler>::is_thread_worker", false);
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  bool boosted_tcp_server<t_protocol_handler>::timed_wait_server_stop(uint64_t wait_mseconds)
  {
    TRY_ENTRY();
    boost::chrono::milliseconds ms(wait_mseconds);
    for (std::size_t i = 0; i < m_threads.size(); ++i)
    {
      if(m_threads[i]->joinable() && !m_threads[i]->try_join_for(ms))
      {
        _dbg1("Interrupting thread " << m_threads[i]->native_handle());
        m_threads[i]->interrupt();
      }
    }
    return true;
    CATCH_ENTRY_L0("boosted_tcp_server<t_protocol_handler>::timed_wait_server_stop", false);
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  bool boosted_tcp_server<t_protocol_handler>::mark_stop_signal_sent()
  {
    if (m_stop_signal_sent.exchange(true))
    {
      MDEBUG("Stop signal already sent");
      return false;
    }
    typename connection<t_protocol_handler>::shared_state *state = static_cast<typename connection<t_protocol_handler>::shared_state*>(m_state.get());
    state->stop_signal_sent = true;
    return true;
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  void boosted_tcp_server<t_protocol_handler>::close_server_connections()
  {
    decltype(connections_) connections;
    {
      boost::unique_lock<boost::mutex> lock(connections_mutex);
      connections.swap(connections_);
    }
 
    for (auto &c: connections)
    {
      c->cancel(true/*wait_for_shutdown*/);
    }
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  void boosted_tcp_server<t_protocol_handler>::stop_io_context()
  {
    {
      boost::unique_lock<boost::mutex> lock(connections_mutex);
      if (!connections_.empty())
      {
        MERROR("Stopping io_context with " << connections_.size() << " server-owned connections still open");
      }
    }
    MDEBUG("Stopping io_context");
    io_context_.stop();
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  void boosted_tcp_server<t_protocol_handler>::send_stop_signal()
  {
    TRY_ENTRY();
    if (!mark_stop_signal_sent())
      return;
    close_server_connections();
    stop_io_context();
    CATCH_ENTRY_L0("boosted_tcp_server<t_protocol_handler>::send_stop_signal()", void());
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  void boosted_tcp_server<t_protocol_handler>::handle_accept_ipv4(const boost::system::error_code& e)
  {
    this->handle_accept(e, false);
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  void boosted_tcp_server<t_protocol_handler>::handle_accept_ipv6(const boost::system::error_code& e)
  {
    this->handle_accept(e, true);
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  void boosted_tcp_server<t_protocol_handler>::handle_accept(const boost::system::error_code& e, bool ipv6)
  {
    MDEBUG("handle_accept");
 
    boost::asio::ip::tcp::acceptor* current_acceptor = &acceptor_;
    connection_ptr* current_new_connection = &new_connection_;
    auto accept_function_pointer = &boosted_tcp_server<t_protocol_handler>::handle_accept_ipv4;
    if (ipv6)
    {
      current_acceptor = &acceptor_ipv6;
      current_new_connection = &new_connection_ipv6;
      accept_function_pointer = &boosted_tcp_server<t_protocol_handler>::handle_accept_ipv6;
    }
 
    bool accept_started = false;
    try
    {
    if (!e)
    {
      if (m_connection_type == e_connection_type_RPC) {
        const char *ssl_message = "unknown";
        switch ((*current_new_connection)->get_ssl_support())
        {
          case epee::net_utils::ssl_support_t::e_ssl_support_disabled: ssl_message = "disabled"; break;
          case epee::net_utils::ssl_support_t::e_ssl_support_enabled: ssl_message = "enabled"; break;
          case epee::net_utils::ssl_support_t::e_ssl_support_autodetect: ssl_message = "autodetection"; break;
        }
        MDEBUG("New server for RPC connections, SSL " << ssl_message);
        (*current_new_connection)->setRpcStation(); // hopefully this is not needed actually
      }
      connection_ptr conn(std::move((*current_new_connection)));
      (*current_new_connection).reset(new connection<t_protocol_handler>(io_context_, m_state, m_connection_type, conn->get_ssl_support()));
      current_acceptor->async_accept((*current_new_connection)->socket(),
          boost::bind(accept_function_pointer, this,
            boost::asio::placeholders::error));
      accept_started = true;
 
      boost::asio::socket_base::keep_alive opt(true);
      conn->socket().set_option(opt);
 
      bool res;
      if (default_remote.get_type_id() == net_utils::address_type::invalid)
        res = conn->start(true, 1 < m_threads_count);
      else
        res = conn->start(true, 1 < m_threads_count, default_remote);
      if (!res)
      {
        conn->cancel();
        return;
      }
      conn->save_dbg_log();
      return;
    }
    else
    {
      MERROR("Error in boosted_tcp_server<t_protocol_handler>::handle_accept: " << e);
    }
    }
    catch (const std::exception &e)
    {
      MERROR("Exception in boosted_tcp_server<t_protocol_handler>::handle_accept: " << e.what());
      if (accept_started)
        return;
    }
 
    // error path, if e or exception
    assert(m_state != nullptr); // always set in constructor
    _erro("Some problems at accept: " << e.message() << ", connections_count = " << m_state->sock_count);
    misc_utils::sleep_no_w(100);
    (*current_new_connection).reset(new connection<t_protocol_handler>(io_context_, m_state, m_connection_type, (*current_new_connection)->get_ssl_support()));
    current_acceptor->async_accept((*current_new_connection)->socket(),
        boost::bind(accept_function_pointer, this,
          boost::asio::placeholders::error));
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  bool boosted_tcp_server<t_protocol_handler>::add_connection(t_connection_context& out, boost::asio::ip::tcp::socket&& sock, network_address real_remote, epee::net_utils::ssl_support_t ssl_support)
  {
    if(std::addressof(get_io_context()) == std::addressof(sock.get_executor().context()))
    {
      connection_ptr conn(new connection<t_protocol_handler>(io_context_, std::move(sock), m_state, m_connection_type, ssl_support));
      if(conn->start(false, 1 < m_threads_count, std::move(real_remote)))
      {
        conn->get_context(out);
        conn->save_dbg_log();
        return true;
      }
    }
    else
    {
    MWARNING(out << " was not added, socket/io_context mismatch");
    }
    return false;
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  typename boosted_tcp_server<t_protocol_handler>::try_connect_result_t boosted_tcp_server<t_protocol_handler>::try_connect(connection_ptr new_connection_l, const std::string& adr, const std::string& port, boost::asio::ip::tcp::socket &sock_, const boost::asio::ip::tcp::endpoint &remote_endpoint, const std::string &bind_ip, uint32_t conn_timeout, epee::net_utils::ssl_support_t ssl_support)
  {
    TRY_ENTRY();
 
    sock_.open(remote_endpoint.protocol());
    if(bind_ip != "0.0.0.0" && bind_ip != "0" && bind_ip != "" )
    {
      boost::asio::ip::tcp::endpoint local_endpoint(boost::asio::ip::make_address(bind_ip), 0);
      boost::system::error_code ec;
      sock_.bind(local_endpoint, ec);
      if (ec)
      {
        MERROR("Error binding to " << bind_ip << ": " << ec.message());
        if (sock_.is_open())
          sock_.close();
        return CONNECT_FAILURE;
      }
    }
 
    /*
    NOTICE: be careful to make sync connection from event handler: in case if all threads suddenly do sync connect, there will be no thread to dispatch events from io service.
    */
 
    boost::system::error_code ec = boost::asio::error::would_block;
 
    //have another free thread(s), work in wait mode, without event handling
    struct local_async_context
    {
      boost::system::error_code ec;
      boost::mutex connect_mut;
      boost::condition_variable cond;
    };
 
    boost::shared_ptr<local_async_context> local_shared_context(new local_async_context());
    local_shared_context->ec = boost::asio::error::would_block;
    boost::unique_lock<boost::mutex> lock(local_shared_context->connect_mut);
    auto connect_callback = [](boost::system::error_code ec_, boost::shared_ptr<local_async_context> shared_context)
    {
      shared_context->connect_mut.lock(); shared_context->ec = ec_; shared_context->cond.notify_one(); shared_context->connect_mut.unlock();
    };
 
    sock_.async_connect(remote_endpoint, std::bind<void>(connect_callback, std::placeholders::_1, local_shared_context));
    while(local_shared_context->ec == boost::asio::error::would_block)
    {
      bool r = local_shared_context->cond.timed_wait(lock, boost::get_system_time() + boost::posix_time::milliseconds(conn_timeout));
      if (m_stop_signal_sent)
      {
        if (sock_.is_open())
          sock_.close();
        return CONNECT_FAILURE;
      }
      if(local_shared_context->ec == boost::asio::error::would_block && !r)
      {
        //timeout
        sock_.close();
        _dbg3("Failed to connect to " << adr << ":" << port << ", because of timeout (" << conn_timeout << ")");
        return CONNECT_FAILURE;
      }
    }
    ec = local_shared_context->ec;
 
    if (ec || !sock_.is_open())
    {
      _dbg3("Some problems at connect, message: " << ec.message());
      if (sock_.is_open())
        sock_.close();
      return CONNECT_FAILURE;
    }
 
    _dbg3("Connected success to " << adr << ':' << port);
 
    const ssl_support_t ssl_support = new_connection_l->get_ssl_support();
    if (ssl_support == epee::net_utils::ssl_support_t::e_ssl_support_enabled || ssl_support == epee::net_utils::ssl_support_t::e_ssl_support_autodetect)
    {
      // Handshake
      MDEBUG("Handshaking SSL...");
      if (!new_connection_l->handshake(boost::asio::ssl::stream_base::client))
      {
        if (ssl_support == epee::net_utils::ssl_support_t::e_ssl_support_autodetect)
        {
          boost::system::error_code ignored_ec;
          sock_.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ignored_ec);
          sock_.close();
          return CONNECT_NO_SSL;
        }
        MERROR("SSL handshake failed");
        if (sock_.is_open())
          sock_.close();
        return CONNECT_FAILURE;
      }
    }
 
    return CONNECT_SUCCESS;
 
    CATCH_ENTRY_L0("boosted_tcp_server<t_protocol_handler>::try_connect", CONNECT_FAILURE);
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler>
  bool boosted_tcp_server<t_protocol_handler>::connect(const std::string& adr, const std::string& port, uint32_t conn_timeout, t_connection_context& conn_context, const std::string& bind_ip, epee::net_utils::ssl_support_t ssl_support)
  {
    TRY_ENTRY();
 
    connection_ptr new_connection_l(new connection<t_protocol_handler>(io_context_, m_state, m_connection_type, ssl_support) );
    connections_mutex.lock();
    connections_.insert(new_connection_l);
    MDEBUG("connections_ size now " << connections_.size());
    connections_mutex.unlock();
    epee::misc_utils::auto_scope_leave_caller scope_exit_handler = epee::misc_utils::create_scope_leave_handler([&](){ CRITICAL_REGION_LOCAL(connections_mutex); connections_.erase(new_connection_l); });
    boost::asio::ip::tcp::socket&  sock_ = new_connection_l->socket();
 
    bool try_ipv6 = false;
 
    boost::asio::ip::tcp::resolver resolver(io_context_);
    boost::asio::ip::tcp::resolver::results_type results{};
    boost::system::error_code resolve_error;
 
    try
    {
      //resolving ipv4 address as ipv6 throws, catch here and move on
      results = resolver.resolve(
        boost::asio::ip::tcp::v4(), adr, port, boost::asio::ip::tcp::resolver::canonical_name, resolve_error
      );
    }
    catch (const boost::system::system_error& e)
    {
      if (!m_use_ipv6 || (resolve_error != boost::asio::error::host_not_found &&
            resolve_error != boost::asio::error::host_not_found_try_again))
      {
        throw;
      }
      try_ipv6 = true;
    }
    catch (...)
    {
      throw;
    }
 
    std::string bind_ip_to_use;
 
    if(results.empty())
    {
      if (!m_use_ipv6)
      {
        _erro("Failed to resolve " << adr);
        return false;
      }
      else
      {
        try_ipv6 = true;
        MINFO("Resolving address as IPv4 failed, trying IPv6");
      }
    }
    else
    {
      bind_ip_to_use = bind_ip;
    }
 
    if (try_ipv6)
    {
      results = resolver.resolve(
        boost::asio::ip::tcp::v6(), adr, port, boost::asio::ip::tcp::resolver::canonical_name, resolve_error
      );
 
      if(results.empty())
      {
        _erro("Failed to resolve " << adr);
        return false;
      }
      else
      {
        if (bind_ip == "0.0.0.0")
        {
          bind_ip_to_use = "::";
        }
        else
        {
          bind_ip_to_use = "";
        }
 
      }
 
    }
 
    const auto iterator = results.begin();
 
    MDEBUG("Trying to connect to " << adr << ":" << port << ", bind_ip = " << bind_ip_to_use);
 
    //boost::asio::ip::tcp::endpoint remote_endpoint(boost::asio::ip::address::from_string(addr.c_str()), port);
    boost::asio::ip::tcp::endpoint remote_endpoint(*iterator);
 
    auto try_connect_result = try_connect(new_connection_l, adr, port, sock_, remote_endpoint, bind_ip_to_use, conn_timeout, ssl_support);
    if (try_connect_result == CONNECT_FAILURE)
      return false;
    if (ssl_support == epee::net_utils::ssl_support_t::e_ssl_support_autodetect && try_connect_result == CONNECT_NO_SSL)
    {
      // we connected, but could not connect with SSL, try without
      MERROR("SSL handshake failed on an autodetect connection, reconnecting without SSL");
      new_connection_l->disable_ssl();
      try_connect_result = try_connect(new_connection_l, adr, port, sock_, remote_endpoint, bind_ip_to_use, conn_timeout, epee::net_utils::ssl_support_t::e_ssl_support_disabled);
      if (try_connect_result != CONNECT_SUCCESS)
        return false;
    }
 
    // start adds the connection to the config object's list, so we don't need to have it locally anymore
    connections_mutex.lock();
    connections_.erase(new_connection_l);
    connections_mutex.unlock();
    bool r = new_connection_l->start(false, 1 < m_threads_count);
    if (r)
    {
      new_connection_l->get_context(conn_context);
    }
    else
    {
      assert(m_state != nullptr); // always set in constructor
      _erro("[sock " << new_connection_l->socket().native_handle() << "] Failed to start connection, connections_count = " << m_state->sock_count);
    }
    
    new_connection_l->save_dbg_log();
 
    return r;
 
    CATCH_ENTRY_L0("boosted_tcp_server<t_protocol_handler>::connect", false);
  }
  //---------------------------------------------------------------------------------
  template<class t_protocol_handler> template<class t_callback>
  bool boosted_tcp_server<t_protocol_handler>::connect_async(const std::string& adr, const std::string& port, uint32_t conn_timeout, const t_callback &cb, const std::string& bind_ip, epee::net_utils::ssl_support_t ssl_support, t_connection_context&& initial)
  {
    TRY_ENTRY();    
    connection_ptr new_connection_l(new connection<t_protocol_handler>(io_context_, m_state, m_connection_type, ssl_support, std::move(initial)) );
    connections_mutex.lock();
    connections_.insert(new_connection_l);
    MDEBUG("connections_ size now " << connections_.size());
    connections_mutex.unlock();
    epee::misc_utils::auto_scope_leave_caller scope_exit_handler = epee::misc_utils::create_scope_leave_handler([&](){ CRITICAL_REGION_LOCAL(connections_mutex); connections_.erase(new_connection_l); });
    boost::asio::ip::tcp::socket&  sock_ = new_connection_l->socket();
    
    bool try_ipv6 = false;
 
    boost::asio::ip::tcp::resolver resolver(io_context_);
    boost::asio::ip::tcp::resolver::results_type results{};
    boost::system::error_code resolve_error;
 
    try
    {
      //resolving ipv4 address as ipv6 throws, catch here and move on
      results = resolver.resolve(
        boost::asio::ip::tcp::v4(), adr, port, boost::asio::ip::tcp::resolver::canonical_name, resolve_error
      );
    }
    catch (const boost::system::system_error& e)
    {
      if (!m_use_ipv6 || (resolve_error != boost::asio::error::host_not_found &&
            resolve_error != boost::asio::error::host_not_found_try_again))
      {
        throw;
      }
      try_ipv6 = true;
    }
    catch (...)
    {
      throw;
    }
 
    if(results.empty())
    {
      if (!try_ipv6)
      {
        _erro("Failed to resolve " << adr);
        return false;
      }
      else
      {
        MINFO("Resolving address as IPv4 failed, trying IPv6");
      }
    }
 
    if (try_ipv6)
    {
      results = resolver.resolve(
        boost::asio::ip::tcp::v6(), adr, port, boost::asio::ip::tcp::resolver::canonical_name, resolve_error
      );
 
      if(results.empty())
      {
        _erro("Failed to resolve " << adr);
        return false;
      }
    }
 
    boost::asio::ip::tcp::endpoint remote_endpoint(*results.begin());
     
    sock_.open(remote_endpoint.protocol());
    if(bind_ip != "0.0.0.0" && bind_ip != "0" && bind_ip != "" )
    {
      boost::asio::ip::tcp::endpoint local_endpoint(boost::asio::ip::make_address(bind_ip.c_str()), 0);
      boost::system::error_code ec;
      sock_.bind(local_endpoint, ec);
      if (ec)
      {
        MERROR("Error binding to " << bind_ip << ": " << ec.message());
        if (sock_.is_open())
          sock_.close();
        return false;
      }
    }
    
    boost::shared_ptr<boost::asio::deadline_timer> sh_deadline(new boost::asio::deadline_timer(io_context_));
    //start deadline
    sh_deadline->expires_from_now(boost::posix_time::milliseconds(conn_timeout));
    sh_deadline->async_wait([=](const boost::system::error_code& error)
      {
          if(error != boost::asio::error::operation_aborted) 
          {
            _dbg3("Failed to connect to " << adr << ':' << port << ", because of timeout (" << conn_timeout << ")");
            new_connection_l->socket().close();
          }
      });
    //start async connect
    sock_.async_connect(remote_endpoint, [=](const boost::system::error_code& ec_)
      {
        t_connection_context conn_context = AUTO_VAL_INIT(conn_context);
        boost::system::error_code ignored_ec;
        boost::asio::ip::tcp::socket::endpoint_type lep = new_connection_l->socket().local_endpoint(ignored_ec);
        if(!ec_)
        {//success
          if(!sh_deadline->cancel())
          {
            cb(conn_context, boost::asio::error::operation_aborted);//this mean that deadline timer already queued callback with cancel operation, rare situation
          }else
          {
            _dbg3("[sock " << new_connection_l->socket().native_handle() << "] Connected success to " << adr << ':' << port <<
              " from " << lep.address().to_string() << ':' << lep.port());
 
            // start adds the connection to the config object's list, so we don't need to have it locally anymore
            connections_mutex.lock();
            connections_.erase(new_connection_l);
            connections_mutex.unlock();
            bool r = new_connection_l->start(false, 1 < m_threads_count);
            if (r)
            {
              new_connection_l->get_context(conn_context);
              cb(conn_context, ec_);
            }
            else
            {
              _dbg3("[sock " << new_connection_l->socket().native_handle() << "] Failed to start connection to " << adr << ':' << port);
              cb(conn_context, boost::asio::error::fault);
            }
          }
        }else
        {
          _dbg3("[sock " << new_connection_l->socket().native_handle() << "] Failed to connect to " << adr << ':' << port <<
            " from " << lep.address().to_string() << ':' << lep.port() << ": " << ec_.message() << ':' << ec_.value());
          cb(conn_context, ec_);
        }
      });
    return true;
    CATCH_ENTRY_L0("boosted_tcp_server<t_protocol_handler>::connect_async", false);
  }
  
} // namespace
} // namespace