p2p: fix hanging shutdown

4 files changed, 132 insertions, 44 deletions

diff --git a/tests/unit_tests/epee_boosted_tcp_server.cpp b/tests/unit_tests/epee_boosted_tcp_server.cpp
index c56174109..6052f7a50 100644
--- a/tests/unit_tests/epee_boosted_tcp_server.cpp
+++ b/tests/unit_tests/epee_boosted_tcp_server.cpp
@@ -37,10 +37,12 @@
 
 #include "gtest/gtest.h"
 
+#include "cryptonote_protocol/cryptonote_protocol_defs.h"
 #include "include_base_utils.h"
 #include "string_tools.h"
 #include "net/abstract_tcp_server2.h"
 #include "net/levin_protocol_handler_async.h"
+#include "p2p/net_node.h"
 
 namespace
 {
@@ -234,7 +236,7 @@ TEST(test_epee_connection, test_lifetime)
     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);
+      shared_state->close(context.m_connection_id, true);
       context.m_remote_address.get_zone();
       return true;
     });
@@ -250,9 +252,9 @@ TEST(test_epee_connection, test_lifetime)
     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(t, true);
 
-      shared_state->close(context.m_connection_id);
+      shared_state->close(context.m_connection_id, true);
       context.m_remote_address.get_zone();
       ++index;
 
@@ -266,7 +268,7 @@ TEST(test_epee_connection, test_lifetime)
 
     index = 0;
     success = shared_state->foreach_connection([&index, shared_state](context_t& context){
-      shared_state->close(context.m_connection_id);
+      shared_state->close(context.m_connection_id, true);
       context.m_remote_address.get_zone();
       ++index;
       return true;
@@ -296,7 +298,7 @@ TEST(test_epee_connection, test_lifetime)
         });
         ASSERT_TRUE(success);
       }
-      shared_state->close(tag);
+      shared_state->close(tag, true);
       ASSERT_TRUE(shared_state->get_connections_count() == 0);
     }
 
@@ -450,7 +452,7 @@ TEST(test_epee_connection, test_lifetime)
           auto tag = context.m_connection_id;
           boost::asio::post(io_context, [conn] { conn->cancel(); });
           conn.reset();
-          s->close(tag);
+          s->close(tag, true);
           while (s->sock_count);
         }
       });
@@ -643,7 +645,7 @@ TEST(boosted_tcp_server, strand_deadlock)
         }
         else if(context.m_recv_cnt == 2) {
           guard.unlock();
-          socket->close();
+          socket->close(false);
         }
       }
       return true;
@@ -711,3 +713,104 @@ TEST(boosted_tcp_server, strand_deadlock)
   server.timed_wait_server_stop(5 * 1000);
   server.deinit_server();
 }
