Merge pull request #2039 from Vort/ssu2_expiration

add expiration for messages in SSU2 send queue

libi2pd/I2NPProtocol.h

@@ -152,6 +152,7 @@ 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_EXPIRATION_TIMEOUT = 8000; // in milliseconds (as initial RTT)
 	const unsigned int I2NP_MESSAGE_CLOCK_SKEW = 60*1000; // 1 minute in milliseconds
 
@@ -161,9 +162,10 @@ namespace tunnel
 		size_t len, offset, maxLen;
 		std::shared_ptr<i2p::tunnel::InboundTunnel> from;
 		std::function<void ()> onDrop;
-		
-		I2NPMessage (): buf (nullptr),len (I2NP_HEADER_SIZE + 2),
-			offset(2), maxLen (0), from (nullptr) {}; // reserve 2 bytes for NTCP header
+		uint64_t localExpiration; // monotonic microseconds
+
+		I2NPMessage (): buf (nullptr), len (I2NP_HEADER_SIZE + 2),
+			offset(2), maxLen (0), from (nullptr), localExpiration(0) {}; // reserve 2 bytes for NTCP header
 
 		// header accessors
 		uint8_t * GetHeader () { return GetBuffer (); };
@@ -173,6 +175,7 @@ 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; };
 		uint64_t GetExpiration () const { return bufbe64toh (GetHeader () + I2NP_HEADER_EXPIRATION_OFFSET); };
 		void SetSize (uint16_t size) { htobe16buf (GetHeader () + I2NP_HEADER_SIZE_OFFSET, size); };
 		uint16_t GetSize () const { return bufbe16toh (GetHeader () + I2NP_HEADER_SIZE_OFFSET); };
@@ -264,6 +267,8 @@ 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_EXPIRATION_TIMEOUT / 2; }; // monotonic microseconds
 
 		void Drop () { if (onDrop) { onDrop (); onDrop = nullptr; }; }
 	};

libi2pd/SSU2Session.cpp

@@ -353,25 +353,32 @@ namespace transport
 	void SSU2Session::PostI2NPMessages (std::vector<std::shared_ptr<I2NPMessage> > msgs)
 	{
 		if (m_State == eSSU2SessionStateTerminated) return;
-		bool isSemiFull = m_SendQueue.size () > SSU2_MAX_OUTGOING_QUEUE_SIZE/2;
+		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);
+			if (isSemiFull)
+			{
+				LogPrint (eLogWarning, "SSU2: Outgoing messages queue to ",
+					GetIdentHashBase64 (), " is semi-full (", m_SendQueue.size (), ")");
+			}
+		}
 		for (auto it: msgs)
+		{
 			if (isSemiFull && it->onDrop)
 				it->Drop (); // drop earlier because we can handle it
 			else
+			{
+				it->SetLocalExpiration (localExpiration);
 				m_SendQueue.push_back (std::move (it));
+			}
+		}
 		SendQueue ();
 
 		if (m_SendQueue.size () > 0) // windows is full
-		{
-			if (m_SendQueue.size () <= SSU2_MAX_OUTGOING_QUEUE_SIZE)
-				Resend (i2p::util::GetMillisecondsSinceEpoch ());
-			else
-			{
-				LogPrint (eLogWarning, "SSU2: Outgoing messages queue size to ",
-					GetIdentHashBase64(), " exceeds ", SSU2_MAX_OUTGOING_QUEUE_SIZE);
-				RequestTermination (eSSU2TerminationReasonTimeout);
-			}
-		}
+			Resend (i2p::util::GetMillisecondsSinceEpoch ());
 		SetSendQueueSize (m_SendQueue.size ());
 	}
 
@@ -380,6 +387,7 @@ namespace transport
 		if (!m_SendQueue.empty () && m_SentPackets.size () <= m_WindowSize)
 		{
 			auto ts = i2p::util::GetMillisecondsSinceEpoch ();
+			uint64_t mts = i2p::util::GetMonotonicMicroseconds ();
 			auto packet = m_Server.GetSentPacketsPool ().AcquireShared ();
 			size_t ackBlockSize = CreateAckBlock (packet->payload, m_MaxPayloadSize);
 			bool ackBlockSent = false;
@@ -387,7 +395,7 @@ namespace transport
 			while (!m_SendQueue.empty () && m_SentPackets.size () <= m_WindowSize)
 			{
 				auto msg = m_SendQueue.front ();
-				if (!msg || msg->IsExpired (ts))
+				if (!msg || msg->IsExpired (ts) || msg->IsLocalExpired (mts))
 				{
 					// drop null or expired message
 					if (msg) msg->Drop ();

libi2pd/SSU2Session.h

@@ -47,7 +47,6 @@ namespace transport
 	const size_t SSU2_MIN_RTO = 100; // in milliseconds
 	const size_t SSU2_MAX_RTO = 2500; // in milliseconds
 	const float SSU2_kAPPA = 1.8;
-	const size_t SSU2_MAX_OUTGOING_QUEUE_SIZE = 500; // in messages
 	const int SSU2_MAX_NUM_ACNT = 255; // acnt, acks or nacks
 	const int SSU2_MAX_NUM_ACK_PACKETS = 511; // ackthrough + acnt + 1 range
 	const int SSU2_MAX_NUM_ACK_RANGES = 32; // to send