diff --git a/contrib/epee/include/net/abstract_tcp_server2.h b/contrib/epee/include/net/abstract_tcp_server2.h index 4bf4b29c89b..fba5d5c4587 100644 --- a/contrib/epee/include/net/abstract_tcp_server2.h +++ b/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; using connection_ptr = boost::shared_ptr; @@ -149,14 +157,6 @@ namespace net_utils boost::optional 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 --------------------- diff --git a/contrib/epee/include/net/abstract_tcp_server2.inl b/contrib/epee/include/net/abstract_tcp_server2.inl index a53531c3d61..c190150b95a 100644 --- a/contrib/epee/include/net/abstract_tcp_server2.inl +++ b/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(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(); } } diff --git a/tests/unit_tests/epee_boosted_tcp_server.cpp b/tests/unit_tests/epee_boosted_tcp_server.cpp index 3fd799c9a5f..2509a2d45e2 100644 --- a/tests/unit_tests/epee_boosted_tcp_server.cpp +++ b/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; + 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 out_connection; + const auto shared = std::make_shared(); + 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( + 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::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()); +} +