|
|
|
|
@ -68,11 +68,12 @@ namespace stream
@@ -68,11 +68,12 @@ 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_LastReceivedSequenceNumber (-1), |
|
|
|
|
m_SendStreamID (0), m_SequenceNumber (0), |
|
|
|
|
m_TunnelsChangeSequenceNumber (0), m_LastReceivedSequenceNumber (-1), |
|
|
|
|
m_Status (eStreamStatusNew), m_IsAckSendScheduled (false), m_LocalDestination (local), |
|
|
|
|
m_RemoteLeaseSet (remote), m_ReceiveTimer (m_Service), m_ResendTimer (m_Service), |
|
|
|
|
m_AckSendTimer (m_Service), m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (port), |
|
|
|
|
m_WindowSize (MIN_WINDOW_SIZE), m_RTT (INITIAL_RTT), m_RTO (INITIAL_RTO), |
|
|
|
|
m_RTT (INITIAL_RTT), m_WindowSize (MIN_WINDOW_SIZE), m_RTO (INITIAL_RTO), |
|
|
|
|
m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()), |
|
|
|
|
m_LastWindowSizeIncreaseTime (0), m_NumResendAttempts (0), m_MTU (STREAMING_MTU) |
|
|
|
|
{ |
|
|
|
|
@ -81,11 +82,12 @@ namespace stream
@@ -81,11 +82,12 @@ namespace stream
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
Stream::Stream (boost::asio::io_service& service, StreamingDestination& local): |
|
|
|
|
m_Service (service), m_SendStreamID (0), m_SequenceNumber (0), m_LastReceivedSequenceNumber (-1), |
|
|
|
|
m_Service (service), m_SendStreamID (0), m_SequenceNumber (0), |
|
|
|
|
m_TunnelsChangeSequenceNumber (0), m_LastReceivedSequenceNumber (-1), |
|
|
|
|
m_Status (eStreamStatusNew), m_IsAckSendScheduled (false), m_LocalDestination (local), |
|
|
|
|
m_ReceiveTimer (m_Service), m_ResendTimer (m_Service), m_AckSendTimer (m_Service), |
|
|
|
|
m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (0), m_WindowSize (MIN_WINDOW_SIZE), |
|
|
|
|
m_RTT (INITIAL_RTT), m_RTO (INITIAL_RTO), m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()), |
|
|
|
|
m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (0), m_RTT (INITIAL_RTT), |
|
|
|
|
m_WindowSize (MIN_WINDOW_SIZE), m_RTO (INITIAL_RTO), m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()), |
|
|
|
|
m_LastWindowSizeIncreaseTime (0), m_NumResendAttempts (0), m_MTU (STREAMING_MTU) |
|
|
|
|
{ |
|
|
|
|
RAND_bytes ((uint8_t *)&m_RecvStreamID, 4); |
|
|
|
|
@ -287,6 +289,8 @@ namespace stream
@@ -287,6 +289,8 @@ namespace stream
|
|
|
|
|
m_AckSendTimer.async_wait (std::bind (&Stream::HandleAckSendTimer, |
|
|
|
|
shared_from_this (), std::placeholders::_1)); |
|
|
|
|
} |
|
|
|
|
if (delayRequested >= DELAY_CHOKING) |
|
|
|
|
m_WindowSize = 1; |
|
|
|
|
} |
|
|
|
|
optionData += 2; |
|
|
|
|
} |
|
|
|
|
@ -404,6 +408,8 @@ namespace stream
@@ -404,6 +408,8 @@ namespace stream
|
|
|
|
|
LogPrint (eLogError, "Streaming: Unexpected ackThrough=", ackThrough, " > seqn=", m_SequenceNumber); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
int rttSample = INT_MAX; |
|
|
|
|
bool firstRttSample = false; |
|
|
|
|
int nackCount = packet->GetNACKCount (); |
|
|
|
|
for (auto it = m_SentPackets.begin (); it != m_SentPackets.end ();) |
|
|
|
|
{ |
|
|
|
|
@ -427,41 +433,51 @@ namespace stream
@@ -427,41 +433,51 @@ namespace stream
|
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
auto sentPacket = *it; |
|
|
|
|
uint64_t rtt = ts - sentPacket->sendTime; |
|
|
|
|
if(ts < sentPacket->sendTime) |
|
|
|
|
int64_t rtt = (int64_t)ts - (int64_t)sentPacket->sendTime; |
|
|
|
|
if (rtt < 0) |
|
|
|
|
LogPrint (eLogError, "Streaming: Packet ", seqn, "sent from the future, sendTime=", sentPacket->sendTime); |
|
|
|
|
if (!seqn) |
|
|
|
|
{ |
|
|
|
|
LogPrint(eLogError, "Streaming: Packet ", seqn, "sent from the future, sendTime=", sentPacket->sendTime); |
|
|
|
|
rtt = 1; |
|
|
|
|
firstRttSample = true; |
|
|
|
|
rttSample = rtt < 0 ? 1 : rtt; |
|
|
|
|
} |
|
|
|
|
if (seqn) |
|
|
|
|
m_RTT = std::round (RTT_EWMA_ALPHA * m_RTT + (1.0 - RTT_EWMA_ALPHA) * rtt); |
|
|
|
|
else |
|
|
|
|
m_RTT = rtt; |
|
|
|
|
m_RTO = m_RTT*1.5; // TODO: implement it better
|
|
|
|
|
else if (!sentPacket->resent && seqn > m_TunnelsChangeSequenceNumber && rtt >= 0) |
|
|
|
|
rttSample = std::min (rttSample, (int)rtt); |
|
|
|
|
LogPrint (eLogDebug, "Streaming: Packet ", seqn, " acknowledged rtt=", rtt, " sentTime=", sentPacket->sendTime); |
|
|
|
|
m_SentPackets.erase (it++); |
|
|
|
|
m_LocalDestination.DeletePacket (sentPacket); |
|
|
|
|
acknowledged = true; |
|
|
|
|
if (m_WindowSize < WINDOW_SIZE) |
|
|
|
|
m_WindowSize++; // slow start
|
|
|
|
|
else |
|
|
|
|
{ |
|
|
|
|
// linear growth
|
|
|
|
|
if (ts > m_LastWindowSizeIncreaseTime + m_RTT) |
|
|
|
|
{ |
|
|
|
|
m_WindowSize++; |
|
|
|
|
if (m_WindowSize > MAX_WINDOW_SIZE) m_WindowSize = MAX_WINDOW_SIZE; |
|
|
|
|
m_LastWindowSizeIncreaseTime = ts; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
if (!seqn && m_RoutingSession) // first message confirmed
|
|
|
|
|
m_RoutingSession->SetSharedRoutingPath ( |
|
|
|
|
std::make_shared<i2p::garlic::GarlicRoutingPath> ( |
|
|
|
|
i2p::garlic::GarlicRoutingPath{m_CurrentOutboundTunnel, m_CurrentRemoteLease, m_RTT, 0, 0})); |
|
|
|
|
} |
|
|
|
|
else |
|
|
|
|
break; |
|
|
|
|
} |
|
|
|
|
if (rttSample != INT_MAX) |
|
|
|
|
{ |
|
|
|
|
if (firstRttSample) |
|
|
|
|
m_RTT = rttSample; |
|
|
|
|
else |
|
|
|
|
m_RTT = RTT_EWMA_ALPHA * rttSample + (1.0 - RTT_EWMA_ALPHA) * m_RTT; |
|
|
|
|
bool wasInitial = m_RTO == INITIAL_RTO; |
|
|
|
|
m_RTO = std::max (MIN_RTO, (int)(m_RTT * 1.5)); // TODO: implement it better
|
|
|
|
|
if (wasInitial) |
|
|
|
|
ScheduleResend (); |
|
|
|
|
} |
|
|
|
|
if (acknowledged && m_WindowSize >= WINDOW_SIZE) |
|
|
|
|
{ |
|
|
|
|
// linear growth
|
|
|
|
|
if (ts > m_LastWindowSizeIncreaseTime + m_RTT) |
|
|
|
|
{ |
|
|
|
|
m_WindowSize++; |
|
|
|
|
if (m_WindowSize > MAX_WINDOW_SIZE) m_WindowSize = MAX_WINDOW_SIZE; |
|
|
|
|
m_LastWindowSizeIncreaseTime = ts; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
if (firstRttSample && m_RoutingSession) |
|
|
|
|
m_RoutingSession->SetSharedRoutingPath ( |
|
|
|
|
std::make_shared<i2p::garlic::GarlicRoutingPath> ( |
|
|
|
|
i2p::garlic::GarlicRoutingPath{m_CurrentOutboundTunnel, m_CurrentRemoteLease, (int)m_RTT, 0, 0})); |
|
|
|
|
if (m_SentPackets.empty ()) |
|
|
|
|
m_ResendTimer.cancel (); |
|
|
|
|
if (acknowledged) |
|
|
|
|
@ -677,6 +693,7 @@ namespace stream
@@ -677,6 +693,7 @@ namespace stream
|
|
|
|
|
htobe32buf (packet + size, lastReceivedSeqn); |
|
|
|
|
size += 4; // ack Through
|
|
|
|
|
uint8_t numNacks = 0; |
|
|
|
|
bool choking = false; |
|
|
|
|
if (lastReceivedSeqn > m_LastReceivedSequenceNumber) |
|
|
|
|
{ |
|
|
|
|
// fill NACKs
|
|
|
|
|
@ -688,7 +705,8 @@ namespace stream
@@ -688,7 +705,8 @@ namespace stream
|
|
|
|
|
if (numNacks + (seqn - nextSeqn) >= 256) |
|
|
|
|
{ |
|
|
|
|
LogPrint (eLogError, "Streaming: Number of NACKs exceeds 256. seqn=", seqn, " nextSeqn=", nextSeqn); |
|
|
|
|
htobe32buf (packet + 12, nextSeqn); // change ack Through
|
|
|
|
|
htobe32buf (packet + 12, nextSeqn - 1); // change ack Through back
|
|
|
|
|
choking = true; |
|
|
|
|
break; |
|
|
|
|
} |
|
|
|
|
for (uint32_t i = nextSeqn; i < seqn; i++) |
|
|
|
|
@ -710,10 +728,17 @@ namespace stream
@@ -710,10 +728,17 @@ namespace stream
|
|
|
|
|
size++; // NACK count
|
|
|
|
|
} |
|
|
|
|
packet[size] = 0; |
|
|
|
|
size++; // resend delay
|
|
|
|
|
htobuf16 (packet + size, 0); // no flags set
|
|
|
|
|
size++; // resend delay
|
|
|
|
|
htobuf16 (packet + size, choking ? PACKET_FLAG_DELAY_REQUESTED : 0); // no flags set or delay
|
|
|
|
|
size += 2; // flags
|
|
|
|
|
htobuf16 (packet + size, 0); // no options
|
|
|
|
|
if (choking) |
|
|
|
|
{ |
|
|
|
|
htobuf16 (packet + size, 2); // 2 bytes delay interval
|
|
|
|
|
htobuf16 (packet + size + 2, DELAY_CHOKING); // set choking interval
|
|
|
|
|
size += 2; |
|
|
|
|
} |
|
|
|
|
else |
|
|
|
|
htobuf16 (packet + size, 0); // no options
|
|
|
|
|
size += 2; // options size
|
|
|
|
|
p.len = size; |
|
|
|
|
|
|
|
|
|
@ -881,7 +906,7 @@ namespace stream
@@ -881,7 +906,7 @@ namespace stream
|
|
|
|
|
m_CurrentOutboundTunnel = routingPath->outboundTunnel; |
|
|
|
|
m_CurrentRemoteLease = routingPath->remoteLease; |
|
|
|
|
m_RTT = routingPath->rtt; |
|
|
|
|
m_RTO = m_RTT*1.5; // TODO: implement it better
|
|
|
|
|
m_RTO = std::max (MIN_RTO, (int)(m_RTT * 1.5)); // TODO: implement it better
|
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -891,20 +916,32 @@ namespace stream
@@ -891,20 +916,32 @@ namespace stream
|
|
|
|
|
UpdateCurrentRemoteLease (true); |
|
|
|
|
if (m_CurrentRemoteLease && ts < m_CurrentRemoteLease->endDate + i2p::data::LEASE_ENDDATE_THRESHOLD) |
|
|
|
|
{ |
|
|
|
|
bool freshTunnel = false; |
|
|
|
|
if (!m_CurrentOutboundTunnel) |
|
|
|
|
{ |
|
|
|
|
auto leaseRouter = i2p::data::netdb.FindRouter (m_CurrentRemoteLease->tunnelGateway); |
|
|
|
|
m_CurrentOutboundTunnel = m_LocalDestination.GetOwner ()->GetTunnelPool ()->GetNextOutboundTunnel (nullptr, |
|
|
|
|
leaseRouter ? leaseRouter->GetCompatibleTransports (false) : (i2p::data::RouterInfo::CompatibleTransports)i2p::data::RouterInfo::eAllTransports); |
|
|
|
|
freshTunnel = true; |
|
|
|
|
} |
|
|
|
|
else if (!m_CurrentOutboundTunnel->IsEstablished ()) |
|
|
|
|
{ |
|
|
|
|
auto oldOutboundTunnel = m_CurrentOutboundTunnel; |
|
|
|
|
m_CurrentOutboundTunnel = m_LocalDestination.GetOwner ()->GetTunnelPool ()->GetNewOutboundTunnel (m_CurrentOutboundTunnel); |
|
|
|
|
if (m_CurrentOutboundTunnel && oldOutboundTunnel->GetEndpointIdentHash() != m_CurrentOutboundTunnel->GetEndpointIdentHash()) |
|
|
|
|
freshTunnel = true; |
|
|
|
|
} |
|
|
|
|
if (!m_CurrentOutboundTunnel) |
|
|
|
|
{ |
|
|
|
|
LogPrint (eLogError, "Streaming: No outbound tunnels in the pool, sSID=", m_SendStreamID); |
|
|
|
|
m_CurrentRemoteLease = nullptr; |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
if (freshTunnel) |
|
|
|
|
{ |
|
|
|
|
m_RTO = INITIAL_RTO; |
|
|
|
|
m_TunnelsChangeSequenceNumber = m_SequenceNumber; // should be determined more precisely
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
std::vector<i2p::tunnel::TunnelMessageBlock> msgs; |
|
|
|
|
for (const auto& it: packets) |
|
|
|
|
@ -936,10 +973,10 @@ namespace stream
@@ -936,10 +973,10 @@ namespace stream
|
|
|
|
|
if (m_RoutingSession->IsLeaseSetNonConfirmed ()) |
|
|
|
|
{ |
|
|
|
|
auto ts = i2p::util::GetMillisecondsSinceEpoch (); |
|
|
|
|
if (ts > m_RoutingSession->GetLeaseSetSubmissionTime () + i2p::garlic::LEASET_CONFIRMATION_TIMEOUT) |
|
|
|
|
if (ts > m_RoutingSession->GetLeaseSetSubmissionTime () + i2p::garlic::LEASESET_CONFIRMATION_TIMEOUT) |
|
|
|
|
{ |
|
|
|
|
// LeaseSet was not confirmed, should try other tunnels
|
|
|
|
|
LogPrint (eLogWarning, "Streaming: LeaseSet was not confirmed in ", i2p::garlic::LEASET_CONFIRMATION_TIMEOUT, " milliseconds. Trying to resubmit"); |
|
|
|
|
LogPrint (eLogWarning, "Streaming: LeaseSet was not confirmed in ", i2p::garlic::LEASESET_CONFIRMATION_TIMEOUT, " milliseconds. Trying to resubmit"); |
|
|
|
|
m_RoutingSession->SetSharedRoutingPath (nullptr); |
|
|
|
|
m_CurrentOutboundTunnel = nullptr; |
|
|
|
|
m_CurrentRemoteLease = nullptr; |
|
|
|
|
@ -989,6 +1026,7 @@ namespace stream
@@ -989,6 +1026,7 @@ namespace stream
|
|
|
|
|
{ |
|
|
|
|
if (ts >= it->sendTime + m_RTO) |
|
|
|
|
{ |
|
|
|
|
it->resent = true; |
|
|
|
|
it->sendTime = ts; |
|
|
|
|
packets.push_back (it); |
|
|
|
|
} |
|
|
|
|
@ -998,31 +1036,31 @@ namespace stream
@@ -998,31 +1036,31 @@ namespace stream
|
|
|
|
|
if (packets.size () > 0) |
|
|
|
|
{ |
|
|
|
|
m_NumResendAttempts++; |
|
|
|
|
m_RTO *= 2; |
|
|
|
|
switch (m_NumResendAttempts) |
|
|
|
|
if (m_NumResendAttempts == 1 && m_RTO != INITIAL_RTO) |
|
|
|
|
{ |
|
|
|
|
case 1: // congestion avoidance
|
|
|
|
|
m_WindowSize -= (m_WindowSize + WINDOW_SIZE_DROP_FRACTION) / WINDOW_SIZE_DROP_FRACTION; // adjustment >= 1
|
|
|
|
|
if (m_WindowSize < MIN_WINDOW_SIZE) m_WindowSize = MIN_WINDOW_SIZE; |
|
|
|
|
break; |
|
|
|
|
case 2: |
|
|
|
|
m_RTO = INITIAL_RTO; // drop RTO to initial upon tunnels pair change first time
|
|
|
|
|
#if (__cplusplus >= 201703L) // C++ 17 or higher
|
|
|
|
|
[[fallthrough]]; |
|
|
|
|
#endif |
|
|
|
|
// no break here
|
|
|
|
|
case 4: |
|
|
|
|
if (m_RoutingSession) m_RoutingSession->SetSharedRoutingPath (nullptr); |
|
|
|
|
UpdateCurrentRemoteLease (); // pick another lease
|
|
|
|
|
LogPrint (eLogWarning, "Streaming: Another remote lease has been selected for stream with rSID=", m_RecvStreamID, ", sSID=", m_SendStreamID); |
|
|
|
|
break; |
|
|
|
|
case 3: |
|
|
|
|
// congestion avoidance
|
|
|
|
|
m_RTO *= 2; |
|
|
|
|
m_WindowSize -= (m_WindowSize + WINDOW_SIZE_DROP_FRACTION) / WINDOW_SIZE_DROP_FRACTION; // adjustment >= 1
|
|
|
|
|
if (m_WindowSize < MIN_WINDOW_SIZE) m_WindowSize = MIN_WINDOW_SIZE; |
|
|
|
|
} |
|
|
|
|
else |
|
|
|
|
{ |
|
|
|
|
m_TunnelsChangeSequenceNumber = m_SequenceNumber; |
|
|
|
|
m_RTO = INITIAL_RTO; // drop RTO to initial upon tunnels pair change
|
|
|
|
|
if (m_RoutingSession) m_RoutingSession->SetSharedRoutingPath (nullptr); |
|
|
|
|
if (m_NumResendAttempts & 1) |
|
|
|
|
{ |
|
|
|
|
// pick another outbound tunnel
|
|
|
|
|
if (m_RoutingSession) m_RoutingSession->SetSharedRoutingPath (nullptr); |
|
|
|
|
m_CurrentOutboundTunnel = m_LocalDestination.GetOwner ()->GetTunnelPool ()->GetNextOutboundTunnel (m_CurrentOutboundTunnel); |
|
|
|
|
LogPrint (eLogWarning, "Streaming: Another outbound tunnel has been selected for stream with sSID=", m_SendStreamID); |
|
|
|
|
break; |
|
|
|
|
default: ; |
|
|
|
|
LogPrint (eLogWarning, "Streaming: Resend #", m_NumResendAttempts, |
|
|
|
|
", another outbound tunnel has been selected for stream with sSID=", m_SendStreamID); |
|
|
|
|
} |
|
|
|
|
else |
|
|
|
|
{ |
|
|
|
|
UpdateCurrentRemoteLease (); // pick another lease
|
|
|
|
|
LogPrint (eLogWarning, "Streaming: Resend #", m_NumResendAttempts, |
|
|
|
|
", another remote lease has been selected for stream with rSID=", m_RecvStreamID, ", sSID=", m_SendStreamID); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
SendPackets (packets); |
|
|
|
|
} |
|
|
|
|
@ -1056,9 +1094,13 @@ namespace stream
@@ -1056,9 +1094,13 @@ namespace stream
|
|
|
|
|
{ |
|
|
|
|
if (m_RoutingSession && m_RoutingSession->IsLeaseSetNonConfirmed ()) |
|
|
|
|
{ |
|
|
|
|
// seems something went wrong and we should re-select tunnels
|
|
|
|
|
m_CurrentOutboundTunnel = nullptr; |
|
|
|
|
m_CurrentRemoteLease = nullptr; |
|
|
|
|
auto ts = i2p::util::GetMillisecondsSinceEpoch (); |
|
|
|
|
if (ts > m_RoutingSession->GetLeaseSetSubmissionTime () + i2p::garlic::LEASESET_CONFIRMATION_TIMEOUT) |
|
|
|
|
{ |
|
|
|
|
// seems something went wrong and we should re-select tunnels
|
|
|
|
|
m_CurrentOutboundTunnel = nullptr; |
|
|
|
|
m_CurrentRemoteLease = nullptr; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
SendQuickAck (); |
|
|
|
|
} |
|
|
|
|
|
orignal
8 months ago
committed by