From 9d320d77eb833983e32e6f12b34a2889c58673df Mon Sep 17 00:00:00 2001 From: acetone <63557806+freeacetone@users.noreply.github.com> Date: Sat, 8 Aug 2026 17:41:36 +0300 Subject: [PATCH 1/2] datagram tunnel window and ack policy --- libi2pd/Datagram.cpp | 6 + libi2pd/Datagram.h | 6 +- libi2pd_client/UDPTunnel.cpp | 254 ++++++++++++++++++++++++++++++++--- libi2pd_client/UDPTunnel.h | 47 ++++++- 4 files changed, 289 insertions(+), 24 deletions(-) diff --git a/libi2pd/Datagram.cpp b/libi2pd/Datagram.cpp index b576b42c..e5560849 100644 --- a/libi2pd/Datagram.cpp +++ b/libi2pd/Datagram.cpp @@ -785,6 +785,12 @@ namespace datagram routingPath->outboundTunnel->SendTunnelDataMsgs(send); } } + else + { + m_NumDroppedNoPath += m_SendQueue.size (); + LogPrint (eLogWarning, "Datagram: No routing path, dropped ", m_SendQueue.size (), " messages, ", + m_NumDroppedNoPath, " total"); + } m_SendQueue.clear(); } } diff --git a/libi2pd/Datagram.h b/libi2pd/Datagram.h index 45be618b..22011153 100644 --- a/libi2pd/Datagram.h +++ b/libi2pd/Datagram.h @@ -84,10 +84,13 @@ namespace datagram DatagramVersion GetVersion () const { return m_Version; } void SetVersion (DatagramVersion version) { m_Version = version; } - void DropSharedRoutingPath () { if (m_RoutingSession) m_RoutingSession->SetSharedRoutingPath (nullptr); } + void DropSharedRoutingPath () { if (m_RoutingSession) { m_RoutingSession->SetSharedRoutingPath (nullptr); m_NumPathDrops++; } } // request LeaseSet update from floodfill in case our copy contains dead leases void RequestUpdatedLeaseSet (); + uint32_t GetNumPathDrops () const { return m_NumPathDrops; } + uint32_t GetNumDroppedNoPath () const { return m_NumDroppedNoPath; } + struct Info { std::shared_ptr IBGW; @@ -123,6 +126,7 @@ namespace datagram uint64_t m_LastUse, m_LastFlush; // milliseconds bool m_RequestingLS; DatagramVersion m_Version; + uint32_t m_NumPathDrops = 0, m_NumDroppedNoPath = 0; }; typedef std::shared_ptr DatagramSession_ptr; diff --git a/libi2pd_client/UDPTunnel.cpp b/libi2pd_client/UDPTunnel.cpp index fad4a501..b01edc4d 100644 --- a/libi2pd_client/UDPTunnel.cpp +++ b/libi2pd_client/UDPTunnel.cpp @@ -30,6 +30,7 @@ namespace client { if (!m_LastSession || m_LastSession->Identity.GetLL()[0] != from.GetIdentHash ().GetLL()[0] || (fromPort && fromPort != m_LastSession->RemotePort)) m_LastSession = ObtainUDPSession(from, toPort, fromPort); + m_LastSession->m_LastReceivedTime = i2p::util::GetMillisecondsSinceEpoch (); boost::system::error_code ec; if (len > 0) m_LastSession->IPSocket.send_to(boost::asio::buffer(buf, len), m_RemoteEndpoint, 0, ec); @@ -74,6 +75,12 @@ namespace client else m_LastSession = nullptr; } + if (!m_LastSession) + { + m_NumRawNoSession++; + LogPrint (eLogWarning, "UDP Server: No session for raw datagram from port ", fromPort, " to ", toPort, + ", dropped, ", m_NumRawNoSession, " total"); + } if (m_LastSession) { boost::system::error_code ec; @@ -147,6 +154,7 @@ namespace client auto s = std::make_shared(boost::asio::ip::udp::endpoint(addr, 0), m_LocalDest, m_RemoteEndpoint, ih, localPort, remotePort); + s->SetMaxWindow (m_MaxWindow); std::lock_guard lock(m_SessionsMutex); m_Sessions.emplace (idx, s); return s; @@ -165,6 +173,7 @@ namespace client if (seqn >= m_AckTimerSeqn) { m_AckTimerSeqn = 0; + m_NumAckTimeoutsInRow = 0; m_AckTimer.cancel (); } } @@ -178,45 +187,164 @@ namespace client if (it->first > seqn) break; if (it->first == seqn && m_IsSendingAllowed) // ignore first ack after path change { - auto rtt = i2p::util::GetMillisecondsSinceEpoch () - it->second; - m_RTT = m_RTT ? (m_RTT + rtt)/2 : rtt; + auto ts = i2p::util::GetMillisecondsSinceEpoch (); + UpdateRTT (ts - it->second, ts); + m_NumAckTimeoutsInRow = 0; acknowledged = true; } it++; } m_UnackedDatagrams.erase (m_UnackedDatagrams.begin (), it); m_IsSendingAllowed = true; // if we recieve ack after path change, now can send new datagrams - if (acknowledged && !m_UnackedDatagrams.empty ()) + if (!m_UnackedDatagrams.empty ()) { - m_AckTimer.cancel (); - m_AckTimerSeqn = 0; + // keep armed while anything is unacked, otherwise a full window can't be unblocked + if (acknowledged) + { + m_AckTimer.cancel (); + m_AckTimerSeqn = 0; + } ScheduleAckTimer (m_UnackedDatagrams.back ().first); } } + void UDPConnection::UpdateRTT (uint64_t rtt, uint64_t ts) + { + if (m_RTT) + { + uint64_t diff = (rtt > m_RTT) ? rtt - m_RTT : m_RTT - rtt; + m_RTTVar = ((I2P_UDP_RTT_VAR_BETA - 1)*m_RTTVar + diff)/I2P_UDP_RTT_VAR_BETA; + m_RTT = ((I2P_UDP_RTT_ALPHA - 1)*m_RTT + rtt)/I2P_UDP_RTT_ALPHA; + } + else + { + m_RTT = rtt; + m_RTTVar = rtt/2; // rfc 6298 for the first measurement + } + if (!m_MinRTT || rtt < m_MinRTT) + { + m_MinRTT = rtt; + m_MinRTTCandidate = rtt; + m_MinRTTUpdateTime = ts; + } + else + { + if (!m_MinRTTCandidate || rtt < m_MinRTTCandidate) m_MinRTTCandidate = rtt; + // windowed minimum, so queueing delay is not taken for path delay + if (ts > m_MinRTTUpdateTime + I2P_UDP_MIN_RTT_EXPIRATION_TIMEOUT) + { + m_MinRTT = m_MinRTTCandidate; + m_MinRTTCandidate = 0; + m_MinRTTUpdateTime = ts; + } + } + } + + void UDPConnection::UpdateSendRate (uint32_t numDatagrams, uint64_t ts) + { + m_NumSentSinceRateUpdate += numDatagrams; + if (!m_SendRateUpdateTime) + { + m_SendRateUpdateTime = ts; + return; + } + uint64_t interval = ts - m_SendRateUpdateTime; + if (interval < I2P_UDP_SEND_RATE_INTERVAL) return; + uint32_t rate = m_NumSentSinceRateUpdate * 1000 / interval; + m_NumSentSinceRateUpdate = 0; + m_SendRateUpdateTime = ts; + // best rate of the last few seconds, a rate following every dip would shrink the window + if (rate >= m_SendRate || ts > m_SendRateMaxTime + I2P_UDP_SEND_RATE_EXPIRATION_TIMEOUT) + { + m_SendRate = rate; + m_SendRateMaxTime = ts; + } + } + + size_t UDPConnection::GetMaxNumUnackedDatagrams () const + { + if (!m_MinRTT || !m_SendRate) return I2P_UDP_MIN_MAX_NUM_UNACKED_DATAGRAMS; + // bandwidth-delay product for one path delay plus one repliable interval + size_t w = I2P_UDP_WINDOW_GAIN * m_SendRate * (m_MinRTT + I2P_UDP_REPLIABLE_DATAGRAM_INTERVAL) / 1000; + if (w < I2P_UDP_MIN_MAX_NUM_UNACKED_DATAGRAMS) w = I2P_UDP_MIN_MAX_NUM_UNACKED_DATAGRAMS; + if (w > m_MaxWindow) w = m_MaxWindow; + return w; + } + + bool UDPConnection::IsWindowFull () const + { + return !m_UnackedDatagrams.empty () && + m_NextSendPacketNum > m_UnackedDatagrams.front ().first + GetMaxNumUnackedDatagrams (); + } + + void UDPConnection::ExpireUnackedDatagrams (uint64_t ts) + { + // such an ack is never coming and would keep the window blocked + while (!m_UnackedDatagrams.empty () && + ts > m_UnackedDatagrams.front ().second + I2P_UDP_MAX_UNACKED_DATAGRAM_TIME) + m_UnackedDatagrams.pop_front (); + } + + uint64_t UDPConnection::GetRTO () const + { + if (!m_RTT) return I2P_UDP_MAX_UNACKED_DATAGRAM_TIME; + uint64_t rto = m_RTT + I2P_UDP_RTO_K*m_RTTVar; + if (rto < I2P_UDP_MIN_ACK_TIMEOUT) rto = I2P_UDP_MIN_ACK_TIMEOUT; + if (rto > I2P_UDP_MAX_UNACKED_DATAGRAM_TIME) rto = I2P_UDP_MAX_UNACKED_DATAGRAM_TIME; + return rto; + } + + uint64_t UDPConnection::GetWindowProbeInterval () const + { + // an ack can't come back sooner than one rtt + if (!m_RTT) return GetRTO (); + return (m_RTT > I2P_UDP_MIN_WINDOW_PROBE_INTERVAL) ? m_RTT : I2P_UDP_MIN_WINDOW_PROBE_INTERVAL; + } + void UDPConnection::ScheduleAckTimer (uint32_t seqn) { if (!m_AckTimerSeqn) { m_AckTimerSeqn = seqn; - m_AckTimer.expires_after (std::chrono::milliseconds (m_RTT ? 2*m_RTT : I2P_UDP_MAX_UNACKED_DATAGRAM_TIME)); + uint32_t shift = m_NumAckTimeoutsInRow; + if (shift > I2P_UDP_MAX_NUM_ACK_TIMEOUTS) shift = I2P_UDP_MAX_NUM_ACK_TIMEOUTS; + uint64_t timeout = GetRTO () << shift; + if (timeout > I2P_UDP_MAX_UNACKED_DATAGRAM_TIME) timeout = I2P_UDP_MAX_UNACKED_DATAGRAM_TIME; + m_AckTimer.expires_after (std::chrono::milliseconds (timeout)); m_AckTimer.async_wait ([this](const boost::system::error_code& ecode) { if (ecode != boost::asio::error::operation_aborted) { - LogPrint (eLogInfo, "UDP Connection: Packet ", m_AckTimerSeqn, " was not acked"); + auto ts = i2p::util::GetMillisecondsSinceEpoch (); + m_NumAckTimeouts++; m_NumAckTimeoutsInRow++; + // timeouts happen under congestion, only silence means the path is dead + bool resetPath = m_NumAckTimeoutsInRow >= I2P_UDP_MAX_NUM_ACK_TIMEOUTS && + ts > m_LastReceivedTime + I2P_UDP_MAX_UNACKED_DATAGRAM_TIME; + bool resetPeerPath = resetPath && ts > m_LastReceivedTime + I2P_UDP_PEER_PATH_RESET_TIMEOUT; + LogPrint (eLogWarning, "UDP Connection: Packet ", m_AckTimerSeqn, " was not acked, rtt ", + m_RTT, "/", m_RTTVar, "ms, ", m_NumAckTimeoutsInRow, " in a row, ", m_NumAckTimeouts, + " total", resetPath ? ", resetting path" : ""); if (m_IsFirstPacket) m_IsSendingAllowed = false; // stop sending only if session is not established yet m_AckTimerSeqn = 0; - m_RTT = 0; - if (!m_UnackedDatagrams.empty ()) ScheduleAckTimer (0); // try again if failed - // send empty packet with reset path flag + // probe the peer, reset path only after several timeouts in a row + uint8_t flags = UDP_SESSION_FLAG_ACK_REQUESTED; + if (resetPeerPath) flags |= UDP_SESSION_FLAG_RESET_PATH | UDP_SESSION_FLAG_RESET_SEQN; i2p::util::Mapping options; - options.Put (UDP_SESSION_FLAGS, UDP_SESSION_FLAG_RESET_PATH | UDP_SESSION_FLAG_ACK_REQUESTED); + options.Put (UDP_SESSION_FLAGS, flags); + // without seqn the peer acks a stale one and a full window stays blocked + options.Put (UDP_SESSION_SEQN, m_NextSendPacketNum); + m_NextSendPacketNum++; + // probe only while real data is outstanding, otherwise probes feed themselves + if (!m_UnackedDatagrams.empty ()) ScheduleAckTimer (0); auto session = GetDatagramSession (); if (session) { - session->DropSharedRoutingPath (); - session->RequestUpdatedLeaseSet (); // in case current leases are dead + if (resetPath) + { + session->DropSharedRoutingPath (); + session->RequestUpdatedLeaseSet (); // in case current leases are dead + m_NumAckTimeoutsInRow = 0; + } m_Destination->SendDatagram (session, nullptr, 0, 0, 0, &options); } } @@ -246,7 +374,8 @@ namespace client { SetIdentity (to); Start (); - IPSocket.set_option (boost::asio::socket_base::receive_buffer_size (I2P_UDP_MAX_MTU )); + IPSocket.set_option (boost::asio::socket_base::receive_buffer_size (I2P_UDP_SOCKET_BUFFER_SIZE)); + IPSocket.set_option (boost::asio::socket_base::send_buffer_size (I2P_UDP_SOCKET_BUFFER_SIZE)); IPSocket.non_blocking (true); Receive(); } @@ -262,23 +391,30 @@ namespace client { if(!ecode) { - if (!m_UnackedDatagrams.empty () && m_NextSendPacketNum > m_UnackedDatagrams.front ().first + I2P_UDP_MAX_NUM_UNACKED_DATAGRAMS) + ExpireUnackedDatagrams (i2p::util::GetMillisecondsSinceEpoch ()); + bool isWindowFull = IsWindowFull (); + if (isWindowFull && i2p::util::GetMillisecondsSinceEpoch () < m_LastWindowProbeTime + GetWindowProbeInterval ()) { // window is full, drop packet + m_NumWindowDrops++; + if (m_NumWindowDrops == 1 || !(m_NumWindowDrops % 100)) + LogPrint (eLogWarning, "UDP Server: Window full (front=", m_UnackedDatagrams.front ().first, + " next=", m_NextSendPacketNum, "), dropped ", m_NumWindowDrops, " packets"); Receive (); return; } LogPrint(eLogDebug, "UDPSession: Forward ", len, "B from ", FromEndpoint); auto ts = i2p::util::GetMillisecondsSinceEpoch(); + if (isWindowFull) m_LastWindowProbeTime = ts; auto session = GetDatagramSession (); uint64_t repliableDatagramInterval = I2P_UDP_REPLIABLE_DATAGRAM_INTERVAL; if (m_RTT && m_RTT >= I2P_UDP_REPLIABLE_DATAGRAM_INTERVAL && m_RTT < I2P_UDP_REPLIABLE_DATAGRAM_INTERVAL*10) repliableDatagramInterval = m_RTT/10; // 10 - 100 ms - if (ts > m_LastRepliableDatagramTime + repliableDatagramInterval) + if (isWindowFull || ts > m_LastRepliableDatagramTime + repliableDatagramInterval) { if (session->GetVersion () == i2p::datagram::eDatagramV3) { uint8_t flags = 0; - if (!m_RTT || !m_AckTimerSeqn || (!m_UnackedDatagrams.empty () && + if (isWindowFull || !m_RTT || !m_AckTimerSeqn || (!m_UnackedDatagrams.empty () && ts > m_UnackedDatagrams.back ().second + repliableDatagramInterval)) // last ack request { flags |= UDP_SESSION_FLAG_ACK_REQUESTED; @@ -315,6 +451,7 @@ namespace client if (numPackets > 0) LogPrint(eLogDebug, "UDPSession: Forward more ", numPackets, "packets B from ", FromEndpoint); m_NextSendPacketNum += numPackets + 1; + UpdateSendRate (numPackets + 1, ts); m_Destination->FlushSendQueue (session); LastActivity = ts; Receive(); @@ -349,10 +486,44 @@ namespace client std::bind (&I2PUDPServerTunnel::HandleRecvFromI2PRaw, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4), m_inPort ); + m_StatsTimer.reset (new boost::asio::steady_timer (m_LocalDest->GetService ())); + ScheduleStatsTimer (); + } + + void I2PUDPServerTunnel::ScheduleStatsTimer () + { + if (m_StatsTimer) + { + m_StatsTimer->expires_after (std::chrono::seconds (10)); + m_StatsTimer->async_wait (std::bind (&I2PUDPServerTunnel::HandleStatsTimer, + this, std::placeholders::_1)); + } + } + + void I2PUDPServerTunnel::HandleStatsTimer (const boost::system::error_code& ecode) + { + if (ecode != boost::asio::error::operation_aborted) + { + std::lock_guard lock (m_SessionsMutex); + for (const auto& it: m_Sessions) + { + auto& s = it.second; + auto session = s->GetDatagramSession (); + LogPrint (eLogDebug, "UDP Server: stats port=", s->RemotePort, " rtt=", s->m_RTT, "/", + s->m_RTTVar, "/", s->m_MinRTT, "ms rto=", s->GetRTO (), "ms rate=", s->m_SendRate, + "/s window=", s->GetMaxNumUnackedDatagrams (), " unacked=", s->m_UnackedDatagrams.size (), + " nextSeqn=", s->m_NextSendPacketNum, " lastRecvSeqn=", s->m_LastReceivedPacketNum, + " winDrops=", s->m_NumWindowDrops, " ackTimeouts=", s->m_NumAckTimeouts, + " pathDrops=", session ? session->GetNumPathDrops () : 0, + " noPathDrops=", session ? session->GetNumDroppedNoPath () : 0); + } + ScheduleStatsTimer (); + } } void I2PUDPServerTunnel::Stop () { + if (m_StatsTimer) m_StatsTimer->cancel (); auto dgram = m_LocalDest->GetDatagramDestination (); if (dgram) { dgram->ResetReceiver (m_inPort); @@ -410,7 +581,8 @@ namespace client if (m_cancel_resolve) m_cancel_resolve = false; m_LocalSocket.reset (new boost::asio::ip::udp::socket (m_LocalDest->GetService (), m_LocalEndpoint)); - m_LocalSocket->set_option (boost::asio::socket_base::receive_buffer_size (I2P_UDP_MAX_MTU)); + m_LocalSocket->set_option (boost::asio::socket_base::receive_buffer_size (I2P_UDP_SOCKET_BUFFER_SIZE)); + m_LocalSocket->set_option (boost::asio::socket_base::send_buffer_size (I2P_UDP_SOCKET_BUFFER_SIZE)); m_LocalSocket->set_option (boost::asio::socket_base::reuse_address (true)); m_LocalSocket->non_blocking (true); @@ -431,11 +603,15 @@ namespace client if (m_KeepAliveInterval) ScheduleKeepAliveTimer (); + + m_StatsTimer.reset (new boost::asio::steady_timer (m_LocalDest->GetService ())); + ScheduleStatsTimer (); } void I2PUDPClientTunnel::Stop () { if (m_KeepAliveTimer) m_KeepAliveTimer->cancel (); + if (m_StatsTimer) m_StatsTimer->cancel (); auto dgram = m_LocalDest->GetDatagramDestination (); if (dgram) @@ -507,9 +683,15 @@ namespace client // else fall through and send new first packet, previous one wasn't acked in time } } - if (!m_UnackedDatagrams.empty () && m_NextSendPacketNum > m_UnackedDatagrams.front ().first + I2P_UDP_MAX_NUM_UNACKED_DATAGRAMS) + ExpireUnackedDatagrams (i2p::util::GetMillisecondsSinceEpoch ()); + bool isWindowFull = IsWindowFull (); + if (isWindowFull && i2p::util::GetMillisecondsSinceEpoch () < m_LastWindowProbeTime + GetWindowProbeInterval ()) { // window is full, drop packet + m_NumWindowDrops++; + if (m_NumWindowDrops == 1 || !(m_NumWindowDrops % 100)) + LogPrint (eLogWarning, "UDP Client: Window full (front=", m_UnackedDatagrams.front ().first, + " next=", m_NextSendPacketNum, "), dropped ", m_NumWindowDrops, " packets"); RecvFromLocal (); return; } @@ -529,16 +711,17 @@ namespace client } // send off to remote i2p destination auto ts = i2p::util::GetMillisecondsSinceEpoch (); + if (isWindowFull) m_LastWindowProbeTime = ts; LogPrint (eLogDebug, "UDP Client: Send ", transferred, " to ", Identity.ToBase32 (), ":", RemotePort); auto session = GetDatagramSession (); uint64_t repliableDatagramInterval = I2P_UDP_REPLIABLE_DATAGRAM_INTERVAL; if (m_RTT && m_RTT >= I2P_UDP_REPLIABLE_DATAGRAM_INTERVAL && m_RTT < I2P_UDP_REPLIABLE_DATAGRAM_INTERVAL*10) repliableDatagramInterval = m_RTT/10; // 10 - 100 ms - if (ts > m_LastRepliableDatagramTime + repliableDatagramInterval) + if (isWindowFull || ts > m_LastRepliableDatagramTime + repliableDatagramInterval) { if (m_DatagramVersion == i2p::datagram::eDatagramV3) { uint8_t flags = 0; - if (!m_RTT || !m_AckTimerSeqn || (!m_UnackedDatagrams.empty () && + if (isWindowFull || !m_RTT || !m_AckTimerSeqn || (!m_UnackedDatagrams.empty () && ts > m_UnackedDatagrams.back ().second + repliableDatagramInterval)) // last ack request { flags |= UDP_SESSION_FLAG_ACK_REQUESTED; @@ -577,6 +760,7 @@ namespace client if (numPackets) LogPrint (eLogDebug, "UDP Client: Sent ", numPackets, " more packets to ", Identity.ToBase32 ()); m_NextSendPacketNum += numPackets + 1; + UpdateSendRate (numPackets + 1, ts); m_Destination->FlushSendQueue (session); // mark convo as active @@ -627,6 +811,7 @@ namespace client { if (isIdentity && from.GetIdentHash() == Identity) { + m_LastReceivedTime = i2p::util::GetMillisecondsSinceEpoch (); if (options) { uint32_t seqn = 0; @@ -658,6 +843,7 @@ namespace client void I2PUDPClientTunnel::HandleRecvFromI2PRaw (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len) { + m_LastReceivedTime = i2p::util::GetMillisecondsSinceEpoch (); std::shared_ptr convo; { std::lock_guard lock (m_SessionsMutex); @@ -711,5 +897,31 @@ namespace client ScheduleKeepAliveTimer (); } } + + void I2PUDPClientTunnel::ScheduleStatsTimer () + { + if (m_StatsTimer) + { + m_StatsTimer->expires_after (std::chrono::seconds (10)); + m_StatsTimer->async_wait (std::bind (&I2PUDPClientTunnel::HandleStatsTimer, + this, std::placeholders::_1)); + } + } + + void I2PUDPClientTunnel::HandleStatsTimer (const boost::system::error_code& ecode) + { + if (ecode != boost::asio::error::operation_aborted) + { + auto session = GetDatagramSession (); + LogPrint (eLogDebug, "UDP Client: stats rtt=", m_RTT, "/", m_RTTVar, "/", m_MinRTT, "ms rto=", GetRTO (), + "ms rate=", m_SendRate, "/s window=", GetMaxNumUnackedDatagrams (), + " unacked=", m_UnackedDatagrams.size (), + " nextSeqn=", m_NextSendPacketNum, " winDrops=", m_NumWindowDrops, + " ackTimeouts=", m_NumAckTimeouts, + " pathDrops=", session ? session->GetNumPathDrops () : 0, + " noPathDrops=", session ? session->GetNumDroppedNoPath () : 0); + ScheduleStatsTimer (); + } + } } } diff --git a/libi2pd_client/UDPTunnel.h b/libi2pd_client/UDPTunnel.h index 6c59c0b3..5e5a1430 100644 --- a/libi2pd_client/UDPTunnel.h +++ b/libi2pd_client/UDPTunnel.h @@ -32,11 +32,27 @@ namespace client const uint64_t I2P_UDP_SESSION_TIMEOUT = 1000 * 60 * 2; const uint64_t I2P_UDP_REPLIABLE_DATAGRAM_INTERVAL = 100; // in milliseconds const uint64_t I2P_UDP_MAX_UNACKED_DATAGRAM_TIME = 8000; // in milliseconds + const uint64_t I2P_UDP_MIN_ACK_TIMEOUT = 500; // in milliseconds + // rfc 6298 rto estimator, as denominators: srtt alpha = 1/8, rttvar beta = 1/4 + const uint64_t I2P_UDP_RTT_ALPHA = 8; + const uint64_t I2P_UDP_RTT_VAR_BETA = 4; + const uint64_t I2P_UDP_RTO_K = 4; // rto = srtt + k*rttvar + const uint64_t I2P_UDP_MIN_WINDOW_PROBE_INTERVAL = 50; // in milliseconds const uint64_t I2P_UDP_FIRST_PACKET_RESEND_INTERVAL = 1000; // in milliseconds - const size_t I2P_UDP_MAX_NUM_UNACKED_DATAGRAMS = 500; + const size_t I2P_UDP_MIN_MAX_NUM_UNACKED_DATAGRAMS = 500; + const size_t I2P_UDP_DEFAULT_MAX_NUM_UNACKED_DATAGRAMS = 1000; + const size_t I2P_UDP_MAX_NUM_UNACKED_DATAGRAMS = 8192; + const size_t I2P_UDP_WINDOW_GAIN = 2; // in bandwidth-delay products + const uint64_t I2P_UDP_MIN_RTT_EXPIRATION_TIMEOUT = 15000; // in milliseconds + const uint64_t I2P_UDP_SEND_RATE_INTERVAL = 200; // in milliseconds + const uint64_t I2P_UDP_SEND_RATE_EXPIRATION_TIMEOUT = 5000; // in milliseconds + const uint32_t I2P_UDP_MAX_NUM_ACK_TIMEOUTS = 3; // in a row, before routing path gets reset + const uint64_t I2P_UDP_PEER_PATH_RESET_TIMEOUT = 4*I2P_UDP_MAX_UNACKED_DATAGRAM_TIME; // silence, in milliseconds /** max size for i2p udp */ const size_t I2P_UDP_MAX_MTU = 64*1024; + /** local socket buffers, not datagram size */ + const size_t I2P_UDP_SOCKET_BUFFER_SIZE = 4*1024*1024; struct UDPConnection { @@ -47,16 +63,25 @@ namespace client bool isIdentity = false; uint32_t m_NextSendPacketNum = 1, m_LastReceivedPacketNum = 0; std::list > m_UnackedDatagrams; // list of sent but not acked repliable datagrams(seqn, timestamp) in ascending order - uint64_t m_RTT = 0; // milliseconds + uint64_t m_RTT = 0, m_RTTVar = 0; // milliseconds, smoothed + uint64_t m_MinRTT = 0, m_MinRTTCandidate = 0, m_MinRTTUpdateTime = 0; // path delay, without queueing + uint64_t m_LastReceivedTime = 0; // milliseconds, any datagram from the peer + uint64_t m_SendRateUpdateTime = 0, m_SendRateMaxTime = 0; // milliseconds + uint32_t m_NumSentSinceRateUpdate = 0, m_SendRate = 0; // datagrams per second + size_t m_MaxWindow = I2P_UDP_MIN_MAX_NUM_UNACKED_DATAGRAMS; boost::asio::steady_timer m_AckTimer; uint32_t m_AckTimerSeqn = 0; bool m_IsSendingAllowed = true; bool m_IsFirstPacket = true; + uint32_t m_NumAckTimeoutsInRow = 0; + uint64_t m_LastWindowProbeTime = 0; // milliseconds + uint32_t m_NumWindowDrops = 0, m_NumAckTimeouts = 0; UDPConnection (boost::asio::io_context& service, std::shared_ptr destination): m_Destination (destination), m_LastRepliableDatagramTime (0), m_AckTimer (service) {}; void SetIdentity (const i2p::data::IdentHash& ident) { Identity = ident; isIdentity = true; }; + void SetMaxWindow (size_t w) { m_MaxWindow = w; }; virtual ~UDPConnection () { Stop (); }; virtual void Start () {}; @@ -64,6 +89,13 @@ namespace client void Acked (uint32_t seqn); void ScheduleAckTimer (uint32_t seqn); + void UpdateRTT (uint64_t rtt, uint64_t ts); + void UpdateSendRate (uint32_t numDatagrams, uint64_t ts); + uint64_t GetRTO () const; + uint64_t GetWindowProbeInterval () const; + size_t GetMaxNumUnackedDatagrams () const; + bool IsWindowFull () const; + void ExpireUnackedDatagrams (uint64_t ts); std::shared_ptr GetDatagramSession (); }; @@ -132,6 +164,7 @@ namespace client std::shared_ptr GetLocalDestination () const { return m_LocalDest; } void SetUniqueLocal (bool isUniqueLocal = true) { m_IsUniqueLocal = isUniqueLocal; } + void SetMaxWindow (size_t w) { m_MaxWindow = w; } private: @@ -141,6 +174,9 @@ namespace client UDPSessionPtr ObtainUDPSession (const i2p::data::IdentityEx& from, uint16_t localPort, uint16_t remotePort); uint32_t GetSessionIndex (uint16_t fromPort, uint16_t toPort) const { return ((uint32_t)fromPort << 16) + toPort; } + void ScheduleStatsTimer (); + void HandleStatsTimer (const boost::system::error_code& ecode); + private: bool m_IsUniqueLocal; @@ -153,6 +189,9 @@ namespace client UDPSessionPtr m_LastSession; uint16_t m_inPort; bool m_Gzip; + uint32_t m_NumRawNoSession = 0; + size_t m_MaxWindow = I2P_UDP_MIN_MAX_NUM_UNACKED_DATAGRAMS; + std::unique_ptr m_StatsTimer; public: @@ -199,6 +238,9 @@ namespace client void ScheduleKeepAliveTimer (); void HandleKeepAliveTimer (const boost::system::error_code& ecode); + void ScheduleStatsTimer (); + void HandleStatsTimer (const boost::system::error_code& ecode); + private: const std::string m_Name; @@ -218,6 +260,7 @@ namespace client std::shared_ptr m_LastSession; uint32_t m_KeepAliveInterval = 0; std::unique_ptr m_KeepAliveTimer; + std::unique_ptr m_StatsTimer; public: From 72b8ef3eb4f49842d25b69b97243a9e1f86c30b6 Mon Sep 17 00:00:00 2001 From: acetone <63557806+freeacetone@users.noreply.github.com> Date: Sat, 8 Aug 2026 17:41:36 +0300 Subject: [PATCH 2/2] configurable udp tunnel window --- libi2pd_client/ClientContext.cpp | 10 ++++++++++ libi2pd_client/ClientContext.h | 1 + 2 files changed, 11 insertions(+) diff --git a/libi2pd_client/ClientContext.cpp b/libi2pd_client/ClientContext.cpp index a8cece86..8748f694 100644 --- a/libi2pd_client/ClientContext.cpp +++ b/libi2pd_client/ClientContext.cpp @@ -564,6 +564,14 @@ namespace client options.Insert (I2CP_PARAM_STREAMING_MAX_RESENDS, value); } + static size_t GetUDPTunnelMaxWindow (const boost::property_tree::ptree& section) + { + size_t w = section.get (UDP_TUNNEL_MAX_WINDOW, I2P_UDP_DEFAULT_MAX_NUM_UNACKED_DATAGRAMS); + if (w < I2P_UDP_MIN_MAX_NUM_UNACKED_DATAGRAMS) w = I2P_UDP_MIN_MAX_NUM_UNACKED_DATAGRAMS; + if (w > I2P_UDP_MAX_NUM_UNACKED_DATAGRAMS) w = I2P_UDP_MAX_NUM_UNACKED_DATAGRAMS; + return w; + } + void ClientContext::ReadTunnels () { int numClientTunnels = 0, numServerTunnels = 0; @@ -686,6 +694,7 @@ namespace client int datagramVersion = (i2p::datagram::DatagramVersion)section.second.get (UDP_CLIENT_TUNNEL_DATAGRAM_VERSION, (int)i2p::datagram::eDatagramV3); auto clientTunnel = std::make_shared (name, dest, end, localDestination, destinationPort, gzip, (i2p::datagram::DatagramVersion)datagramVersion); + clientTunnel->SetMaxWindow (GetUDPTunnelMaxWindow (section.second)); uint32_t keepAlive = section.second.get(I2P_CLIENT_TUNNEL_KEEP_ALIVE_INTERVAL, 0); if (keepAlive) @@ -864,6 +873,7 @@ namespace client } auto localAddress = boost::asio::ip::make_address(address); auto serverTunnel = std::make_shared(name, localDestination, localAddress, endpoint, inPort, gzip); + serverTunnel->SetMaxWindow (GetUDPTunnelMaxWindow (section.second)); if(!isUniqueLocal) { LogPrint(eLogInfo, "Clients: Disabling loopback address mapping"); diff --git a/libi2pd_client/ClientContext.h b/libi2pd_client/ClientContext.h index 0970dd42..ad1a6af7 100644 --- a/libi2pd_client/ClientContext.h +++ b/libi2pd_client/ClientContext.h @@ -69,6 +69,7 @@ namespace client const char I2P_SERVER_TUNNEL_ENABLE_UNIQUE_LOCAL[] = "enableuniquelocal"; const char I2P_SERVER_TUNNEL_SSL[] = "ssl"; const char UDP_CLIENT_TUNNEL_DATAGRAM_VERSION[] = "datagramversion"; + const char UDP_TUNNEL_MAX_WINDOW[] = "maxwindow"; const char TORRENTS_TUNNEL_KEYS[] = "keys"; const char TORRENTS_TUNNEL_SIGNATURE_TYPE[] = "signaturetype"; const char TORRENTS_TUNNEL_TORRENTS_DIR[] = "torrentsdir";