connection: fix implementation

CMakeLists.txt

@@ -1076,6 +1076,7 @@ if(STATIC)
   set(Boost_USE_STATIC_RUNTIME ON)
 endif()
 find_package(Boost 1.58 QUIET REQUIRED COMPONENTS system filesystem thread date_time chrono regex serialization program_options locale)
+add_definitions(-DBOOST_ASIO_ENABLE_SEQUENTIAL_STRAND_ALLOCATION)
 
 set(CMAKE_FIND_LIBRARY_SUFFIXES ${OLD_LIB_SUFFIXES})
 if(NOT Boost_FOUND)

contrib/epee/include/net/abstract_tcp_server2.h

@@ -44,12 +44,16 @@
 #include <cassert>
 #include <map>
 #include <memory>
+#include <condition_variable>
 
 #include <boost/asio.hpp>
 #include <boost/asio/ssl.hpp>
+#include <boost/asio/strand.hpp>
+#include <boost/asio/steady_timer.hpp>
 #include <boost/array.hpp>
 #include <boost/enable_shared_from_this.hpp>
 #include <boost/thread/thread.hpp>
+#include <boost/optional.hpp>
 #include "byte_slice.h"
 #include "net_utils_base.h"
 #include "syncobj.h"
@@ -85,6 +89,181 @@ namespace net_utils
     public i_service_endpoint,
     public connection_basic
   {
+  private:
+    using string_t = std::string;
+    using handler_t = t_protocol_handler;
+    using context_t = typename handler_t::connection_context;
+    using connection_t = connection<handler_t>;
+    using connection_ptr = boost::shared_ptr<connection_t>;
+    using ssl_support_t = epee::net_utils::ssl_support_t;
+    using timer_t = boost::asio::steady_timer;
+    using duration_t = timer_t::duration;
+    using lock_t = std::mutex;
+    using condition_t = std::condition_variable_any;
+    using lock_guard_t = std::lock_guard<lock_t>;
+    using unique_lock_t = std::unique_lock<lock_t>;
+    using byte_slice_t = epee::byte_slice;
+    using ec_t = boost::system::error_code;
+    using handshake_t = boost::asio::ssl::stream_base::handshake_type;
+
+    using io_context_t = boost::asio::io_service;
+    using strand_t = boost::asio::io_service::strand;
+    using socket_t = boost::asio::ip::tcp::socket;
+
+    using write_queue_t = std::deque<byte_slice_t>;
+    using read_buffer_t = std::array<uint8_t, 0x2000>;
+    using network_throttle_t = epee::net_utils::network_throttle;
+    using network_throttle_manager_t = epee::net_utils::network_throttle_manager;
+
+    unsigned int host_count(int delta = 0);
+    duration_t get_default_timeout();
+    duration_t get_timeout_from_bytes_read(size_t bytes) const;
+
+    void start_timer(duration_t duration, bool add = {});
+    void async_wait_timer();
+    void cancel_timer();
+
+    void start_handshake();
+    void start_read();
+    void start_write();
+    void start_shutdown();
+    void cancel_socket();
+
+    void cancel_handler();
+
+    void interrupt();
+    void on_interrupted();
+
+    void terminate();
+    void on_terminating();
+
+    bool send(byte_slice_t message);
+    bool start_internal(
+      bool is_income,
+      bool is_multithreaded,
+      boost::optional<network_address> real_remote
+    );
+
+    struct state_t {
+      struct stat_t {
+        struct {
+          network_throttle_t throttle{"speed_in", "throttle_speed_in"};
+        } in;
+        struct {
+          network_throttle_t throttle{"speed_out", "throttle_speed_out"};
+        } out;
+      };
+
+      struct data_t {
+        struct {
+          read_buffer_t buffer;
+        } read;
+        struct {
+          write_queue_t queue;
+          bool wait_consume;
+        } write;
+      };
+
+      struct ssl_t {
+        bool enabled;
+        bool forced;
+        bool detected;
+        bool handshaked;
+      };
+
+      struct socket_t {
+        bool connected;
+
+        bool wait_handshake;
+        bool cancel_handshake;
+
+        bool wait_read;
+        bool handle_read;
+        bool cancel_read;
+
+        bool wait_write;
+        bool handle_write;
+        bool cancel_write;
+
+        bool wait_shutdown;
+        bool cancel_shutdown;
+      };
+
+      struct timer_t {
+        bool wait_expire;
+        bool cancel_expire;
+        bool reset_expire;
+      };
+
+      struct timers_t {
+        struct throttle_t {
+          timer_t in;
+          timer_t out;
+        };
+
+        timer_t general;
+        throttle_t throttle;
+      };
+
+      enum status_t {
+        TERMINATED,
+        RUNNING,
+        INTERRUPTED,
+        TERMINATING,
+        WASTED,
+      };
+
+      struct protocol_t {
+        size_t reference_counter;
+        bool released;
+        bool initialized;
+
+        bool wait_release;
+        bool wait_init;
+        size_t wait_callback;
+      };
+
+      lock_t lock;
+      condition_t condition;
+      status_t status;
+      socket_t socket;
+      ssl_t ssl;
+      timers_t timers;
+      protocol_t protocol;
+      stat_t stat;
+      data_t data;
+    };
+
+    using status_t = typename state_t::status_t;
+
+    struct timers_t {
+      timers_t(io_context_t &io_context):
+        general(io_context),
+        throttle(io_context)
+      {}
+      struct throttle_t {
+        throttle_t(io_context_t &io_context):
+          in(io_context),
+          out(io_context)
+        {}
+        timer_t in;
+        timer_t out;
+      };
+
+      timer_t general;
+      throttle_t throttle;
+    };
+
+    io_context_t &io_context;
+    t_connection_type connection_type;
+    context_t context{};
+    strand_t strand;
+    timers_t timers;
+    connection_ptr self{};
+    bool local{};
+    string_t host{};
+    state_t state{};
+    handler_t handler;
   public:
     typedef typename t_protocol_handler::connection_context t_connection_context;
 
@@ -141,58 +320,6 @@ namespace net_utils
     virtual bool add_ref();
     virtual bool release();
     //------------------------------------------------------
-    bool do_send_chunk(byte_slice chunk); ///< will send (or queue) a part of data. internal use only
-
-    boost::shared_ptr<connection<t_protocol_handler> > safe_shared_from_this();
-    bool shutdown();
-    /// Handle completion of a receive operation.
-    void handle_receive(const boost::system::error_code& e,
-      std::size_t bytes_transferred);
-
-    /// Handle completion of a read operation.
-    void handle_read(const boost::system::error_code& e,
-      std::size_t bytes_transferred);
-
-    /// Handle completion of a write operation.
-    void handle_write(const boost::system::error_code& e, size_t cb);
-
-    /// reset connection timeout timer and callback
-    void reset_timer(boost::posix_time::milliseconds ms, bool add);
-    boost::posix_time::milliseconds get_default_timeout();
-    boost::posix_time::milliseconds get_timeout_from_bytes_read(size_t bytes);
-
-    /// host connection count tracking
-    unsigned int host_count(const std::string &host, int delta = 0);
-
-    /// Buffer for incoming data.
-    boost::array<char, 8192> buffer_;
-    size_t buffer_ssl_init_fill;
-
-    t_connection_context context;
-
-	// TODO what do they mean about wait on destructor?? --rfree :
-    //this should be the last one, because it could be wait on destructor, while other activities possible on other threads
-    t_protocol_handler m_protocol_handler;
-    //typename t_protocol_handler::config_type m_dummy_config;
-    size_t m_reference_count = 0; // reference count managed through add_ref/release support
-    boost::shared_ptr<connection<t_protocol_handler> > m_self_ref; // the reference to hold
-    critical_section m_self_refs_lock;
-    critical_section m_chunking_lock; // held while we add small chunks of the big do_send() to small do_send_chunk()
-    critical_section m_shutdown_lock; // held while shutting down
-    
-    t_connection_type m_connection_type;
-    
-    // for calculate speed (last 60 sec)
-    network_throttle m_throttle_speed_in;
-    network_throttle m_throttle_speed_out;
-    boost::mutex m_throttle_speed_in_mutex;
-    boost::mutex m_throttle_speed_out_mutex;
-
-    boost::asio::deadline_timer m_timer;
-    bool m_local;
-    bool m_ready_to_close;
-    std::string m_host;
-
 	public:
 			void setRpcStation();
   };

contrib/epee/include/net/net_ssl.h

@@ -110,6 +110,11 @@ namespace net_utils
     //! Search against internal fingerprints. Always false if `behavior() != user_certificate_check`.
     bool has_fingerprint(boost::asio::ssl::verify_context &ctx) const;
 
+    //! configure ssl_stream handshake verification
+    void configure(
+      boost::asio::ssl::stream<boost::asio::ip::tcp::socket> &socket,
+      boost::asio::ssl::stream_base::handshake_type type,
+      const std::string& host = {}) const;
     boost::asio::ssl::context create_context() const;
 
     /*! \note If `this->support == autodetect && this->verification != none`,

contrib/epee/src/net_ssl.cpp

@@ -32,6 +32,8 @@
 #include <boost/asio/ssl.hpp>
 #include <boost/cerrno.hpp>
 #include <boost/filesystem/operations.hpp>
+#include <boost/asio/strand.hpp>
+#include <condition_variable>
 #include <boost/lambda/lambda.hpp>
 #include <openssl/ssl.h>
 #include <openssl/pem.h>
@@ -488,12 +490,10 @@ bool ssl_options_t::has_fingerprint(boost::asio::ssl::verify_context &ctx) const
   return false;
 }
 
-bool ssl_options_t::handshake(
+void ssl_options_t::configure(
   boost::asio::ssl::stream<boost::asio::ip::tcp::socket> &socket,
   boost::asio::ssl::stream_base::handshake_type type,
-  boost::asio::const_buffer buffer,
+  const std::string& host) const
-  const std::string& host,
-  std::chrono::milliseconds timeout) const
 {
   socket.next_layer().set_option(boost::asio::ip::tcp::no_delay(true));
 
@@ -538,30 +538,101 @@ bool ssl_options_t::handshake(
       return true;
     });
   }
+}
 
-  auto& io_service = GET_IO_SERVICE(socket);
+bool ssl_options_t::handshake(
-  boost::asio::steady_timer deadline(io_service, timeout);
+  boost::asio::ssl::stream<boost::asio::ip::tcp::socket> &socket,
-  deadline.async_wait([&socket](const boost::system::error_code& error) {
+  boost::asio::ssl::stream_base::handshake_type type,
-    if (error != boost::asio::error::operation_aborted)
+  boost::asio::const_buffer buffer,
+  const std::string& host,
+  std::chrono::milliseconds timeout) const
 {
-      socket.next_layer().close();
+  configure(socket, type, host);
-    }
-  });
 
-  boost::system::error_code ec = boost::asio::error::would_block;
+  auto start_handshake = [&]{
-  socket.async_handshake(type, boost::asio::buffer(buffer), boost::lambda::var(ec) = boost::lambda::_1);
+    using ec_t = boost::system::error_code;
-  if (io_service.stopped())
+    using timer_t = boost::asio::steady_timer;
-  {
+    using strand_t = boost::asio::io_service::strand;
-    io_service.reset();
+    using lock_t = std::mutex;
+    using lock_guard_t = std::lock_guard<lock_t>;
+    using condition_t = std::condition_variable_any;
+    using socket_t = boost::asio::ip::tcp::socket;
+
+    auto &io_context = GET_IO_SERVICE(socket);
+    if (io_context.stopped())
+      io_context.reset();
+    strand_t strand(io_context);
+    timer_t deadline(io_context, timeout);
+
+    struct state_t {
+      lock_t lock;
+      condition_t condition;
+      ec_t result;
+      bool wait_timer;
+      bool wait_handshake;
+      bool cancel_timer;
+      bool cancel_handshake;
+    };
+    state_t state{};
+
+    state.wait_timer = true;
+    auto on_timer = [&](const ec_t &ec){
+      lock_guard_t guard(state.lock);
+      state.wait_timer = false;
+      state.condition.notify_all();
+      if (not state.cancel_timer) {
+        state.cancel_handshake = true;
+        ec_t ec;
+        socket.next_layer().cancel(ec);
       }
-  while (ec == boost::asio::error::would_block && !io_service.stopped())
+    };
-  {
+
-    // should poll_one(), can't run_one() because it can block if there is
+    state.wait_handshake = true;
-    // another worker thread executing io_service's tasks
+    auto on_handshake = [&](const ec_t &ec, size_t bytes_transferred){
-    // TODO: once we get Boost 1.66+, replace with run_one_for/run_until
+      lock_guard_t guard(state.lock);
-    std::this_thread::sleep_for(std::chrono::milliseconds(30));
+      state.wait_handshake = false;
-    io_service.poll_one();
+      state.condition.notify_all();
+      state.result = ec;
+      if (not state.cancel_handshake) {
+        state.cancel_timer = true;
+        ec_t ec;
+        deadline.cancel(ec);
       }
+    };
+
+    deadline.async_wait(on_timer);
+    strand.post(
+      [&]{
+        socket.async_handshake(
+          type,
+          boost::asio::buffer(buffer),
+          strand.wrap(on_handshake)
+        );
+      }
+    );
+
+    while (!io_context.stopped())
+    {
+      io_context.poll_one();
+      lock_guard_t guard(state.lock);
+      state.condition.wait_for(
+        state.lock,
+        std::chrono::milliseconds(30),
+        [&]{
+          return not state.wait_timer and not state.wait_handshake;
+        }
+      );
+      if (not state.wait_timer and not state.wait_handshake)
+        break;
+    }
+    if (state.result.value()) {
+      ec_t ec;
+      socket.next_layer().shutdown(socket_t::shutdown_both, ec);
+      socket.next_layer().close(ec);
+    }
+    return state.result;
+  };
+  const auto ec = start_handshake();
 
   if (ec)
   {