+
+TEST(boosted_tcp_server, 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 config_t : epee::levin::async_protocol_handler_config<context_t> {
+    void received_handshake() { handshake_received.raise(); }
+    epee::simple_event handshake_received;
+  };
+
+  struct handler_t : epee::levin::async_protocol_handler<context_t> {
+    using config_type = config_t;
+    using connection_context = context_t;
+    using epee::levin::async_protocol_handler<context_t>::async_protocol_handler;
+
+    bool handle_recv(const void *data, size_t bytes_transferred)
+    {
+      // We don't respond to the handshake (the async_invoke_remote_command2 is waiting for a response)
+      MINFO("handle_recv just came in");
+      config_t* config = dynamic_cast<config_t*>(&m_config);
+      if (config == nullptr)
+        throw std::runtime_error("m_config must be of type config_t");
+      config->received_handshake();
+      return true;
+    }
+  };
+
+  boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::make_address("127.0.0.1"), 5262);
+  epee::net_utils::boosted_tcp_server<handler_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
+  );
+
+  // Run the server in a thread and wait for it to start
+  MINFO("Starting the server");
+  std::thread running_server([&]{server.run_server(2, true/*wait*/);} );
+
+  // Have the server connect to itself
+  MINFO("Connecting the server to itself");
+  context_t context;
+  {
+    epee::simple_event connected;
+    server.async_call(
+      [&]{
+        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
+          )
+        );
+        connected.raise();
+      }
+    );
+    connected.wait();
+  }
+
+  // Invoke handshake to the connection, and wait for cb cancel in a separate thread
+  MINFO("Invoking handshake");
+  epee::simple_event ev;
+  {
+    using COMMAND_HANDSHAKE = nodetool::COMMAND_HANDSHAKE_T<cryptonote::CORE_SYNC_DATA>;
+    COMMAND_HANDSHAKE::request arg;
+    bool r = epee::net_utils::async_invoke_remote_command2<COMMAND_HANDSHAKE::response>(context, COMMAND_HANDSHAKE::ID, arg, server.get_config_object(),
+      [&ev](int code, const COMMAND_HANDSHAKE::response&, context_t&)
+    {
+      ASSERT_EQ(code, LEVIN_ERROR_CONNECTION_DESTROYED);
+      ev.raise();
+    }, P2P_DEFAULT_HANDSHAKE_INVOKE_TIMEOUT);
+    ASSERT_TRUE(r);
+
+    MINFO("Waiting for handshake invocation to be received");
+    server.get_config_object().handshake_received.wait();
+  }
+
+  // Now stop the server, providing the callback necessary to wait for all connections to shutdown
+  const auto close_all_connections = [&]()
+  {
+    server.get_config_object().close(context.m_connection_id, true/*wait_for_shutdown*/);
+  };
+
+  MINFO("Stopping the server");
+  server.send_stop_signal(close_all_connections);
+  running_server.join();
+
+  MINFO("Waiting for handshake to cancel");
+  ev.wait();
+}
diff --git a/tests/unit_tests/epee_levin_protocol_handler_async.cpp b/tests/unit_tests/epee_levin_protocol_handler_async.cpp
index 4ea6fde90..103ee9cfa 100644
--- a/tests/unit_tests/epee_levin_protocol_handler_async.cpp
+++ b/tests/unit_tests/epee_levin_protocol_handler_async.cpp
@@ -151,7 +151,7 @@ namespace
       return m_send_return;
     }
 
-    virtual bool close()                              { /*std::cout << "test_connection::close()" << std::endl; */return true; }
+    virtual bool close(const bool wait_for_shutdown)  { /*std::cout << "test_connection::close()" << std::endl; */return true; }
     virtual bool send_done()                          { /*std::cout << "test_connection::send_done()" << std::endl; */return true; }
     virtual bool call_run_once_service_io()           { std::cout << "test_connection::call_run_once_service_io()" << std::endl; return true; }
     virtual bool request_callback()                   { std::cout << "test_connection::request_callback()" << std::endl; return true; }
@@ -572,7 +572,7 @@ TEST_F(test_levin_protocol_handler__hanle_recv_with_invalid_data, does_not_handl
 {
   prepare_buf();
 
-  ASSERT_TRUE(m_conn->m_protocol_handler.close());
+  ASSERT_TRUE(m_conn->m_protocol_handler.close(true));
   ASSERT_FALSE(m_conn->m_protocol_handler.handle_recv(m_buf.data(), m_buf.size()));
 }
 
diff --git a/tests/unit_tests/levin.cpp b/tests/unit_tests/levin.cpp
index 25b67907f..1a734181b 100644
--- a/tests/unit_tests/levin.cpp
+++ b/tests/unit_tests/levin.cpp
@@ -63,7 +63,7 @@ namespace
             return true;
         }
 
-        virtual bool close() override final
+        virtual bool close(const bool wait_for_shutdown) override final
         {
             return true;
         }
diff --git a/tests/unit_tests/node_server.cpp b/tests/unit_tests/node_server.cpp
index e55b43911..11853c0fd 100644
--- a/tests/unit_tests/node_server.cpp
+++ b/tests/unit_tests/node_server.cpp
@@ -42,6 +42,21 @@
 #define MAKE_IPV4_ADDRESS_PORT(a,b,c,d,e) epee::net_utils::ipv4_network_address{MAKE_IP(a,b,c,d),e}
 #define MAKE_IPV4_SUBNET(a,b,c,d,e) epee::net_utils::ipv4_network_subnet{MAKE_IP(a,b,c,d),e}
 
