aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--contrib/epee/include/net/levin_protocol_handler_async.h52
-rw-r--r--tests/unit_tests/epee_boosted_tcp_server.cpp110
2 files changed, 134 insertions, 28 deletions
diff --git a/contrib/epee/include/net/levin_protocol_handler_async.h b/contrib/epee/include/net/levin_protocol_handler_async.h
index f6b73a2d5..d062fa877 100644
--- a/contrib/epee/include/net/levin_protocol_handler_async.h
+++ b/contrib/epee/include/net/levin_protocol_handler_async.h
@@ -769,36 +769,32 @@ void async_protocol_handler_config<t_connection_context>::del_connection(async_p
template<class t_connection_context>
void async_protocol_handler_config<t_connection_context>::delete_connections(size_t count, bool incoming)
{
- std::vector <boost::uuids::uuid> connections;
+ std::vector<typename connections_map::mapped_type> connections;
+
+ auto scope_exit_handler = misc_utils::create_scope_leave_handler([&connections]{
+ for (auto &aph: connections)
+ aph->finish_outer_call();
+ });
+
CRITICAL_REGION_BEGIN(m_connects_lock);
for (auto& c: m_connects)
{
if (c.second->m_connection_context.m_is_income == incoming)
- connections.push_back(c.first);
+ if (c.second->start_outer_call())
+ connections.push_back(c.second);
}
// close random connections from the provided set
// TODO or better just keep removing random elements (performance)
unsigned seed = std::chrono::system_clock::now().time_since_epoch().count();
shuffle(connections.begin(), connections.end(), std::default_random_engine(seed));
- while (count > 0 && connections.size() > 0)
- {
- try
- {
- auto i = connections.end() - 1;
- async_protocol_handler<t_connection_context> *conn = m_connects.at(*i);
- m_connects.erase(*i);
- conn->close();
- connections.erase(i);
- }
- catch (const std::out_of_range &e)
- {
- MWARNING("Connection not found in m_connects, continuing");
- }
- --count;
- }
+ for (size_t i = 0; i < connections.size() && i < count; ++i)
+ m_connects.erase(connections[i]->get_connection_id());
CRITICAL_REGION_END();
+
+ for (size_t i = 0; i < connections.size() && i < count; ++i)
+ connections[i]->close();
}
//------------------------------------------------------------------------------------------
template<class t_connection_context>
@@ -860,18 +856,19 @@ int async_protocol_handler_config<t_connection_context>::invoke_async(int comman
template<class t_connection_context> template<class callback_t>
bool async_protocol_handler_config<t_connection_context>::foreach_connection(const callback_t &cb)
{
- CRITICAL_REGION_LOCAL(m_connects_lock);
std::vector<typename connections_map::mapped_type> 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();
});
+ CRITICAL_REGION_BEGIN(m_connects_lock);
+ conn.reserve(m_connects.size());
for (auto &e: m_connects)
if (e.second->start_outer_call())
conn.push_back(e.second);
+ CRITICAL_REGION_END()
for (auto &aph: conn)
if (!cb(aph->get_context_ref()))
@@ -883,11 +880,8 @@ bool async_protocol_handler_config<t_connection_context>::foreach_connection(con
template<class t_connection_context> template<class callback_t>
bool async_protocol_handler_config<t_connection_context>::for_connection(const boost::uuids::uuid &connection_id, const callback_t &cb)
{
- CRITICAL_REGION_LOCAL(m_connects_lock);
- async_protocol_handler<t_connection_context>* aph = find_connection(connection_id);
- if (!aph)
- return false;
- if (!aph->start_outer_call())
+ async_protocol_handler<t_connection_context>* aph = nullptr;
+ if (find_and_lock_connection(connection_id, aph) != LEVIN_OK)
return false;
auto scope_exit_handler = misc_utils::create_scope_leave_handler(
boost::bind(&async_protocol_handler<t_connection_context>::finish_outer_call, aph));
@@ -953,12 +947,14 @@ int async_protocol_handler_config<t_connection_context>::send(byte_slice message
template<class t_connection_context>
bool async_protocol_handler_config<t_connection_context>::close(boost::uuids::uuid connection_id)
{
- CRITICAL_REGION_LOCAL(m_connects_lock);
- async_protocol_handler<t_connection_context>* aph = find_connection(connection_id);
- if (!aph)
+ async_protocol_handler<t_connection_context>* aph = nullptr;
+ if (find_and_lock_connection(connection_id, aph) != LEVIN_OK)
return false;
+ auto scope_exit_handler = misc_utils::create_scope_leave_handler(
+ boost::bind(&async_protocol_handler<t_connection_context>::finish_outer_call, aph));
if (!aph->close())
return false;
+ CRITICAL_REGION_LOCAL(m_connects_lock);
m_connects.erase(connection_id);
return true;
}
diff --git a/tests/unit_tests/epee_boosted_tcp_server.cpp b/tests/unit_tests/epee_boosted_tcp_server.cpp
index 06e076a3a..457c05c15 100644
--- a/tests/unit_tests/epee_boosted_tcp_server.cpp
+++ b/tests/unit_tests/epee_boosted_tcp_server.cpp
@@ -170,6 +170,7 @@ TEST(test_epee_connection, test_lifetime)
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;
@@ -344,6 +345,115 @@ TEST(test_epee_connection, test_lifetime)
shared_state->del_out_connections(1);
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();
+
});
for (auto& w: workers) {
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-15 09:24:30 +0000