diff --git a/contrib/epee/include/net/abstract_tcp_server2.inl b/contrib/epee/include/net/abstract_tcp_server2.inl index cb1388f3b..c1d261659 100644 --- a/contrib/epee/include/net/abstract_tcp_server2.inl +++ b/contrib/epee/include/net/abstract_tcp_server2.inl @@ -269,8 +269,6 @@ PRAGMA_WARNING_DISABLE_VS(4355) //_dbg3("[sock " << socket().native_handle() << "] add_ref, m_peer_number=" << mI->m_peer_number); CRITICAL_REGION_LOCAL(self->m_self_refs_lock); //_dbg3("[sock " << socket().native_handle() << "] add_ref 2, m_peer_number=" << mI->m_peer_number); - if(m_was_shutdown) - return false; ++m_reference_count; m_self_ref = std::move(self); return true; diff --git a/contrib/epee/include/net/levin_protocol_handler_async.h b/contrib/epee/include/net/levin_protocol_handler_async.h index 790b64aa5..debe22208 100644 --- a/contrib/epee/include/net/levin_protocol_handler_async.h +++ b/contrib/epee/include/net/levin_protocol_handler_async.h @@ -892,12 +892,22 @@ template template bool async_protocol_handler_config::foreach_connection(const callback_t &cb) { CRITICAL_REGION_LOCAL(m_connects_lock); - for(auto& c: m_connects) - { - async_protocol_handler* aph = c.second; - if(!cb(aph->get_context_ref())) + std::vector conn; + conn.reserve(m_connects.size()); + + auto scope_exit_handler = misc_utils::create_scope_leave_handler([&conn]{ + for (auto &aph: conn) + aph->finish_outer_call(); + }); + + for (auto &e: m_connects) + if (e.second->start_outer_call()) + conn.push_back(e.second); + + for (auto &aph: conn) + if (!cb(aph->get_context_ref())) return false; - } + return true; } //------------------------------------------------------------------------------------------ @@ -908,6 +918,10 @@ bool async_protocol_handler_config::for_connection(const b async_protocol_handler* aph = find_connection(connection_id); if (!aph) return false; + if (!aph->start_outer_call()) + return false; + auto scope_exit_handler = misc_utils::create_scope_leave_handler( + boost::bind(&async_protocol_handler::finish_outer_call, aph)); if(!cb(aph->get_context_ref())) return false; return true; diff --git a/tests/unit_tests/epee_boosted_tcp_server.cpp b/tests/unit_tests/epee_boosted_tcp_server.cpp index 71098f612..ed3ac757f 100644 --- a/tests/unit_tests/epee_boosted_tcp_server.cpp +++ b/tests/unit_tests/epee_boosted_tcp_server.cpp @@ -37,6 +37,7 @@ #include "include_base_utils.h" #include "string_tools.h" #include "net/abstract_tcp_server2.h" +#include "net/levin_protocol_handler_async.h" namespace { @@ -132,3 +133,152 @@ TEST(boosted_tcp_server, worker_threads_are_exception_resistant) 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; } + }; + + struct command_handler_t: epee::levin::levin_commands_handler { + size_t delay; + command_handler_t(size_t delay = 0): delay(delay) {} + virtual int invoke(int, const epee::span, std::string&, context_t&) override { epee::misc_utils::sleep_no_w(delay); return {}; } + virtual int notify(int, const epee::span, 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* ptr) { delete ptr; } + }; + + using handler_t = epee::levin::async_protocol_handler; + using connection_t = epee::net_utils::connection; + using connection_ptr = boost::shared_ptr; + using shared_state_t = typename connection_t::shared_state; + using shared_state_ptr = std::shared_ptr; + using tag_t = boost::uuids::uuid; + using tags_t = std::vector; + 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; + using workers_t = std::vector; + using server_t = epee::net_utils::boosted_tcp_server; + + io_context_t io_context; + work_ptr work(std::make_shared(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]{ + auto scope_exit_handler = epee::misc_utils::create_scope_leave_handler([&work]{ + work.reset(); + }); + + shared_state_ptr shared_state(std::make_shared()); + 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); + + ASSERT_TRUE(shared_state->get_connections_count() == 0); + constexpr auto DELAY = 30; + constexpr auto TIMEOUT = 1; + 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, {}, tag, [](int, const epee::span, 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); + } + }); + + for (auto& w: workers) { + w.join(); + } + server.send_stop_signal(); + server.timed_wait_server_stop(5 * 1000); + server.deinit_server(); +}