mirror of https://github.com/PurpleI2P/i2pd.git
I2P: End-to-End encrypted and anonymous Internet
https://i2pd.website/
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
2050 lines
66 KiB
2050 lines
66 KiB
/* |
|
* Copyright (c) 2013-2024, The PurpleI2P Project |
|
* |
|
* This file is part of Purple i2pd project and licensed under BSD3 |
|
* |
|
* See full license text in LICENSE file at top of project tree |
|
*/ |
|
|
|
#include "Crypto.h" |
|
#include "Log.h" |
|
#include "RouterInfo.h" |
|
#include "RouterContext.h" |
|
#include "Tunnel.h" |
|
#include "Timestamp.h" |
|
#include "Destination.h" |
|
#include "Streaming.h" |
|
|
|
namespace i2p |
|
{ |
|
namespace stream |
|
{ |
|
void SendBufferQueue::Add (std::shared_ptr<SendBuffer> buf) |
|
{ |
|
if (buf) |
|
{ |
|
m_Buffers.push_back (buf); |
|
m_Size += buf->len; |
|
} |
|
} |
|
|
|
size_t SendBufferQueue::Get (uint8_t * buf, size_t len) |
|
{ |
|
size_t offset = 0; |
|
while (!m_Buffers.empty () && offset < len) |
|
{ |
|
auto nextBuffer = m_Buffers.front (); |
|
auto rem = nextBuffer->GetRemainingSize (); |
|
if (offset + rem <= len) |
|
{ |
|
// whole buffer |
|
memcpy (buf + offset, nextBuffer->GetRemaningBuffer (), rem); |
|
offset += rem; |
|
m_Buffers.pop_front (); // delete it |
|
} |
|
else |
|
{ |
|
// partially |
|
rem = len - offset; |
|
memcpy (buf + offset, nextBuffer->GetRemaningBuffer (), rem); |
|
nextBuffer->offset += rem; |
|
offset = len; // break |
|
} |
|
} |
|
m_Size -= offset; |
|
return offset; |
|
} |
|
|
|
void SendBufferQueue::CleanUp () |
|
{ |
|
if (!m_Buffers.empty ()) |
|
{ |
|
for (auto it: m_Buffers) |
|
it->Cancel (); |
|
m_Buffers.clear (); |
|
m_Size = 0; |
|
} |
|
} |
|
|
|
Stream::Stream (boost::asio::io_context& service, StreamingDestination& local, |
|
std::shared_ptr<const i2p::data::LeaseSet> remote, int port): 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_IsIncoming (false), m_IsAckSendScheduled (false), m_IsNAcked (false), m_IsFirstACK (false), |
|
m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), m_IsWinDropped (false), m_IsClientChoked (false), |
|
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), |
|
m_WindowDropTargetSize (0), m_WindowIncCounter (0), m_RTO (INITIAL_RTO), |
|
m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()), m_PrevRTTSample (INITIAL_RTT), m_WindowSizeTail (0), |
|
m_Jitter (0), m_MinPacingTime (0), |
|
m_PacingTime (INITIAL_PACING_TIME), m_PacingTimeRem (0), m_LastSendTime (0), m_LastACKRecieveTime (0), m_ACKRecieveInterval (local.GetOwner ()->GetStreamingAckDelay ()), m_RemoteLeaseChangeTime (0), |
|
m_LastACKSendTime (0), m_PacketACKInterval (1), m_PacketACKIntervalRem (0), // for limit inbound speed |
|
m_NumResendAttempts (0), m_NumPacketsToSend (0), m_MTU (STREAMING_MTU) |
|
{ |
|
RAND_bytes ((uint8_t *)&m_RecvStreamID, 4); |
|
m_RemoteIdentity = remote->GetIdentity (); |
|
auto outboundSpeed = local.GetOwner ()->GetStreamingOutboundSpeed (); |
|
if (outboundSpeed) |
|
m_MinPacingTime = (1000000LL*STREAMING_MTU)/outboundSpeed; |
|
|
|
auto inboundSpeed = local.GetOwner ()->GetStreamingInboundSpeed (); // for limit inbound speed |
|
if (inboundSpeed) |
|
m_PacketACKInterval = (1000000LL*STREAMING_MTU)/inboundSpeed; |
|
} |
|
|
|
Stream::Stream (boost::asio::io_context& service, StreamingDestination& local): |
|
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_IsIncoming (true), m_IsAckSendScheduled (false), m_IsNAcked (false), m_IsFirstACK (false), |
|
m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), m_IsWinDropped (false), m_IsClientChoked (false), |
|
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), |
|
m_RTO (INITIAL_RTO), m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()), |
|
m_PrevRTTSample (INITIAL_RTT), m_WindowSizeTail (0), m_Jitter (0), m_MinPacingTime (0), |
|
m_PacingTime (INITIAL_PACING_TIME), m_PacingTimeRem (0), m_LastSendTime (0), m_LastACKRecieveTime (0), m_ACKRecieveInterval (local.GetOwner ()->GetStreamingAckDelay ()), m_RemoteLeaseChangeTime (0), |
|
m_LastACKSendTime (0), m_PacketACKInterval (1), m_PacketACKIntervalRem (0), // for limit inbound speed |
|
m_NumResendAttempts (0), m_NumPacketsToSend (0), m_MTU (STREAMING_MTU) |
|
{ |
|
RAND_bytes ((uint8_t *)&m_RecvStreamID, 4); |
|
auto outboundSpeed = local.GetOwner ()->GetStreamingOutboundSpeed (); |
|
if (outboundSpeed) |
|
m_MinPacingTime = (1000000LL*STREAMING_MTU)/outboundSpeed; |
|
|
|
auto inboundSpeed = local.GetOwner ()->GetStreamingInboundSpeed (); // for limit inbound speed |
|
if (inboundSpeed) |
|
m_PacketACKInterval = (1000000LL*STREAMING_MTU)/inboundSpeed; |
|
} |
|
|
|
Stream::~Stream () |
|
{ |
|
CleanUp (); |
|
LogPrint (eLogDebug, "Streaming: Stream deleted"); |
|
} |
|
|
|
void Stream::Terminate (bool deleteFromDestination) // should be called from StreamingDestination::Stop only |
|
{ |
|
m_Status = eStreamStatusTerminated; |
|
m_AckSendTimer.cancel (); |
|
m_ReceiveTimer.cancel (); |
|
m_ResendTimer.cancel (); |
|
m_SendTimer.cancel (); |
|
//CleanUp (); /* Need to recheck - broke working on windows */ |
|
if (deleteFromDestination) |
|
m_LocalDestination.DeleteStream (shared_from_this ()); |
|
} |
|
|
|
void Stream::CleanUp () |
|
{ |
|
m_SendBuffer.CleanUp (); |
|
while (!m_ReceiveQueue.empty ()) |
|
{ |
|
auto packet = m_ReceiveQueue.front (); |
|
m_ReceiveQueue.pop (); |
|
m_LocalDestination.DeletePacket (packet); |
|
} |
|
|
|
m_NACKedPackets.clear (); |
|
|
|
for (auto it: m_SentPackets) |
|
m_LocalDestination.DeletePacket (it); |
|
m_SentPackets.clear (); |
|
|
|
for (auto it: m_SavedPackets) |
|
m_LocalDestination.DeletePacket (it); |
|
m_SavedPackets.clear (); |
|
} |
|
|
|
void Stream::HandleNextPacket (Packet * packet) |
|
{ |
|
if (m_Status == eStreamStatusTerminated) |
|
{ |
|
m_LocalDestination.DeletePacket (packet); |
|
return; |
|
} |
|
m_NumReceivedBytes += packet->GetLength (); |
|
if (!m_SendStreamID) |
|
{ |
|
m_SendStreamID = packet->GetReceiveStreamID (); |
|
if (!m_RemoteIdentity && packet->GetNACKCount () == 8 && // first incoming packet |
|
memcmp (packet->GetNACKs (), m_LocalDestination.GetOwner ()->GetIdentHash (), 32)) |
|
{ |
|
LogPrint (eLogWarning, "Streaming: Destination mismatch for ", m_LocalDestination.GetOwner ()->GetIdentHash ().ToBase32 ()); |
|
m_LocalDestination.DeletePacket (packet); |
|
return; |
|
} |
|
} |
|
|
|
if (!packet->IsNoAck ()) // ack received |
|
ProcessAck (packet); |
|
|
|
int32_t receivedSeqn = packet->GetSeqn (); |
|
if (!receivedSeqn && m_LastReceivedSequenceNumber >= 0) |
|
{ |
|
uint16_t flags = packet->GetFlags (); |
|
if (flags) |
|
// plain ack with options |
|
ProcessOptions (flags, packet); |
|
else |
|
// plain ack |
|
{ |
|
LogPrint (eLogDebug, "Streaming: Plain ACK received"); |
|
if (m_IsImmediateAckRequested) |
|
{ |
|
auto ts = i2p::util::GetMillisecondsSinceEpoch (); |
|
if (m_IsFirstRttSample) |
|
{ |
|
m_RTT = ts - m_LastSendTime; |
|
m_IsFirstRttSample = false; |
|
} |
|
else |
|
m_RTT = (m_RTT + (ts - m_LastSendTime)) / 2; |
|
m_IsImmediateAckRequested = false; |
|
} |
|
} |
|
m_LocalDestination.DeletePacket (packet); |
|
return; |
|
} |
|
|
|
LogPrint (eLogDebug, "Streaming: Received seqn=", receivedSeqn, " on sSID=", m_SendStreamID); |
|
if (receivedSeqn == m_LastReceivedSequenceNumber + 1) |
|
{ |
|
// we have received next in sequence message |
|
ProcessPacket (packet); |
|
if (m_Status == eStreamStatusTerminated) return; |
|
|
|
// we should also try stored messages if any |
|
for (auto it = m_SavedPackets.begin (); it != m_SavedPackets.end ();) |
|
{ |
|
if ((*it)->GetSeqn () == (uint32_t)(m_LastReceivedSequenceNumber + 1)) |
|
{ |
|
Packet * savedPacket = *it; |
|
m_SavedPackets.erase (it++); |
|
|
|
ProcessPacket (savedPacket); |
|
if (m_Status == eStreamStatusTerminated) return; |
|
} |
|
else |
|
break; |
|
} |
|
|
|
// schedule ack for last message |
|
if (m_Status == eStreamStatusOpen) |
|
{ |
|
if (!m_IsAckSendScheduled) |
|
{ |
|
auto ackTimeout = m_RTT/10; |
|
if (ackTimeout > m_AckDelay) ackTimeout = m_AckDelay; |
|
ScheduleAck (ackTimeout); |
|
} |
|
} |
|
else if (packet->IsSYN ()) |
|
// we have to send SYN back to incoming connection |
|
SendBuffer (); // also sets m_IsOpen |
|
} |
|
else |
|
{ |
|
if (receivedSeqn <= m_LastReceivedSequenceNumber) |
|
{ |
|
// we have received duplicate |
|
LogPrint (eLogWarning, "Streaming: Duplicate message ", receivedSeqn, " on sSID=", m_SendStreamID); |
|
if (receivedSeqn <= m_PreviousReceivedSequenceNumber || receivedSeqn == m_LastReceivedSequenceNumber) |
|
{ |
|
m_CurrentOutboundTunnel = m_LocalDestination.GetOwner ()->GetTunnelPool ()->GetNextOutboundTunnel (m_CurrentOutboundTunnel); |
|
CancelRemoteLeaseChange (); |
|
UpdateCurrentRemoteLease (); |
|
} |
|
m_PreviousReceivedSequenceNumber = receivedSeqn; |
|
m_LocalDestination.DeletePacket (packet); // packet dropped |
|
if (!m_IsAckSendScheduled) |
|
{ |
|
SendQuickAck (); // resend ack for previous message again |
|
auto ackTimeout = m_RTT/10; |
|
if (ackTimeout > m_AckDelay) ackTimeout = m_AckDelay; |
|
ScheduleAck (ackTimeout); |
|
} |
|
} |
|
else |
|
{ |
|
LogPrint (eLogWarning, "Streaming: Missing messages on sSID=", m_SendStreamID, ": from ", m_LastReceivedSequenceNumber + 1, " to ", receivedSeqn - 1); |
|
// save message and wait for missing message again |
|
SavePacket (packet); |
|
if (m_LastReceivedSequenceNumber >= 0) |
|
{ |
|
if (!m_IsAckSendScheduled) |
|
{ |
|
// send NACKs for missing messages |
|
SendQuickAck (); |
|
auto ackTimeout = m_RTT/10; |
|
if (ackTimeout > m_AckDelay) ackTimeout = m_AckDelay; |
|
ScheduleAck (ackTimeout); |
|
} |
|
} |
|
else |
|
// wait for SYN |
|
ScheduleAck (SYN_TIMEOUT); |
|
} |
|
} |
|
} |
|
|
|
void Stream::SavePacket (Packet * packet) |
|
{ |
|
if (!m_SavedPackets.insert (packet).second) |
|
m_LocalDestination.DeletePacket (packet); |
|
} |
|
|
|
void Stream::ProcessPacket (Packet * packet) |
|
{ |
|
uint32_t receivedSeqn = packet->GetSeqn (); |
|
uint16_t flags = packet->GetFlags (); |
|
LogPrint (eLogDebug, "Streaming: Process seqn=", receivedSeqn, ", flags=", flags); |
|
|
|
if (!ProcessOptions (flags, packet)) |
|
{ |
|
m_LocalDestination.DeletePacket (packet); |
|
Terminate (); |
|
return; |
|
} |
|
|
|
packet->offset = packet->GetPayload () - packet->buf; |
|
if (packet->GetLength () > 0) |
|
{ |
|
m_ReceiveQueue.push (packet); |
|
m_ReceiveTimer.cancel (); |
|
} |
|
else |
|
m_LocalDestination.DeletePacket (packet); |
|
|
|
m_LastReceivedSequenceNumber = receivedSeqn; |
|
|
|
if (flags & PACKET_FLAG_RESET) |
|
{ |
|
LogPrint (eLogDebug, "Streaming: closing stream sSID=", m_SendStreamID, ", rSID=", m_RecvStreamID, ": reset flag received in packet #", receivedSeqn); |
|
m_Status = eStreamStatusReset; |
|
Close (); |
|
} |
|
else if (flags & PACKET_FLAG_CLOSE) |
|
{ |
|
if (m_Status != eStreamStatusClosed) |
|
SendClose (); |
|
m_Status = eStreamStatusClosed; |
|
Terminate (); |
|
} |
|
} |
|
|
|
bool Stream::ProcessOptions (uint16_t flags, Packet * packet) |
|
{ |
|
const uint8_t * optionData = packet->GetOptionData (); |
|
size_t optionSize = packet->GetOptionSize (); |
|
if (optionSize > packet->len) |
|
{ |
|
LogPrint (eLogInfo, "Streaming: Invalid option size ", optionSize, " Discarded"); |
|
return false; |
|
} |
|
if (!flags) return true; |
|
bool immediateAckRequested = false; |
|
if (flags & PACKET_FLAG_DELAY_REQUESTED) |
|
{ |
|
uint16_t delayRequested = bufbe16toh (optionData); |
|
if (!delayRequested) // 0 requests an immediate ack |
|
immediateAckRequested = true; |
|
else if (!m_IsAckSendScheduled) |
|
{ |
|
if (delayRequested < m_RTT) |
|
{ |
|
m_IsAckSendScheduled = true; |
|
m_AckSendTimer.expires_from_now (boost::posix_time::milliseconds(delayRequested)); |
|
m_AckSendTimer.async_wait (std::bind (&Stream::HandleAckSendTimer, |
|
shared_from_this (), std::placeholders::_1)); |
|
} |
|
if (delayRequested >= DELAY_CHOKING) |
|
{ |
|
if (!m_IsClientChoked) |
|
{ |
|
LogPrint (eLogDebug, "Streaming: Client choked, set min. window size"); |
|
m_WindowDropTargetSize = MIN_WINDOW_SIZE; |
|
m_LastWindowDropSize = 0; |
|
m_WindowIncCounter = 0; |
|
m_IsClientChoked = true; |
|
m_IsWinDropped = false; |
|
m_DropWindowDelaySequenceNumber = m_SequenceNumber; |
|
UpdatePacingTime (); |
|
} |
|
} |
|
} |
|
optionData += 2; |
|
} |
|
|
|
if (flags & PACKET_FLAG_FROM_INCLUDED) |
|
{ |
|
if (m_RemoteLeaseSet) m_RemoteIdentity = m_RemoteLeaseSet->GetIdentity (); |
|
if (!m_RemoteIdentity) |
|
m_RemoteIdentity = std::make_shared<i2p::data::IdentityEx>(optionData, optionSize); |
|
if (m_RemoteIdentity->IsRSA ()) |
|
{ |
|
LogPrint (eLogInfo, "Streaming: Incoming stream from RSA destination ", m_RemoteIdentity->GetIdentHash ().ToBase64 (), " Discarded"); |
|
return false; |
|
} |
|
optionData += m_RemoteIdentity->GetFullLen (); |
|
if (!m_RemoteLeaseSet) |
|
LogPrint (eLogDebug, "Streaming: Incoming stream from ", m_RemoteIdentity->GetIdentHash ().ToBase64 (), ", sSID=", m_SendStreamID, ", rSID=", m_RecvStreamID); |
|
} |
|
|
|
if (flags & PACKET_FLAG_MAX_PACKET_SIZE_INCLUDED) |
|
{ |
|
uint16_t maxPacketSize = bufbe16toh (optionData); |
|
LogPrint (eLogDebug, "Streaming: Max packet size ", maxPacketSize); |
|
optionData += 2; |
|
} |
|
|
|
if (flags & PACKET_FLAG_OFFLINE_SIGNATURE) |
|
{ |
|
if (!m_RemoteIdentity) |
|
{ |
|
LogPrint (eLogInfo, "Streaming: offline signature without identity"); |
|
return false; |
|
} |
|
// if we have it in LeaseSet already we don't need to parse it again |
|
if (m_RemoteLeaseSet) m_TransientVerifier = m_RemoteLeaseSet->GetTransientVerifier (); |
|
if (m_TransientVerifier) |
|
{ |
|
// skip option data |
|
optionData += 6; // timestamp and key type |
|
optionData += m_TransientVerifier->GetPublicKeyLen (); // public key |
|
optionData += m_RemoteIdentity->GetSignatureLen (); // signature |
|
} |
|
else |
|
{ |
|
// transient key |
|
size_t offset = 0; |
|
m_TransientVerifier = i2p::data::ProcessOfflineSignature (m_RemoteIdentity, optionData, optionSize - (optionData - packet->GetOptionData ()), offset); |
|
optionData += offset; |
|
if (!m_TransientVerifier) |
|
{ |
|
LogPrint (eLogError, "Streaming: offline signature failed"); |
|
return false; |
|
} |
|
} |
|
} |
|
|
|
if (flags & PACKET_FLAG_SIGNATURE_INCLUDED) |
|
{ |
|
uint8_t signature[256]; |
|
auto signatureLen = m_TransientVerifier ? m_TransientVerifier->GetSignatureLen () : m_RemoteIdentity->GetSignatureLen (); |
|
if(signatureLen <= sizeof(signature)) |
|
{ |
|
memcpy (signature, optionData, signatureLen); |
|
memset (const_cast<uint8_t *>(optionData), 0, signatureLen); |
|
bool verified = m_TransientVerifier ? |
|
m_TransientVerifier->Verify (packet->GetBuffer (), packet->GetLength (), signature) : |
|
m_RemoteIdentity->Verify (packet->GetBuffer (), packet->GetLength (), signature); |
|
if (!verified) |
|
{ |
|
LogPrint (eLogError, "Streaming: Signature verification failed, sSID=", m_SendStreamID, ", rSID=", m_RecvStreamID); |
|
Close (); |
|
flags |= PACKET_FLAG_CLOSE; |
|
} |
|
memcpy (const_cast<uint8_t *>(optionData), signature, signatureLen); |
|
optionData += signatureLen; |
|
} |
|
else |
|
{ |
|
LogPrint (eLogError, "Streaming: Signature too big, ", signatureLen, " bytes"); |
|
return false; |
|
} |
|
} |
|
if (immediateAckRequested) |
|
SendQuickAck (); |
|
return true; |
|
} |
|
|
|
void Stream::HandlePing (Packet * packet) |
|
{ |
|
uint16_t flags = packet->GetFlags (); |
|
if (ProcessOptions (flags, packet) && m_RemoteIdentity) |
|
{ |
|
// send pong |
|
Packet p; |
|
memset (p.buf, 0, 22); // minimal header all zeroes |
|
memcpy (p.buf + 4, packet->buf, 4); // but receiveStreamID is the sendStreamID from the ping |
|
htobe16buf (p.buf + 18, PACKET_FLAG_ECHO); // and echo flag |
|
auto payloadLen = int(packet->len) - (packet->GetPayload () - packet->buf); |
|
if (payloadLen > 0) |
|
memcpy (p.buf + 22, packet->GetPayload (), payloadLen); |
|
else |
|
payloadLen = 0; |
|
p.len = payloadLen + 22; |
|
SendPackets (std::vector<Packet *> { &p }); |
|
LogPrint (eLogDebug, "Streaming: Pong of ", p.len, " bytes sent"); |
|
} |
|
m_LocalDestination.DeletePacket (packet); |
|
} |
|
|
|
void Stream::ProcessAck (Packet * packet) |
|
{ |
|
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); |
|
return; |
|
} |
|
int rttSample = INT_MAX; |
|
int incCounter = 0; |
|
m_IsNAcked = false; |
|
m_IsResendNeeded = false; |
|
int nackCount = packet->GetNACKCount (); |
|
for (auto it = m_SentPackets.begin (); it != m_SentPackets.end ();) |
|
{ |
|
auto seqn = (*it)->GetSeqn (); |
|
if (seqn <= ackThrough) |
|
{ |
|
if (nackCount > 0) |
|
{ |
|
bool nacked = false; |
|
for (int i = 0; i < nackCount; i++) |
|
if (seqn == packet->GetNACK (i)) |
|
{ |
|
m_NACKedPackets.insert (*it); |
|
m_IsNAcked = true; |
|
nacked = true; |
|
break; |
|
} |
|
if (nacked) |
|
{ |
|
LogPrint (eLogDebug, "Streaming: Packet ", seqn, " NACK"); |
|
++it; |
|
continue; |
|
} |
|
} |
|
auto sentPacket = *it; |
|
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) |
|
{ |
|
m_IsFirstRttSample = true; |
|
rttSample = rtt < 0 ? 1 : rtt; |
|
} |
|
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_WindowIncCounter < MAX_WINDOW_SIZE && !m_IsFirstACK && !m_IsWinDropped) |
|
incCounter++; |
|
} |
|
else |
|
break; |
|
} |
|
if (m_LastACKRecieveTime) |
|
{ |
|
uint64_t interval = ts - m_LastACKRecieveTime; |
|
if (m_ACKRecieveInterval) |
|
m_ACKRecieveInterval = (m_ACKRecieveInterval + interval) / 2; |
|
else |
|
m_ACKRecieveInterval = interval; |
|
} |
|
m_LastACKRecieveTime = ts; |
|
if (rttSample != INT_MAX) |
|
{ |
|
if (m_IsFirstRttSample && !m_IsFirstACK) |
|
{ |
|
m_RTT = rttSample; |
|
m_SlowRTT = rttSample; |
|
m_SlowRTT2 = rttSample; |
|
m_PrevRTTSample = rttSample; |
|
m_Jitter = rttSample / 10; // 10% |
|
m_Jitter += 15; // for low-latency connections |
|
m_IsFirstRttSample = false; |
|
} |
|
else |
|
m_RTT = (m_PrevRTTSample + rttSample) / 2; |
|
if (!m_IsWinDropped) |
|
{ |
|
m_SlowRTT = SLOWRTT_EWMA_ALPHA * m_RTT + (1.0 - SLOWRTT_EWMA_ALPHA) * m_SlowRTT; |
|
m_SlowRTT2 = RTT_EWMA_ALPHA * m_RTT + (1.0 - RTT_EWMA_ALPHA) * m_SlowRTT2; |
|
// calculate jitter |
|
double jitter = 0; |
|
if (rttSample > m_PrevRTTSample) |
|
jitter = rttSample - m_PrevRTTSample; |
|
else if (rttSample < m_PrevRTTSample) |
|
jitter = m_PrevRTTSample - rttSample; |
|
else |
|
jitter = rttSample / 10; // 10% |
|
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 && !m_IsClientChoked) // Drop window if RTT grows too fast, late detection |
|
{ |
|
LogPrint (eLogDebug, "Streaming: Congestion detected, reduce window size"); |
|
ProcessWindowDrop (); |
|
} |
|
UpdatePacingTime (); |
|
m_PrevRTTSample = rttSample; |
|
|
|
bool wasInitial = m_RTO == INITIAL_RTO; |
|
m_RTO = std::max (MIN_RTO, (int)(m_RTT * 1.3 + m_Jitter + m_ACKRecieveInterval)); // TODO: implement it better |
|
|
|
if (wasInitial) |
|
ScheduleResend (); |
|
} |
|
if (m_IsClientChoked && ackThrough > m_DropWindowDelaySequenceNumber) |
|
{ |
|
m_IsClientChoked = false; |
|
} |
|
if (m_IsWinDropped && ackThrough > m_DropWindowDelaySequenceNumber) |
|
{ |
|
m_IsFirstRttSample = true; |
|
m_IsWinDropped = false; |
|
} |
|
if (m_WindowDropTargetSize && int(m_SentPackets.size ()) <= m_WindowDropTargetSize) |
|
{ |
|
m_WindowSize = m_WindowDropTargetSize; |
|
m_WindowDropTargetSize = 0; |
|
} |
|
if (acknowledged || m_IsNAcked) |
|
{ |
|
ScheduleResend (); |
|
} |
|
if (m_SendBuffer.IsEmpty () && m_SentPackets.size () > 0) // tail loss |
|
{ |
|
m_IsResendNeeded = true; |
|
m_RTO = std::max (MIN_RTO, (int)(m_RTT * 1.5 + m_Jitter + m_ACKRecieveInterval)); // to prevent spurious retransmit |
|
} |
|
if (m_SentPackets.empty () && m_SendBuffer.IsEmpty ()) |
|
{ |
|
m_ResendTimer.cancel (); |
|
m_SendTimer.cancel (); |
|
m_LastACKRecieveTime = 0; |
|
m_ACKRecieveInterval = m_AckDelay; |
|
} |
|
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_IsTimeOutResend = false; |
|
SendBuffer (); |
|
} |
|
if (m_Status == eStreamStatusClosed) |
|
Terminate (); |
|
else if (m_Status == eStreamStatusClosing) |
|
Close (); // check is all outgoing messages have been sent and we can send close |
|
} |
|
|
|
size_t Stream::Receive (uint8_t * buf, size_t len, int timeout) |
|
{ |
|
if (!len) return 0; |
|
size_t ret = 0; |
|
volatile bool done = false; |
|
std::condition_variable newDataReceived; |
|
std::mutex newDataReceivedMutex; |
|
AsyncReceive (boost::asio::buffer (buf, len), |
|
[&ret, &done, &newDataReceived, &newDataReceivedMutex](const boost::system::error_code& ecode, std::size_t bytes_transferred) |
|
{ |
|
if (ecode == boost::asio::error::timed_out) |
|
ret = 0; |
|
else |
|
ret = bytes_transferred; |
|
std::unique_lock<std::mutex> l(newDataReceivedMutex); |
|
newDataReceived.notify_all (); |
|
done = true; |
|
}, |
|
timeout); |
|
if (!done) |
|
{ std::unique_lock<std::mutex> l(newDataReceivedMutex); |
|
if (!done && newDataReceived.wait_for (l, std::chrono::seconds (timeout)) == std::cv_status::timeout) |
|
ret = 0; |
|
} |
|
if (!done) |
|
{ |
|
// make sure that AsycReceive complete |
|
auto s = shared_from_this(); |
|
boost::asio::post (m_Service, [s]() |
|
{ |
|
s->m_ReceiveTimer.cancel (); |
|
}); |
|
int i = 0; |
|
while (!done && i < 100) // 1 sec |
|
{ |
|
std::this_thread::sleep_for (std::chrono::milliseconds(10)); |
|
i++; |
|
} |
|
} |
|
return ret; |
|
} |
|
|
|
size_t Stream::Send (const uint8_t * buf, size_t len) |
|
{ |
|
AsyncSend (buf, len, nullptr); |
|
return len; |
|
} |
|
|
|
void Stream::AsyncSend (const uint8_t * buf, size_t len, SendHandler handler) |
|
{ |
|
std::shared_ptr<i2p::stream::SendBuffer> buffer; |
|
if (len > 0 && buf) |
|
buffer = std::make_shared<i2p::stream::SendBuffer>(buf, len, handler); |
|
else if (handler) |
|
handler(boost::system::error_code ()); |
|
auto s = shared_from_this (); |
|
boost::asio::post (m_Service, [s, buffer]() |
|
{ |
|
if (buffer) |
|
s->m_SendBuffer.Add (buffer); |
|
s->SendBuffer (); |
|
}); |
|
} |
|
|
|
void Stream::SendBuffer () |
|
{ |
|
if (m_RemoteLeaseSet) // don't scheudle send for first SYN for incoming stream |
|
ScheduleSend (); |
|
auto ts = i2p::util::GetMillisecondsSinceEpoch (); |
|
int numMsgs = m_WindowSize - m_SentPackets.size (); |
|
if (numMsgs <= 0 || !m_IsSendTime) // window is full |
|
{ |
|
m_LastSendTime = ts; |
|
return; |
|
} |
|
else if (numMsgs > m_NumPacketsToSend) |
|
numMsgs = m_NumPacketsToSend; |
|
bool isNoAck = m_LastReceivedSequenceNumber < 0; // first packet |
|
std::vector<Packet *> packets; |
|
while ((m_Status == eStreamStatusNew) || (IsEstablished () && !m_SendBuffer.IsEmpty () && numMsgs > 0)) |
|
{ |
|
Packet * p = m_LocalDestination.NewPacket (); |
|
uint8_t * packet = p->GetBuffer (); |
|
// TODO: implement setters |
|
size_t size = 0; |
|
htobe32buf (packet + size, m_SendStreamID); |
|
size += 4; // sendStreamID |
|
htobe32buf (packet + size, m_RecvStreamID); |
|
size += 4; // receiveStreamID |
|
htobe32buf (packet + size, m_SequenceNumber++); |
|
size += 4; // sequenceNum |
|
if (isNoAck) |
|
htobuf32 (packet + size, 0); |
|
else |
|
htobe32buf (packet + size, m_LastReceivedSequenceNumber); |
|
size += 4; // ack Through |
|
if (m_Status == eStreamStatusNew && !m_SendStreamID && m_RemoteIdentity) |
|
{ |
|
// first SYN packet |
|
packet[size] = 8; |
|
size++; // NACK count |
|
memcpy (packet + size, m_RemoteIdentity->GetIdentHash (), 32); |
|
size += 32; |
|
} |
|
else |
|
{ |
|
packet[size] = 0; |
|
size++; // NACK count |
|
} |
|
packet[size] = m_RTO/1000; |
|
size++; // resend delay |
|
if (m_Status == eStreamStatusNew) |
|
{ |
|
// initial packet |
|
m_Status = eStreamStatusOpen; |
|
if (!m_RemoteLeaseSet) m_RemoteLeaseSet = m_LocalDestination.GetOwner ()->FindLeaseSet (m_RemoteIdentity->GetIdentHash ());; |
|
if (m_RemoteLeaseSet) |
|
{ |
|
m_RoutingSession = m_LocalDestination.GetOwner ()->GetRoutingSession (m_RemoteLeaseSet, true, !m_IsIncoming); |
|
m_MTU = (m_RoutingSession && m_RoutingSession->IsRatchets ()) ? STREAMING_MTU_RATCHETS : STREAMING_MTU; |
|
} |
|
uint16_t flags = PACKET_FLAG_SYNCHRONIZE | PACKET_FLAG_FROM_INCLUDED | |
|
PACKET_FLAG_SIGNATURE_INCLUDED | PACKET_FLAG_MAX_PACKET_SIZE_INCLUDED; |
|
if (isNoAck) flags |= PACKET_FLAG_NO_ACK; |
|
bool isOfflineSignature = m_LocalDestination.GetOwner ()->GetPrivateKeys ().IsOfflineSignature (); |
|
if (isOfflineSignature) flags |= PACKET_FLAG_OFFLINE_SIGNATURE; |
|
htobe16buf (packet + size, flags); |
|
size += 2; // flags |
|
size_t identityLen = m_LocalDestination.GetOwner ()->GetIdentity ()->GetFullLen (); |
|
size_t signatureLen = m_LocalDestination.GetOwner ()->GetPrivateKeys ().GetSignatureLen (); |
|
uint8_t * optionsSize = packet + size; // set options size later |
|
size += 2; // options size |
|
m_LocalDestination.GetOwner ()->GetIdentity ()->ToBuffer (packet + size, identityLen); |
|
size += identityLen; // from |
|
htobe16buf (packet + size, m_MTU); |
|
size += 2; // max packet size |
|
if (isOfflineSignature) |
|
{ |
|
const auto& offlineSignature = m_LocalDestination.GetOwner ()->GetPrivateKeys ().GetOfflineSignature (); |
|
memcpy (packet + size, offlineSignature.data (), offlineSignature.size ()); |
|
size += offlineSignature.size (); // offline signature |
|
} |
|
uint8_t * signature = packet + size; // set it later |
|
memset (signature, 0, signatureLen); // zeroes for now |
|
size += signatureLen; // signature |
|
htobe16buf (optionsSize, packet + size - 2 - optionsSize); // actual options size |
|
size += m_SendBuffer.Get (packet + size, m_MTU); // payload |
|
m_LocalDestination.GetOwner ()->Sign (packet, size, signature); |
|
} |
|
else |
|
{ |
|
// follow on packet |
|
htobuf16 (packet + size, 0); |
|
size += 2; // flags |
|
htobuf16 (packet + size, 0); // no options |
|
size += 2; // options size |
|
size += m_SendBuffer.Get(packet + size, m_MTU); // payload |
|
} |
|
p->len = size; |
|
packets.push_back (p); |
|
numMsgs--; |
|
} |
|
if (packets.size () > 0) |
|
{ |
|
if (m_SavedPackets.empty ()) // no NACKS |
|
{ |
|
m_IsAckSendScheduled = false; |
|
m_AckSendTimer.cancel (); |
|
} |
|
bool isEmpty = m_SentPackets.empty (); |
|
// auto ts = i2p::util::GetMillisecondsSinceEpoch (); |
|
for (auto& it: packets) |
|
{ |
|
it->sendTime = ts; |
|
m_SentPackets.insert (it); |
|
} |
|
SendPackets (packets); |
|
m_LastSendTime = ts; |
|
m_IsSendTime = false; |
|
if (m_Status == eStreamStatusClosing && m_SendBuffer.IsEmpty ()) |
|
SendClose (); |
|
if (isEmpty) |
|
ScheduleResend (); |
|
} |
|
} |
|
|
|
void Stream::SendQuickAck () |
|
{ |
|
int32_t lastReceivedSeqn = m_LastReceivedSequenceNumber; |
|
// for limit inbound speed |
|
auto ts = i2p::util::GetMillisecondsSinceEpoch (); |
|
int numPackets = 0; |
|
bool lostPackets = false; |
|
int64_t passedTime = m_PacketACKInterval * INITIAL_WINDOW_SIZE; // in microseconds // while m_LastACKSendTime == 0 |
|
if (m_LastACKSendTime) |
|
passedTime = (ts - m_LastACKSendTime)*1000; // in microseconds |
|
numPackets = (passedTime + m_PacketACKIntervalRem) / m_PacketACKInterval; |
|
m_PacketACKIntervalRem = (passedTime + m_PacketACKIntervalRem) - (numPackets * m_PacketACKInterval); |
|
if (m_LastConfirmedReceivedSequenceNumber + numPackets < m_LastReceivedSequenceNumber) |
|
{ |
|
lastReceivedSeqn = m_LastConfirmedReceivedSequenceNumber + numPackets; |
|
if (!m_IsAckSendScheduled) |
|
{ |
|
auto ackTimeout = m_RTT/10; |
|
if (ackTimeout > m_AckDelay) ackTimeout = m_AckDelay; |
|
ScheduleAck (ackTimeout); |
|
} |
|
} |
|
if (numPackets == 0) return; |
|
// for limit inbound speed |
|
if (!m_SavedPackets.empty ()) |
|
{ |
|
for (auto it: m_SavedPackets) |
|
{ |
|
auto seqn = it->GetSeqn (); |
|
// for limit inbound speed |
|
if (m_LastConfirmedReceivedSequenceNumber + numPackets < int(seqn)) |
|
{ |
|
if (!m_IsAckSendScheduled) |
|
{ |
|
auto ackTimeout = m_RTT/10; |
|
if (ackTimeout > m_AckDelay) ackTimeout = m_AckDelay; |
|
ScheduleAck (ackTimeout); |
|
} |
|
if (lostPackets) |
|
break; |
|
else |
|
return; |
|
} |
|
// for limit inbound speed |
|
if ((int)seqn > lastReceivedSeqn) |
|
{ |
|
lastReceivedSeqn = seqn; |
|
lostPackets = true; // for limit inbound speed |
|
} |
|
} |
|
} |
|
if (lastReceivedSeqn < 0) |
|
{ |
|
LogPrint (eLogError, "Streaming: No packets have been received yet"); |
|
return; |
|
} |
|
|
|
Packet p; |
|
uint8_t * packet = p.GetBuffer (); |
|
size_t size = 0; |
|
htobe32buf (packet + size, m_SendStreamID); |
|
size += 4; // sendStreamID |
|
htobe32buf (packet + size, m_RecvStreamID); |
|
size += 4; // receiveStreamID |
|
htobuf32 (packet + size, 0); // this is plain Ack message |
|
size += 4; // sequenceNum |
|
htobe32buf (packet + size, lastReceivedSeqn); |
|
size += 4; // ack Through |
|
uint8_t numNacks = 0; |
|
bool choking = false; |
|
if (lastReceivedSeqn > m_LastReceivedSequenceNumber) |
|
{ |
|
// fill NACKs |
|
uint8_t * nacks = packet + size + 1; |
|
auto nextSeqn = m_LastReceivedSequenceNumber + 1; |
|
for (auto it: m_SavedPackets) |
|
{ |
|
auto seqn = it->GetSeqn (); |
|
if (m_LastConfirmedReceivedSequenceNumber + numPackets < int(seqn)) // for limit inbound speed |
|
{ |
|
htobe32buf (packet + 12, nextSeqn - 1); |
|
break; |
|
} |
|
if (numNacks + (seqn - nextSeqn) >= 256) |
|
{ |
|
LogPrint (eLogError, "Streaming: Number of NACKs exceeds 256. seqn=", seqn, " nextSeqn=", nextSeqn); |
|
htobe32buf (packet + 12, nextSeqn - 1); // change ack Through back |
|
choking = true; |
|
break; |
|
} |
|
for (uint32_t i = nextSeqn; i < seqn; i++) |
|
{ |
|
htobe32buf (nacks, i); |
|
nacks += 4; |
|
numNacks++; |
|
} |
|
nextSeqn = seqn + 1; |
|
} |
|
packet[size] = numNacks; |
|
size++; // NACK count |
|
size += numNacks*4; // NACKs |
|
} |
|
else |
|
{ |
|
// No NACKs |
|
packet[size] = 0; |
|
size++; // NACK count |
|
} |
|
packet[size] = 0; |
|
size++; // resend delay |
|
bool requestImmediateAck = false; |
|
if (!choking) |
|
requestImmediateAck = m_LastSendTime && ts > m_LastSendTime + REQUEST_IMMEDIATE_ACK_INTERVAL && |
|
ts > m_LastSendTime + REQUEST_IMMEDIATE_ACK_INTERVAL + m_LocalDestination.GetRandom () % REQUEST_IMMEDIATE_ACK_INTERVAL_VARIANCE; |
|
htobe16buf (packet + size, (choking || requestImmediateAck) ? PACKET_FLAG_DELAY_REQUESTED : 0); // no flags set or delay requested |
|
size += 2; // flags |
|
if (choking || requestImmediateAck) |
|
{ |
|
htobe16buf (packet + size, 2); // 2 bytes delay interval |
|
htobe16buf (packet + size + 2, choking ? DELAY_CHOKING : 0); // set choking or immediate ack interval |
|
size += 2; |
|
if (requestImmediateAck) // ack request sent |
|
{ |
|
m_LastSendTime = ts; |
|
m_IsImmediateAckRequested = true; |
|
} |
|
} |
|
else |
|
htobuf16 (packet + size, 0); // no options |
|
size += 2; // options size |
|
p.len = size; |
|
|
|
SendPackets (std::vector<Packet *> { &p }); |
|
m_LastACKSendTime = ts; // for limit inbound speed |
|
m_LastConfirmedReceivedSequenceNumber = lastReceivedSeqn; // for limit inbound speed |
|
LogPrint (eLogDebug, "Streaming: Quick Ack sent. ", (int)numNacks, " NACKs"); |
|
} |
|
|
|
void Stream::SendPing () |
|
{ |
|
Packet p; |
|
uint8_t * packet = p.GetBuffer (); |
|
size_t size = 0; |
|
htobe32buf (packet, m_RecvStreamID); |
|
size += 4; // sendStreamID |
|
memset (packet + size, 0, 14); |
|
size += 14; // all zeroes |
|
uint16_t flags = PACKET_FLAG_ECHO | PACKET_FLAG_SIGNATURE_INCLUDED | PACKET_FLAG_FROM_INCLUDED; |
|
bool isOfflineSignature = m_LocalDestination.GetOwner ()->GetPrivateKeys ().IsOfflineSignature (); |
|
if (isOfflineSignature) flags |= PACKET_FLAG_OFFLINE_SIGNATURE; |
|
htobe16buf (packet + size, flags); |
|
size += 2; // flags |
|
size_t identityLen = m_LocalDestination.GetOwner ()->GetIdentity ()->GetFullLen (); |
|
size_t signatureLen = m_LocalDestination.GetOwner ()->GetPrivateKeys ().GetSignatureLen (); |
|
uint8_t * optionsSize = packet + size; // set options size later |
|
size += 2; // options size |
|
m_LocalDestination.GetOwner ()->GetIdentity ()->ToBuffer (packet + size, identityLen); |
|
size += identityLen; // from |
|
if (isOfflineSignature) |
|
{ |
|
const auto& offlineSignature = m_LocalDestination.GetOwner ()->GetPrivateKeys ().GetOfflineSignature (); |
|
memcpy (packet + size, offlineSignature.data (), offlineSignature.size ()); |
|
size += offlineSignature.size (); // offline signature |
|
} |
|
uint8_t * signature = packet + size; // set it later |
|
memset (signature, 0, signatureLen); // zeroes for now |
|
size += signatureLen; // signature |
|
htobe16buf (optionsSize, packet + size - 2 - optionsSize); // actual options size |
|
m_LocalDestination.GetOwner ()->Sign (packet, size, signature); |
|
p.len = size; |
|
SendPackets (std::vector<Packet *> { &p }); |
|
LogPrint (eLogDebug, "Streaming: Ping of ", p.len, " bytes sent"); |
|
} |
|
|
|
void Stream::Close () |
|
{ |
|
LogPrint(eLogDebug, "Streaming: closing stream with sSID=", m_SendStreamID, ", rSID=", m_RecvStreamID, ", status=", m_Status); |
|
switch (m_Status) |
|
{ |
|
case eStreamStatusOpen: |
|
m_Status = eStreamStatusClosing; |
|
Close (); // recursion |
|
if (m_Status == eStreamStatusClosing) //still closing |
|
LogPrint (eLogDebug, "Streaming: Trying to send stream data before closing, sSID=", m_SendStreamID); |
|
break; |
|
case eStreamStatusReset: |
|
// TODO: send reset |
|
Terminate (); |
|
break; |
|
case eStreamStatusClosing: |
|
if (m_SentPackets.empty () && m_SendBuffer.IsEmpty ()) // nothing to send |
|
{ |
|
m_Status = eStreamStatusClosed; |
|
SendClose(); |
|
} |
|
break; |
|
case eStreamStatusClosed: |
|
// already closed |
|
Terminate (); |
|
break; |
|
default: |
|
LogPrint (eLogWarning, "Streaming: Unexpected stream status=", (int)m_Status, " for sSID=", m_SendStreamID); |
|
}; |
|
} |
|
|
|
void Stream::SendClose () |
|
{ |
|
Packet * p = m_LocalDestination.NewPacket (); |
|
uint8_t * packet = p->GetBuffer (); |
|
size_t size = 0; |
|
htobe32buf (packet + size, m_SendStreamID); |
|
size += 4; // sendStreamID |
|
htobe32buf (packet + size, m_RecvStreamID); |
|
size += 4; // receiveStreamID |
|
htobe32buf (packet + size, m_SequenceNumber++); |
|
size += 4; // sequenceNum |
|
htobe32buf (packet + size, m_LastReceivedSequenceNumber >= 0 ? m_LastReceivedSequenceNumber : 0); |
|
size += 4; // ack Through |
|
packet[size] = 0; |
|
size++; // NACK count |
|
packet[size] = 0; |
|
size++; // resend delay |
|
htobe16buf (packet + size, PACKET_FLAG_CLOSE | PACKET_FLAG_SIGNATURE_INCLUDED); |
|
size += 2; // flags |
|
size_t signatureLen = m_LocalDestination.GetOwner ()->GetPrivateKeys ().GetSignatureLen (); |
|
htobe16buf (packet + size, signatureLen); // signature only |
|
size += 2; // options size |
|
uint8_t * signature = packet + size; |
|
memset (packet + size, 0, signatureLen); |
|
size += signatureLen; // signature |
|
m_LocalDestination.GetOwner ()->Sign (packet, size, signature); |
|
|
|
p->len = size; |
|
boost::asio::post (m_Service, std::bind (&Stream::SendPacket, shared_from_this (), p)); |
|
LogPrint (eLogDebug, "Streaming: FIN sent, sSID=", m_SendStreamID); |
|
} |
|
|
|
size_t Stream::ConcatenatePackets (uint8_t * buf, size_t len) |
|
{ |
|
size_t pos = 0; |
|
while (pos < len && !m_ReceiveQueue.empty ()) |
|
{ |
|
Packet * packet = m_ReceiveQueue.front (); |
|
size_t l = std::min (packet->GetLength (), len - pos); |
|
memcpy (buf + pos, packet->GetBuffer (), l); |
|
pos += l; |
|
packet->offset += l; |
|
if (!packet->GetLength ()) |
|
{ |
|
m_ReceiveQueue.pop (); |
|
m_LocalDestination.DeletePacket (packet); |
|
} |
|
} |
|
return pos; |
|
} |
|
|
|
bool Stream::SendPacket (Packet * packet) |
|
{ |
|
if (packet) |
|
{ |
|
if (m_IsAckSendScheduled) |
|
{ |
|
m_IsAckSendScheduled = false; |
|
m_AckSendTimer.cancel (); |
|
} |
|
if (!packet->sendTime) packet->sendTime = i2p::util::GetMillisecondsSinceEpoch (); |
|
SendPackets (std::vector<Packet *> { packet }); |
|
bool isEmpty = m_SentPackets.empty (); |
|
m_SentPackets.insert (packet); |
|
if (isEmpty) |
|
ScheduleResend (); |
|
return true; |
|
} |
|
else |
|
return false; |
|
} |
|
|
|
void Stream::SendPackets (const std::vector<Packet *>& packets) |
|
{ |
|
if (!m_RemoteLeaseSet) |
|
{ |
|
CancelRemoteLeaseChange (); |
|
UpdateCurrentRemoteLease (); |
|
if (!m_RemoteLeaseSet) |
|
{ |
|
LogPrint (eLogError, "Streaming: Can't send packets, missing remote LeaseSet, sSID=", m_SendStreamID); |
|
return; |
|
} |
|
} |
|
if (!m_RoutingSession || m_RoutingSession->IsTerminated () || !m_RoutingSession->IsReadyToSend ()) // expired and detached or new session sent |
|
{ |
|
m_RoutingSession = m_LocalDestination.GetOwner ()->GetRoutingSession (m_RemoteLeaseSet, true, !m_IsIncoming || m_SequenceNumber > 1); |
|
if (!m_RoutingSession) |
|
{ |
|
LogPrint (eLogError, "Streaming: Can't obtain routing session, sSID=", m_SendStreamID); |
|
Terminate (); |
|
return; |
|
} |
|
} |
|
if (!m_CurrentOutboundTunnel && m_RoutingSession) // first message to send |
|
{ |
|
// try to get shared path first |
|
auto routingPath = m_RoutingSession->GetSharedRoutingPath (); |
|
if (routingPath) |
|
{ |
|
m_CurrentOutboundTunnel = routingPath->outboundTunnel; |
|
m_CurrentRemoteLease = routingPath->remoteLease; |
|
m_RTT = routingPath->rtt; |
|
} |
|
} |
|
|
|
auto ts = i2p::util::GetMillisecondsSinceEpoch (); |
|
if (!m_CurrentRemoteLease || !m_CurrentRemoteLease->endDate) // excluded from LeaseSet |
|
{ |
|
CancelRemoteLeaseChange (); |
|
UpdateCurrentRemoteLease (true); |
|
} |
|
if (m_RemoteLeaseChangeTime && m_IsRemoteLeaseChangeInProgress && ts > m_RemoteLeaseChangeTime + INITIAL_RTO) |
|
{ |
|
LogPrint (eLogDebug, "Streaming: RemoteLease changed, set initial window size"); |
|
CancelRemoteLeaseChange (); |
|
m_CurrentRemoteLease = m_NextRemoteLease; |
|
ResetWindowSize (); |
|
} |
|
auto currentRemoteLease = m_CurrentRemoteLease; |
|
if (!m_IsRemoteLeaseChangeInProgress && m_RemoteLeaseSet && m_CurrentRemoteLease && ts >= m_CurrentRemoteLease->endDate - i2p::data::LEASE_ENDDATE_THRESHOLD) |
|
{ |
|
auto leases = m_RemoteLeaseSet->GetNonExpiredLeases (false); |
|
if (leases.size ()) |
|
{ |
|
m_IsRemoteLeaseChangeInProgress = true; |
|
UpdateCurrentRemoteLease (true); |
|
m_NextRemoteLease = m_CurrentRemoteLease; |
|
} |
|
else |
|
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 ()) |
|
std::tie(m_CurrentOutboundTunnel, freshTunnel) = m_LocalDestination.GetOwner ()->GetTunnelPool ()->GetNewOutboundTunnel (m_CurrentOutboundTunnel); |
|
if (!m_CurrentOutboundTunnel) |
|
{ |
|
LogPrint (eLogError, "Streaming: No outbound tunnels in the pool, sSID=", m_SendStreamID); |
|
m_CurrentRemoteLease = nullptr; |
|
return; |
|
} |
|
if (freshTunnel) |
|
{ |
|
LogPrint (eLogDebug, "Streaming: OutboundTunnel changed, set initial window size"); |
|
ResetWindowSize (); |
|
// m_TunnelsChangeSequenceNumber = m_SequenceNumber; // should be determined more precisely |
|
} |
|
|
|
std::vector<i2p::tunnel::TunnelMessageBlock> msgs; |
|
for (const auto& it: packets) |
|
{ |
|
auto msg = m_RoutingSession->WrapSingleMessage (m_LocalDestination.CreateDataMessage ( |
|
it->GetBuffer (), it->GetLength (), m_Port, !m_RoutingSession->IsRatchets (), it->IsSYN ())); |
|
msgs.push_back (i2p::tunnel::TunnelMessageBlock |
|
{ |
|
i2p::tunnel::eDeliveryTypeTunnel, |
|
m_CurrentRemoteLease->tunnelGateway, m_CurrentRemoteLease->tunnelID, |
|
msg |
|
}); |
|
m_NumSentBytes += it->GetLength (); |
|
if (m_IsRemoteLeaseChangeInProgress && !m_RemoteLeaseChangeTime) |
|
{ |
|
m_RemoteLeaseChangeTime = ts; |
|
m_CurrentRemoteLease = currentRemoteLease; // change it back before new lease is confirmed |
|
} |
|
} |
|
m_CurrentOutboundTunnel->SendTunnelDataMsgs (msgs); |
|
} |
|
else |
|
{ |
|
LogPrint (eLogWarning, "Streaming: Remote lease is not available, sSID=", m_SendStreamID); |
|
if (m_RoutingSession) |
|
m_RoutingSession->SetSharedRoutingPath (nullptr); // invalidate routing path |
|
} |
|
} |
|
|
|
void Stream::SendUpdatedLeaseSet () |
|
{ |
|
if (m_RoutingSession && !m_RoutingSession->IsTerminated ()) |
|
{ |
|
if (m_RoutingSession->IsLeaseSetNonConfirmed ()) |
|
{ |
|
auto ts = i2p::util::GetMillisecondsSinceEpoch (); |
|
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::LEASESET_CONFIRMATION_TIMEOUT, " milliseconds. Trying to resubmit"); |
|
m_RoutingSession->SetSharedRoutingPath (nullptr); |
|
m_CurrentOutboundTunnel = nullptr; |
|
m_CurrentRemoteLease = nullptr; |
|
SendQuickAck (); |
|
} |
|
} |
|
else if (m_RoutingSession->IsLeaseSetUpdated ()) |
|
{ |
|
LogPrint (eLogDebug, "Streaming: sending updated LeaseSet"); |
|
SendQuickAck (); |
|
} |
|
} |
|
else |
|
SendQuickAck (); |
|
} |
|
|
|
void Stream::ScheduleSend () |
|
{ |
|
if (m_Status != eStreamStatusTerminated) |
|
{ |
|
m_SendTimer.cancel (); |
|
m_SendTimer.expires_from_now (boost::posix_time::microseconds( |
|
SEND_INTERVAL + m_LocalDestination.GetRandom () % SEND_INTERVAL_VARIANCE)); |
|
m_SendTimer.async_wait (std::bind (&Stream::HandleSendTimer, |
|
shared_from_this (), std::placeholders::_1)); |
|
} |
|
} |
|
|
|
void Stream::HandleSendTimer (const boost::system::error_code& ecode) |
|
{ |
|
if (ecode != boost::asio::error::operation_aborted) |
|
{ |
|
auto ts = i2p::util::GetMillisecondsSinceEpoch (); |
|
if (m_LastSendTime && ts*1000 > m_LastSendTime*1000 + m_PacingTime) |
|
{ |
|
if (m_PacingTime) |
|
{ |
|
auto numPackets = std::lldiv (m_PacingTimeRem + ts*1000 - m_LastSendTime*1000, m_PacingTime); |
|
m_NumPacketsToSend = numPackets.quot; |
|
m_PacingTimeRem = numPackets.rem; |
|
} |
|
else |
|
{ |
|
LogPrint (eLogError, "Streaming: pacing time is zero"); |
|
m_NumPacketsToSend = 1; m_PacingTimeRem = 0; |
|
} |
|
m_IsSendTime = true; |
|
if (m_WindowIncCounter && (m_WindowSize < MAX_WINDOW_SIZE || m_WindowDropTargetSize) && !m_SendBuffer.IsEmpty () && m_PacingTime > m_MinPacingTime) |
|
{ |
|
for (int i = 0; i < m_NumPacketsToSend; i++) |
|
{ |
|
if (m_WindowIncCounter) |
|
{ |
|
if (m_WindowDropTargetSize) |
|
{ |
|
if (m_LastWindowDropSize && (m_LastWindowDropSize >= m_WindowDropTargetSize)) |
|
m_WindowDropTargetSize += 1 - (1 / ((m_LastWindowDropSize + PREV_SPEED_KEEP_TIME_COEFF) / m_WindowDropTargetSize)); // some magic here |
|
else if (m_LastWindowDropSize && (m_LastWindowDropSize < m_WindowSize)) |
|
m_WindowDropTargetSize += (m_WindowDropTargetSize - (m_LastWindowDropSize - PREV_SPEED_KEEP_TIME_COEFF)) / m_WindowDropTargetSize; // some magic here |
|
else |
|
m_WindowDropTargetSize += (m_WindowDropTargetSize - (1 - PREV_SPEED_KEEP_TIME_COEFF)) / m_WindowDropTargetSize; |
|
if (m_WindowDropTargetSize > MAX_WINDOW_SIZE) m_WindowDropTargetSize = MAX_WINDOW_SIZE; |
|
m_WindowIncCounter--; |
|
} |
|
else |
|
{ |
|
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--; |
|
} |
|
} |
|
else |
|
break; |
|
} |
|
UpdatePacingTime (); |
|
} |
|
else if (m_WindowIncCounter && m_WindowSize == MAX_WINDOW_SIZE && !m_SendBuffer.IsEmpty () && m_PacingTime > m_MinPacingTime) |
|
{ |
|
m_WindowSizeTail = m_WindowSizeTail + m_WindowIncCounter; |
|
if (m_WindowSizeTail > MAX_WINDOW_SIZE) m_WindowSizeTail = MAX_WINDOW_SIZE; |
|
} |
|
if (m_IsNAcked || m_IsResendNeeded || m_IsClientChoked) // resend packets |
|
ResendPacket (); |
|
else if (m_WindowSize > int(m_SentPackets.size ())) // send packets |
|
SendBuffer (); |
|
} |
|
else // pass |
|
ScheduleSend (); |
|
} |
|
} |
|
|
|
void Stream::ScheduleResend () |
|
{ |
|
if (m_Status != eStreamStatusTerminated) |
|
{ |
|
m_ResendTimer.cancel (); |
|
// check for invalid value |
|
if (m_RTO <= 0) m_RTO = INITIAL_RTO; |
|
m_ResendTimer.expires_from_now (boost::posix_time::milliseconds(m_RTO)); |
|
m_ResendTimer.async_wait (std::bind (&Stream::HandleResendTimer, |
|
shared_from_this (), std::placeholders::_1)); |
|
} |
|
} |
|
|
|
void Stream::HandleResendTimer (const boost::system::error_code& ecode) |
|
{ |
|
if (ecode != boost::asio::error::operation_aborted) |
|
{ |
|
m_IsSendTime = true; |
|
if (m_RTO > INITIAL_RTO) m_RTO = INITIAL_RTO; |
|
m_SendTimer.cancel (); // if no ack's in RTO, disable fast retransmit |
|
m_IsTimeOutResend = true; |
|
m_IsNAcked = false; |
|
m_IsResendNeeded = false; |
|
m_NumPacketsToSend = 1; |
|
ResendPacket (); // send one packet per RTO, waiting for ack |
|
} |
|
} |
|
|
|
void Stream::ResendPacket () |
|
{ |
|
// check for resend attempts |
|
if (m_IsIncoming && m_SequenceNumber == 1 && m_NumResendAttempts > 0) |
|
{ |
|
LogPrint (eLogWarning, "Streaming: SYNACK packet was not ACKed after ", m_NumResendAttempts, " attempts, terminate, rSID=", m_RecvStreamID, ", sSID=", m_SendStreamID); |
|
m_Status = eStreamStatusReset; |
|
Close (); |
|
return; |
|
} |
|
if (m_NumResendAttempts >= MAX_NUM_RESEND_ATTEMPTS) |
|
{ |
|
LogPrint (eLogWarning, "Streaming: packet was not ACKed after ", MAX_NUM_RESEND_ATTEMPTS, " attempts, terminate, rSID=", m_RecvStreamID, ", sSID=", m_SendStreamID); |
|
m_Status = eStreamStatusReset; |
|
Close (); |
|
return; |
|
} |
|
|
|
// collect packets to resend |
|
auto ts = i2p::util::GetMillisecondsSinceEpoch (); |
|
std::vector<Packet *> packets; |
|
if (m_IsNAcked) |
|
{ |
|
for (auto it : m_NACKedPackets) |
|
{ |
|
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) |
|
{ |
|
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; |
|
} |
|
} |
|
} |
|
|
|
// select tunnels if necessary and send |
|
if (packets.size () > 0 && m_IsSendTime) |
|
{ |
|
if (m_IsNAcked) m_NumResendAttempts = 1; |
|
else if (m_IsTimeOutResend) m_NumResendAttempts++; |
|
if (m_NumResendAttempts == 1 && m_RTO != INITIAL_RTO) |
|
{ |
|
// loss-based CC |
|
if (!m_IsWinDropped && LOSS_BASED_CONTROL_ENABLED && !m_IsClientChoked) |
|
{ |
|
LogPrint (eLogDebug, "Streaming: Packet loss, reduce window size"); |
|
ProcessWindowDrop (); |
|
} |
|
} |
|
else if (m_IsTimeOutResend) |
|
{ |
|
m_IsTimeOutResend = false; |
|
m_RTO = INITIAL_RTO; // drop RTO to initial upon tunnels pair change |
|
m_WindowDropTargetSize = INITIAL_WINDOW_SIZE; |
|
m_LastWindowDropSize = 0; |
|
m_WindowIncCounter = 0; |
|
m_IsWinDropped = true; |
|
m_IsFirstRttSample = true; |
|
m_DropWindowDelaySequenceNumber = 0; |
|
m_IsFirstACK = true; |
|
m_LastACKRecieveTime = 0; |
|
m_ACKRecieveInterval = m_AckDelay; |
|
UpdatePacingTime (); |
|
if (m_RoutingSession) m_RoutingSession->SetSharedRoutingPath (nullptr); |
|
if (m_NumResendAttempts & 1) |
|
{ |
|
// pick another outbound tunnel |
|
m_CurrentOutboundTunnel = m_LocalDestination.GetOwner ()->GetTunnelPool ()->GetNextOutboundTunnel (m_CurrentOutboundTunnel); |
|
LogPrint (eLogWarning, "Streaming: Resend #", m_NumResendAttempts, |
|
", another outbound tunnel has been selected for stream with sSID=", m_SendStreamID); |
|
} |
|
else |
|
{ |
|
CancelRemoteLeaseChange (); |
|
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); |
|
m_LastSendTime = ts; |
|
m_IsSendTime = false; |
|
if (m_IsNAcked || m_IsResendNeeded || m_IsClientChoked) ScheduleSend (); |
|
} |
|
else if (!m_IsClientChoked) |
|
SendBuffer (); |
|
if (!m_IsNAcked && !m_IsResendNeeded) ScheduleResend (); |
|
if (m_IsClientChoked) ScheduleSend (); |
|
} |
|
|
|
void Stream::ScheduleAck (int timeout) |
|
{ |
|
if (m_IsAckSendScheduled) |
|
m_AckSendTimer.cancel (); |
|
m_IsAckSendScheduled = true; |
|
if (timeout < MIN_SEND_ACK_TIMEOUT) timeout = MIN_SEND_ACK_TIMEOUT; |
|
m_AckSendTimer.expires_from_now (boost::posix_time::milliseconds(timeout)); |
|
m_AckSendTimer.async_wait (std::bind (&Stream::HandleAckSendTimer, |
|
shared_from_this (), std::placeholders::_1)); |
|
} |
|
|
|
void Stream::HandleAckSendTimer (const boost::system::error_code& ecode) |
|
{ |
|
if (m_IsAckSendScheduled) |
|
{ |
|
if (m_LastReceivedSequenceNumber < 0) |
|
{ |
|
LogPrint (eLogWarning, "Streaming: SYN has not been received after ", SYN_TIMEOUT, " milliseconds after follow on, terminate rSID=", m_RecvStreamID, ", sSID=", m_SendStreamID); |
|
m_Status = eStreamStatusReset; |
|
Close (); |
|
return; |
|
} |
|
if (m_Status == eStreamStatusOpen) |
|
{ |
|
if (m_RoutingSession && m_RoutingSession->IsLeaseSetNonConfirmed ()) |
|
{ |
|
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 (); |
|
} |
|
m_IsAckSendScheduled = false; |
|
} |
|
} |
|
|
|
void Stream::UpdateCurrentRemoteLease (bool expired) |
|
{ |
|
bool isLeaseChanged = true; |
|
if (!m_RemoteLeaseSet || m_RemoteLeaseSet->IsExpired ()) |
|
{ |
|
auto remoteLeaseSet = m_LocalDestination.GetOwner ()->FindLeaseSet (m_RemoteIdentity->GetIdentHash ()); |
|
if (!remoteLeaseSet) |
|
{ |
|
LogPrint (eLogWarning, "Streaming: LeaseSet ", m_RemoteIdentity->GetIdentHash ().ToBase64 (), m_RemoteLeaseSet ? " expired" : " not found"); |
|
if (!m_IsIncoming) // outgoing |
|
{ |
|
if (m_RemoteLeaseSet && m_RemoteLeaseSet->IsPublishedEncrypted ()) |
|
{ |
|
m_LocalDestination.GetOwner ()->RequestDestinationWithEncryptedLeaseSet ( |
|
std::make_shared<i2p::data::BlindedPublicKey>(m_RemoteIdentity)); |
|
return; // we keep m_RemoteLeaseSet for possible next request |
|
} |
|
else |
|
{ |
|
m_RemoteLeaseSet = nullptr; |
|
m_LocalDestination.GetOwner ()->RequestDestination (m_RemoteIdentity->GetIdentHash ()); // try to request for a next attempt |
|
} |
|
} |
|
else // incoming |
|
{ |
|
// just close the socket without sending FIN or RST |
|
m_Status = eStreamStatusClosed; |
|
AsyncClose (); |
|
} |
|
} |
|
else |
|
{ |
|
// LeaseSet updated |
|
m_RemoteLeaseSet = remoteLeaseSet; |
|
m_RemoteIdentity = m_RemoteLeaseSet->GetIdentity (); |
|
m_TransientVerifier = m_RemoteLeaseSet->GetTransientVerifier (); |
|
} |
|
} |
|
if (m_RemoteLeaseSet) |
|
{ |
|
if (!m_RoutingSession) |
|
m_RoutingSession = m_LocalDestination.GetOwner ()->GetRoutingSession (m_RemoteLeaseSet, true); |
|
auto leases = m_RemoteLeaseSet->GetNonExpiredLeases (false); // try without threshold first |
|
if (leases.empty ()) |
|
{ |
|
expired = false; |
|
// time to request |
|
if (m_RemoteLeaseSet->IsPublishedEncrypted ()) |
|
m_LocalDestination.GetOwner ()->RequestDestinationWithEncryptedLeaseSet ( |
|
std::make_shared<i2p::data::BlindedPublicKey>(m_RemoteIdentity)); |
|
else |
|
m_LocalDestination.GetOwner ()->RequestDestination (m_RemoteIdentity->GetIdentHash ()); |
|
leases = m_RemoteLeaseSet->GetNonExpiredLeases (true); // then with threshold |
|
} |
|
if (!leases.empty ()) |
|
{ |
|
bool updated = false; |
|
if (expired && m_CurrentRemoteLease) |
|
{ |
|
for (const auto& it: leases) |
|
if ((it->tunnelGateway == m_CurrentRemoteLease->tunnelGateway) && (it->tunnelID != m_CurrentRemoteLease->tunnelID)) |
|
{ |
|
m_CurrentRemoteLease = it; |
|
updated = true; |
|
break; |
|
} |
|
} |
|
if (!updated) |
|
{ |
|
uint32_t i = m_LocalDestination.GetRandom () % leases.size (); |
|
if (m_CurrentRemoteLease && leases[i]->tunnelID == m_CurrentRemoteLease->tunnelID) |
|
{ |
|
// make sure we don't select previous |
|
if (leases.size () > 1) |
|
i = (i + 1) % leases.size (); // if so, pick next |
|
else |
|
isLeaseChanged = false; |
|
} |
|
m_CurrentRemoteLease = leases[i]; |
|
} |
|
} |
|
else |
|
{ |
|
LogPrint (eLogWarning, "Streaming: All remote leases are expired"); |
|
m_RemoteLeaseSet = nullptr; |
|
m_CurrentRemoteLease = nullptr; |
|
// we have requested expired before, no need to do it twice |
|
} |
|
} |
|
else |
|
{ |
|
LogPrint (eLogWarning, "Streaming: Remote LeaseSet not found"); |
|
m_CurrentRemoteLease = nullptr; |
|
} |
|
if (isLeaseChanged && !m_IsRemoteLeaseChangeInProgress) |
|
{ |
|
LogPrint (eLogDebug, "Streaming: RemoteLease changed, set initial window size"); |
|
ResetWindowSize (); |
|
} |
|
} |
|
|
|
void Stream::ResetRoutingPath () |
|
{ |
|
m_CurrentOutboundTunnel = nullptr; |
|
m_CurrentRemoteLease = nullptr; |
|
m_RTT = INITIAL_RTT; |
|
m_RTO = INITIAL_RTO; |
|
if (m_RoutingSession) |
|
m_RoutingSession->SetSharedRoutingPath (nullptr); // TODO: count failures |
|
} |
|
|
|
void Stream::UpdatePacingTime () |
|
{ |
|
if (m_WindowDropTargetSize) |
|
m_PacingTime = std::round (m_RTT*1000/m_WindowDropTargetSize); |
|
else |
|
m_PacingTime = std::round (m_RTT*1000/m_WindowSize); |
|
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 + m_WindowSizeTail) / 2; |
|
if (m_LastWindowDropSize > MAX_WINDOW_SIZE) m_LastWindowDropSize = MAX_WINDOW_SIZE; |
|
} |
|
else |
|
m_LastWindowDropSize = m_WindowSize; |
|
m_WindowDropTargetSize = m_LastWindowDropSize - (m_LastWindowDropSize / 4); // -25%; |
|
if (m_WindowDropTargetSize < MIN_WINDOW_SIZE) |
|
m_WindowDropTargetSize = MIN_WINDOW_SIZE; |
|
m_WindowIncCounter = 0; // disable window growth |
|
m_DropWindowDelaySequenceNumber = m_SequenceNumber + int(m_WindowDropTargetSize); |
|
m_IsFirstACK = true; // ignore first RTT sample |
|
m_IsWinDropped = true; // don't drop window twice |
|
m_WindowSizeTail = 0; |
|
UpdatePacingTime (); |
|
} |
|
|
|
void Stream::ResetWindowSize () |
|
{ |
|
m_RTO = INITIAL_RTO; |
|
if (!m_IsClientChoked) |
|
{ |
|
if (m_WindowSize > INITIAL_WINDOW_SIZE) |
|
{ |
|
m_WindowDropTargetSize = (float)INITIAL_WINDOW_SIZE; |
|
m_IsWinDropped = true; |
|
} |
|
else |
|
m_WindowSize = INITIAL_WINDOW_SIZE; |
|
} |
|
m_LastWindowDropSize = 0; |
|
m_WindowIncCounter = 0; |
|
m_IsFirstRttSample = true; |
|
m_IsFirstACK = true; |
|
m_WindowSizeTail = 0; |
|
UpdatePacingTime (); |
|
} |
|
|
|
void Stream::CancelRemoteLeaseChange () |
|
{ |
|
m_RemoteLeaseChangeTime = 0; |
|
m_IsRemoteLeaseChangeInProgress = false; |
|
} |
|
|
|
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 ()), |
|
m_LastCleanupTime (i2p::util::GetSecondsSinceEpoch ()) |
|
{ |
|
} |
|
|
|
StreamingDestination::~StreamingDestination () |
|
{ |
|
for (auto& it: m_SavedPackets) |
|
{ |
|
for (auto it1: it.second) DeletePacket (it1); |
|
it.second.clear (); |
|
} |
|
m_SavedPackets.clear (); |
|
} |
|
|
|
void StreamingDestination::Start () |
|
{ |
|
} |
|
|
|
void StreamingDestination::Stop () |
|
{ |
|
ResetAcceptor (); |
|
m_PendingIncomingTimer.cancel (); |
|
m_PendingIncomingStreams.clear (); |
|
{ |
|
std::unique_lock<std::mutex> l(m_StreamsMutex); |
|
for (auto it: m_Streams) |
|
it.second->Terminate (false); // we delete here |
|
m_Streams.clear (); |
|
m_IncomingStreams.clear (); |
|
m_LastStream = nullptr; |
|
} |
|
} |
|
|
|
void StreamingDestination::HandleNextPacket (Packet * packet) |
|
{ |
|
uint32_t sendStreamID = packet->GetSendStreamID (); |
|
if (sendStreamID) |
|
{ |
|
if (!m_LastStream || sendStreamID != m_LastStream->GetRecvStreamID ()) |
|
{ |
|
auto it = m_Streams.find (sendStreamID); |
|
if (it != m_Streams.end ()) |
|
m_LastStream = it->second; |
|
else |
|
m_LastStream = nullptr; |
|
} |
|
if (m_LastStream) |
|
m_LastStream->HandleNextPacket (packet); |
|
else if (packet->IsEcho () && m_Owner->IsStreamingAnswerPings ()) |
|
{ |
|
// ping |
|
LogPrint (eLogInfo, "Streaming: Ping received sSID=", sendStreamID); |
|
auto s = std::make_shared<Stream> (m_Owner->GetService (), *this); |
|
s->HandlePing (packet); |
|
} |
|
else |
|
{ |
|
LogPrint (eLogInfo, "Streaming: Unknown stream sSID=", sendStreamID); |
|
DeletePacket (packet); |
|
} |
|
} |
|
else |
|
{ |
|
if (packet->IsEcho ()) |
|
{ |
|
// pong |
|
LogPrint (eLogInfo, "Streaming: Pong received rSID=", packet->GetReceiveStreamID ()); |
|
DeletePacket (packet); |
|
return; |
|
} |
|
if (packet->IsSYN () && !packet->GetSeqn ()) // new incoming stream |
|
{ |
|
uint32_t receiveStreamID = packet->GetReceiveStreamID (); |
|
auto it1 = m_IncomingStreams.find (receiveStreamID); |
|
if (it1 != m_IncomingStreams.end ()) |
|
{ |
|
// already pending |
|
LogPrint(eLogWarning, "Streaming: Incoming streaming with rSID=", receiveStreamID, " already exists"); |
|
it1->second->ResetRoutingPath (); // Ack was not delivered, changing path |
|
DeletePacket (packet); // drop it, because previous should be connected |
|
return; |
|
} |
|
if (m_Owner->GetStreamingMaxConcurrentStreams () > 0 && (int)m_Streams.size () > m_Owner->GetStreamingMaxConcurrentStreams ()) |
|
{ |
|
LogPrint(eLogWarning, "Streaming: Number of streams exceeds ", m_Owner->GetStreamingMaxConcurrentStreams ()); |
|
DeletePacket (packet); |
|
return; |
|
} |
|
auto incomingStream = CreateNewIncomingStream (receiveStreamID); |
|
incomingStream->HandleNextPacket (packet); // SYN |
|
if (!incomingStream->GetRemoteLeaseSet ()) |
|
{ |
|
LogPrint (eLogWarning, "Streaming: No remote LeaseSet for incoming stream. Terminated"); |
|
incomingStream->Terminate (); // can't send FIN anyway |
|
return; |
|
} |
|
|
|
// handle saved packets if any |
|
{ |
|
auto it = m_SavedPackets.find (receiveStreamID); |
|
if (it != m_SavedPackets.end ()) |
|
{ |
|
LogPrint (eLogDebug, "Streaming: Processing ", it->second.size (), " saved packets for rSID=", receiveStreamID); |
|
for (auto it1: it->second) |
|
incomingStream->HandleNextPacket (it1); |
|
m_SavedPackets.erase (it); |
|
} |
|
} |
|
// accept |
|
if (m_Acceptor != nullptr) |
|
m_Acceptor (incomingStream); |
|
else |
|
{ |
|
LogPrint (eLogWarning, "Streaming: Acceptor for incoming stream is not set"); |
|
if (m_PendingIncomingStreams.size () < MAX_PENDING_INCOMING_BACKLOG) |
|
{ |
|
m_PendingIncomingStreams.push_back (incomingStream); |
|
m_PendingIncomingTimer.cancel (); |
|
m_PendingIncomingTimer.expires_from_now (boost::posix_time::seconds(PENDING_INCOMING_TIMEOUT)); |
|
m_PendingIncomingTimer.async_wait (std::bind (&StreamingDestination::HandlePendingIncomingTimer, |
|
shared_from_this (), std::placeholders::_1)); |
|
LogPrint (eLogDebug, "Streaming: Pending incoming stream added, rSID=", receiveStreamID); |
|
} |
|
else |
|
{ |
|
LogPrint (eLogWarning, "Streaming: Pending incoming streams backlog exceeds ", MAX_PENDING_INCOMING_BACKLOG); |
|
incomingStream->Close (); |
|
} |
|
} |
|
} |
|
else // follow on packet without SYN |
|
{ |
|
uint32_t receiveStreamID = packet->GetReceiveStreamID (); |
|
auto it1 = m_IncomingStreams.find (receiveStreamID); |
|
if (it1 != m_IncomingStreams.end ()) |
|
{ |
|
// found |
|
it1->second->HandleNextPacket (packet); |
|
return; |
|
} |
|
// save follow on packet |
|
auto it = m_SavedPackets.find (receiveStreamID); |
|
if (it != m_SavedPackets.end ()) |
|
it->second.push_back (packet); |
|
else |
|
{ |
|
m_SavedPackets[receiveStreamID] = std::list<Packet *>{ packet }; |
|
auto timer = std::make_shared<boost::asio::deadline_timer> (m_Owner->GetService ()); |
|
timer->expires_from_now (boost::posix_time::seconds(PENDING_INCOMING_TIMEOUT)); |
|
auto s = shared_from_this (); |
|
timer->async_wait ([s,timer,receiveStreamID](const boost::system::error_code& ecode) |
|
{ |
|
if (ecode != boost::asio::error::operation_aborted) |
|
{ |
|
auto it = s->m_SavedPackets.find (receiveStreamID); |
|
if (it != s->m_SavedPackets.end ()) |
|
{ |
|
for (auto it1: it->second) s->DeletePacket (it1); |
|
it->second.clear (); |
|
s->m_SavedPackets.erase (it); |
|
} |
|
} |
|
}); |
|
} |
|
} |
|
} |
|
} |
|
|
|
std::shared_ptr<Stream> StreamingDestination::CreateNewOutgoingStream (std::shared_ptr<const i2p::data::LeaseSet> remote, int port) |
|
{ |
|
auto s = std::make_shared<Stream> (m_Owner->GetService (), *this, remote, port); |
|
std::unique_lock<std::mutex> l(m_StreamsMutex); |
|
m_Streams.emplace (s->GetRecvStreamID (), s); |
|
return s; |
|
} |
|
|
|
void StreamingDestination::SendPing (std::shared_ptr<const i2p::data::LeaseSet> remote) |
|
{ |
|
auto s = std::make_shared<Stream> (m_Owner->GetService (), *this, remote, 0); |
|
s->SendPing (); |
|
} |
|
|
|
std::shared_ptr<Stream> StreamingDestination::CreateNewIncomingStream (uint32_t receiveStreamID) |
|
{ |
|
auto s = std::make_shared<Stream> (m_Owner->GetService (), *this); |
|
std::unique_lock<std::mutex> l(m_StreamsMutex); |
|
m_Streams.emplace (s->GetRecvStreamID (), s); |
|
m_IncomingStreams.emplace (receiveStreamID, s); |
|
return s; |
|
} |
|
|
|
void StreamingDestination::DeleteStream (std::shared_ptr<Stream> stream) |
|
{ |
|
if (stream) |
|
{ |
|
std::unique_lock<std::mutex> l(m_StreamsMutex); |
|
m_Streams.erase (stream->GetRecvStreamID ()); |
|
if (stream->IsIncoming ()) |
|
m_IncomingStreams.erase (stream->GetSendStreamID ()); |
|
if (m_LastStream == stream) m_LastStream = nullptr; |
|
} |
|
auto ts = i2p::util::GetSecondsSinceEpoch (); |
|
if (m_Streams.empty () || ts > m_LastCleanupTime + STREAMING_DESTINATION_POOLS_CLEANUP_INTERVAL) |
|
{ |
|
m_PacketsPool.CleanUp (); |
|
m_I2NPMsgsPool.CleanUp (); |
|
m_LastCleanupTime = ts; |
|
} |
|
} |
|
|
|
bool StreamingDestination::DeleteStream (uint32_t recvStreamID) |
|
{ |
|
auto it = m_Streams.find (recvStreamID); |
|
if (it == m_Streams.end ()) |
|
return false; |
|
auto s = it->second; |
|
boost::asio::post (m_Owner->GetService (), [this, s] () |
|
{ |
|
s->Close (); // try to send FIN |
|
s->Terminate (false); |
|
DeleteStream (s); |
|
}); |
|
return true; |
|
} |
|
|
|
void StreamingDestination::SetAcceptor (const Acceptor& acceptor) |
|
{ |
|
m_Acceptor = acceptor; // we must set it immediately for IsAcceptorSet |
|
auto s = shared_from_this (); |
|
boost::asio::post (m_Owner->GetService (), [s](void) |
|
{ |
|
// take care about incoming queue |
|
for (auto& it: s->m_PendingIncomingStreams) |
|
if (it->GetStatus () == eStreamStatusOpen) // still open? |
|
s->m_Acceptor (it); |
|
s->m_PendingIncomingStreams.clear (); |
|
s->m_PendingIncomingTimer.cancel (); |
|
}); |
|
} |
|
|
|
void StreamingDestination::ResetAcceptor () |
|
{ |
|
if (m_Acceptor) m_Acceptor (nullptr); |
|
m_Acceptor = nullptr; |
|
} |
|
|
|
void StreamingDestination::AcceptOnce (const Acceptor& acceptor) |
|
{ |
|
boost::asio::post (m_Owner->GetService (), [acceptor, this](void) |
|
{ |
|
if (!m_PendingIncomingStreams.empty ()) |
|
{ |
|
acceptor (m_PendingIncomingStreams.front ()); |
|
m_PendingIncomingStreams.pop_front (); |
|
if (m_PendingIncomingStreams.empty ()) |
|
m_PendingIncomingTimer.cancel (); |
|
} |
|
else // we must save old acceptor and set it back |
|
{ |
|
m_Acceptor = std::bind (&StreamingDestination::AcceptOnceAcceptor, this, |
|
std::placeholders::_1, acceptor, m_Acceptor); |
|
} |
|
}); |
|
} |
|
|
|
void StreamingDestination::AcceptOnceAcceptor (std::shared_ptr<Stream> stream, Acceptor acceptor, Acceptor prev) |
|
{ |
|
m_Acceptor = prev; |
|
acceptor (stream); |
|
} |
|
|
|
std::shared_ptr<Stream> StreamingDestination::AcceptStream (int timeout) |
|
{ |
|
std::shared_ptr<i2p::stream::Stream> stream; |
|
std::condition_variable streamAccept; |
|
std::mutex streamAcceptMutex; |
|
std::unique_lock<std::mutex> l(streamAcceptMutex); |
|
AcceptOnce ( |
|
[&streamAccept, &streamAcceptMutex, &stream](std::shared_ptr<i2p::stream::Stream> s) |
|
{ |
|
stream = s; |
|
std::unique_lock<std::mutex> l(streamAcceptMutex); |
|
streamAccept.notify_all (); |
|
}); |
|
if (timeout) |
|
streamAccept.wait_for (l, std::chrono::seconds (timeout)); |
|
else |
|
streamAccept.wait (l); |
|
return stream; |
|
} |
|
|
|
void StreamingDestination::HandlePendingIncomingTimer (const boost::system::error_code& ecode) |
|
{ |
|
if (ecode != boost::asio::error::operation_aborted) |
|
{ |
|
LogPrint (eLogWarning, "Streaming: Pending incoming timeout expired"); |
|
for (auto& it: m_PendingIncomingStreams) |
|
it->Close (); |
|
m_PendingIncomingStreams.clear (); |
|
} |
|
} |
|
|
|
void StreamingDestination::HandleDataMessagePayload (const uint8_t * buf, size_t len) |
|
{ |
|
// unzip it |
|
Packet * uncompressed = NewPacket (); |
|
uncompressed->offset = 0; |
|
uncompressed->len = m_Inflator.Inflate (buf, len, uncompressed->buf, MAX_PACKET_SIZE); |
|
if (uncompressed->len) |
|
HandleNextPacket (uncompressed); |
|
else |
|
DeletePacket (uncompressed); |
|
} |
|
|
|
std::shared_ptr<I2NPMessage> StreamingDestination::CreateDataMessage ( |
|
const uint8_t * payload, size_t len, uint16_t toPort, bool checksum, bool gzip) |
|
{ |
|
size_t size; |
|
auto msg = (len <= STREAMING_MTU_RATCHETS) ? m_I2NPMsgsPool.AcquireShared () : NewI2NPMessage (); |
|
uint8_t * buf = msg->GetPayload (); |
|
buf += 4; // reserve for lengthlength |
|
msg->len += 4; |
|
|
|
if (m_Gzip || gzip) |
|
size = m_Deflator.Deflate (payload, len, buf, msg->maxLen - msg->len); |
|
else |
|
size = i2p::data::GzipNoCompression (payload, len, buf, msg->maxLen - msg->len); |
|
|
|
if (size) |
|
{ |
|
htobe32buf (msg->GetPayload (), size); // length |
|
htobe16buf (buf + 4, m_LocalPort); // source port |
|
htobe16buf (buf + 6, toPort); // destination port |
|
buf[9] = i2p::client::PROTOCOL_TYPE_STREAMING; // streaming protocol |
|
msg->len += size; |
|
msg->FillI2NPMessageHeader (eI2NPData, 0, checksum); |
|
} |
|
else |
|
msg = nullptr; |
|
return msg; |
|
} |
|
|
|
uint32_t StreamingDestination::GetRandom () |
|
{ |
|
if (m_Owner) |
|
{ |
|
auto pool = m_Owner->GetTunnelPool (); |
|
if (pool) |
|
return pool->GetRng ()(); |
|
} |
|
return rand (); |
|
} |
|
} |
|
}
|
|
|