improved window size reculculation algorithm

libi2pd/Streaming.cpp

@@ -68,18 +68,19 @@ namespace stream
 
 	Stream::Stream (boost::asio::io_service& service, StreamingDestination& local,
 		std::shared_ptr<const i2p::data::LeaseSet> remote, int port): m_Service (service),
-		m_SendStreamID (0), m_SequenceNumber (0),
+		m_SendStreamID (0), m_SequenceNumber (0), m_DropWindowDelaySequenceNumber (0),
 		m_TunnelsChangeSequenceNumber (0), m_LastReceivedSequenceNumber (-1), m_PreviousReceivedSequenceNumber (-1),
 		m_LastConfirmedReceivedSequenceNumber (0), // for limit inbound speed
 		m_Status (eStreamStatusNew), m_IsAckSendScheduled (false), m_IsNAcked (false), m_IsFirstACK (false), 
-		m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), m_IsWinDropped (true),
+		m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), m_IsWinDropped (false),
 		m_IsTimeOutResend (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_WindowSize (INITIAL_WINDOW_SIZE), m_LastWindowDropSize  (0), m_WindowIncCounter (0), m_RTO (INITIAL_RTO),
+		m_RTT (INITIAL_RTT), m_SlowRTT (INITIAL_RTT), m_WindowSize (INITIAL_WINDOW_SIZE), m_LastWindowDropSize  (0),
+		m_WindowDropTargetSize (0), m_WindowIncCounter (0), m_RTO (INITIAL_RTO),
 		m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()), m_PrevRTTSample (INITIAL_RTT), 
-		m_PrevRTT (INITIAL_RTT), m_Jitter (0), m_MinPacingTime (0),
+		m_Jitter (0), m_MinPacingTime (0),
-		m_PacingTime (INITIAL_PACING_TIME), m_PacingTimeRem (0), m_DropWindowDelayTime (0), m_LastSendTime (0),
+		m_PacingTime (INITIAL_PACING_TIME), m_PacingTimeRem (0), m_LastSendTime (0),
 		m_LastACKSendTime (0), m_PacketACKInterval (1), m_PacketACKIntervalRem (0), // for limit inbound speed
 		m_NumResendAttempts (0), m_NumPacketsToSend (0), m_MTU (STREAMING_MTU)
 	{
@@ -95,18 +96,18 @@ namespace stream
 	}
 
 	Stream::Stream (boost::asio::io_service& service, StreamingDestination& local):
-		m_Service (service), m_SendStreamID (0), m_SequenceNumber (0),
+		m_Service (service), m_SendStreamID (0), m_SequenceNumber (0), m_DropWindowDelaySequenceNumber (0),
 		m_TunnelsChangeSequenceNumber (0), m_LastReceivedSequenceNumber (-1), m_PreviousReceivedSequenceNumber (-1),
 		m_LastConfirmedReceivedSequenceNumber (0), // for limit inbound speed
 		m_Status (eStreamStatusNew), m_IsAckSendScheduled (false), m_IsNAcked (false), m_IsFirstACK (false), 
-		m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), m_IsWinDropped (true),
+		m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), m_IsWinDropped (false),
 		m_IsTimeOutResend (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_WindowSize (INITIAL_WINDOW_SIZE), m_LastWindowDropSize  (0), m_WindowIncCounter (0), 
+		m_WindowSize (INITIAL_WINDOW_SIZE), m_LastWindowDropSize  (0), m_WindowDropTargetSize (0), m_WindowIncCounter (0), 
 		m_RTO (INITIAL_RTO), m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()),
-		m_PrevRTTSample (INITIAL_RTT), m_PrevRTT (INITIAL_RTT), m_Jitter (0), m_MinPacingTime (0),
+		m_PrevRTTSample (INITIAL_RTT), m_Jitter (0), m_MinPacingTime (0),
-		m_PacingTime (INITIAL_PACING_TIME), m_PacingTimeRem (0), m_DropWindowDelayTime (0), m_LastSendTime (0),
+		m_PacingTime (INITIAL_PACING_TIME), m_PacingTimeRem (0), m_LastSendTime (0),
 		m_LastACKSendTime (0), m_PacketACKInterval (1), m_PacketACKIntervalRem (0), // for limit inbound speed
 		m_NumResendAttempts (0), m_NumPacketsToSend (0), m_MTU (STREAMING_MTU)
 	{
@@ -339,8 +340,15 @@ namespace stream
 				}
 				if (delayRequested >= DELAY_CHOKING)
 				{
-					m_WindowSize = 1;
+					if (!m_IsWinDropped)
-					m_WindowIncCounter = 0;
+					{
+						m_WindowDropTargetSize = MIN_WINDOW_SIZE;
+						m_LastWindowDropSize = 0;
+						m_WindowIncCounter = 0;
+						m_IsWinDropped = true; // don't drop window twice
+						m_DropWindowDelaySequenceNumber = m_SequenceNumber;
+						UpdatePacingTime ();
+					}
 				}
 			}
 			optionData += 2;
