diff --git a/libi2pd/Streaming.cpp b/libi2pd/Streaming.cpp index d6f0441c..2866f945 100644 --- a/libi2pd/Streaming.cpp +++ b/libi2pd/Streaming.cpp @@ -68,18 +68,19 @@ namespace stream Stream::Stream (boost::asio::io_service& service, StreamingDestination& local, std::shared_ptr 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_PacingTime (INITIAL_PACING_TIME), m_PacingTimeRem (0), m_DropWindowDelayTime (0), m_LastSendTime (0), + m_Jitter (0), m_MinPacingTime (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_PacingTime (INITIAL_PACING_TIME), m_PacingTimeRem (0), m_DropWindowDelayTime (0), m_LastSendTime (0), + m_PrevRTTSample (INITIAL_RTT), m_Jitter (0), m_MinPacingTime (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; - m_WindowIncCounter = 0; + if (!m_IsWinDropped) + { + 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{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 - m_Jitter = std::round (RTT_EWMA_ALPHA * jitter + (1.0 - RTT_EWMA_ALPHA) * m_Jitter); - m_PrevRTTSample = rttSample; + jitter += 1; // for low-latency connections + m_Jitter = (0.025 * jitter) + (1.0 - 0.025) * m_Jitter; // // 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_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 - } + 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 (); UpdatePacingTime (); - if (rttSample < m_RTT) // need for delay-based CC - m_SlowRTT = RTT_EWMA_ALPHA * rttSample + (1.0 - RTT_EWMA_ALPHA) * m_SlowRTT; - else - m_SlowRTT = RTT_EWMA_ALPHA * m_RTT + (1.0 - RTT_EWMA_ALPHA) * m_SlowRTT; - m_PrevRTT = m_RTT; + m_SlowRTT = SLOWRTT_EWMA_ALPHA * rttSample + (1.0 - SLOWRTT_EWMA_ALPHA) * m_SlowRTT; + m_PrevRTTSample = rttSample; 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{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 - { - 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 (); - } + ProcessWindowDrop (); else if (m_WindowSize > int(m_SentPackets.size ())) // send packets SendBuffer (); } @@ -1266,30 +1272,18 @@ namespace stream { // loss-based CC if (!m_IsWinDropped) - { - 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 (); - } + ProcessWindowDrop (); } 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 (); } @@ -1464,6 +1462,24 @@ namespace stream if (m_MinPacingTime && m_PacingTime < m_MinPacingTime) 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 owner, uint16_t localPort, bool gzip): m_Owner (owner), m_LocalPort (localPort), m_Gzip (gzip), diff --git a/libi2pd/Streaming.h b/libi2pd/Streaming.h index a0229483..204d4ef9 100644 --- a/libi2pd/Streaming.h +++ b/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; - int m_WindowIncCounter, m_RTO, m_AckDelay, m_PrevRTTSample, m_PrevRTT, m_Jitter; + float m_WindowSize, m_LastWindowDropSize, m_WindowDropTargetSize; + 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;