Fix stall issues with p2p [stage]

contrib/epee/include/net/abstract_tcp_server2.h

@@ -100,6 +100,14 @@ namespace net_utils
   {
   public:
     typedef typename t_protocol_handler::connection_context t_connection_context;
+
+    enum status_t {
+      TERMINATED,
+      RUNNING,
+      INTERRUPTED,
+      TERMINATING,
+      WASTED,
+    };
   private:
     using connection_t = connection<t_protocol_handler>;
     using connection_ptr = boost::shared_ptr<connection_t>;
@@ -149,14 +157,6 @@ namespace net_utils
       boost::optional<network_address> real_remote
     );
 
-    enum status_t {
-      TERMINATED,
-      RUNNING,
-      INTERRUPTED,
-      TERMINATING,
-      WASTED,
-    };
-
     struct state_t {
       struct stat_t {
         struct {
@@ -321,6 +321,8 @@ namespace net_utils
 		bool speed_limit_is_enabled() const; ///< tells us should we be sleeping here (e.g. do not sleep on RPC connections)
 
     bool cancel();
+
+    status_t get_status() const noexcept { return m_state.status; }
 
   private:
     //----------------- i_service_endpoint ---------------------

contrib/epee/include/net/abstract_tcp_server2.inl

@@ -826,7 +826,8 @@ namespace net_utils
       return false;
 
     // Wait for the write queue to fall below the max. If it doesn't after a
-    // randomized delay, drop the connection.
+    // 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;
@@ -850,6 +851,12 @@ namespace net_utils
       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,
@@ -888,7 +895,11 @@ namespace net_utils
     };
     if (!wait_sender())
       return false;
-    constexpr size_t CHUNK_SIZE = 32 * 1024;
+    /* 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
     ) {
@@ -900,13 +911,18 @@ namespace net_utils
       start_write();
     }
     else {
+      std::size_t soft_limit = 0;
+      const scope_guard scope_exit_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)
         );
-        m_state.data.write.total_bytes += m_state.data.write.queue.front().size();
+        soft_limit += m_state.data.write.queue.front().size();
         start_write();
       }
     }

tests/unit_tests/epee_boosted_tcp_server.cpp

@@ -674,7 +674,7 @@ TEST(boosted_tcp_server, strand_deadlock)
   using endpoint_t = boost::asio::ip::tcp::endpoint;
 
   endpoint_t endpoint(boost::asio::ip::make_address("127.0.0.1"), 5262);
-  server_t server(epee::net_utils::e_connection_type_P2P);
+  server_t server(epee::net_utils::e_connection_type_RPC);
   server.init_server(
     endpoint.port(),
     endpoint.address().to_string(),
@@ -827,3 +827,105 @@ TEST(boosted_tcp_server, shutdown)
   MINFO("Waiting for handshake to cancel");
   ev.wait();
 }
+
+TEST(boosted_tcp_server, write_failure)
+{
+  using context_t = epee::net_utils::connection_context_base;
+
+  struct config_t {};
+
+  struct handler_t {
+    using config_type = config_t;
+    using connection_context = context_t;
+    using socket_t = epee::net_utils::i_service_endpoint;
+
+    handler_t(socket_t *socket, config_t &config, context_t &):
+      config(config)
+    {}
+    void after_init_connection()
+    {}
+
+    void handle_qued_callback()
+    {}
+
+    bool handle_recv(const char *data, size_t bytes_transferred)
+    {
+      throw std::runtime_error{"UNEXPECTED!"};
+    }
+
+    void release_protocol()
+    {}
+
+    config_t &config;
+  };
+
+
+  using byte_slice_t = epee::byte_slice;
+  using connection_t = epee::net_utils::connection<handler_t>;
+  using shared_t = connection_t::shared_state;
+  using tcp_t = boost::asio::ip::tcp;
+  using endpoint_t = tcp_t::endpoint;
+  using socket_t = tcp_t::socket;
+  using acceptor_t = tcp_t::acceptor;
+
+  const endpoint_t endpoint{boost::asio::ip::make_address("127.0.0.1"), 5262};
+  boost::asio::io_context context{};
+  acceptor_t acceptor{context};
+  acceptor.open(endpoint.protocol());
+#if !defined(_WIN32)
+  acceptor.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
+#endif
+  acceptor.bind(endpoint);
+  acceptor.listen();
+
+  socket_t in_socket{context};
+
+  boost::shared_ptr<connection_t> out_connection;
+  const auto shared = std::make_shared<shared_t>();
+  const auto make_connection = [&] {
+    in_socket = socket_t{context};
+    acceptor.async_accept(in_socket, [] (auto error) { EXPECT_TRUE(!error); });
+
+    socket_t out_socket{context};
+    out_socket.async_connect(endpoint, [] (auto error) { EXPECT_TRUE(!error); });
+
+    context.restart();
+    ASSERT_EQ(2u, context.run()); // connect and accept
+
+    out_connection = boost::make_shared<connection_t>(
+      context,
+      std::move(out_socket),
+      shared,
+      epee::net_utils::e_connection_type_P2P,
+      epee::net_utils::ssl_support_t::e_ssl_support_disabled
+    );
+    EXPECT_TRUE(out_connection->start(false, true));
+  };
+
+  make_connection();
+  {
+    const byte_slice_t payload{"."};
+    epee::net_utils::i_service_endpoint& out{*out_connection};
+    static_assert(ABSTRACT_SERVER_SEND_QUE_MAX_COUNT < std::numeric_limits<std::size_t>::max(), "");
+    for (std::size_t i = 0; i <= ABSTRACT_SERVER_SEND_QUE_MAX_COUNT; ++i)
+      EXPECT_TRUE(out.do_send(payload.clone()));
+    EXPECT_FALSE(out.do_send(payload.clone()));
+  }
+  context.restart();
+  EXPECT_LE(1u, context.run());
+  EXPECT_EQ(connection_t::WASTED, out_connection->get_status());
+
+  make_connection();
+  {
+    const byte_slice_t spayload{"."};
+    const byte_slice_t lpayload{std::string(std::size_t(3 * 128 * 1024), '.')};
+    epee::net_utils::i_service_endpoint& out{*out_connection};
+    for (std::size_t i = 0; i < ABSTRACT_SERVER_SEND_QUE_MAX_COUNT; ++i)
+      EXPECT_TRUE(out.do_send(spayload.clone()));
+    EXPECT_FALSE(out.do_send(lpayload.clone()));
+  }
+  context.restart();
+  EXPECT_LE(1u, context.run());
+  EXPECT_EQ(connection_t::WASTED, out_connection->get_status());
+}
+