@@ -451,6 +459,7 @@ namespace stream
 
 	void Stream::ProcessAck (Packet * packet)
 	{
+		srand (time(NULL));
 		bool acknowledged = false;
 		auto ts = i2p::util::GetMillisecondsSinceEpoch ();
 		uint32_t ackThrough = packet->GetAckThrough ();
@@ -464,6 +473,7 @@ namespace stream
 		m_IsNAcked = false;
 		m_IsResendNeeded = false;
 		int nackCount = packet->GetNACKCount ();
+		int ackCount = 0;
 		for (auto it = m_SentPackets.begin (); it != m_SentPackets.end ();)
 		{
 			auto seqn = (*it)->GetSeqn ();
@@ -502,59 +512,42 @@ namespace stream
 				m_SentPackets.erase (it++);
 				m_LocalDestination.DeletePacket (sentPacket);
 				acknowledged = true;
+				ackCount++;
 				if (m_WindowSize < MAX_WINDOW_SIZE && !m_IsFirstACK)
-					m_WindowIncCounter++;
+					if (m_RTT < rand () % INITIAL_RTT) // dirty
+						m_WindowIncCounter++;
 			}
 			else
 				break;
 		}
 		if (rttSample != INT_MAX)
 		{
-			if (m_IsFirstRttSample)
+			if (m_IsFirstRttSample && !m_IsFirstACK)
 			{
 				m_RTT = rttSample;
 				m_SlowRTT = rttSample;
 				m_PrevRTTSample = rttSample;
-				if (m_RoutingSession)
-					m_RoutingSession->SetSharedRoutingPath (
-						std::make_shared<i2p::garlic::GarlicRoutingPath> (
-							i2p::garlic::GarlicRoutingPath{m_CurrentOutboundTunnel, m_CurrentRemoteLease, (int)m_RTT, 0}));
 				m_IsFirstRttSample = false;
 			}
 			else
-				m_RTT = RTT_EWMA_ALPHA * m_RTT + (1.0 - RTT_EWMA_ALPHA) * rttSample;
+				m_RTT = RTT_EWMA_ALPHA * rttSample + (1.0 - RTT_EWMA_ALPHA) * m_RTT;
 			// calculate jitter
-			int jitter = 0;
+			double jitter = 0;
 			if (rttSample > m_PrevRTTSample)
 				jitter = rttSample - m_PrevRTTSample;
 			else if (rttSample < m_PrevRTTSample)
 				jitter = m_PrevRTTSample - rttSample;
 			else
 				jitter = std::round (rttSample / 10); // 10%
-			jitter += 5; // for low-latency connections
+			jitter += 1; // for low-latency connections
-			m_Jitter = std::round (RTT_EWMA_ALPHA * jitter + (1.0 - RTT_EWMA_ALPHA) * m_Jitter);
+			m_Jitter = (0.025 * jitter) + (1.0 - 0.025) * m_Jitter;
-			m_PrevRTTSample = rttSample;
 			//
 			// delay-based CC
-			if ((m_PrevRTT > m_SlowRTT + m_Jitter) && (m_RTT > m_SlowRTT + m_Jitter) && !m_IsWinDropped) // Drop window if RTT grows too fast, late detection
+			if ((m_RTT > m_SlowRTT + m_Jitter && rttSample > m_RTT && rttSample > m_PrevRTTSample) && !m_IsWinDropped) // Drop window if RTT grows too fast, late detection
-			{
+				ProcessWindowDrop ();
-				if (m_LastWindowDropSize)
-					m_LastWindowDropSize = (m_LastWindowDropSize + m_WindowSize) / 2;
-				else
-					m_LastWindowDropSize = m_WindowSize;
-				m_WindowSize = m_WindowSize / 2; // /2
-				if (m_WindowSize < MIN_WINDOW_SIZE) m_WindowSize = MIN_WINDOW_SIZE;
-				m_WindowIncCounter = 0;
-				m_DropWindowDelayTime = ts + m_SlowRTT;
-				m_IsFirstACK = true;
-				m_IsWinDropped = true; // don't drop window twice
-			}
 			UpdatePacingTime ();
-			if (rttSample < m_RTT) // need for delay-based CC
+			m_SlowRTT = SLOWRTT_EWMA_ALPHA * rttSample + (1.0 - SLOWRTT_EWMA_ALPHA) * m_SlowRTT;
-				m_SlowRTT = RTT_EWMA_ALPHA * rttSample + (1.0 - RTT_EWMA_ALPHA) * m_SlowRTT;
+			m_PrevRTTSample = rttSample;
-			else
-				m_SlowRTT = RTT_EWMA_ALPHA * m_RTT + (1.0 - RTT_EWMA_ALPHA) * m_SlowRTT;
-			m_PrevRTT = m_RTT;
 			
 			bool wasInitial = m_RTO == INITIAL_RTO;
 			m_RTO = std::max (MIN_RTO, (int)(m_RTT * 1.3 + m_Jitter)); // TODO: implement it better
@@ -562,8 +555,21 @@ namespace stream
 			if (wasInitial)
 				ScheduleResend ();
 		}
