From 3987318f09f4aabc21ca5cb89394038248973e42 Mon Sep 17 00:00:00 2001 From: sfan5 Date: Fri, 5 Jan 2024 12:24:14 +0100 Subject: [PATCH] Time out when reliables can't be delivered If one of the channels stalls for whatever reason we can't pretend the connection is fine. --- src/network/connection.cpp | 50 ++++++++++++++++++++++++++----- src/network/connection.h | 8 +++-- src/network/connectionthreads.cpp | 13 ++++---- 3 files changed, 55 insertions(+), 16 deletions(-) diff --git a/src/network/connection.cpp b/src/network/connection.cpp index f75364f5c..e3d8d3999 100644 --- a/src/network/connection.cpp +++ b/src/network/connection.cpp @@ -363,8 +363,19 @@ void ReliablePacketBuffer::incrementTimeouts(float dtime) } } +u32 ReliablePacketBuffer::getTimedOuts(float timeout) +{ + MutexAutoLock listlock(m_list_mutex); + u32 count = 0; + for (auto &packet : m_list) { + if (packet->totaltime >= timeout) + count++; + } + return count; +} + std::vector> - ReliablePacketBuffer::getTimedOuts(float timeout, u32 max_packets) + ReliablePacketBuffer::getResend(float timeout, u32 max_packets) { MutexAutoLock listlock(m_list_mutex); std::vector> timed_outs; @@ -939,17 +950,22 @@ void Peer::RTTStatistics(float rtt, const std::string &profiler_id, m_last_rtt = rtt; } -bool Peer::isTimedOut(float timeout) +bool Peer::isTimedOut(float timeout, std::string &reason) { MutexAutoLock lock(m_exclusive_access_mutex); - u64 current_time = porting::getTimeMs(); - float dtime = CALC_DTIME(m_last_timeout_check,current_time); - m_last_timeout_check = current_time; + { + u64 current_time = porting::getTimeMs(); + float dtime = CALC_DTIME(m_last_timeout_check, current_time); + m_last_timeout_check = current_time; + m_timeout_counter += dtime; + } + if (m_timeout_counter > timeout) { + reason = "timeout counter"; + return true; + } - m_timeout_counter += dtime; - - return m_timeout_counter > timeout; + return false; } void Peer::Drop() @@ -980,6 +996,24 @@ UDPPeer::UDPPeer(session_t a_id, Address a_address, Connection* connection) : channel.setWindowSize(START_RELIABLE_WINDOW_SIZE); } +bool UDPPeer::isTimedOut(float timeout, std::string &reason) +{ + if (Peer::isTimedOut(timeout, reason)) + return true; + + MutexAutoLock lock(m_exclusive_access_mutex); + + for (int i = 0; i < CHANNEL_COUNT; i++) { + Channel &channel = channels[i]; + if (channel.outgoing_reliables_sent.getTimedOuts(timeout) > 0) { + reason = "outgoing reliables channel=" + itos(i); + return true; + } + } + + return false; +} + bool UDPPeer::getAddress(MTProtocols type,Address& toset) { if ((type == MTP_UDP) || (type == MTP_MINETEST_RELIABLE_UDP) || (type == MTP_PRIMARY)) diff --git a/src/network/connection.h b/src/network/connection.h index d5664e8bb..a14ebb9e9 100644 --- a/src/network/connection.h +++ b/src/network/connection.h @@ -262,7 +262,9 @@ public: void insert(BufferedPacketPtr &p_ptr, u16 next_expected); void incrementTimeouts(float dtime); - std::vector> getTimedOuts(float timeout, u32 max_packets); + u32 getTimedOuts(float timeout); + // timeout relative to last resend + std::vector> getResend(float timeout, u32 max_packets); void print(); bool empty(); @@ -525,7 +527,7 @@ class Peer { bool isHalfOpen() const { return m_half_open; } void SetFullyOpen() { m_half_open = false; } - bool isTimedOut(float timeout); + virtual bool isTimedOut(float timeout, std::string &reason); unsigned int m_increment_packets_remaining = 0; @@ -636,6 +638,8 @@ public: SharedBuffer addSplitPacket(u8 channel, BufferedPacketPtr &toadd, bool reliable); + bool isTimedOut(float timeout, std::string &reason) override; + protected: /* Calculates avg_rtt and resend_timeout. diff --git a/src/network/connectionthreads.cpp b/src/network/connectionthreads.cpp index afe897918..970b7d1ed 100644 --- a/src/network/connectionthreads.cpp +++ b/src/network/connectionthreads.cpp @@ -195,10 +195,11 @@ void ConnectionSendThread::runTimeouts(float dtime) // Note that this time is also fixed since the timeout is not reset in half-open state. const float peer_timeout = peer->isHalfOpen() ? MYMAX(5.0f, m_timeout / 4) : m_timeout; - if (peer->isTimedOut(peer_timeout)) { + std::string reason; + if (peer->isTimedOut(peer_timeout, reason)) { infostream << m_connection->getDesc() << "RunTimeouts(): Peer " << peer->id - << " has timed out." + << " has timed out (" << reason << ")" << std::endl; // Add peer to the list timeouted_peers.push_back(peer->id); @@ -216,7 +217,7 @@ void ConnectionSendThread::runTimeouts(float dtime) channel.outgoing_reliables_sent.incrementTimeouts(dtime); // Re-send timed out outgoing reliables - auto timed_outs = channel.outgoing_reliables_sent.getTimedOuts(resend_timeout, + auto timed_outs = channel.outgoing_reliables_sent.getResend(resend_timeout, (m_max_data_packets_per_iteration / numpeers)); channel.UpdatePacketLossCounter(timed_outs.size()); @@ -424,10 +425,10 @@ void ConnectionSendThread::processReliableCommand(ConnectionCommandPtr &c) return; Channel &channel = dynamic_cast(&peer)->channels[c->channelnum]; - auto timed_outs = channel.outgoing_reliables_sent.getTimedOuts(0, 1); + auto list = channel.outgoing_reliables_sent.getResend(0, 1); - if (!timed_outs.empty()) - resendReliable(channel, timed_outs.front().get(), -1); + if (!list.empty()) + resendReliable(channel, list.front().get(), -1); return; }