+namespace
+{
+  boost::filesystem::path create_temp_dir()
+  {
+    boost::system::error_code ec;
+    auto path = boost::filesystem::temp_directory_path() / boost::filesystem::unique_path("daemon-%%%%%%%%%%%%%%%%", ec);
+    if (ec)
+      return boost::filesystem::path{};
+    auto success = boost::filesystem::create_directory(path, ec);
+    if (!ec && success)
+      return path;
+    return boost::filesystem::path{};
+  }
+}
+
 namespace cryptonote {
   class blockchain_storage;
 }
@@ -300,17 +315,7 @@ TEST(ban, file_banlist)
   Server server(cprotocol);
   cprotocol.set_p2p_endpoint(&server);
 
-  auto create_node_dir = [](){
-    boost::system::error_code ec;
-    auto path = boost::filesystem::temp_directory_path() / boost::filesystem::unique_path("daemon-%%%%%%%%%%%%%%%%", ec);
-    if (ec)
-      return boost::filesystem::path{};
-    auto success = boost::filesystem::create_directory(path, ec);
-    if (!ec && success)
-      return path;
-    return boost::filesystem::path{};
-  };
-  const auto node_dir = create_node_dir();
+  const auto node_dir = create_temp_dir();
   ASSERT_TRUE(!node_dir.empty());
   auto auto_remove_node_dir = epee::misc_utils::create_scope_leave_handler([&node_dir](){
       boost::filesystem::remove_all(node_dir);
@@ -615,7 +620,7 @@ TEST(cryptonote_protocol_handler, race_condition)
     }
     virtual bool drop_connection(const contexts::basic& context) override {
       if (shared_state)
-        return shared_state->close(context.m_connection_id);
+        return shared_state->close(context.m_connection_id, true);
       else
         return {};
     }
@@ -697,16 +702,6 @@ TEST(cryptonote_protocol_handler, race_condition)
     handshaked.wait();
   };
   using path_t = boost::filesystem::path;
-  auto create_dir = []{
-    ec_t ec;
-    path_t path = boost::filesystem::temp_directory_path() / boost::filesystem::unique_path("daemon-%%%%%%%%%%%%%%%%", ec);
-    if (ec)
-      return path_t{};
-    auto success = boost::filesystem::create_directory(path, ec);
-    if (not ec && success)
-      return path;
-    return path_t{};
-  };
   auto remove_tree = [](const path_t &path){
     ec_t ec;
     boost::filesystem::remove_all(path, ec);
@@ -726,7 +721,7 @@ TEST(cryptonote_protocol_handler, race_condition)
   };
   using options_description_t = boost::program_options::options_description;
 
-  const auto dir = create_dir();
+  const auto dir = create_temp_dir();
   ASSERT_TRUE(not dir.empty());
 
   daemons_t daemon{
@@ -1215,21 +1210,11 @@ TEST(node_server, race_condition)
   };
   using path_t = boost::filesystem::path;
   using ec_t = boost::system::error_code;
-  auto create_dir = []{
-    ec_t ec;
-    path_t path = boost::filesystem::temp_directory_path() / boost::filesystem::unique_path("daemon-%%%%%%%%%%%%%%%%", ec);
-    if (ec)
-      return path_t{};
-    auto success = boost::filesystem::create_directory(path, ec);
-    if (not ec && success)
-      return path;
-    return path_t{};
-  };
   auto remove_tree = [](const path_t &path){
     ec_t ec;
     boost::filesystem::remove_all(path, ec);
   };
-  const auto dir = create_dir();
+  const auto dir = create_temp_dir();
   ASSERT_TRUE(not dir.empty());
   protocol_t protocol{};
   node_server_t node_server(protocol);