-		if ( ts > m_DropWindowDelayTime)
+		if (ackThrough > m_DropWindowDelaySequenceNumber && m_WindowSize <= m_WindowDropTargetSize)
 			m_IsWinDropped = false;
+		if (acknowledged && m_IsWinDropped && m_WindowSize > m_WindowDropTargetSize)
+		{
+			if (ackCount > 1)
+				m_WindowSize = m_SentPackets.size () + 1;
+			else
+			{
+				m_WindowSize = m_SentPackets.size ();
+				m_IsResendNeeded = true;
+			}
+			if (m_WindowSize < MIN_WINDOW_SIZE) m_WindowSize = MIN_WINDOW_SIZE;
+			m_WindowIncCounter = 0;
+			UpdatePacingTime ();
+		}
 		if (acknowledged || m_IsNAcked)
 		{
 			ScheduleResend ();
@@ -578,10 +584,18 @@ namespace stream
 			m_ResendTimer.cancel ();
 			m_SendTimer.cancel ();
 		}
+		if (acknowledged && m_IsFirstACK)
+		{
+			if (m_RoutingSession)
+				m_RoutingSession->SetSharedRoutingPath (
+					std::make_shared<i2p::garlic::GarlicRoutingPath> (
+						i2p::garlic::GarlicRoutingPath{m_CurrentOutboundTunnel, m_CurrentRemoteLease, (int)m_RTT, 0}));
+			m_IsFirstACK = false;
+		}
 		if (acknowledged)
 		{
 			m_NumResendAttempts = 0;
-			m_IsFirstACK = false;
+			m_IsTimeOutResend = false;
 			SendBuffer ();
 		}
 		if (m_Status == eStreamStatusClosed)
@@ -1135,45 +1149,37 @@ namespace stream
 	{
 		if (ecode != boost::asio::error::operation_aborted)
 		{
-			if (m_WindowIncCounter && m_WindowSize < MAX_WINDOW_SIZE)
-			{
-				if (m_LastWindowDropSize && (m_LastWindowDropSize > m_WindowSize))
-				{
-					m_WindowSize += 2.001-(2/((m_LastWindowDropSize+(1/m_WindowSize))/m_WindowSize)); // some magic here
-					m_WindowIncCounter --;
-				}
-				else
-				{
-					m_WindowSize += 1;
-					m_WindowIncCounter --;
-				}
-				if (m_WindowSize > MAX_WINDOW_SIZE) m_WindowSize = MAX_WINDOW_SIZE;
-				UpdatePacingTime ();
-			}
 			auto ts = i2p::util::GetMillisecondsSinceEpoch ();
 			if (m_LastSendTime && ts*1000 > m_LastSendTime*1000 + m_PacingTime)
 			{
 				m_NumPacketsToSend = ((ts*1000 - m_LastSendTime*1000) + m_PacingTimeRem) / m_PacingTime;
 				m_PacingTimeRem = ((ts*1000 - m_LastSendTime*1000) + m_PacingTimeRem) - (m_NumPacketsToSend * m_PacingTime);
 				m_IsSendTime = true;
-				if (m_IsNAcked || m_IsResendNeeded) // resend packets
+				if (m_WindowIncCounter && m_WindowSize < MAX_WINDOW_SIZE && !m_SendBuffer.IsEmpty ())
+				{
+					for (int i = 0; i < m_NumPacketsToSend; i++)
+					{
+						if (m_WindowIncCounter)
+						{
+							if (m_LastWindowDropSize && (m_LastWindowDropSize >= m_WindowSize))
+								m_WindowSize += 1 - (1 / ((m_LastWindowDropSize + PREV_SPEED_KEEP_TIME_COEFF) / m_WindowSize)); // some magic here
+							else if (m_LastWindowDropSize && (m_LastWindowDropSize < m_WindowSize))
+								m_WindowSize += (m_WindowSize - (m_LastWindowDropSize - PREV_SPEED_KEEP_TIME_COEFF)) / m_WindowSize; // some magic here
+							else
+								m_WindowSize += (m_WindowSize - (1 - PREV_SPEED_KEEP_TIME_COEFF)) / m_WindowSize;
+							if (m_WindowSize > MAX_WINDOW_SIZE) m_WindowSize = MAX_WINDOW_SIZE;
+							m_WindowIncCounter --;
+							UpdatePacingTime ();
+						}
+					}
+				}
+				if (m_IsNAcked)
+					ResendPacket ();
+				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 ();
-					auto ts = i2p::util::GetMillisecondsSinceEpoch ();
-					if (m_LastWindowDropSize)
-						m_LastWindowDropSize = (m_LastWindowDropSize + m_WindowSize) / 2;
-					else
-						m_LastWindowDropSize = m_WindowSize;
-					m_WindowSize = m_WindowSize / 2; // /2
-					if (m_WindowSize < MIN_WINDOW_SIZE) m_WindowSize = MIN_WINDOW_SIZE;
-					m_WindowIncCounter = 0;
-					m_DropWindowDelayTime = ts + m_SlowRTT;
-					m_IsFirstACK = true;
-					m_IsWinDropped = true; // don't drop window twice
-					UpdatePacingTime ();
-				}
 				else if (m_WindowSize > int(m_SentPackets.size ())) // send packets
 					SendBuffer ();
 			}
@@ -1266,30 +1272,18 @@ namespace stream
 				{
 					// loss-based CC
 					if (!m_IsWinDropped)
-					{
+						ProcessWindowDrop ();
-						if (m_LastWindowDropSize)
-							m_LastWindowDropSize = (m_LastWindowDropSize + m_WindowSize) / 2;
-						else
-							m_LastWindowDropSize = m_WindowSize;
-						m_WindowSize = m_WindowSize / 2; // /2
-						if (m_WindowSize < MIN_WINDOW_SIZE) m_WindowSize = MIN_WINDOW_SIZE;
-						m_WindowIncCounter = 0;
-						m_IsWinDropped = true; // don't drop window twice
-						m_DropWindowDelayTime = ts + m_SlowRTT;
-						m_IsFirstACK = true;
-						UpdatePacingTime ();
-					}
 				}
 				else if (m_IsTimeOutResend)
 				{
 					m_IsTimeOutResend = false;
 					m_RTO = INITIAL_RTO; // drop RTO to initial upon tunnels pair change
-					m_WindowSize = INITIAL_WINDOW_SIZE;
+					m_WindowDropTargetSize = INITIAL_WINDOW_SIZE;
 					m_LastWindowDropSize = 0;
 					m_WindowIncCounter = 0;
 					m_IsWinDropped = true;
 					m_IsFirstRttSample = true;
-					m_DropWindowDelayTime = 0;
+					m_DropWindowDelaySequenceNumber = 0;
 					m_IsFirstACK = true;
 					UpdatePacingTime ();
 					if (m_RoutingSession) m_RoutingSession->SetSharedRoutingPath (nullptr);
@@ -1438,12 +1432,16 @@ namespace stream
 		}
 		// drop window to initial upon RemoteLease change
 		m_RTO = INITIAL_RTO;
-		m_WindowSize = INITIAL_WINDOW_SIZE;
+		if (m_WindowSize > INITIAL_WINDOW_SIZE)
+		{
+			m_WindowDropTargetSize = INITIAL_WINDOW_SIZE;
+			m_IsWinDropped = true;
+		}
+		else
+			m_WindowSize = INITIAL_WINDOW_SIZE;
 		m_LastWindowDropSize = 0;
 		m_WindowIncCounter = 0;
-		m_IsWinDropped = true;
 		m_IsFirstRttSample = true;
-		m_DropWindowDelayTime = 0;
 		m_IsFirstACK = true;
 		UpdatePacingTime ();
 	}
@@ -1465,6 +1463,24 @@ namespace stream
 			m_PacingTime = m_MinPacingTime;
 	}	
 
+	void Stream::ProcessWindowDrop ()
+	{
+		if (m_WindowSize > m_LastWindowDropSize)
+			m_LastWindowDropSize = (m_LastWindowDropSize + m_WindowSize) / 2;
+		else
+			m_LastWindowDropSize = m_WindowSize;
+		m_WindowDropTargetSize = m_LastWindowDropSize - (m_LastWindowDropSize / 4); // -25%;
+		if (m_WindowDropTargetSize < MIN_WINDOW_SIZE + 1)
+			m_WindowDropTargetSize = MIN_WINDOW_SIZE + 1;
+		m_WindowSize = m_SentPackets.size (); // stop sending now
+		if (m_WindowSize < MIN_WINDOW_SIZE) m_WindowSize = MIN_WINDOW_SIZE;
+		m_WindowIncCounter = 0; // disable window growth
+		m_DropWindowDelaySequenceNumber = m_SequenceNumber;
+		m_IsFirstACK = true; // ignore first RTT sample
+		m_IsWinDropped = true; // don't drop window twice
+		UpdatePacingTime ();
+	}
+		
 	StreamingDestination::StreamingDestination (std::shared_ptr<i2p::client::ClientDestination> owner, uint16_t localPort, bool gzip):
 		m_Owner (owner), m_LocalPort (localPort), m_Gzip (gzip),
 		m_PendingIncomingTimer (m_Owner->GetService ())

libi2pd/Streaming.h

@@ -56,7 +56,9 @@ namespace stream
 	const int INITIAL_WINDOW_SIZE = 10;
 	const int MIN_WINDOW_SIZE = 1;
 	const int MAX_WINDOW_SIZE = 1024;
-	const double RTT_EWMA_ALPHA = 0.125;
+	const double RTT_EWMA_ALPHA = 0.25;
+	const double SLOWRTT_EWMA_ALPHA = 0.125;
+	const double PREV_SPEED_KEEP_TIME_COEFF = 0.1; // 0.1 - 1 // how long will the window size stay around the previous drop level, less is longer
 	const int MIN_RTO = 20; // in milliseconds
 	const int INITIAL_RTT = 8000; // in milliseconds
 	const int INITIAL_RTO = 9000; // in milliseconds
@@ -241,11 +243,13 @@ namespace stream
 			void HandleAckSendTimer (const boost::system::error_code& ecode);
 
 			void UpdatePacingTime ();
+			void ProcessWindowDrop ();
 			
 		private:
 
 			boost::asio::io_service& m_Service;
 			uint32_t m_SendStreamID, m_RecvStreamID, m_SequenceNumber;
+			uint32_t m_DropWindowDelaySequenceNumber;
 			uint32_t m_TunnelsChangeSequenceNumber;
 			int32_t m_LastReceivedSequenceNumber;
 			int32_t m_PreviousReceivedSequenceNumber;
@@ -276,8 +280,9 @@ namespace stream
 
 			SendBufferQueue m_SendBuffer;
 			double m_RTT, m_SlowRTT;
-			float m_WindowSize, m_LastWindowDropSize;
+			float m_WindowSize, m_LastWindowDropSize, m_WindowDropTargetSize;
-			int m_WindowIncCounter, m_RTO, m_AckDelay, m_PrevRTTSample, m_PrevRTT, m_Jitter;
+			int m_WindowIncCounter, m_RTO, m_AckDelay, m_PrevRTTSample;
+			double m_Jitter;
 			uint64_t m_MinPacingTime, m_PacingTime, m_PacingTimeRem, m_DropWindowDelayTime, m_LastSendTime; // microseconds
 			uint64_t m_LastACKSendTime, m_PacketACKInterval, m_PacketACKIntervalRem; // for limit inbound speed
 			int m_NumResendAttempts, m_NumPacketsToSend;