shedule ack timer for repliable datagram3

libi2pd_client/UDPTunnel.cpp

@@ -283,7 +283,8 @@ namespace client
 		m_Name (name), m_RemoteDest (remoteDest), m_LocalDest (localDestination), m_LocalEndpoint (localEndpoint),
 		m_ResolveThread (nullptr), m_LocalSocket (nullptr), RemotePort (remotePort),
 		m_LastPort (0), m_cancel_resolve (false), m_Gzip (gzip), m_DatagramVersion (datagramVersion),
-		m_NextSendPacketNum (1), m_LastReceivedPacketNum (0), m_RTT (0)
+		m_NextSendPacketNum (1), m_LastReceivedPacketNum (0), m_RTT (0),
+		m_AckTimer (localDestination->GetService ()), m_AckTimerSeqn (0)
 	{
 	}
 
@@ -341,6 +342,7 @@ namespace client
 			m_ResolveThread = nullptr;
 		}
 		m_RemoteAddr = nullptr;
+		m_AckTimer.cancel ();
 	}
 
 	void I2PUDPClientTunnel::RecvFromLocal ()
@@ -390,8 +392,7 @@ namespace client
 			if (m_DatagramVersion == i2p::datagram::eDatagramV3)
 			{	
 				uint8_t flags = 0;
-				if (!m_UnackedDatagrams.empty () && (ts > m_UnackedDatagrams.front ().second + I2P_UDP_MAX_UNACKED_DATAGRAM_TIME ||
-				 	m_NextSendPacketNum > m_UnackedDatagrams.front ().first + I2P_UDP_MAX_NUM_UNACKED_DATAGRAMS))
+				if (!m_UnackedDatagrams.empty () && m_NextSendPacketNum > m_UnackedDatagrams.front ().first + I2P_UDP_MAX_NUM_UNACKED_DATAGRAMS)
 				{
 					m_UnackedDatagrams.clear ();
 					session->DropSharedRoutingPath ();
@@ -406,6 +407,7 @@ namespace client
 				if (flags)
 					options.Put (UDP_SESSION_FLAGS, flags);
 				m_LocalDest->GetDatagramDestination ()->SendDatagram (session, m_RecvBuff, transferred, remotePort, RemotePort, &options);
+				ScheduleAckTimer (m_NextSendPacketNum);
 			}
 			else
 				m_LocalDest->GetDatagramDestination ()->SendDatagram (session, m_RecvBuff, transferred, remotePort, RemotePort);
@@ -527,6 +529,11 @@ namespace client
 
 	void I2PUDPClientTunnel::Acked (uint32_t seqn)
 	{
+		if (m_AckTimerSeqn && seqn >= m_AckTimerSeqn)
+		{
+			m_AckTimerSeqn = 0;
+			m_AckTimer.cancel ();
+		}	
 		if (m_UnackedDatagrams.empty () && seqn < m_UnackedDatagrams.front ().first) return;
 		auto it = m_UnackedDatagrams.begin ();
 		while (it != m_UnackedDatagrams.end ())
@@ -538,6 +545,30 @@ namespace client
 		}	
 		m_UnackedDatagrams.erase (m_UnackedDatagrams.begin (), it);
 	}	
+
+	void I2PUDPClientTunnel::ScheduleAckTimer (uint32_t seqn)
+	{
+		if (!m_AckTimerSeqn)
+		{
+			m_AckTimerSeqn = seqn;
+			m_AckTimer.expires_from_now (boost::posix_time::milliseconds (m_RTT ? 2*m_RTT : I2P_UDP_MAX_UNACKED_DATAGRAM_TIME));
+			m_AckTimer.async_wait ([this](const boost::system::error_code& ecode)
+				{
+					if (ecode != boost::asio::error::operation_aborted)
+					{
+						LogPrint (eLogInfo, "UDP Client: Packet ", m_AckTimerSeqn, " was not acked");
+						m_AckTimerSeqn = 0;
+						m_RTT = 0;
+						// send empty packet with reset path flag
+						i2p::util::Mapping options;
+						options.Put (UDP_SESSION_FLAGS, UDP_SESSION_FLAG_RESET_PATH);
+						auto session = GetDatagramSession ();
+						session->DropSharedRoutingPath ();
+						m_LocalDest->GetDatagramDestination ()->SendDatagram (session, nullptr, 0, 0, 0, &options);
+					}
+				});
+		}	
+	}	
 		
 }
 }

libi2pd_client/UDPTunnel.h

@@ -36,7 +36,7 @@ namespace client
 	/** max size for i2p udp */
 	const size_t I2P_UDP_MAX_MTU = 64*1024;
 	
-	struct UDPSession
+	struct UDPSession // for server side
 	{
 		i2p::datagram::DatagramDestination * m_Destination;
 		std::weak_ptr<i2p::datagram::DatagramSession> m_LastDatagramSession;
@@ -169,6 +169,7 @@ namespace client
 			void TryResolving ();
 			std::shared_ptr<i2p::datagram::DatagramSession> GetDatagramSession ();
 			void Acked (uint32_t seqn);
+			void ScheduleAckTimer (uint32_t seqn);
 			
 		private:
 
@@ -192,6 +193,8 @@ namespace client
 			uint32_t m_NextSendPacketNum, m_LastReceivedPacketNum;
 			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; // milliseconds
+			boost::asio::deadline_timer m_AckTimer;
+			uint32_t m_AckTimerSeqn;
 			
 		public: