Changing the window calculation algorithm, increasing the minimum window size

libi2pd/NTCP2.cpp

@@ -344,6 +344,7 @@ namespace transport
 		m_Server (server), m_Socket (m_Server.GetService ()),
 		m_IsEstablished (false), m_IsTerminated (false),
 		m_Establisher (new NTCP2Establisher),
+		m_SendKey (nullptr), m_ReceiveKey (nullptr),
 #if OPENSSL_SIPHASH
 		m_SendMDCtx(nullptr), m_ReceiveMDCtx (nullptr),
 #else
@@ -721,6 +722,7 @@ namespace transport
 
 	void NTCP2Session::HandleSessionConfirmedReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred)
 	{
+		(void) bytes_transferred;
 		if (ecode)
 		{
 			LogPrint (eLogWarning, "NTCP2: SessionConfirmed read error: ", ecode.message ());
@@ -729,123 +731,143 @@ namespace transport
 		else
 		{
 			m_HandshakeInterval = i2p::util::GetMillisecondsSinceEpoch () - m_HandshakeInterval;
-			LogPrint (eLogDebug, "NTCP2: SessionConfirmed received");
-			// part 1
-			uint8_t nonce[12];
-			CreateNonce (1, nonce);
-			if (m_Establisher->ProcessSessionConfirmedMessagePart1 (nonce))
-			{
-				// part 2
-				std::vector<uint8_t> buf(m_Establisher->m3p2Len - 16); // -MAC
-				memset (nonce, 0, 12); // set nonce to 0 again
-				if (m_Establisher->ProcessSessionConfirmedMessagePart2 (nonce, buf.data ())) // TODO:handle in establisher thread
+			boost::asio::post (m_Server.GetEstablisherService (), 
+				[s = shared_from_this ()] ()
 				{
-					// payload 
-					// RI block must be first 
-					if (buf[0] != eNTCP2BlkRouterInfo)
-					{
-						LogPrint (eLogWarning, "NTCP2: Unexpected block ", (int)buf[0], " in SessionConfirmed");
-						Terminate ();
-						return;
-					}
-					auto size = bufbe16toh (buf.data () + 1);
-					if (size > buf.size () - 3 || size > i2p::data::MAX_RI_BUFFER_SIZE + 1)
-					{
-						LogPrint (eLogError, "NTCP2: Unexpected RouterInfo size ", size, " in SessionConfirmed");
-						Terminate ();
-						return;
-					}
-					// TODO: check flag
-					i2p::data::RouterInfo ri (buf.data () + 4, size - 1); // 1 byte block type + 2 bytes size + 1 byte flag
-					if (ri.IsUnreachable ())
-					{
-						LogPrint (eLogError, "NTCP2: RouterInfo verification failed in SessionConfirmed from ", GetRemoteEndpoint ());
-						SendTerminationAndTerminate (eNTCP2RouterInfoSignatureVerificationFail);
-						return;
-					}
-					LogPrint(eLogDebug, "NTCP2: SessionConfirmed from ", GetRemoteEndpoint (),
-						" (", i2p::data::GetIdentHashAbbreviation (ri.GetIdentHash ()), ")");
-					auto ts = i2p::util::GetMillisecondsSinceEpoch ();
-					if (ts > ri.GetTimestamp () + i2p::data::NETDB_MIN_EXPIRATION_TIMEOUT*1000LL) // 90 minutes
-					{
-						LogPrint (eLogError, "NTCP2: RouterInfo is too old in SessionConfirmed for ", (ts - ri.GetTimestamp ())/1000LL, " seconds");
-						SendTerminationAndTerminate (eNTCP2Message3Error);
-						return;
-					}
-					if (ts + i2p::data::NETDB_EXPIRATION_TIMEOUT_THRESHOLD*1000LL < ri.GetTimestamp ()) // 2 minutes
-					{
-						LogPrint (eLogError, "NTCP2: RouterInfo is from future for ", (ri.GetTimestamp () - ts)/1000LL, " seconds");
-						SendTerminationAndTerminate (eNTCP2Message3Error);
-						return;
-					}	
-					// update RouterInfo in netdb
-					auto ri1 = i2p::data::netdb.AddRouterInfo (ri.GetBuffer (), ri.GetBufferLen ()); // ri1 points to one from netdb now
-					if (!ri1)
-					{
-						LogPrint (eLogError, "NTCP2: Couldn't update RouterInfo from SessionConfirmed in netdb");
-						Terminate ();
-						return;
-					}
-					std::shared_ptr<i2p::data::RouterProfile> profile; // not null if older 
-					if (ri.GetTimestamp () + i2p::data::NETDB_EXPIRATION_TIMEOUT_THRESHOLD*1000LL < ri1->GetTimestamp ())
-					{	
-						// received RouterInfo is older than one in netdb
-						profile = i2p::data::GetRouterProfile (ri1->GetIdentHash ()); // retrieve profile	
-						if (profile && profile->IsDuplicated ())
-						{	
-							SendTerminationAndTerminate (eNTCP2Banned);
-							return;
-						}	
-					}
-					
-					auto addr = m_RemoteEndpoint.address ().is_v4 () ? ri1->GetNTCP2V4Address () :
-						(i2p::util::net::IsYggdrasilAddress (m_RemoteEndpoint.address ()) ? ri1->GetYggdrasilAddress () : ri1->GetNTCP2V6Address ());
-					if (!addr || memcmp (m_Establisher->m_RemoteStaticKey, addr->s, 32))
-					{
-						LogPrint (eLogError, "NTCP2: Wrong static key in SessionConfirmed");
-						Terminate ();
-						return;
-					}
-					if (addr->IsPublishedNTCP2 () && m_RemoteEndpoint.address () != addr->host &&
-					    (!m_RemoteEndpoint.address ().is_v6 () || (i2p::util::net::IsYggdrasilAddress (m_RemoteEndpoint.address ()) ?
-					     memcmp (m_RemoteEndpoint.address ().to_v6 ().to_bytes ().data () + 1, addr->host.to_v6 ().to_bytes ().data () + 1, 7) : // from the same yggdrasil subnet
-					     memcmp (m_RemoteEndpoint.address ().to_v6 ().to_bytes ().data (), addr->host.to_v6 ().to_bytes ().data (), 8)))) // temporary address
-					{
-						if (profile) // older router?
-							profile->Duplicated (); // mark router as duplicated in profile
-						else
-							LogPrint (eLogInfo, "NTCP2: Host mismatch between published address ", addr->host, " and actual endpoint ", m_RemoteEndpoint.address ());
-						SendTerminationAndTerminate (eNTCP2Banned);
-						return;
-					}
-					// TODO: process options block
-
-					// ready to communicate
-					SetRemoteIdentity (ri1->GetRouterIdentity ());
-					if (m_Server.AddNTCP2Session (shared_from_this (), true))
-					{
-						KeyDerivationFunctionDataPhase ();
-						// Bob data phase keys
-						m_SendKey = m_Kba;
-						m_ReceiveKey = m_Kab;
-						SetSipKeys (m_Sipkeysba, m_Sipkeysab);
-						memcpy (m_ReceiveIV.buf, m_Sipkeysab + 16, 8);
-						memcpy (m_SendIV.buf, m_Sipkeysba + 16, 8);
-						
-						Established ();
-						ReceiveLength ();
-					}
-					else
-						Terminate ();
-				}
-				else
-					Terminate ();
-			}
-			else
-				Terminate ();
+					s->ProcessSessionConfirmed ();;
+				});	
 		}
 	}
 
+	void NTCP2Session::ProcessSessionConfirmed ()
+	{
+		// run on establisher thread
+		LogPrint (eLogDebug, "NTCP2: SessionConfirmed received");
+		// part 1
+		uint8_t nonce[12];
+		CreateNonce (1, nonce);
+		if (m_Establisher->ProcessSessionConfirmedMessagePart1 (nonce))
+		{
+			// part 2
+			auto buf = std::make_shared<std::vector<uint8_t> > (m_Establisher->m3p2Len - 16); // -MAC
+			memset (nonce, 0, 12); // set nonce to 0 again
+			if (m_Establisher->ProcessSessionConfirmedMessagePart2 (nonce, buf->data ())) // TODO:handle in establisher thread
+			{
+				// payload 
+				// RI block must be first 
+				if ((*buf)[0] != eNTCP2BlkRouterInfo)
+				{
+					LogPrint (eLogWarning, "NTCP2: Unexpected block ", (int)(*buf)[0], " in SessionConfirmed");
+					boost::asio::post (m_Server.GetService (), std::bind (&NTCP2Session::Terminate, shared_from_this ()));
+					return;
+				}
+				auto size = bufbe16toh (buf->data () + 1);
+				if (size > buf->size () - 3 || size > i2p::data::MAX_RI_BUFFER_SIZE + 1)
+				{
+					LogPrint (eLogError, "NTCP2: Unexpected RouterInfo size ", size, " in SessionConfirmed");
+					boost::asio::post (m_Server.GetService (), std::bind (&NTCP2Session::Terminate, shared_from_this ()));
+					return;
+				}
+				boost::asio::post (m_Server.GetService (), 
+					[s = shared_from_this (), buf, size] ()
+					{
+						s->EstablishSessionAfterSessionConfirmed (buf, size);
+					});
+			}
+			else
+				boost::asio::post (m_Server.GetService (), std::bind (&NTCP2Session::Terminate, shared_from_this ()));
+		}
+		else
+			boost::asio::post (m_Server.GetService (), std::bind (&NTCP2Session::Terminate, shared_from_this ()));	
+	}	
+
+	void NTCP2Session::EstablishSessionAfterSessionConfirmed (std::shared_ptr<std::vector<uint8_t> > buf, size_t size)
+	{
+		// run on main NTCP2 thread
+		KeyDerivationFunctionDataPhase ();
+		// Bob data phase keys
+		m_SendKey = m_Kba;
+		m_ReceiveKey = m_Kab;
+		SetSipKeys (m_Sipkeysba, m_Sipkeysab);
+		memcpy (m_ReceiveIV.buf, m_Sipkeysab + 16, 8);
+		memcpy (m_SendIV.buf, m_Sipkeysba + 16, 8);
+		// we need to set keys for SendTerminationAndTerminate
+		// TODO: check flag
+		i2p::data::RouterInfo ri (buf->data () + 4, size - 1); // 1 byte block type + 2 bytes size + 1 byte flag
+		if (ri.IsUnreachable ())
+		{
+			LogPrint (eLogError, "NTCP2: RouterInfo verification failed in SessionConfirmed from ", GetRemoteEndpoint ());
+			SendTerminationAndTerminate (eNTCP2RouterInfoSignatureVerificationFail);
+			return;
+		}
+		LogPrint(eLogDebug, "NTCP2: SessionConfirmed from ", GetRemoteEndpoint (),
+			" (", i2p::data::GetIdentHashAbbreviation (ri.GetIdentHash ()), ")");
+		auto ts = i2p::util::GetMillisecondsSinceEpoch ();
+		if (ts > ri.GetTimestamp () + i2p::data::NETDB_MIN_EXPIRATION_TIMEOUT*1000LL) // 90 minutes
+		{
+			LogPrint (eLogError, "NTCP2: RouterInfo is too old in SessionConfirmed for ", (ts - ri.GetTimestamp ())/1000LL, " seconds");
+			SendTerminationAndTerminate (eNTCP2Message3Error);
+			return;
+		}
+		if (ts + i2p::data::NETDB_EXPIRATION_TIMEOUT_THRESHOLD*1000LL < ri.GetTimestamp ()) // 2 minutes
+		{
+			LogPrint (eLogError, "NTCP2: RouterInfo is from future for ", (ri.GetTimestamp () - ts)/1000LL, " seconds");
+			SendTerminationAndTerminate (eNTCP2Message3Error);
+			return;
+		}	
+		// update RouterInfo in netdb
+		auto ri1 = i2p::data::netdb.AddRouterInfo (ri.GetBuffer (), ri.GetBufferLen ()); // ri1 points to one from netdb now
+		if (!ri1)
+		{
+			LogPrint (eLogError, "NTCP2: Couldn't update RouterInfo from SessionConfirmed in netdb");
+			Terminate ();
+			return;
+		}
+		std::shared_ptr<i2p::data::RouterProfile> profile; // not null if older 
+		if (ri.GetTimestamp () + i2p::data::NETDB_EXPIRATION_TIMEOUT_THRESHOLD*1000LL < ri1->GetTimestamp ())
+		{	
+			// received RouterInfo is older than one in netdb
+			profile = i2p::data::GetRouterProfile (ri1->GetIdentHash ()); // retrieve profile	
+			if (profile && profile->IsDuplicated ())
+			{	
+				SendTerminationAndTerminate (eNTCP2Banned);
+				return;
+			}	
+		}
+		
+		auto addr = m_RemoteEndpoint.address ().is_v4 () ? ri1->GetNTCP2V4Address () :
+			(i2p::util::net::IsYggdrasilAddress (m_RemoteEndpoint.address ()) ? ri1->GetYggdrasilAddress () : ri1->GetNTCP2V6Address ());
+		if (!addr || memcmp (m_Establisher->m_RemoteStaticKey, addr->s, 32))
+		{
+			LogPrint (eLogError, "NTCP2: Wrong static key in SessionConfirmed");
+			Terminate ();
+			return;
+		}
+		if (addr->IsPublishedNTCP2 () && m_RemoteEndpoint.address () != addr->host &&
+		    (!m_RemoteEndpoint.address ().is_v6 () || (i2p::util::net::IsYggdrasilAddress (m_RemoteEndpoint.address ()) ?
+		     memcmp (m_RemoteEndpoint.address ().to_v6 ().to_bytes ().data () + 1, addr->host.to_v6 ().to_bytes ().data () + 1, 7) : // from the same yggdrasil subnet
+		     memcmp (m_RemoteEndpoint.address ().to_v6 ().to_bytes ().data (), addr->host.to_v6 ().to_bytes ().data (), 8)))) // temporary address
+		{
+			if (profile) // older router?
+				profile->Duplicated (); // mark router as duplicated in profile
+			else
+				LogPrint (eLogInfo, "NTCP2: Host mismatch between published address ", addr->host, " and actual endpoint ", m_RemoteEndpoint.address ());
+			SendTerminationAndTerminate (eNTCP2Banned);
+			return;
+		}
+		// TODO: process options block
+
+		// ready to communicate
+		SetRemoteIdentity (ri1->GetRouterIdentity ());
+		if (m_Server.AddNTCP2Session (shared_from_this (), true))
+		{
+			Established ();
+			ReceiveLength ();
+		}
+		else
+			Terminate ();
+	}	
+		
 	void NTCP2Session::SetSipKeys (const uint8_t * sendSipKey, const uint8_t * receiveSipKey)
 	{
 #if OPENSSL_SIPHASH

libi2pd/NTCP2.h

@@ -180,7 +180,9 @@ namespace transport
 			void HandleSessionCreatedPaddingReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred);
 			void HandleSessionConfirmedSent (const boost::system::error_code& ecode, std::size_t bytes_transferred);
 			void HandleSessionConfirmedReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred);
-
+			void ProcessSessionConfirmed ();
+			void EstablishSessionAfterSessionConfirmed (std::shared_ptr<std::vector<uint8_t> > buf, size_t size);
+			
 			// data
 			void ReceiveLength ();
 			void HandleReceivedLength (const boost::system::error_code& ecode, std::size_t bytes_transferred);

libi2pd/Streaming.cpp

@@ -73,7 +73,7 @@ namespace stream
 		m_LastConfirmedReceivedSequenceNumber (0), // for limit inbound speed
 		m_Status (eStreamStatusNew), m_IsIncoming (false), m_IsAckSendScheduled (false), m_IsNAcked (false), m_IsFirstACK (false), 
 		m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), m_IsWinDropped (false),
-		m_IsTimeOutResend (false), m_IsImmediateAckRequested (false), m_IsRemoteLeaseChangeInProgress (false), m_LocalDestination (local),
+		m_IsTimeOutResend (false), m_IsImmediateAckRequested (false), m_IsRemoteLeaseChangeInProgress (false), m_DoubleWinIncCounter (false), m_LocalDestination (local),
 		m_RemoteLeaseSet (remote), m_ReceiveTimer (m_Service), m_SendTimer (m_Service), m_ResendTimer (m_Service),
 		m_AckSendTimer (m_Service), m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (port),
 		m_RTT (INITIAL_RTT), m_SlowRTT (INITIAL_RTT), m_SlowRTT2 (INITIAL_RTT), m_WindowSize (INITIAL_WINDOW_SIZE), m_LastWindowDropSize  (0),
@@ -101,7 +101,7 @@ namespace stream
 		m_LastConfirmedReceivedSequenceNumber (0), // for limit inbound speed
 		m_Status (eStreamStatusNew), m_IsIncoming (true), m_IsAckSendScheduled (false), m_IsNAcked (false), m_IsFirstACK (false), 
 		m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), m_IsWinDropped (false),
-		m_IsTimeOutResend (false), m_IsImmediateAckRequested (false), m_IsRemoteLeaseChangeInProgress (false), m_LocalDestination (local),
+		m_IsTimeOutResend (false), m_IsImmediateAckRequested (false), m_IsRemoteLeaseChangeInProgress (false), m_DoubleWinIncCounter (false), m_LocalDestination (local),
 		m_ReceiveTimer (m_Service), m_SendTimer (m_Service), m_ResendTimer (m_Service), m_AckSendTimer (m_Service),
 		m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (0), m_RTT (INITIAL_RTT), m_SlowRTT (INITIAL_RTT), m_SlowRTT2 (INITIAL_RTT),
 		m_WindowSize (INITIAL_WINDOW_SIZE), m_LastWindowDropSize  (0), m_WindowDropTargetSize (0), m_WindowIncCounter (0), 
@@ -495,6 +495,7 @@ namespace stream
 			return;
 		}
 		int rttSample = INT_MAX;
+		int incCounter = 0;
 		m_IsNAcked = false;
 		m_IsResendNeeded = false;
 		int nackCount = packet->GetNACKCount ();
@@ -537,8 +538,7 @@ namespace stream
 				m_LocalDestination.DeletePacket (sentPacket);
 				acknowledged = true;
 				if (m_WindowSize < MAX_WINDOW_SIZE && !m_IsFirstACK)
-					if (m_RTT < m_LocalDestination.GetRandom () % INITIAL_RTO) // dirty
-						m_WindowIncCounter++;
+					incCounter++;
 			}
 			else
 				break;
@@ -552,7 +552,7 @@ namespace stream
 				m_SlowRTT2 = rttSample;
 				m_PrevRTTSample = rttSample;
 				m_Jitter = rttSample / 10; // 10%
-				m_Jitter += 5; // for low-latency connections
+				m_Jitter += 15; // for low-latency connections
 				m_IsFirstRttSample = false;
 			}
 			else
@@ -569,9 +569,23 @@ namespace stream
 					jitter = m_PrevRTTSample - rttSample;
 				else
 					jitter = rttSample / 10; // 10%
-				jitter += 5; // for low-latency connections
+				jitter += 15; // for low-latency connections
 				m_Jitter = (0.05 * jitter) + (1.0 - 0.05) * m_Jitter;
 			}
+			if (rttSample > m_SlowRTT)
+			{
+				incCounter = 0;
+				m_DoubleWinIncCounter = 1;
+			}
+			else if (rttSample < m_SlowRTT)
+			{
+				if (m_DoubleWinIncCounter)
+				{
+					incCounter = incCounter * 2;
+					m_DoubleWinIncCounter = 0;
+				}
+			}
+			m_WindowIncCounter = m_WindowIncCounter + incCounter;
 			//
 			// delay-based CC
 			if ((m_SlowRTT2 > m_SlowRTT + m_Jitter && rttSample > m_SlowRTT2 && rttSample > m_PrevRTTSample) && !m_IsWinDropped) // Drop window if RTT grows too fast, late detection
@@ -1297,8 +1311,6 @@ namespace stream
 				else if (m_IsResendNeeded) // resend packets
 					ResendPacket ();
 				// delay-based CC
-				else if (!m_IsWinDropped && int(m_SentPackets.size ()) == m_WindowSize) // we sending packets too fast, early detection
-					ProcessWindowDrop ();
 				else if (m_WindowSize > int(m_SentPackets.size ())) // send packets
 					SendBuffer ();
 			}

libi2pd/Streaming.h

@@ -19,6 +19,7 @@
 #include <mutex>
 #include <boost/asio.hpp>
 #include "Base.h"
+#include "Gzip.h"
 #include "I2PEndian.h"
 #include "Identity.h"
 #include "LeaseSet.h"
@@ -54,7 +55,7 @@ namespace stream
 	const size_t COMPRESSION_THRESHOLD_SIZE = 66;
 	const int MAX_NUM_RESEND_ATTEMPTS = 10;
 	const int INITIAL_WINDOW_SIZE = 10;
-	const int MIN_WINDOW_SIZE = 2;
+	const int MIN_WINDOW_SIZE = 3;
 	const int MAX_WINDOW_SIZE = 512;
 	const double RTT_EWMA_ALPHA = 0.25;
 	const double SLOWRTT_EWMA_ALPHA = 0.05;
@@ -274,6 +275,7 @@ namespace stream
 			bool m_IsTimeOutResend;
 			bool m_IsImmediateAckRequested;
 			bool m_IsRemoteLeaseChangeInProgress;
+			bool m_DoubleWinIncCounter;
 			StreamingDestination& m_LocalDestination;
 			std::shared_ptr<const i2p::data::IdentityEx> m_RemoteIdentity;
 			std::shared_ptr<const i2p::crypto::Verifier> m_TransientVerifier; // in case of offline key