initial import of Monero base code

tests/unit_tests/epee_boosted_tcp_server.cpp

@@ -0,0 +1,717 @@
+// Copyright (c) 2014-2023, The Monero Project
+// 
+// All rights reserved.
+// 
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+// 
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+//    conditions and the following disclaimer.
+// 
+// 2. 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.
+// 
+// 3. Neither the name of the copyright holder 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 HOLDER OR CONTRIBUTORS 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.
+// 
+// Parts of this file are originally copyright (c) 2012-2013 The Cryptonote developers
+
+#include <boost/chrono/chrono.hpp>
+#include <boost/thread/condition_variable.hpp>
+#include <boost/thread/mutex.hpp>
+#include <condition_variable>
+#include <mutex>
+
+#include "gtest/gtest.h"
+
+#include "include_base_utils.h"
+#include "string_tools.h"
+#include "net/abstract_tcp_server2.h"
+#include "net/levin_protocol_handler_async.h"
+
+namespace
+{
+  const uint32_t test_server_port = 5626;
+  const std::string test_server_host("127.0.0.1");
+
+  struct test_connection_context : public epee::net_utils::connection_context_base
+  {
+  };
+
+  struct test_protocol_handler_config
+  {
+  };
+
+  struct test_protocol_handler
+  {
+    typedef test_connection_context connection_context;
+    typedef test_protocol_handler_config config_type;
+
+    test_protocol_handler(epee::net_utils::i_service_endpoint* /*psnd_hndlr*/, config_type& /*config*/, connection_context& /*conn_context*/)
+    {
+    }
+
+    void after_init_connection()
+    {
+    }
+
+    void handle_qued_callback()
+    {
+    }
+
+    bool release_protocol()
+    {
+      return true;
+    }
+
+    bool handle_recv(const void* /*data*/, size_t /*size*/)
+    {
+      return false;
+    }
+  };
+
+  typedef epee::net_utils::boosted_tcp_server<test_protocol_handler> test_tcp_server;
+}
+
+TEST(boosted_tcp_server, worker_threads_are_exception_resistant)
+{
+  test_tcp_server srv(epee::net_utils::e_connection_type_RPC); // RPC disables network limit for unit tests
+  ASSERT_TRUE(srv.init_server(test_server_port, test_server_host));
+
+  boost::mutex mtx;
+  boost::condition_variable cond;
+  int counter = 0;
+
+  auto counter_incrementer = [&counter, &cond, &mtx]()
+  {
+    boost::unique_lock<boost::mutex> lock(mtx);
+    ++counter;
+    if (4 <= counter)
+    {
+      cond.notify_one();
+    }
+  };
+
+  // 2 theads, but 4 exceptions
+  ASSERT_TRUE(srv.run_server(2, false));
+  ASSERT_TRUE(srv.async_call([&counter_incrementer]() { counter_incrementer(); throw std::runtime_error("test 1"); }));
+  ASSERT_TRUE(srv.async_call([&counter_incrementer]() { counter_incrementer(); throw std::string("test 2"); }));
+  ASSERT_TRUE(srv.async_call([&counter_incrementer]() { counter_incrementer(); throw "test 3"; }));
+  ASSERT_TRUE(srv.async_call([&counter_incrementer]() { counter_incrementer(); throw 4; }));
+
+  {
+    boost::unique_lock<boost::mutex> lock(mtx);
+    ASSERT_TRUE(cond.wait_for(lock, boost::chrono::seconds(5), [&counter]{ return counter == 4; }));
+  }
+
+  // Check if threads are alive
+  counter = 0;
+  //auto counter_incrementer = [&counter]() { counter.fetch_add(1); epee::misc_utils::sleep_no_w(counter.load() * 10); };
+  ASSERT_TRUE(srv.async_call(counter_incrementer));
+  ASSERT_TRUE(srv.async_call(counter_incrementer));
+  ASSERT_TRUE(srv.async_call(counter_incrementer));
+  ASSERT_TRUE(srv.async_call(counter_incrementer));
+
+  {
+    boost::unique_lock<boost::mutex> lock(mtx);
+    ASSERT_TRUE(cond.wait_for(lock, boost::chrono::seconds(5), [&counter]{ return counter == 4; }));
+  }
+
+  srv.send_stop_signal();
+  ASSERT_TRUE(srv.timed_wait_server_stop(5 * 1000));
+  ASSERT_TRUE(srv.deinit_server());
+}
+
+
+TEST(test_epee_connection, test_lifetime)
+{
+  struct context_t: epee::net_utils::connection_context_base {
+    static constexpr size_t get_max_bytes(int) noexcept { return -1; }
+    static constexpr int handshake_command() noexcept { return 1001; }
+    static constexpr bool handshake_complete() noexcept { return true; }
+  };
+
+  using functional_obj_t = std::function<void ()>;
+  struct command_handler_t: epee::levin::levin_commands_handler<context_t> {
+    size_t delay;
+    functional_obj_t on_connection_close_f;
+    command_handler_t(size_t delay = 0,
+      functional_obj_t on_connection_close_f = nullptr
+    ):
+      delay(delay),
+      on_connection_close_f(on_connection_close_f)
+    {}
+    virtual int invoke(int, const epee::span<const uint8_t>, epee::byte_stream&, context_t&) override { epee::misc_utils::sleep_no_w(delay); return {}; }
+    virtual int notify(int, const epee::span<const uint8_t>, context_t&) override { return {}; }
+    virtual void callback(context_t&) override {}
+    virtual void on_connection_new(context_t&) override {}
+    virtual void on_connection_close(context_t&) override {
+      if (on_connection_close_f)
+        on_connection_close_f();
+    }
+    virtual ~command_handler_t() override {}
+    static void destroy(epee::levin::levin_commands_handler<context_t>* ptr) { delete ptr; }
+  };
+
+  using handler_t = epee::levin::async_protocol_handler<context_t>;
+  using connection_t = epee::net_utils::connection<handler_t>;
+  using connection_ptr = boost::shared_ptr<connection_t>;
+  using shared_state_t = typename connection_t::shared_state;
+  using shared_state_ptr = std::shared_ptr<shared_state_t>;
+  using shared_states_t = std::vector<shared_state_ptr>;
+  using tag_t = boost::uuids::uuid;
+  using tags_t = std::vector<tag_t>;
+  using io_context_t = boost::asio::io_service;
+  using endpoint_t = boost::asio::ip::tcp::endpoint;
+  using work_t = boost::asio::io_service::work;
+  using work_ptr = std::shared_ptr<work_t>;
+  using workers_t = std::vector<std::thread>;
+  using server_t = epee::net_utils::boosted_tcp_server<handler_t>;
+  using lock_t = std::mutex;
+  using lock_guard_t = std::lock_guard<lock_t>;
+  using connection_weak_ptr = boost::weak_ptr<connection_t>;
+  struct shared_conn_t {
+    lock_t lock;
+    connection_weak_ptr conn;
+  };
+  using shared_conn_ptr = std::shared_ptr<shared_conn_t>;
+
+  io_context_t io_context;
+  work_ptr work(std::make_shared<work_t>(io_context));
+
+  workers_t workers;
+  while (workers.size() < 4) {
+    workers.emplace_back([&io_context]{
+      io_context.run();
+    });
+  }
+
+  endpoint_t endpoint(boost::asio::ip::address::from_string("127.0.0.1"), 5262);
+  server_t server(epee::net_utils::e_connection_type_P2P);
+  server.init_server(endpoint.port(),
+    endpoint.address().to_string(),
+    0,
+    "",
+    false,
+    true,
+    epee::net_utils::ssl_support_t::e_ssl_support_disabled
+  );
+  server.run_server(2, false);
+  server.get_config_shared()->set_handler(new command_handler_t, &command_handler_t::destroy);
+
+  io_context.post([&io_context, &work, &endpoint, &server]{
+    shared_state_ptr shared_state;
+    auto scope_exit_handler = epee::misc_utils::create_scope_leave_handler([&work, &shared_state]{
+      work.reset();
+      if (shared_state)
+        shared_state->set_handler(nullptr, nullptr);
+    });
+
+    shared_state = std::make_shared<shared_state_t>();
+    shared_state->set_handler(new command_handler_t, &command_handler_t::destroy);
+
+    auto create_connection = [&io_context, &endpoint, &shared_state] {
+        connection_ptr conn(new connection_t(io_context, shared_state, {}, {}));
+        conn->socket().connect(endpoint);
+        conn->start({}, {});
+        context_t context;
+        conn->get_context(context);
+        auto tag = context.m_connection_id;
+        return tag;
+    };
+
+    ASSERT_TRUE(shared_state->get_connections_count() == 0);
+    auto tag = create_connection();
+    ASSERT_TRUE(shared_state->get_connections_count() == 1);
+    bool success = shared_state->for_connection(tag, [shared_state](context_t& context){
+      shared_state->close(context.m_connection_id);
+      context.m_remote_address.get_zone();
+      return true;
+    });
+    ASSERT_TRUE(success);
+
+    ASSERT_TRUE(shared_state->get_connections_count() == 0);
+    constexpr auto N = 8;
+    tags_t tags(N);
+    for(auto &t: tags)
+      t = create_connection();
+    ASSERT_TRUE(shared_state->get_connections_count() == N);
+    size_t index = 0;
+    success = shared_state->foreach_connection([&index, shared_state, &tags, &create_connection](context_t& context){
+      if (!index)
+        for (const auto &t: tags)
+          shared_state->close(t);
+
+      shared_state->close(context.m_connection_id);
+      context.m_remote_address.get_zone();
+      ++index;
+
+      for(auto i = 0; i < N; ++i)
+        create_connection();
+      return true;
+    });
+    ASSERT_TRUE(success);
+    ASSERT_TRUE(index == N);
+    ASSERT_TRUE(shared_state->get_connections_count() == N * N);
+
+    index = 0;
+    success = shared_state->foreach_connection([&index, shared_state](context_t& context){
+      shared_state->close(context.m_connection_id);
+      context.m_remote_address.get_zone();
+      ++index;
+      return true;
+    });
+    ASSERT_TRUE(success);
+    ASSERT_TRUE(index == N * N);
+    ASSERT_TRUE(shared_state->get_connections_count() == 0);
+
+    while (shared_state->sock_count);
+    ASSERT_TRUE(shared_state->get_connections_count() == 0);
+    constexpr auto DELAY = 30;
+    constexpr auto TIMEOUT = 1;
+    while (server.get_connections_count()) {
+      server.get_config_shared()->del_in_connections(
+        server.get_config_shared()->get_in_connections_count()
+      );
+    }
+    server.get_config_shared()->set_handler(new command_handler_t(DELAY), &command_handler_t::destroy);
+    for (auto i = 0; i < N; ++i) {
+      tag = create_connection();
+      ASSERT_TRUE(shared_state->get_connections_count() == 1);
+      success = shared_state->invoke_async(1, epee::levin::message_writer{}, tag, [](int, const epee::span<const uint8_t>, context_t&){}, TIMEOUT);
+      ASSERT_TRUE(success);
+      while (shared_state->sock_count == 1) {
+        success = shared_state->foreach_connection([&shared_state, &tag](context_t&){
+          return shared_state->request_callback(tag);
+        });
+        ASSERT_TRUE(success);
+      }
+      shared_state->close(tag);
+      ASSERT_TRUE(shared_state->get_connections_count() == 0);
+    }
+
+    while (shared_state->sock_count);
+    constexpr auto ZERO_DELAY = 0;
+    size_t counter = 0;
+    shared_state->set_handler(new command_handler_t(ZERO_DELAY,
+        [&counter]{
+          ASSERT_TRUE(counter++ == 0);
+        }
+      ),
+      &command_handler_t::destroy
+    );
+    connection_ptr conn(new connection_t(io_context, shared_state, {}, {}));
+    conn->socket().connect(endpoint);
+    conn->start({}, {});
+    ASSERT_TRUE(shared_state->get_connections_count() == 1);
+    shared_state->del_out_connections(1);
+    ASSERT_TRUE(shared_state->get_connections_count() == 0);
+    conn.reset();
+
+    while (shared_state->sock_count);
+    shared_conn_ptr shared_conn(std::make_shared<shared_conn_t>());
+    shared_state->set_handler(new command_handler_t(ZERO_DELAY,
+        [shared_state, shared_conn]{
+          {
+            connection_ptr conn;
+            {
+              lock_guard_t guard(shared_conn->lock);
+              conn = shared_conn->conn.lock();
+              shared_conn->conn.reset();
+            }
+            if (conn)
+              conn->cancel();
+          }
+          const auto success = shared_state->foreach_connection([](context_t&){
+            return true;
+          });
+          ASSERT_TRUE(success);
+        }
+      ),
+      &command_handler_t::destroy
+    );
+    for (auto i = 0; i < N * N * N; ++i) {
+      {
+        connection_ptr conn(new connection_t(io_context, shared_state, {}, {}));
+        conn->socket().connect(endpoint);
+        conn->start({}, {});
+        lock_guard_t guard(shared_conn->lock);
+        shared_conn->conn = conn;
+      }
+      ASSERT_TRUE(shared_state->get_connections_count() == 1);
+      shared_state->del_out_connections(1);
+      while (shared_state->sock_count);
+      ASSERT_TRUE(shared_state->get_connections_count() == 0);
+    }
+
+    shared_states_t shared_states;
+    while (shared_states.size() < 2) {
+      shared_states.emplace_back(std::make_shared<shared_state_t>());
+      shared_states.back()->set_handler(new command_handler_t(ZERO_DELAY,
+          [&shared_states]{
+            for (auto &s: shared_states) {
+              auto success = s->foreach_connection([](context_t&){
+                return true;
+              });
+              ASSERT_TRUE(success);
+            }
+          }
+        ),
+        &command_handler_t::destroy
+      );
+    }
+    workers_t workers;
+
+    for (auto &s: shared_states) {
+      workers.emplace_back([&io_context, &s, &endpoint]{
+        for (auto i = 0; i < N * N; ++i) {
+          connection_ptr conn(new connection_t(io_context, s, {}, {}));
+          conn->socket().connect(endpoint);
+          conn->start({}, {});
+          io_context.post([conn]{
+            conn->cancel();
+          });
+          conn.reset();
+          s->del_out_connections(1);
+          while (s->sock_count);
+        }
+      });
+    }
+    for (;workers.size(); workers.pop_back())
+      workers.back().join();
+
+    for (auto &s: shared_states) {
+      workers.emplace_back([&io_context, &s, &endpoint]{
+        for (auto i = 0; i < N * N; ++i) {
+          connection_ptr conn(new connection_t(io_context, s, {}, {}));
+          conn->socket().connect(endpoint);
+          conn->start({}, {});
+          conn->cancel();
+          while (conn.use_count() > 1);
+          s->foreach_connection([&io_context, &s, &endpoint, &conn](context_t& context){
+            conn.reset(new connection_t(io_context, s, {}, {}));
+            conn->socket().connect(endpoint);
+            conn->start({}, {});
+            conn->cancel();
+            while (conn.use_count() > 1);
+            conn.reset();
+            return true;
+          });
+          while (s->sock_count);
+        }
+      });
+    }
+    for (;workers.size(); workers.pop_back())
+      workers.back().join();
+
+    for (auto &s: shared_states) {
+      workers.emplace_back([&io_context, &s, &endpoint]{
+        for (auto i = 0; i < N; ++i) {
+          connection_ptr conn(new connection_t(io_context, s, {}, {}));
+          conn->socket().connect(endpoint);
+          conn->start({}, {});
+          context_t context;
+          conn->get_context(context);
+          auto tag = context.m_connection_id;
+          conn->cancel();
+          while (conn.use_count() > 1);
+          s->for_connection(tag, [&io_context, &s, &endpoint, &conn](context_t& context){
+            conn.reset(new connection_t(io_context, s, {}, {}));
+            conn->socket().connect(endpoint);
+            conn->start({}, {});
+            conn->cancel();
+            while (conn.use_count() > 1);
+            conn.reset();
+            return true;
+          });
+          while (s->sock_count);
+        }
+      });
+    }
+    for (;workers.size(); workers.pop_back())
+      workers.back().join();
+
+    for (auto &s: shared_states) {
+      workers.emplace_back([&io_context, &s, &endpoint]{
+        for (auto i = 0; i < N; ++i) {
+          connection_ptr conn(new connection_t(io_context, s, {}, {}));
+          conn->socket().connect(endpoint);
+          conn->start({}, {});
+          context_t context;
+          conn->get_context(context);
+          auto tag = context.m_connection_id;
+          io_context.post([conn]{
+            conn->cancel();
+          });
+          conn.reset();
+          s->close(tag);
+          while (s->sock_count);
+        }
+      });
+    }
+    for (;workers.size(); workers.pop_back())
+      workers.back().join();
+    while (server.get_connections_count()) {
+      server.get_config_shared()->del_in_connections(
+        server.get_config_shared()->get_in_connections_count()
+      );
+    }
+  });
+
+  for (auto& w: workers) {
+    w.join();
+  }
+  server.send_stop_signal();
+  server.timed_wait_server_stop(5 * 1000);
+  server.deinit_server();
+}
+
+TEST(test_epee_connection, ssl_shutdown)
+{
+  struct context_t: epee::net_utils::connection_context_base {
+    static constexpr size_t get_max_bytes(int) noexcept { return -1; }
+    static constexpr int handshake_command() noexcept { return 1001; }
+    static constexpr bool handshake_complete() noexcept { return true; }
+  };
+
+  struct command_handler_t: epee::levin::levin_commands_handler<context_t> {
+    virtual int invoke(int, const epee::span<const uint8_t>, epee::byte_stream&, context_t&) override { return {}; }
+    virtual int notify(int, const epee::span<const uint8_t>, context_t&) override { return {}; }
+    virtual void callback(context_t&) override {}
+    virtual void on_connection_new(context_t&) override {}
+    virtual void on_connection_close(context_t&) override { }
+    virtual ~command_handler_t() override {}
+    static void destroy(epee::levin::levin_commands_handler<context_t>* ptr) { delete ptr; }
+  };
+
+  using handler_t = epee::levin::async_protocol_handler<context_t>;
+  using io_context_t = boost::asio::io_service;
+  using endpoint_t = boost::asio::ip::tcp::endpoint;
+  using server_t = epee::net_utils::boosted_tcp_server<handler_t>;
+  using socket_t = boost::asio::ip::tcp::socket;
+  using ssl_socket_t = boost::asio::ssl::stream<socket_t>;
+  using ssl_context_t = boost::asio::ssl::context;
+  using ec_t = boost::system::error_code;
+
+  endpoint_t endpoint(boost::asio::ip::address::from_string("127.0.0.1"), 5263);
+  server_t server(epee::net_utils::e_connection_type_P2P);
+  server.init_server(endpoint.port(),
+    endpoint.address().to_string(),
+    0,
+    "",
+    false,
+    true,
+    epee::net_utils::ssl_support_t::e_ssl_support_enabled
+  );
+  server.get_config_shared()->set_handler(new command_handler_t, &command_handler_t::destroy);
+  server.run_server(2, false);
+
+  ssl_context_t ssl_context{boost::asio::ssl::context::sslv23};
+  io_context_t io_context;
+  ssl_socket_t socket(io_context, ssl_context);
+  ec_t ec;
+  socket.next_layer().connect(endpoint, ec);
+  EXPECT_EQ(ec.value(), 0);
+  socket.handshake(boost::asio::ssl::stream_base::client, ec);
+  EXPECT_EQ(ec.value(), 0);
+  std::this_thread::sleep_for(std::chrono::milliseconds(100));
+  while (server.get_config_shared()->get_connections_count() < 1);
+  server.get_config_shared()->del_in_connections(1);
+  while (server.get_config_shared()->get_connections_count() > 0);
+  server.send_stop_signal();
+  EXPECT_TRUE(server.timed_wait_server_stop(5 * 1000));
+  server.deinit_server();
+  socket.next_layer().shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
+  socket.next_layer().close(ec);
+  socket.shutdown(ec);
+}
+
+TEST(test_epee_connection, ssl_handshake)
+{
+  using io_context_t = boost::asio::io_service;
+  using work_t = boost::asio::io_service::work;
+  using work_ptr = std::shared_ptr<work_t>;
+  using workers_t = std::vector<std::thread>;
+  using socket_t = boost::asio::ip::tcp::socket;
+  using ssl_socket_t = boost::asio::ssl::stream<socket_t>;
+  using ssl_socket_ptr = std::unique_ptr<ssl_socket_t>;
+  using ssl_options_t = epee::net_utils::ssl_options_t;
+  io_context_t io_context;
+  work_ptr work(std::make_shared<work_t>(io_context));
+  workers_t workers;
+  auto constexpr N = 2;
+  while (workers.size() < N) {
+    workers.emplace_back([&io_context]{
+      io_context.run();
+    });
+  }
+  ssl_options_t ssl_options{{}};
+  auto ssl_context = ssl_options.create_context();
+  for (size_t i = 0; i < N * N * N; ++i) {
+    ssl_socket_ptr ssl_socket(new ssl_socket_t(io_context, ssl_context));
+    ssl_socket->next_layer().open(boost::asio::ip::tcp::v4());
+    for (size_t i = 0; i < N; ++i) {
+      io_context.post([]{
+        std::this_thread::sleep_for(std::chrono::milliseconds(50));
+      });
+    }
+    EXPECT_EQ(
+      ssl_options.handshake(
+        *ssl_socket,
+        ssl_socket_t::server,
+        {},
+        {},
+        std::chrono::milliseconds(0)
+      ),
+      false
+    );
+    ssl_socket->next_layer().close();
+    ssl_socket.reset();
+  }
+  work.reset();
+  for (;workers.size(); workers.pop_back())
+    workers.back().join();
+}
+
+
+TEST(boosted_tcp_server, strand_deadlock)
+{
+  using context_t = epee::net_utils::connection_context_base;
+  using lock_t = std::mutex;
+  using unique_lock_t = std::unique_lock<lock_t>;
+
+  struct config_t {
+    using condition_t = std::condition_variable_any;
+    using lock_guard_t = std::lock_guard<lock_t>;
+    void notify_success()
+    {
+      lock_guard_t guard(lock);
+      success = true;
+      condition.notify_all();
+    }
+    lock_t lock;
+    condition_t condition;
+    bool success;
+  };
+
+  struct handler_t {
+    using config_type = config_t;
+    using connection_context = context_t;
+    using byte_slice_t = epee::byte_slice;
+    using socket_t = epee::net_utils::i_service_endpoint;
+
+    handler_t(socket_t *socket, config_t &config, context_t &context):
+      socket(socket),
+      config(config),
+      context(context)
+    {}
+    void after_init_connection()
+    {
+      unique_lock_t guard(lock);
+      if (!context.m_is_income) {
+        guard.unlock();
+        socket->do_send(byte_slice_t{"."});
+      }
+    }
+    void handle_qued_callback()
+    {
+    }
+    bool handle_recv(const char *data, size_t bytes_transferred)
+    {
+      unique_lock_t guard(lock);
+      if (!context.m_is_income) {
+        if (context.m_recv_cnt == 1024) {
+          guard.unlock();
+          socket->do_send(byte_slice_t{"."});
+        }
+      }
+      else {
+        if (context.m_recv_cnt == 1) {
+          for(size_t i = 0; i < 1024; ++i) {
+            guard.unlock();
+            socket->do_send(byte_slice_t{"."});
+            guard.lock();
+          }
+        }
+        else if(context.m_recv_cnt == 2) {
+          guard.unlock();
+          socket->close();
+        }
+      }
+      return true;
+    }
+    void release_protocol()
+    {
+      unique_lock_t guard(lock);
+      if(!context.m_is_income
+        && context.m_recv_cnt == 1024
+        && context.m_send_cnt == 2
+      ) {
+        guard.unlock();
+        config.notify_success();
+      }
+    }
+
+    lock_t lock;
+    socket_t *socket;
+    config_t &config;
+    context_t &context;
+  };
+
+  using server_t = epee::net_utils::boosted_tcp_server<handler_t>;
+  using endpoint_t = boost::asio::ip::tcp::endpoint;
+
+  endpoint_t endpoint(boost::asio::ip::address::from_string("127.0.0.1"), 5262);
+  server_t server(epee::net_utils::e_connection_type_P2P);
+  server.init_server(
+    endpoint.port(),
+    endpoint.address().to_string(),
+    {},
+    {},
+    {},
+    true,
+    epee::net_utils::ssl_support_t::e_ssl_support_disabled
+  );
+  server.run_server(2, {});
+  server.async_call(
+    [&]{
+      context_t context;
+      ASSERT_TRUE(
+        server.connect(
+          endpoint.address().to_string(),
+          std::to_string(endpoint.port()),
+          5,
+          context,
+          "0.0.0.0",
+          epee::net_utils::ssl_support_t::e_ssl_support_disabled
+        )
+      );
+    }
+  );
+  {
+    unique_lock_t guard(server.get_config_object().lock);
+    EXPECT_TRUE(
+      server.get_config_object().condition.wait_for(
+        guard,
+        std::chrono::seconds(5),
+        [&] { return server.get_config_object().success; }
+      )
+    );
+  }
+
+  server.send_stop_signal();
+  server.timed_wait_server_stop(5 * 1000);
+  server.deinit_server();
+}