select maximum SSU2 queue size depending on RTT value

libi2pd/I2NPProtocol.h

@@ -152,7 +152,9 @@ namespace tunnel
 	const size_t I2NP_MAX_MESSAGE_SIZE = 62708;
 	const size_t I2NP_MAX_SHORT_MESSAGE_SIZE = 4096;
 	const size_t I2NP_MAX_MEDIUM_MESSAGE_SIZE = 16384;
-	const unsigned int I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT = 2000000; // in microseconds
+	const unsigned int I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_FACTOR = 3; // multiples of RTT
+	const unsigned int I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_MIN = 200000; // in microseconds
+	const unsigned int I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_MAX = 2000000; // in microseconds
 	const unsigned int I2NP_MESSAGE_EXPIRATION_TIMEOUT = 8000; // in milliseconds (as initial RTT)
 	const unsigned int I2NP_MESSAGE_CLOCK_SKEW = 60*1000; // 1 minute in milliseconds
 
@@ -162,10 +164,10 @@ namespace tunnel
 		size_t len, offset, maxLen;
 		std::shared_ptr<i2p::tunnel::InboundTunnel> from;
 		std::function<void ()> onDrop;
-		uint64_t localExpiration; // monotonic microseconds
+		uint64_t enqueueTime; // monotonic microseconds
 
 		I2NPMessage (): buf (nullptr), len (I2NP_HEADER_SIZE + 2),
-			offset(2), maxLen (0), from (nullptr), localExpiration(0) {}; // reserve 2 bytes for NTCP header
+			offset(2), maxLen (0), from (nullptr), enqueueTime (0) {}; // reserve 2 bytes for NTCP header
 
 		// header accessors
 		uint8_t * GetHeader () { return GetBuffer (); };
@@ -175,8 +177,9 @@ namespace tunnel
 		void SetMsgID (uint32_t msgID) { htobe32buf (GetHeader () + I2NP_HEADER_MSGID_OFFSET, msgID); };
 		uint32_t GetMsgID () const { return bufbe32toh (GetHeader () + I2NP_HEADER_MSGID_OFFSET); };
 		void SetExpiration (uint64_t expiration) { htobe64buf (GetHeader () + I2NP_HEADER_EXPIRATION_OFFSET, expiration); };
-		void SetLocalExpiration (uint64_t expiration) { localExpiration = expiration; };
+		void SetEnqueueTime (uint64_t mts) { enqueueTime = mts; };
 		uint64_t GetExpiration () const { return bufbe64toh (GetHeader () + I2NP_HEADER_EXPIRATION_OFFSET); };
+		uint64_t GetEnqueueTime () const { return enqueueTime; };
 		void SetSize (uint16_t size) { htobe16buf (GetHeader () + I2NP_HEADER_SIZE_OFFSET, size); };
 		uint16_t GetSize () const { return bufbe16toh (GetHeader () + I2NP_HEADER_SIZE_OFFSET); };
 		void UpdateSize () { SetSize (GetPayloadLength ()); };
@@ -267,8 +270,6 @@ namespace tunnel
 		void RenewI2NPMessageHeader ();
 		bool IsExpired () const;
 		bool IsExpired (uint64_t ts) const; // in milliseconds
-		bool IsLocalExpired (uint64_t mts) const { return mts > localExpiration; }; // monotonic microseconds
-		bool IsLocalSemiExpired (uint64_t mts) const { return mts > localExpiration - I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT / 2; }; // monotonic microseconds
 
 		void Drop () { if (onDrop) { onDrop (); onDrop = nullptr; }; }
 	};

libi2pd/SSU2Session.cpp

@@ -84,7 +84,10 @@ namespace transport
 		m_Server (server), m_Address (addr), m_RemoteTransports (0), m_RemotePeerTestTransports (0),
 		m_DestConnID (0), m_SourceConnID (0), m_State (eSSU2SessionStateUnknown),
 		m_SendPacketNum (0), m_ReceivePacketNum (0), m_LastDatetimeSentPacketNum (0),
-		m_IsDataReceived (false), m_RTT (SSU2_UNKNOWN_RTT), m_WindowSize (SSU2_MIN_WINDOW_SIZE), 
+		m_IsDataReceived (false), m_RTT (SSU2_UNKNOWN_RTT),
+		m_MsgLocalExpirationTimeout (I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_MAX),
+		m_MsgLocalSemiExpirationTimeout (I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_MAX / 2),
+		m_WindowSize (SSU2_MIN_WINDOW_SIZE),
 		m_RTO (SSU2_INITIAL_RTO), m_RelayTag (0),m_ConnectTimer (server.GetService ()), 
 		m_TerminationReason (eSSU2TerminationReasonNormalClose),
 		m_MaxPayloadSize (SSU2_MIN_PACKET_SIZE - IPV6_HEADER_SIZE - UDP_HEADER_SIZE - 32) // min size
@@ -356,15 +359,16 @@ namespace transport
 	{
 		if (m_State == eSSU2SessionStateTerminated) return;
 		uint64_t mts = i2p::util::GetMonotonicMicroseconds ();
-		uint64_t localExpiration = mts + I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT;
 		bool isSemiFull = false;
 		if (m_SendQueue.size ())
 		{
-			isSemiFull = m_SendQueue.front ()->IsLocalSemiExpired (mts);
+			int64_t queueLag = (int64_t)mts - (int64_t)m_SendQueue.front ()->GetEnqueueTime ();
+			isSemiFull = queueLag > m_MsgLocalSemiExpirationTimeout;
 			if (isSemiFull)
 			{
 				LogPrint (eLogWarning, "SSU2: Outgoing messages queue to ",
-					GetIdentHashBase64 (), " is semi-full (", m_SendQueue.size (), ")");
+					i2p::data::GetIdentHashAbbreviation (GetRemoteIdentity ()->GetIdentHash ()),
+					" is semi-full (size = ", m_SendQueue.size (), ", lag = ", queueLag / 1000, ", rtt = ", (int)m_RTT, ")");
 			}
 		}
 		for (auto it: msgs)
@@ -373,7 +377,7 @@ namespace transport
 				it->Drop (); // drop earlier because we can handle it
 			else
 			{
-				it->SetLocalExpiration (localExpiration);
+				it->SetEnqueueTime (mts);
 				m_SendQueue.push_back (std::move (it));
 			}
 		}
@@ -397,7 +401,7 @@ namespace transport
 			while (!m_SendQueue.empty () && m_SentPackets.size () <= m_WindowSize)
 			{
 				auto msg = m_SendQueue.front ();
-				if (!msg || msg->IsExpired (ts) || msg->IsLocalExpired (mts))
+				if (!msg || msg->IsExpired (ts) || msg->GetEnqueueTime() + m_MsgLocalExpirationTimeout < mts)
 				{
 					// drop null or expired message
 					if (msg) msg->Drop ();
@@ -1758,6 +1762,10 @@ namespace transport
 					else
 						m_RTT = rtt;
 					m_RTO = m_RTT*SSU2_kAPPA;
+					m_MsgLocalExpirationTimeout = std::max (I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_MIN,
+						std::min (I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_MAX,
+						(unsigned int)(m_RTT * 1000 * I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_FACTOR)));
+					m_MsgLocalSemiExpirationTimeout = m_MsgLocalExpirationTimeout / 2;
 					if (m_RTO < SSU2_MIN_RTO) m_RTO = SSU2_MIN_RTO;
 					if (m_RTO > SSU2_MAX_RTO) m_RTO = SSU2_MAX_RTO;
 				}

libi2pd/SSU2Session.h

@@ -359,6 +359,8 @@ namespace transport
 			i2p::I2NPMessagesHandler m_Handler;
 			bool m_IsDataReceived;
 			double m_RTT;
+			int m_MsgLocalExpirationTimeout;
+			int m_MsgLocalSemiExpirationTimeout;
 			size_t m_WindowSize, m_RTO;
 			uint32_t m_RelayTag; // between Bob and Charlie
 			OnEstablished m_OnEstablished; // callback from Established