hanlde Ack block

libi2pd/SSU2.cpp

@@ -582,9 +582,11 @@ namespace transport
 		}	
 		m_LastActivityTimestamp = i2p::util::GetSecondsSinceEpoch ();
 		m_NumReceivedBytes += len;
-		UpdateReceivePacketNum (packetNum);
+		if (UpdateReceivePacketNum (packetNum))
-		if (HandlePayload (payload, payloadSize))
+		{	
-			SendQuickAck (); // TODO: don't send too requently
+			if (HandlePayload (payload, payloadSize))
+				SendQuickAck (); // TODO: don't send too requently
+		}	
 	}	
 		
 	bool SSU2Session::HandlePayload (const uint8_t * buf, size_t len)
@@ -652,6 +654,8 @@ namespace transport
 				case eSSU2BlkNextNonce:
 				break;	
 				case eSSU2BlkAck:
+					LogPrint (eLogDebug, "SSU2: Ack");
+					HandleAck (buf + offset, size);
 				break;	
 				case eSSU2BlkAddress:
 				{
@@ -692,6 +696,26 @@ namespace transport
 		return isData;
 	}	
 
+	void SSU2Session::HandleAck (const uint8_t * buf, size_t len)
+	{
+		if (m_SentPackets.empty ()) return;
+		uint32_t ackThrough = bufbe32toh (buf);
+		auto it = m_SentPackets.rbegin ();
+		while (it != m_SentPackets.rend () && it->first > ackThrough) ++it; // find first less pack <= ackThrough
+		if (it == m_SentPackets.rend ()) return;
+		int32_t firstPacketNum = ackThrough - buf[4]; // acnt
+		if (firstPacketNum < 0) firstPacketNum = 0;
+		auto it1 = it;
+		while (it1 != m_SentPackets.rend () && it1->first >= (uint32_t)firstPacketNum) it1++;
+		if (it1 == m_SentPackets.rend ())
+		{
+			m_SentPackets.erase (m_SentPackets.begin (), it.base ());
+			return;
+		}	
+		m_SentPackets.erase (it1.base (), it.base ());
+		// TODO: handle ranges
+	}
+		
 	bool SSU2Session::ExtractEndpoint (const uint8_t * buf, size_t size, boost::asio::ip::udp::endpoint& ep)
 	{
 		if (size < 2) return false;
@@ -800,9 +824,9 @@ namespace transport
 		htole64buf (nonce + 4, seqn);
 	}
 
-	void SSU2Session::UpdateReceivePacketNum (uint32_t packetNum)
+	bool SSU2Session::UpdateReceivePacketNum (uint32_t packetNum)
 	{
-		if (packetNum <= m_ReceivePacketNum) return; // duplicate 
+		if (packetNum <= m_ReceivePacketNum) return false; // duplicate 
 		if (packetNum == m_ReceivePacketNum + 1)
 		{
 			for (auto it = m_OutOfSequencePackets.begin (); it != m_OutOfSequencePackets.end ();)
@@ -819,6 +843,7 @@ namespace transport
 		}	
 		else
 			m_OutOfSequencePackets.insert (packetNum);
+		return true;
 	}	
 		
 	void SSU2Session::SendQuickAck ()

libi2pd/SSU2.h

@@ -95,6 +95,15 @@ namespace transport
 			} h;
 		};
 
+		struct SentPacket
+		{
+			Header h;
+			uint8_t payload[SSU2_MTU];
+			size_t payloadLen;            
+			uint32_t nextResendTime; // in seconds
+			int numResends;
+		};	
+		
 		public:
 
 			SSU2Session (SSU2Server& server, std::shared_ptr<const i2p::data::RouterInfo> in_RemoteRouter = nullptr,
@@ -136,10 +145,11 @@ namespace transport
 			void SendTermination ();
 			
 			bool HandlePayload (const uint8_t * buf, size_t len); // returns true is contains data
+			void HandleAck (const uint8_t * buf, size_t len);
 			bool ExtractEndpoint (const uint8_t * buf, size_t size, boost::asio::ip::udp::endpoint& ep);
 			std::shared_ptr<const i2p::data::RouterInfo> ExtractRouterInfo (const uint8_t * buf, size_t size);
 			void CreateNonce (uint64_t seqn, uint8_t * nonce);
-			void UpdateReceivePacketNum (uint32_t packetNum); // for Ack
+			bool UpdateReceivePacketNum (uint32_t packetNum); // for Ack, returns false if duplicate
 
 			size_t CreateAddressBlock (const boost::asio::ip::udp::endpoint& ep, uint8_t * buf, size_t len);
 			size_t CreateAckBlock (uint8_t * buf, size_t len);
@@ -157,6 +167,7 @@ namespace transport
 			uint8_t m_KeyDataSend[64], m_KeyDataReceive[64]; 
 			uint32_t m_SendPacketNum, m_ReceivePacketNum;
 			std::set<uint32_t> m_OutOfSequencePackets; // packet nums > receive packet num
+			std::map<uint32_t, std::shared_ptr<SentPacket> > m_SentPackets; // packetNum -> packet
 			i2p::I2NPMessagesHandler m_Handler;
 	};