Merge pull request #2464 from freeacetone/udp-tunnel-performance

Udp tunnel performance (2)

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();
 	}
 }

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<const i2p::data::IdentHash> 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> DatagramSession_ptr;

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<size_t> (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<I2PUDPClientTunnel> (name, dest, end,
 							localDestination, destinationPort, gzip, (i2p::datagram::DatagramVersion)datagramVersion);
+						clientTunnel->SetMaxWindow (GetUDPTunnelMaxWindow (section.second));
 
 						uint32_t keepAlive = section.second.get<uint32_t>(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<I2PUDPServerTunnel>(name, localDestination, localAddress, endpoint, inPort, gzip);
+						serverTunnel->SetMaxWindow (GetUDPTunnelMaxWindow (section.second));
 						if(!isUniqueLocal)
 						{
 							LogPrint(eLogInfo, "Clients: Disabling loopback address mapping");

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";

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<UDPSession>(boost::asio::ip::udp::endpoint(addr, 0),
 			m_LocalDest, m_RemoteEndpoint, ih, localPort, remotePort);
+		s->SetMaxWindow (m_MaxWindow);
 		std::lock_guard<std::mutex> 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<std::mutex> 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<UDPConvo> convo;
 		{
 			std::lock_guard<std::mutex> 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 ();
+		}
+	}
 }
 }

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<std::pair<uint32_t, uint64_t> > 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<i2p::datagram::DatagramDestination> 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<i2p::datagram::DatagramSession> GetDatagramSession ();
 	};
@@ -132,6 +164,7 @@ namespace client
 			std::shared_ptr<ClientDestination> 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<boost::asio::steady_timer> 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<UDPConvo> m_LastSession;
 			uint32_t m_KeepAliveInterval = 0;
 			std::unique_ptr<boost::asio::steady_timer> m_KeepAliveTimer;
+			std::unique_ptr<boost::asio::steady_timer> m_StatsTimer;
 
 		public: