diff --git a/libi2pd/Streaming.cpp b/libi2pd/Streaming.cpp index 8f6e567b..d6f0441c 100644 --- a/libi2pd/Streaming.cpp +++ b/libi2pd/Streaming.cpp @@ -81,7 +81,7 @@ namespace stream 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_LastACKSendTime (0), m_PacketACKInterval (1), m_PacketACKIntervalRem (0), // for limit inbound speed - m_NumResendAttempts (0), m_NumPacketsToSend (0), m_NumPacketsToResend (0), m_MTU (STREAMING_MTU) + m_NumResendAttempts (0), m_NumPacketsToSend (0), m_MTU (STREAMING_MTU) { RAND_bytes ((uint8_t *)&m_RecvStreamID, 4); m_RemoteIdentity = remote->GetIdentity (); @@ -108,7 +108,7 @@ namespace stream 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_LastACKSendTime (0), m_PacketACKInterval (1), m_PacketACKIntervalRem (0), // for limit inbound speed - m_NumResendAttempts (0), m_NumPacketsToSend (0), m_NumPacketsToResend (0), m_MTU (STREAMING_MTU) + m_NumResendAttempts (0), m_NumPacketsToSend (0), m_MTU (STREAMING_MTU) { RAND_bytes ((uint8_t *)&m_RecvStreamID, 4); auto outboundSpeed = local.GetOwner ()->GetStreamingOutboundSpeed (); @@ -147,6 +147,8 @@ namespace stream m_ReceiveQueue.pop (); m_LocalDestination.DeletePacket (packet); } + + m_NACKedPackets.clear (); for (auto it: m_SentPackets) m_LocalDestination.DeletePacket (it); @@ -452,6 +454,7 @@ namespace stream bool acknowledged = false; auto ts = i2p::util::GetMillisecondsSinceEpoch (); uint32_t ackThrough = packet->GetAckThrough (); + m_NACKedPackets.clear (); if (ackThrough > m_SequenceNumber) { LogPrint (eLogError, "Streaming: Unexpected ackThrough=", ackThrough, " > seqn=", m_SequenceNumber); @@ -461,7 +464,6 @@ namespace stream m_IsNAcked = false; m_IsResendNeeded = false; int nackCount = packet->GetNACKCount (); - m_NumPacketsToResend = nackCount; for (auto it = m_SentPackets.begin (); it != m_SentPackets.end ();) { auto seqn = (*it)->GetSeqn (); @@ -473,6 +475,7 @@ namespace stream for (int i = 0; i < nackCount; i++) if (seqn == packet->GetNACK (i)) { + m_NACKedPackets.insert (*it); m_IsNAcked = true; nacked = true; break; @@ -1219,23 +1222,38 @@ namespace stream } // collect packets to resend - int numPacketsToResend = m_NumPacketsToResend; auto ts = i2p::util::GetMillisecondsSinceEpoch (); std::vector packets; - for (auto it : m_SentPackets) + if (m_IsNAcked) { - if (ts >= it->sendTime + m_RTO) + for (auto it : m_NACKedPackets) { - if (ts < it->sendTime + m_RTO*2) - it->resent = true; - else + if (ts >= it->sendTime + m_RTO) + { + if (ts < it->sendTime + m_RTO*2) + it->resent = true; + else + it->resent = false; + it->sendTime = ts; + packets.push_back (it); + if ((int)packets.size () >= m_NumPacketsToSend) break; + } + } + } + else + { + for (auto it : m_SentPackets) + { + if (ts >= it->sendTime + m_RTO) { - it->resent = false; + if (ts < it->sendTime + m_RTO*2) + it->resent = true; + else + it->resent = false; it->sendTime = ts; + packets.push_back (it); + if ((int)packets.size () >= m_NumPacketsToSend) break; } - packets.push_back (it); - m_NumPacketsToResend--; - if ((int)packets.size () >= m_NumPacketsToSend || (int)packets.size () >= numPacketsToResend) break; } } diff --git a/libi2pd/Streaming.h b/libi2pd/Streaming.h index 665c73f4..a0229483 100644 --- a/libi2pd/Streaming.h +++ b/libi2pd/Streaming.h @@ -269,6 +269,7 @@ namespace stream std::queue m_ReceiveQueue; std::set m_SavedPackets; std::set m_SentPackets; + std::set m_NACKedPackets; boost::asio::deadline_timer m_ReceiveTimer, m_SendTimer, m_ResendTimer, m_AckSendTimer; size_t m_NumSentBytes, m_NumReceivedBytes; uint16_t m_Port; @@ -279,7 +280,7 @@ namespace stream int m_WindowIncCounter, m_RTO, m_AckDelay, m_PrevRTTSample, m_PrevRTT, 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, m_NumPacketsToResend; + int m_NumResendAttempts, m_NumPacketsToSend; size_t m_MTU; };