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.
1213 lines
37 KiB
1213 lines
37 KiB
9 years ago
|
#include "Crypto.h"
|
||
11 years ago
|
#include "Log.h"
|
||
|
#include "RouterInfo.h"
|
||
11 years ago
|
#include "RouterContext.h"
|
||
11 years ago
|
#include "Tunnel.h"
|
||
|
#include "Timestamp.h"
|
||
10 years ago
|
#include "Destination.h"
|
||
11 years ago
|
#include "Streaming.h"
|
||
|
|
||
|
namespace i2p
|
||
|
{
|
||
|
namespace stream
|
||
|
{
|
||
8 years ago
|
void SendBufferQueue::Add (const uint8_t * buf, size_t len, SendHandler handler)
|
||
|
{
|
||
|
m_Buffers.push_back (std::make_shared<SendBuffer>(buf, len, handler));
|
||
|
m_Size += 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 (), len - offset);
|
||
|
nextBuffer->offset += (len - offset);
|
||
|
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;
|
||
|
}
|
||
|
}
|
||
|
|
||
8 years ago
|
Stream::Stream (boost::asio::io_service& service, StreamingDestination& local,
|
||
10 years ago
|
std::shared_ptr<const i2p::data::LeaseSet> remote, int port): m_Service (service),
|
||
8 years ago
|
m_SendStreamID (0), m_SequenceNumber (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),
|
||
10 years ago
|
m_WindowSize (MIN_WINDOW_SIZE), m_RTT (INITIAL_RTT), m_RTO (INITIAL_RTO),
|
||
10 years ago
|
m_LastWindowSizeIncreaseTime (0), m_NumResendAttempts (0)
|
||
11 years ago
|
{
|
||
9 years ago
|
RAND_bytes ((uint8_t *)&m_RecvStreamID, 4);
|
||
10 years ago
|
m_RemoteIdentity = remote->GetIdentity ();
|
||
8 years ago
|
}
|
||
11 years ago
|
|
||
10 years ago
|
Stream::Stream (boost::asio::io_service& service, StreamingDestination& local):
|
||
8 years ago
|
m_Service (service), m_SendStreamID (0), m_SequenceNumber (0), m_LastReceivedSequenceNumber (-1),
|
||
10 years ago
|
m_Status (eStreamStatusNew), m_IsAckSendScheduled (false), m_LocalDestination (local),
|
||
8 years ago
|
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),
|
||
10 years ago
|
m_RTT (INITIAL_RTT), m_RTO (INITIAL_RTO), m_LastWindowSizeIncreaseTime (0), m_NumResendAttempts (0)
|
||
10 years ago
|
{
|
||
9 years ago
|
RAND_bytes ((uint8_t *)&m_RecvStreamID, 4);
|
||
10 years ago
|
}
|
||
|
|
||
11 years ago
|
Stream::~Stream ()
|
||
8 years ago
|
{
|
||
|
CleanUp ();
|
||
9 years ago
|
LogPrint (eLogDebug, "Streaming: Stream deleted");
|
||
8 years ago
|
}
|
||
10 years ago
|
|
||
|
void Stream::Terminate ()
|
||
|
{
|
||
|
m_AckSendTimer.cancel ();
|
||
|
m_ReceiveTimer.cancel ();
|
||
|
m_ResendTimer.cancel ();
|
||
8 years ago
|
//CleanUp (); /* Need to recheck - broke working on windows */
|
||
|
m_LocalDestination.DeleteStream (shared_from_this ());
|
||
|
}
|
||
8 years ago
|
|
||
|
void Stream::CleanUp ()
|
||
|
{
|
||
8 years ago
|
{
|
||
|
std::unique_lock<std::mutex> l(m_SendBufferMutex);
|
||
|
m_SendBuffer.CleanUp ();
|
||
|
}
|
||
8 years ago
|
while (!m_ReceiveQueue.empty ())
|
||
|
{
|
||
|
auto packet = m_ReceiveQueue.front ();
|
||
|
m_ReceiveQueue.pop ();
|
||
|
m_LocalDestination.DeletePacket (packet);
|
||
|
}
|
||
8 years ago
|
|
||
8 years ago
|
for (auto it: m_SentPackets)
|
||
|
m_LocalDestination.DeletePacket (it);
|
||
|
m_SentPackets.clear ();
|
||
8 years ago
|
|
||
8 years ago
|
for (auto it: m_SavedPackets)
|
||
|
m_LocalDestination.DeletePacket (it);
|
||
|
m_SavedPackets.clear ();
|
||
8 years ago
|
}
|
||
|
|
||
11 years ago
|
void Stream::HandleNextPacket (Packet * packet)
|
||
|
{
|
||
10 years ago
|
m_NumReceivedBytes += packet->GetLength ();
|
||
8 years ago
|
if (!m_SendStreamID)
|
||
|
m_SendStreamID = packet->GetReceiveStreamID ();
|
||
10 years ago
|
|
||
10 years ago
|
if (!packet->IsNoAck ()) // ack received
|
||
|
ProcessAck (packet);
|
||
8 years ago
|
|
||
10 years ago
|
int32_t receivedSeqn = packet->GetSeqn ();
|
||
10 years ago
|
bool isSyn = packet->IsSYN ();
|
||
|
if (!receivedSeqn && !isSyn)
|
||
10 years ago
|
{
|
||
|
// plain ack
|
||
9 years ago
|
LogPrint (eLogDebug, "Streaming: Plain ACK received");
|
||
8 years ago
|
m_LocalDestination.DeletePacket (packet);
|
||
10 years ago
|
return;
|
||
|
}
|
||
|
|
||
9 years ago
|
LogPrint (eLogDebug, "Streaming: Received seqn=", receivedSeqn, " on sSID=", m_SendStreamID);
|
||
9 years ago
|
if (receivedSeqn == m_LastReceivedSequenceNumber + 1)
|
||
8 years ago
|
{
|
||
11 years ago
|
// we have received next in sequence message
|
||
|
ProcessPacket (packet);
|
||
8 years ago
|
|
||
11 years ago
|
// we should also try stored messages if any
|
||
|
for (auto it = m_SavedPackets.begin (); it != m_SavedPackets.end ();)
|
||
8 years ago
|
{
|
||
10 years ago
|
if ((*it)->GetSeqn () == (uint32_t)(m_LastReceivedSequenceNumber + 1))
|
||
11 years ago
|
{
|
||
11 years ago
|
Packet * savedPacket = *it;
|
||
11 years ago
|
m_SavedPackets.erase (it++);
|
||
|
|
||
11 years ago
|
ProcessPacket (savedPacket);
|
||
11 years ago
|
}
|
||
|
else
|
||
|
break;
|
||
11 years ago
|
}
|
||
|
|
||
10 years ago
|
// schedule ack for last message
|
||
10 years ago
|
if (m_Status == eStreamStatusOpen)
|
||
10 years ago
|
{
|
||
|
if (!m_IsAckSendScheduled)
|
||
|
{
|
||
|
m_IsAckSendScheduled = true;
|
||
9 years ago
|
auto ackTimeout = m_RTT/10;
|
||
|
if (ackTimeout > ACK_SEND_TIMEOUT) ackTimeout = ACK_SEND_TIMEOUT;
|
||
|
m_AckSendTimer.expires_from_now (boost::posix_time::milliseconds(ackTimeout));
|
||
10 years ago
|
m_AckSendTimer.async_wait (std::bind (&Stream::HandleAckSendTimer,
|
||
|
shared_from_this (), std::placeholders::_1));
|
||
10 years ago
|
}
|
||
8 years ago
|
}
|
||
10 years ago
|
else if (isSyn)
|
||
|
// we have to send SYN back to incoming connection
|
||
8 years ago
|
SendBuffer (); // also sets m_IsOpen
|
||
|
}
|
||
|
else
|
||
|
{
|
||
11 years ago
|
if (receivedSeqn <= m_LastReceivedSequenceNumber)
|
||
|
{
|
||
10 years ago
|
// we have received duplicate
|
||
9 years ago
|
LogPrint (eLogWarning, "Streaming: Duplicate message ", receivedSeqn, " on sSID=", m_SendStreamID);
|
||
11 years ago
|
SendQuickAck (); // resend ack for previous message again
|
||
8 years ago
|
m_LocalDestination.DeletePacket (packet); // packet dropped
|
||
8 years ago
|
}
|
||
11 years ago
|
else
|
||
|
{
|
||
9 years ago
|
LogPrint (eLogWarning, "Streaming: Missing messages on sSID=", m_SendStreamID, ": from ", m_LastReceivedSequenceNumber + 1, " to ", receivedSeqn - 1);
|
||
11 years ago
|
// save message and wait for missing message again
|
||
|
SavePacket (packet);
|
||
10 years ago
|
if (m_LastReceivedSequenceNumber >= 0)
|
||
8 years ago
|
{
|
||
10 years ago
|
// send NACKs for missing messages ASAP
|
||
|
if (m_IsAckSendScheduled)
|
||
|
{
|
||
8 years ago
|
m_IsAckSendScheduled = false;
|
||
10 years ago
|
m_AckSendTimer.cancel ();
|
||
|
}
|
||
|
SendQuickAck ();
|
||
8 years ago
|
}
|
||
10 years ago
|
else
|
||
10 years ago
|
{
|
||
10 years ago
|
// wait for SYN
|
||
|
m_IsAckSendScheduled = true;
|
||
|
m_AckSendTimer.expires_from_now (boost::posix_time::milliseconds(ACK_SEND_TIMEOUT));
|
||
|
m_AckSendTimer.async_wait (std::bind (&Stream::HandleAckSendTimer,
|
||
|
shared_from_this (), std::placeholders::_1));
|
||
8 years ago
|
}
|
||
|
}
|
||
|
}
|
||
|
}
|
||
11 years ago
|
|
||
|
void Stream::SavePacket (Packet * packet)
|
||
|
{
|
||
9 years ago
|
if (!m_SavedPackets.insert (packet).second)
|
||
8 years ago
|
m_LocalDestination.DeletePacket (packet);
|
||
8 years ago
|
}
|
||
11 years ago
|
|
||
|
void Stream::ProcessPacket (Packet * packet)
|
||
|
{
|
||
|
// process flags
|
||
|
uint32_t receivedSeqn = packet->GetSeqn ();
|
||
|
uint16_t flags = packet->GetFlags ();
|
||
9 years ago
|
LogPrint (eLogDebug, "Streaming: Process seqn=", receivedSeqn, ", flags=", flags);
|
||
8 years ago
|
|
||
11 years ago
|
const uint8_t * optionData = packet->GetOptionData ();
|
||
|
|
||
10 years ago
|
if (flags & PACKET_FLAG_DELAY_REQUESTED)
|
||
|
optionData += 2;
|
||
8 years ago
|
|
||
11 years ago
|
if (flags & PACKET_FLAG_FROM_INCLUDED)
|
||
|
{
|
||
9 years ago
|
m_RemoteIdentity = std::make_shared<i2p::data::IdentityEx>(optionData, packet->GetOptionSize ());
|
||
|
optionData += m_RemoteIdentity->GetFullLen ();
|
||
10 years ago
|
if (!m_RemoteLeaseSet)
|
||
9 years ago
|
LogPrint (eLogDebug, "Streaming: Incoming stream from ", m_RemoteIdentity->GetIdentHash ().ToBase64 (), ", sSID=", m_SendStreamID, ", rSID=", m_RecvStreamID);
|
||
8 years ago
|
}
|
||
11 years ago
|
|
||
10 years ago
|
if (flags & PACKET_FLAG_MAX_PACKET_SIZE_INCLUDED)
|
||
|
{
|
||
10 years ago
|
uint16_t maxPacketSize = bufbe16toh (optionData);
|
||
9 years ago
|
LogPrint (eLogDebug, "Streaming: Max packet size ", maxPacketSize);
|
||
10 years ago
|
optionData += 2;
|
||
8 years ago
|
}
|
||
|
|
||
10 years ago
|
if (flags & PACKET_FLAG_SIGNATURE_INCLUDED)
|
||
|
{
|
||
8 years ago
|
uint8_t signature[256];
|
||
9 years ago
|
auto signatureLen = m_RemoteIdentity->GetSignatureLen ();
|
||
10 years ago
|
memcpy (signature, optionData, signatureLen);
|
||
|
memset (const_cast<uint8_t *>(optionData), 0, signatureLen);
|
||
9 years ago
|
if (!m_RemoteIdentity->Verify (packet->GetBuffer (), packet->GetLength (), signature))
|
||
8 years ago
|
{
|
||
8 years ago
|
LogPrint (eLogError, "Streaming: Signature verification failed, sSID=", m_SendStreamID, ", rSID=", m_RecvStreamID);
|
||
9 years ago
|
Close ();
|
||
10 years ago
|
flags |= PACKET_FLAG_CLOSE;
|
||
8 years ago
|
}
|
||
10 years ago
|
memcpy (const_cast<uint8_t *>(optionData), signature, signatureLen);
|
||
|
optionData += signatureLen;
|
||
8 years ago
|
}
|
||
10 years ago
|
|
||
11 years ago
|
packet->offset = packet->GetPayload () - packet->buf;
|
||
|
if (packet->GetLength () > 0)
|
||
8 years ago
|
{
|
||
11 years ago
|
m_ReceiveQueue.push (packet);
|
||
11 years ago
|
m_ReceiveTimer.cancel ();
|
||
8 years ago
|
}
|
||
11 years ago
|
else
|
||
8 years ago
|
m_LocalDestination.DeletePacket (packet);
|
||
8 years ago
|
|
||
11 years ago
|
m_LastReceivedSequenceNumber = receivedSeqn;
|
||
|
|
||
9 years ago
|
if (flags & PACKET_FLAG_RESET)
|
||
11 years ago
|
{
|
||
8 years ago
|
LogPrint (eLogDebug, "Streaming: closing stream sSID=", m_SendStreamID, ", rSID=", m_RecvStreamID, ": reset flag received in packet #", receivedSeqn);
|
||
10 years ago
|
m_Status = eStreamStatusReset;
|
||
10 years ago
|
Close ();
|
||
11 years ago
|
}
|
||
9 years ago
|
else if (flags & PACKET_FLAG_CLOSE)
|
||
|
{
|
||
|
if (m_Status != eStreamStatusClosed)
|
||
|
SendClose ();
|
||
|
m_Status = eStreamStatusClosed;
|
||
8 years ago
|
Terminate ();
|
||
9 years ago
|
}
|
||
8 years ago
|
}
|
||
10 years ago
|
|
||
|
void Stream::ProcessAck (Packet * packet)
|
||
|
{
|
||
10 years ago
|
bool acknowledged = false;
|
||
10 years ago
|
auto ts = i2p::util::GetMillisecondsSinceEpoch ();
|
||
10 years ago
|
uint32_t ackThrough = packet->GetAckThrough ();
|
||
9 years ago
|
if (ackThrough > m_SequenceNumber)
|
||
|
{
|
||
|
LogPrint (eLogError, "Streaming: Unexpected ackThrough=", ackThrough, " > seqn=", m_SequenceNumber);
|
||
|
return;
|
||
8 years ago
|
}
|
||
10 years ago
|
int nackCount = packet->GetNACKCount ();
|
||
10 years ago
|
for (auto it = m_SentPackets.begin (); it != m_SentPackets.end ();)
|
||
8 years ago
|
{
|
||
10 years ago
|
auto seqn = (*it)->GetSeqn ();
|
||
|
if (seqn <= ackThrough)
|
||
10 years ago
|
{
|
||
10 years ago
|
if (nackCount > 0)
|
||
|
{
|
||
|
bool nacked = false;
|
||
|
for (int i = 0; i < nackCount; i++)
|
||
|
if (seqn == packet->GetNACK (i))
|
||
|
{
|
||
|
nacked = true;
|
||
|
break;
|
||
|
}
|
||
|
if (nacked)
|
||
|
{
|
||
9 years ago
|
LogPrint (eLogDebug, "Streaming: Packet ", seqn, " NACK");
|
||
8 years ago
|
++it;
|
||
10 years ago
|
continue;
|
||
8 years ago
|
}
|
||
10 years ago
|
}
|
||
10 years ago
|
auto sentPacket = *it;
|
||
8 years ago
|
uint64_t rtt = ts - sentPacket->sendTime;
|
||
|
if(ts < sentPacket->sendTime)
|
||
|
{
|
||
|
LogPrint(eLogError, "Streaming: Packet ", seqn, "sent from the future, sendTime=", sentPacket->sendTime);
|
||
|
rtt = 1;
|
||
|
}
|
||
10 years ago
|
m_RTT = (m_RTT*seqn + rtt)/(seqn + 1);
|
||
10 years ago
|
m_RTO = m_RTT*1.5; // TODO: implement it better
|
||
8 years ago
|
LogPrint (eLogDebug, "Streaming: Packet ", seqn, " acknowledged rtt=", rtt, " sentTime=", sentPacket->sendTime);
|
||
10 years ago
|
m_SentPackets.erase (it++);
|
||
8 years ago
|
m_LocalDestination.DeletePacket (sentPacket);
|
||
10 years ago
|
acknowledged = true;
|
||
10 years ago
|
if (m_WindowSize < WINDOW_SIZE)
|
||
|
m_WindowSize++; // slow start
|
||
|
else
|
||
|
{
|
||
|
// linear growth
|
||
10 years ago
|
if (ts > m_LastWindowSizeIncreaseTime + m_RTT)
|
||
10 years ago
|
{
|
||
|
m_WindowSize++;
|
||
|
if (m_WindowSize > MAX_WINDOW_SIZE) m_WindowSize = MAX_WINDOW_SIZE;
|
||
|
m_LastWindowSizeIncreaseTime = ts;
|
||
|
}
|
||
|
}
|
||
9 years ago
|
if (!seqn && m_RoutingSession) // first message confirmed
|
||
|
m_RoutingSession->SetSharedRoutingPath (
|
||
|
std::make_shared<i2p::garlic::GarlicRoutingPath> (
|
||
9 years ago
|
i2p::garlic::GarlicRoutingPath{m_CurrentOutboundTunnel, m_CurrentRemoteLease, m_RTT, 0, 0}));
|
||
10 years ago
|
}
|
||
|
else
|
||
|
break;
|
||
|
}
|
||
|
if (m_SentPackets.empty ())
|
||
|
m_ResendTimer.cancel ();
|
||
10 years ago
|
if (acknowledged)
|
||
10 years ago
|
{
|
||
|
m_NumResendAttempts = 0;
|
||
10 years ago
|
SendBuffer ();
|
||
8 years ago
|
}
|
||
9 years ago
|
if (m_Status == eStreamStatusClosed)
|
||
|
Terminate ();
|
||
|
else if (m_Status == eStreamStatusClosing)
|
||
9 years ago
|
Close (); // check is all outgoing messages have been sent and we can send close
|
||
8 years ago
|
}
|
||
|
|
||
10 years ago
|
size_t Stream::Send (const uint8_t * buf, size_t len)
|
||
11 years ago
|
{
|
||
8 years ago
|
// TODO: check max buffer size
|
||
|
AsyncSend (buf, len, nullptr);
|
||
11 years ago
|
return len;
|
||
8 years ago
|
}
|
||
11 years ago
|
|
||
10 years ago
|
void Stream::AsyncSend (const uint8_t * buf, size_t len, SendHandler handler)
|
||
|
{
|
||
8 years ago
|
if (len > 0 && buf)
|
||
|
{
|
||
|
std::unique_lock<std::mutex> l(m_SendBufferMutex);
|
||
|
m_SendBuffer.Add (buf, len, handler);
|
||
|
}
|
||
|
else if (handler)
|
||
|
handler(boost::system::error_code ());
|
||
|
m_Service.post (std::bind (&Stream::SendBuffer, shared_from_this ()));
|
||
10 years ago
|
}
|
||
|
|
||
10 years ago
|
void Stream::SendBuffer ()
|
||
8 years ago
|
{
|
||
10 years ago
|
int numMsgs = m_WindowSize - m_SentPackets.size ();
|
||
8 years ago
|
if (numMsgs <= 0) return; // window is full
|
||
|
|
||
10 years ago
|
bool isNoAck = m_LastReceivedSequenceNumber < 0; // first packet
|
||
|
std::vector<Packet *> packets;
|
||
|
{
|
||
|
std::unique_lock<std::mutex> l(m_SendBufferMutex);
|
||
8 years ago
|
while ((m_Status == eStreamStatusNew) || (IsEstablished () && !m_SendBuffer.IsEmpty () && numMsgs > 0))
|
||
10 years ago
|
{
|
||
8 years ago
|
Packet * p = m_LocalDestination.NewPacket ();
|
||
10 years ago
|
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
|
||
8 years ago
|
if (isNoAck)
|
||
10 years ago
|
htobuf32 (packet + size, 0);
|
||
9 years ago
|
else
|
||
|
htobe32buf (packet + size, m_LastReceivedSequenceNumber);
|
||
10 years ago
|
size += 4; // ack Through
|
||
8 years ago
|
packet[size] = 0;
|
||
10 years ago
|
size++; // NACK count
|
||
10 years ago
|
packet[size] = m_RTO/1000;
|
||
10 years ago
|
size++; // resend delay
|
||
10 years ago
|
if (m_Status == eStreamStatusNew)
|
||
8 years ago
|
{
|
||
10 years ago
|
// initial packet
|
||
10 years ago
|
m_Status = eStreamStatusOpen;
|
||
8 years ago
|
uint16_t flags = PACKET_FLAG_SYNCHRONIZE | PACKET_FLAG_FROM_INCLUDED |
|
||
10 years ago
|
PACKET_FLAG_SIGNATURE_INCLUDED | PACKET_FLAG_MAX_PACKET_SIZE_INCLUDED;
|
||
|
if (isNoAck) flags |= PACKET_FLAG_NO_ACK;
|
||
|
htobe16buf (packet + size, flags);
|
||
|
size += 2; // flags
|
||
9 years ago
|
size_t identityLen = m_LocalDestination.GetOwner ()->GetIdentity ()->GetFullLen ();
|
||
|
size_t signatureLen = m_LocalDestination.GetOwner ()->GetIdentity ()->GetSignatureLen ();
|
||
10 years ago
|
htobe16buf (packet + size, identityLen + signatureLen + 2); // identity + signature + packet size
|
||
|
size += 2; // options size
|
||
8 years ago
|
m_LocalDestination.GetOwner ()->GetIdentity ()->ToBuffer (packet + size, identityLen);
|
||
10 years ago
|
size += identityLen; // from
|
||
|
htobe16buf (packet + size, STREAMING_MTU);
|
||
|
size += 2; // max packet size
|
||
|
uint8_t * signature = packet + size; // set it later
|
||
|
memset (signature, 0, signatureLen); // zeroes for now
|
||
|
size += signatureLen; // signature
|
||
8 years ago
|
size += m_SendBuffer.Get (packet + size, STREAMING_MTU - size); // payload
|
||
9 years ago
|
m_LocalDestination.GetOwner ()->Sign (packet, size, signature);
|
||
8 years ago
|
}
|
||
10 years ago
|
else
|
||
|
{
|
||
|
// follow on packet
|
||
|
htobuf16 (packet + size, 0);
|
||
|
size += 2; // flags
|
||
|
htobuf16 (packet + size, 0); // no options
|
||
|
size += 2; // options size
|
||
8 years ago
|
size += m_SendBuffer.Get(packet + size, STREAMING_MTU - size); // payload
|
||
8 years ago
|
}
|
||
10 years ago
|
p->len = size;
|
||
|
packets.push_back (p);
|
||
10 years ago
|
numMsgs--;
|
||
10 years ago
|
}
|
||
8 years ago
|
}
|
||
10 years ago
|
if (packets.size () > 0)
|
||
|
{
|
||
9 years ago
|
if (m_SavedPackets.empty ()) // no NACKS
|
||
8 years ago
|
{
|
||
|
m_IsAckSendScheduled = false;
|
||
9 years ago
|
m_AckSendTimer.cancel ();
|
||
8 years ago
|
}
|
||
10 years ago
|
bool isEmpty = m_SentPackets.empty ();
|
||
10 years ago
|
auto ts = i2p::util::GetMillisecondsSinceEpoch ();
|
||
8 years ago
|
for (auto& it: packets)
|
||
10 years ago
|
{
|
||
|
it->sendTime = ts;
|
||
10 years ago
|
m_SentPackets.insert (it);
|
||
10 years ago
|
}
|
||
10 years ago
|
SendPackets (packets);
|
||
8 years ago
|
if (m_Status == eStreamStatusClosing && m_SendBuffer.IsEmpty ())
|
||
10 years ago
|
SendClose ();
|
||
10 years ago
|
if (isEmpty)
|
||
|
ScheduleResend ();
|
||
8 years ago
|
}
|
||
10 years ago
|
}
|
||
8 years ago
|
|
||
10 years ago
|
void Stream::SendQuickAck ()
|
||
11 years ago
|
{
|
||
10 years ago
|
int32_t lastReceivedSeqn = m_LastReceivedSequenceNumber;
|
||
|
if (!m_SavedPackets.empty ())
|
||
|
{
|
||
|
int32_t seqn = (*m_SavedPackets.rbegin ())->GetSeqn ();
|
||
|
if (seqn > lastReceivedSeqn) lastReceivedSeqn = seqn;
|
||
8 years ago
|
}
|
||
|
if (lastReceivedSeqn < 0)
|
||
|
{
|
||
9 years ago
|
LogPrint (eLogError, "Streaming: No packets have been received yet");
|
||
10 years ago
|
return;
|
||
|
}
|
||
8 years ago
|
|
||
10 years ago
|
Packet p;
|
||
8 years ago
|
uint8_t * packet = p.GetBuffer ();
|
||
11 years ago
|
size_t size = 0;
|
||
10 years ago
|
htobe32buf (packet + size, m_SendStreamID);
|
||
11 years ago
|
size += 4; // sendStreamID
|
||
10 years ago
|
htobe32buf (packet + size, m_RecvStreamID);
|
||
11 years ago
|
size += 4; // receiveStreamID
|
||
10 years ago
|
htobuf32 (packet + size, 0); // this is plain Ack message
|
||
11 years ago
|
size += 4; // sequenceNum
|
||
10 years ago
|
htobe32buf (packet + size, lastReceivedSeqn);
|
||
11 years ago
|
size += 4; // ack Through
|
||
10 years ago
|
uint8_t numNacks = 0;
|
||
8 years ago
|
if (lastReceivedSeqn > m_LastReceivedSequenceNumber)
|
||
|
{
|
||
10 years ago
|
// fill NACKs
|
||
|
uint8_t * nacks = packet + size + 1;
|
||
|
auto nextSeqn = m_LastReceivedSequenceNumber + 1;
|
||
|
for (auto it: m_SavedPackets)
|
||
|
{
|
||
|
auto seqn = it->GetSeqn ();
|
||
10 years ago
|
if (numNacks + (seqn - nextSeqn) >= 256)
|
||
|
{
|
||
9 years ago
|
LogPrint (eLogError, "Streaming: Number of NACKs exceeds 256. seqn=", seqn, " nextSeqn=", nextSeqn);
|
||
10 years ago
|
htobe32buf (packet + 12, nextSeqn); // change ack Through
|
||
|
break;
|
||
8 years ago
|
}
|
||
10 years ago
|
for (uint32_t i = nextSeqn; i < seqn; i++)
|
||
|
{
|
||
10 years ago
|
htobe32buf (nacks, i);
|
||
10 years ago
|
nacks += 4;
|
||
|
numNacks++;
|
||
8 years ago
|
}
|
||
10 years ago
|
nextSeqn = seqn + 1;
|
||
|
}
|
||
8 years ago
|
packet[size] = numNacks;
|
||
|
size++; // NACK count
|
||
10 years ago
|
size += numNacks*4; // NACKs
|
||
8 years ago
|
}
|
||
10 years ago
|
else
|
||
|
{
|
||
|
// No NACKs
|
||
8 years ago
|
packet[size] = 0;
|
||
|
size++; // NACK count
|
||
|
}
|
||
11 years ago
|
size++; // resend delay
|
||
10 years ago
|
htobuf16 (packet + size, 0); // nof flags set
|
||
11 years ago
|
size += 2; // flags
|
||
10 years ago
|
htobuf16 (packet + size, 0); // no options
|
||
11 years ago
|
size += 2; // options size
|
||
8 years ago
|
p.len = size;
|
||
10 years ago
|
|
||
10 years ago
|
SendPackets (std::vector<Packet *> { &p });
|
||
9 years ago
|
LogPrint (eLogDebug, "Streaming: Quick Ack sent. ", (int)numNacks, " NACKs");
|
||
8 years ago
|
}
|
||
11 years ago
|
|
||
|
void Stream::Close ()
|
||
|
{
|
||
8 years ago
|
LogPrint(eLogDebug, "Streaming: closing stream with sSID=", m_SendStreamID, ", rSID=", m_RecvStreamID, ", status=", m_Status);
|
||
10 years ago
|
switch (m_Status)
|
||
|
{
|
||
|
case eStreamStatusOpen:
|
||
|
m_Status = eStreamStatusClosing;
|
||
|
Close (); // recursion
|
||
|
if (m_Status == eStreamStatusClosing) //still closing
|
||
9 years ago
|
LogPrint (eLogDebug, "Streaming: Trying to send stream data before closing, sSID=", m_SendStreamID);
|
||
10 years ago
|
break;
|
||
|
case eStreamStatusReset:
|
||
9 years ago
|
// TODO: send reset
|
||
8 years ago
|
Terminate ();
|
||
10 years ago
|
break;
|
||
|
case eStreamStatusClosing:
|
||
8 years ago
|
if (m_SentPackets.empty () && m_SendBuffer.IsEmpty ()) // nothing to send
|
||
10 years ago
|
{
|
||
10 years ago
|
m_Status = eStreamStatusClosed;
|
||
10 years ago
|
SendClose ();
|
||
|
}
|
||
|
break;
|
||
|
case eStreamStatusClosed:
|
||
|
// already closed
|
||
|
Terminate ();
|
||
8 years ago
|
break;
|
||
10 years ago
|
default:
|
||
9 years ago
|
LogPrint (eLogWarning, "Streaming: Unexpected stream status ", (int)m_Status, "sSID=", m_SendStreamID);
|
||
8 years ago
|
};
|
||
11 years ago
|
}
|
||
11 years ago
|
|
||
10 years ago
|
void Stream::SendClose ()
|
||
10 years ago
|
{
|
||
8 years ago
|
Packet * p = m_LocalDestination.NewPacket ();
|
||
10 years ago
|
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
|
||
9 years ago
|
htobe32buf (packet + size, m_LastReceivedSequenceNumber >= 0 ? m_LastReceivedSequenceNumber : 0);
|
||
10 years ago
|
size += 4; // ack Through
|
||
8 years ago
|
packet[size] = 0;
|
||
10 years ago
|
size++; // NACK count
|
||
|
size++; // resend delay
|
||
10 years ago
|
htobe16buf (packet + size, PACKET_FLAG_CLOSE | PACKET_FLAG_SIGNATURE_INCLUDED);
|
||
10 years ago
|
size += 2; // flags
|
||
9 years ago
|
size_t signatureLen = m_LocalDestination.GetOwner ()->GetIdentity ()->GetSignatureLen ();
|
||
10 years ago
|
htobe16buf (packet + size, signatureLen); // signature only
|
||
|
size += 2; // options size
|
||
|
uint8_t * signature = packet + size;
|
||
|
memset (packet + size, 0, signatureLen);
|
||
|
size += signatureLen; // signature
|
||
9 years ago
|
m_LocalDestination.GetOwner ()->Sign (packet, size, signature);
|
||
8 years ago
|
|
||
10 years ago
|
p->len = size;
|
||
|
m_Service.post (std::bind (&Stream::SendPacket, shared_from_this (), p));
|
||
9 years ago
|
LogPrint (eLogDebug, "Streaming: FIN sent, sSID=", m_SendStreamID);
|
||
8 years ago
|
}
|
||
|
|
||
11 years ago
|
size_t Stream::ConcatenatePackets (uint8_t * buf, size_t len)
|
||
|
{
|
||
11 years ago
|
size_t pos = 0;
|
||
11 years ago
|
while (pos < len && !m_ReceiveQueue.empty ())
|
||
11 years ago
|
{
|
||
11 years ago
|
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 ())
|
||
11 years ago
|
{
|
||
11 years ago
|
m_ReceiveQueue.pop ();
|
||
8 years ago
|
m_LocalDestination.DeletePacket (packet);
|
||
8 years ago
|
}
|
||
|
}
|
||
|
return pos;
|
||
11 years ago
|
}
|
||
11 years ago
|
|
||
11 years ago
|
bool Stream::SendPacket (Packet * packet)
|
||
|
{
|
||
|
if (packet)
|
||
8 years ago
|
{
|
||
10 years ago
|
if (m_IsAckSendScheduled)
|
||
|
{
|
||
8 years ago
|
m_IsAckSendScheduled = false;
|
||
10 years ago
|
m_AckSendTimer.cancel ();
|
||
|
}
|
||
10 years ago
|
SendPackets (std::vector<Packet *> { packet });
|
||
9 years ago
|
bool isEmpty = m_SentPackets.empty ();
|
||
|
m_SentPackets.insert (packet);
|
||
|
if (isEmpty)
|
||
|
ScheduleResend ();
|
||
8 years ago
|
return true;
|
||
|
}
|
||
11 years ago
|
else
|
||
|
return false;
|
||
8 years ago
|
}
|
||
|
|
||
10 years ago
|
void Stream::SendPackets (const std::vector<Packet *>& packets)
|
||
|
{
|
||
|
if (!m_RemoteLeaseSet)
|
||
|
{
|
||
8 years ago
|
UpdateCurrentRemoteLease ();
|
||
10 years ago
|
if (!m_RemoteLeaseSet)
|
||
|
{
|
||
9 years ago
|
LogPrint (eLogError, "Streaming: Can't send packets, missing remote LeaseSet, sSID=", m_SendStreamID);
|
||
10 years ago
|
return;
|
||
|
}
|
||
|
}
|
||
8 years ago
|
if (!m_RoutingSession || !m_RoutingSession->GetOwner ()) // expired and detached
|
||
8 years ago
|
m_RoutingSession = m_LocalDestination.GetOwner ()->GetRoutingSession (m_RemoteLeaseSet, true);
|
||
8 years ago
|
if (!m_CurrentOutboundTunnel && m_RoutingSession) // first message to send
|
||
9 years ago
|
{
|
||
|
// try to get shared path first
|
||
8 years ago
|
auto routingPath = m_RoutingSession->GetSharedRoutingPath ();
|
||
|
if (routingPath)
|
||
|
{
|
||
|
m_CurrentOutboundTunnel = routingPath->outboundTunnel;
|
||
|
m_CurrentRemoteLease = routingPath->remoteLease;
|
||
|
m_RTT = routingPath->rtt;
|
||
|
m_RTO = m_RTT*1.5; // TODO: implement it better
|
||
|
}
|
||
8 years ago
|
}
|
||
10 years ago
|
if (!m_CurrentOutboundTunnel || !m_CurrentOutboundTunnel->IsEstablished ())
|
||
9 years ago
|
m_CurrentOutboundTunnel = m_LocalDestination.GetOwner ()->GetTunnelPool ()->GetNewOutboundTunnel (m_CurrentOutboundTunnel);
|
||
10 years ago
|
if (!m_CurrentOutboundTunnel)
|
||
|
{
|
||
9 years ago
|
LogPrint (eLogError, "Streaming: No outbound tunnels in the pool, sSID=", m_SendStreamID);
|
||
10 years ago
|
return;
|
||
|
}
|
||
10 years ago
|
|
||
8 years ago
|
auto ts = i2p::util::GetMillisecondsSinceEpoch ();
|
||
8 years ago
|
if (!m_CurrentRemoteLease || !m_CurrentRemoteLease->endDate || // excluded from LeaseSet
|
||
|
ts >= m_CurrentRemoteLease->endDate - i2p::data::LEASE_ENDDATE_THRESHOLD)
|
||
10 years ago
|
UpdateCurrentRemoteLease (true);
|
||
9 years ago
|
if (m_CurrentRemoteLease && ts < m_CurrentRemoteLease->endDate + i2p::data::LEASE_ENDDATE_THRESHOLD)
|
||
8 years ago
|
{
|
||
10 years ago
|
std::vector<i2p::tunnel::TunnelMessageBlock> msgs;
|
||
|
for (auto it: packets)
|
||
8 years ago
|
{
|
||
9 years ago
|
auto msg = m_RoutingSession->WrapSingleMessage (m_LocalDestination.CreateDataMessage (it->GetBuffer (), it->GetLength (), m_Port));
|
||
8 years ago
|
msgs.push_back (i2p::tunnel::TunnelMessageBlock
|
||
|
{
|
||
10 years ago
|
i2p::tunnel::eDeliveryTypeTunnel,
|
||
9 years ago
|
m_CurrentRemoteLease->tunnelGateway, m_CurrentRemoteLease->tunnelID,
|
||
10 years ago
|
msg
|
||
8 years ago
|
});
|
||
10 years ago
|
m_NumSentBytes += it->GetLength ();
|
||
10 years ago
|
}
|
||
10 years ago
|
m_CurrentOutboundTunnel->SendTunnelDataMsg (msgs);
|
||
8 years ago
|
}
|
||
10 years ago
|
else
|
||
8 years ago
|
{
|
||
8 years ago
|
LogPrint (eLogWarning, "Streaming: Remote lease is not available, sSID=", m_SendStreamID);
|
||
8 years ago
|
if (m_RoutingSession)
|
||
8 years ago
|
m_RoutingSession->SetSharedRoutingPath (nullptr); // invalidate routing path
|
||
|
}
|
||
10 years ago
|
}
|
||
|
|
||
8 years ago
|
void Stream::SendUpdatedLeaseSet ()
|
||
|
{
|
||
8 years ago
|
if (m_RoutingSession)
|
||
8 years ago
|
{
|
||
|
if (m_RoutingSession->IsLeaseSetNonConfirmed ())
|
||
8 years ago
|
{
|
||
|
auto ts = i2p::util::GetMillisecondsSinceEpoch ();
|
||
|
if (ts > m_RoutingSession->GetLeaseSetSubmissionTime () + i2p::garlic::LEASET_CONFIRMATION_TIMEOUT)
|
||
|
{
|
||
|
// LeaseSet was not confirmed, should try other tunnels
|
||
|
LogPrint (eLogWarning, "Streaming: LeaseSet was not confrimed in ", i2p::garlic::LEASET_CONFIRMATION_TIMEOUT, " milliseconds. Trying to resubmit");
|
||
|
m_RoutingSession->SetSharedRoutingPath (nullptr);
|
||
8 years ago
|
m_CurrentOutboundTunnel = nullptr;
|
||
8 years ago
|
m_CurrentRemoteLease = nullptr;
|
||
|
SendQuickAck ();
|
||
8 years ago
|
}
|
||
|
}
|
||
8 years ago
|
else if (m_RoutingSession->IsLeaseSetUpdated ())
|
||
8 years ago
|
{
|
||
8 years ago
|
LogPrint (eLogDebug, "Streaming: sending updated LeaseSet");
|
||
|
SendQuickAck ();
|
||
8 years ago
|
}
|
||
|
}
|
||
|
}
|
||
|
|
||
10 years ago
|
void Stream::ScheduleResend ()
|
||
|
{
|
||
|
m_ResendTimer.cancel ();
|
||
8 years ago
|
// check for invalid value
|
||
8 years ago
|
if (m_RTO <= 0) m_RTO = INITIAL_RTO;
|
||
10 years ago
|
m_ResendTimer.expires_from_now (boost::posix_time::milliseconds(m_RTO));
|
||
10 years ago
|
m_ResendTimer.async_wait (std::bind (&Stream::HandleResendTimer,
|
||
|
shared_from_this (), std::placeholders::_1));
|
||
10 years ago
|
}
|
||
8 years ago
|
|
||
10 years ago
|
void Stream::HandleResendTimer (const boost::system::error_code& ecode)
|
||
|
{
|
||
8 years ago
|
if (ecode != boost::asio::error::operation_aborted)
|
||
|
{
|
||
10 years ago
|
// check for resend attempts
|
||
|
if (m_NumResendAttempts >= MAX_NUM_RESEND_ATTEMPTS)
|
||
|
{
|
||
8 years ago
|
LogPrint (eLogWarning, "Streaming: packet was not ACKed after ", MAX_NUM_RESEND_ATTEMPTS, " attempts, terminate, rSID=", m_RecvStreamID, ", sSID=", m_SendStreamID);
|
||
10 years ago
|
m_Status = eStreamStatusReset;
|
||
|
Close ();
|
||
|
return;
|
||
8 years ago
|
}
|
||
10 years ago
|
|
||
|
// collect packets to resend
|
||
10 years ago
|
auto ts = i2p::util::GetMillisecondsSinceEpoch ();
|
||
10 years ago
|
std::vector<Packet *> packets;
|
||
10 years ago
|
for (auto it : m_SentPackets)
|
||
10 years ago
|
{
|
||
10 years ago
|
if (ts >= it->sendTime + m_RTO)
|
||
10 years ago
|
{
|
||
|
it->sendTime = ts;
|
||
10 years ago
|
packets.push_back (it);
|
||
8 years ago
|
}
|
||
|
}
|
||
10 years ago
|
|
||
|
// select tunnels if necessary and send
|
||
10 years ago
|
if (packets.size () > 0)
|
||
10 years ago
|
{
|
||
10 years ago
|
m_NumResendAttempts++;
|
||
10 years ago
|
m_RTO *= 2;
|
||
10 years ago
|
switch (m_NumResendAttempts)
|
||
8 years ago
|
{
|
||
10 years ago
|
case 1: // congesion avoidance
|
||
|
m_WindowSize /= 2;
|
||
|
if (m_WindowSize < MIN_WINDOW_SIZE) m_WindowSize = MIN_WINDOW_SIZE;
|
||
|
break;
|
||
|
case 2:
|
||
10 years ago
|
m_RTO = INITIAL_RTO; // drop RTO to initial upon tunnels pair change first time
|
||
|
// no break here
|
||
9 years ago
|
case 4:
|
||
|
if (m_RoutingSession) m_RoutingSession->SetSharedRoutingPath (nullptr);
|
||
10 years ago
|
UpdateCurrentRemoteLease (); // pick another lease
|
||
8 years ago
|
LogPrint (eLogWarning, "Streaming: Another remote lease has been selected for stream with rSID=", m_RecvStreamID, ", sSID=", m_SendStreamID);
|
||
8 years ago
|
break;
|
||
10 years ago
|
case 3:
|
||
8 years ago
|
// pick another outbound tunnel
|
||
9 years ago
|
if (m_RoutingSession) m_RoutingSession->SetSharedRoutingPath (nullptr);
|
||
8 years ago
|
m_CurrentOutboundTunnel = m_LocalDestination.GetOwner ()->GetTunnelPool ()->GetNextOutboundTunnel (m_CurrentOutboundTunnel);
|
||
9 years ago
|
LogPrint (eLogWarning, "Streaming: Another outbound tunnel has been selected for stream with sSID=", m_SendStreamID);
|
||
10 years ago
|
break;
|
||
8 years ago
|
default: ;
|
||
|
}
|
||
10 years ago
|
SendPackets (packets);
|
||
8 years ago
|
}
|
||
10 years ago
|
ScheduleResend ();
|
||
8 years ago
|
}
|
||
|
}
|
||
|
|
||
10 years ago
|
void Stream::HandleAckSendTimer (const boost::system::error_code& ecode)
|
||
|
{
|
||
10 years ago
|
if (m_IsAckSendScheduled)
|
||
10 years ago
|
{
|
||
10 years ago
|
if (m_LastReceivedSequenceNumber < 0)
|
||
|
{
|
||
8 years ago
|
LogPrint (eLogWarning, "Streaming: SYN has not been received after ", ACK_SEND_TIMEOUT, " milliseconds after follow on, terminate rSID=", m_RecvStreamID, ", sSID=", m_SendStreamID);
|
||
10 years ago
|
m_Status = eStreamStatusReset;
|
||
|
Close ();
|
||
|
return;
|
||
8 years ago
|
}
|
||
10 years ago
|
if (m_Status == eStreamStatusOpen)
|
||
8 years ago
|
{
|
||
8 years ago
|
if (m_RoutingSession && m_RoutingSession->IsLeaseSetNonConfirmed ())
|
||
8 years ago
|
{
|
||
|
// seems something went wrong and we should re-select tunnels
|
||
|
m_CurrentOutboundTunnel = nullptr;
|
||
|
m_CurrentRemoteLease = nullptr;
|
||
|
}
|
||
10 years ago
|
SendQuickAck ();
|
||
8 years ago
|
}
|
||
10 years ago
|
m_IsAckSendScheduled = false;
|
||
8 years ago
|
}
|
||
10 years ago
|
}
|
||
|
|
||
10 years ago
|
void Stream::UpdateCurrentRemoteLease (bool expired)
|
||
11 years ago
|
{
|
||
8 years ago
|
if (!m_RemoteLeaseSet || m_RemoteLeaseSet->IsExpired ())
|
||
10 years ago
|
{
|
||
9 years ago
|
m_RemoteLeaseSet = m_LocalDestination.GetOwner ()->FindLeaseSet (m_RemoteIdentity->GetIdentHash ());
|
||
8 years ago
|
if (!m_RemoteLeaseSet)
|
||
|
{
|
||
9 years ago
|
LogPrint (eLogWarning, "Streaming: LeaseSet ", m_RemoteIdentity->GetIdentHash ().ToBase64 (), " not found");
|
||
8 years ago
|
m_LocalDestination.GetOwner ()->RequestDestination (m_RemoteIdentity->GetIdentHash ()); // try to request for a next attempt
|
||
8 years ago
|
}
|
||
10 years ago
|
}
|
||
10 years ago
|
if (m_RemoteLeaseSet)
|
||
|
{
|
||
10 years ago
|
if (!m_RoutingSession)
|
||
9 years ago
|
m_RoutingSession = m_LocalDestination.GetOwner ()->GetRoutingSession (m_RemoteLeaseSet, true);
|
||
10 years ago
|
auto leases = m_RemoteLeaseSet->GetNonExpiredLeases (false); // try without threshold first
|
||
|
if (leases.empty ())
|
||
|
{
|
||
10 years ago
|
expired = false;
|
||
9 years ago
|
m_LocalDestination.GetOwner ()->RequestDestination (m_RemoteIdentity->GetIdentHash ()); // time to request
|
||
8 years ago
|
leases = m_RemoteLeaseSet->GetNonExpiredLeases (true); // then with threshold
|
||
10 years ago
|
}
|
||
10 years ago
|
if (!leases.empty ())
|
||
8 years ago
|
{
|
||
|
bool updated = false;
|
||
9 years ago
|
if (expired && m_CurrentRemoteLease)
|
||
10 years ago
|
{
|
||
8 years ago
|
for (const auto& it: leases)
|
||
9 years ago
|
if ((it->tunnelGateway == m_CurrentRemoteLease->tunnelGateway) && (it->tunnelID != m_CurrentRemoteLease->tunnelID))
|
||
10 years ago
|
{
|
||
|
m_CurrentRemoteLease = it;
|
||
|
updated = true;
|
||
|
break;
|
||
8 years ago
|
}
|
||
10 years ago
|
}
|
||
|
if (!updated)
|
||
|
{
|
||
9 years ago
|
uint32_t i = rand () % leases.size ();
|
||
9 years ago
|
if (m_CurrentRemoteLease && leases[i]->tunnelID == m_CurrentRemoteLease->tunnelID)
|
||
8 years ago
|
// make sure we don't select previous
|
||
10 years ago
|
i = (i + 1) % leases.size (); // if so, pick next
|
||
8 years ago
|
m_CurrentRemoteLease = leases[i];
|
||
10 years ago
|
}
|
||
8 years ago
|
}
|
||
10 years ago
|
else
|
||
8 years ago
|
{
|
||
8 years ago
|
LogPrint (eLogWarning, "Streaming: All remote leases are expired");
|
||
10 years ago
|
m_RemoteLeaseSet = nullptr;
|
||
9 years ago
|
m_CurrentRemoteLease = nullptr;
|
||
9 years ago
|
// we have requested expired before, no need to do it twice
|
||
8 years ago
|
}
|
||
10 years ago
|
}
|
||
11 years ago
|
else
|
||
8 years ago
|
{
|
||
|
LogPrint (eLogWarning, "Streaming: Remote LeaseSet not found");
|
||
9 years ago
|
m_CurrentRemoteLease = nullptr;
|
||
8 years ago
|
}
|
||
|
}
|
||
10 years ago
|
|
||
8 years ago
|
StreamingDestination::StreamingDestination (std::shared_ptr<i2p::client::ClientDestination> owner, uint16_t localPort, bool gzip):
|
||
|
m_Owner (owner), m_LocalPort (localPort), m_Gzip (gzip),
|
||
8 years ago
|
m_LastIncomingReceiveStreamID (0),
|
||
8 years ago
|
m_PendingIncomingTimer (m_Owner->GetService ()),
|
||
|
m_ConnTrackTimer(m_Owner->GetService()),
|
||
|
m_ConnsPerMinute(DEFAULT_MAX_CONNS_PER_MIN),
|
||
|
m_LastBanClear(i2p::util::GetMillisecondsSinceEpoch())
|
||
9 years ago
|
{
|
||
|
}
|
||
8 years ago
|
|
||
9 years ago
|
StreamingDestination::~StreamingDestination ()
|
||
|
{
|
||
8 years ago
|
for (auto& it: m_SavedPackets)
|
||
9 years ago
|
{
|
||
8 years ago
|
for (auto it1: it.second) DeletePacket (it1);
|
||
8 years ago
|
it.second.clear ();
|
||
9 years ago
|
}
|
||
|
m_SavedPackets.clear ();
|
||
9 years ago
|
}
|
||
|
|
||
10 years ago
|
void StreamingDestination::Start ()
|
||
8 years ago
|
{
|
||
|
ScheduleConnTrack();
|
||
10 years ago
|
}
|
||
8 years ago
|
|
||
10 years ago
|
void StreamingDestination::Stop ()
|
||
8 years ago
|
{
|
||
10 years ago
|
ResetAcceptor ();
|
||
9 years ago
|
m_PendingIncomingTimer.cancel ();
|
||
8 years ago
|
m_PendingIncomingStreams.clear ();
|
||
8 years ago
|
m_ConnTrackTimer.cancel();
|
||
10 years ago
|
{
|
||
|
std::unique_lock<std::mutex> l(m_StreamsMutex);
|
||
|
m_Streams.clear ();
|
||
8 years ago
|
}
|
||
|
{
|
||
|
std::unique_lock<std::mutex> l(m_ConnsMutex);
|
||
|
m_Conns.clear ();
|
||
|
}
|
||
8 years ago
|
}
|
||
|
|
||
10 years ago
|
void StreamingDestination::HandleNextPacket (Packet * packet)
|
||
|
{
|
||
|
uint32_t sendStreamID = packet->GetSendStreamID ();
|
||
|
if (sendStreamID)
|
||
8 years ago
|
{
|
||
10 years ago
|
auto it = m_Streams.find (sendStreamID);
|
||
|
if (it != m_Streams.end ())
|
||
|
it->second->HandleNextPacket (packet);
|
||
|
else
|
||
8 years ago
|
{
|
||
8 years ago
|
LogPrint (eLogInfo, "Streaming: Unknown stream sSID=", sendStreamID);
|
||
8 years ago
|
DeletePacket (packet);
|
||
10 years ago
|
}
|
||
8 years ago
|
}
|
||
|
else
|
||
10 years ago
|
{
|
||
10 years ago
|
if (packet->IsSYN () && !packet->GetSeqn ()) // new incoming stream
|
||
8 years ago
|
{
|
||
8 years ago
|
uint32_t receiveStreamID = packet->GetReceiveStreamID ();
|
||
8 years ago
|
if (receiveStreamID == m_LastIncomingReceiveStreamID)
|
||
8 years ago
|
{
|
||
|
// already pending
|
||
|
LogPrint(eLogWarning, "Streaming: Incoming streaming with rSID=", receiveStreamID, " already exists");
|
||
8 years ago
|
DeletePacket (packet); // drop it, because previous should be connected
|
||
8 years ago
|
return;
|
||
8 years ago
|
}
|
||
8 years ago
|
auto incomingStream = CreateNewIncomingStream ();
|
||
8 years ago
|
incomingStream->HandleNextPacket (packet); // SYN
|
||
8 years ago
|
auto ident = incomingStream->GetRemoteIdentity();
|
||
|
if(ident)
|
||
|
{
|
||
|
auto ih = ident->GetIdentHash();
|
||
|
if(DropNewStream(ih))
|
||
|
{
|
||
|
// drop
|
||
8 years ago
|
LogPrint(eLogWarning, "Streaming: Dropping connection, too many inbound streams from ", ih.ToBase32());
|
||
8 years ago
|
incomingStream->Terminate();
|
||
8 years ago
|
return;
|
||
|
}
|
||
8 years ago
|
}
|
||
8 years ago
|
m_LastIncomingReceiveStreamID = receiveStreamID;
|
||
8 years ago
|
|
||
8 years ago
|
// 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);
|
||
8 years ago
|
}
|
||
8 years ago
|
}
|
||
9 years ago
|
// accept
|
||
10 years ago
|
if (m_Acceptor != nullptr)
|
||
|
m_Acceptor (incomingStream);
|
||
|
else
|
||
|
{
|
||
9 years ago
|
LogPrint (eLogWarning, "Streaming: Acceptor for incoming stream is not set");
|
||
9 years ago
|
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));
|
||
8 years ago
|
m_PendingIncomingTimer.async_wait (std::bind (&StreamingDestination::HandlePendingIncomingTimer,
|
||
9 years ago
|
shared_from_this (), std::placeholders::_1));
|
||
9 years ago
|
LogPrint (eLogDebug, "Streaming: Pending incoming stream added, rSID=", receiveStreamID);
|
||
9 years ago
|
}
|
||
|
else
|
||
8 years ago
|
{
|
||
9 years ago
|
LogPrint (eLogWarning, "Streaming: Pending incoming streams backlog exceeds ", MAX_PENDING_INCOMING_BACKLOG);
|
||
9 years ago
|
incomingStream->Close ();
|
||
8 years ago
|
}
|
||
|
}
|
||
|
}
|
||
10 years ago
|
else // follow on packet without SYN
|
||
|
{
|
||
|
uint32_t receiveStreamID = packet->GetReceiveStreamID ();
|
||
8 years ago
|
for (auto& it: m_Streams)
|
||
10 years ago
|
if (it.second->GetSendStreamID () == receiveStreamID)
|
||
|
{
|
||
|
// found
|
||
|
it.second->HandleNextPacket (packet);
|
||
|
return;
|
||
|
}
|
||
9 years ago
|
// save follow on packet
|
||
|
auto it = m_SavedPackets.find (receiveStreamID);
|
||
|
if (it != m_SavedPackets.end ())
|
||
|
it->second.push_back (packet);
|
||
|
else
|
||
|
{
|
||
9 years ago
|
m_SavedPackets[receiveStreamID] = std::list<Packet *>{ packet };
|
||
9 years ago
|
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)
|
||
|
{
|
||
9 years ago
|
if (ecode != boost::asio::error::operation_aborted)
|
||
9 years ago
|
{
|
||
|
auto it = s->m_SavedPackets.find (receiveStreamID);
|
||
|
if (it != s->m_SavedPackets.end ())
|
||
|
{
|
||
8 years ago
|
for (auto it1: it->second) s->DeletePacket (it1);
|
||
9 years ago
|
it->second.clear ();
|
||
|
s->m_SavedPackets.erase (it);
|
||
|
}
|
||
|
}
|
||
|
});
|
||
|
}
|
||
8 years ago
|
}
|
||
|
}
|
||
|
}
|
||
|
|
||
10 years ago
|
std::shared_ptr<Stream> StreamingDestination::CreateNewOutgoingStream (std::shared_ptr<const i2p::data::LeaseSet> remote, int port)
|
||
10 years ago
|
{
|
||
9 years ago
|
auto s = std::make_shared<Stream> (m_Owner->GetService (), *this, remote, port);
|
||
10 years ago
|
std::unique_lock<std::mutex> l(m_StreamsMutex);
|
||
|
m_Streams[s->GetRecvStreamID ()] = s;
|
||
|
return s;
|
||
8 years ago
|
}
|
||
10 years ago
|
|
||
10 years ago
|
std::shared_ptr<Stream> StreamingDestination::CreateNewIncomingStream ()
|
||
10 years ago
|
{
|
||
9 years ago
|
auto s = std::make_shared<Stream> (m_Owner->GetService (), *this);
|
||
10 years ago
|
std::unique_lock<std::mutex> l(m_StreamsMutex);
|
||
|
m_Streams[s->GetRecvStreamID ()] = s;
|
||
|
return s;
|
||
|
}
|
||
|
|
||
10 years ago
|
void StreamingDestination::DeleteStream (std::shared_ptr<Stream> stream)
|
||
10 years ago
|
{
|
||
|
if (stream)
|
||
8 years ago
|
{
|
||
10 years ago
|
std::unique_lock<std::mutex> l(m_StreamsMutex);
|
||
|
auto it = m_Streams.find (stream->GetRecvStreamID ());
|
||
|
if (it != m_Streams.end ())
|
||
|
m_Streams.erase (it);
|
||
8 years ago
|
}
|
||
|
}
|
||
10 years ago
|
|
||
8 years ago
|
void StreamingDestination::SetAcceptor (const Acceptor& acceptor)
|
||
|
{
|
||
8 years ago
|
m_Acceptor = acceptor; // we must set it immediately for IsAcceptorSet
|
||
|
auto s = shared_from_this ();
|
||
|
m_Owner->GetService ().post([s](void)
|
||
8 years ago
|
{
|
||
8 years ago
|
// take care about incoming queue
|
||
|
for (auto& it: s->m_PendingIncomingStreams)
|
||
9 years ago
|
if (it->GetStatus () == eStreamStatusOpen) // still open?
|
||
8 years ago
|
s->m_Acceptor (it);
|
||
|
s->m_PendingIncomingStreams.clear ();
|
||
|
s->m_PendingIncomingTimer.cancel ();
|
||
8 years ago
|
});
|
||
9 years ago
|
}
|
||
|
|
||
8 years ago
|
void StreamingDestination::ResetAcceptor ()
|
||
|
{
|
||
|
if (m_Acceptor) m_Acceptor (nullptr);
|
||
|
m_Acceptor = nullptr;
|
||
9 years ago
|
}
|
||
|
|
||
8 years ago
|
void StreamingDestination::AcceptOnce (const Acceptor& acceptor)
|
||
|
{
|
||
|
m_Owner->GetService ().post([acceptor, this](void)
|
||
8 years ago
|
{
|
||
8 years ago
|
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
|
||
|
{
|
||
8 years ago
|
m_Acceptor = std::bind (&StreamingDestination::AcceptOnceAcceptor, this,
|
||
|
std::placeholders::_1, acceptor, m_Acceptor);
|
||
8 years ago
|
}
|
||
8 years ago
|
});
|
||
8 years ago
|
}
|
||
|
|
||
8 years ago
|
void StreamingDestination::AcceptOnceAcceptor (std::shared_ptr<Stream> stream, Acceptor acceptor, Acceptor prev)
|
||
|
{
|
||
|
m_Acceptor = prev;
|
||
|
acceptor (stream);
|
||
|
}
|
||
|
|
||
9 years ago
|
void StreamingDestination::HandlePendingIncomingTimer (const boost::system::error_code& ecode)
|
||
|
{
|
||
|
if (ecode != boost::asio::error::operation_aborted)
|
||
|
{
|
||
9 years ago
|
LogPrint (eLogWarning, "Streaming: Pending incoming timeout expired");
|
||
8 years ago
|
for (auto& it: m_PendingIncomingStreams)
|
||
9 years ago
|
it->Close ();
|
||
|
m_PendingIncomingStreams.clear ();
|
||
8 years ago
|
}
|
||
|
}
|
||
|
|
||
10 years ago
|
void StreamingDestination::HandleDataMessagePayload (const uint8_t * buf, size_t len)
|
||
|
{
|
||
|
// unzip it
|
||
8 years ago
|
Packet * uncompressed = NewPacket ();
|
||
10 years ago
|
uncompressed->offset = 0;
|
||
9 years ago
|
uncompressed->len = m_Inflator.Inflate (buf, len, uncompressed->buf, MAX_PACKET_SIZE);
|
||
|
if (uncompressed->len)
|
||
8 years ago
|
HandleNextPacket (uncompressed);
|
||
10 years ago
|
else
|
||
8 years ago
|
DeletePacket (uncompressed);
|
||
10 years ago
|
}
|
||
9 years ago
|
|
||
|
std::shared_ptr<I2NPMessage> StreamingDestination::CreateDataMessage (const uint8_t * payload, size_t len, uint16_t toPort)
|
||
|
{
|
||
|
auto msg = NewI2NPShortMessage ();
|
||
|
if (!m_Gzip || len <= i2p::stream::COMPRESSION_THRESHOLD_SIZE)
|
||
|
m_Deflator.SetCompressionLevel (Z_NO_COMPRESSION);
|
||
|
else
|
||
|
m_Deflator.SetCompressionLevel (Z_DEFAULT_COMPRESSION);
|
||
|
uint8_t * buf = msg->GetPayload ();
|
||
|
buf += 4; // reserve for lengthlength
|
||
|
msg->len += 4;
|
||
|
size_t size = m_Deflator.Deflate (payload, len, buf, msg->maxLen - msg->len);
|
||
|
if (size)
|
||
|
{
|
||
|
htobe32buf (msg->GetPayload (), size); // length
|
||
|
htobe16buf (buf + 4, m_LocalPort); // source port
|
||
8 years ago
|
htobe16buf (buf + 6, toPort); // destination port
|
||
9 years ago
|
buf[9] = i2p::client::PROTOCOL_TYPE_STREAMING; // streaming protocol
|
||
8 years ago
|
msg->len += size;
|
||
9 years ago
|
msg->FillI2NPMessageHeader (eI2NPData);
|
||
8 years ago
|
}
|
||
9 years ago
|
else
|
||
|
msg = nullptr;
|
||
|
return msg;
|
||
8 years ago
|
}
|
||
|
|
||
|
void StreamingDestination::SetMaxConnsPerMinute(const uint32_t conns)
|
||
|
{
|
||
|
m_ConnsPerMinute = conns;
|
||
8 years ago
|
LogPrint(eLogDebug, "Streaming: Set max conns per minute per destination to ", conns);
|
||
8 years ago
|
}
|
||
|
|
||
|
bool StreamingDestination::DropNewStream(const i2p::data::IdentHash & ih)
|
||
|
{
|
||
|
std::lock_guard<std::mutex> lock(m_ConnsMutex);
|
||
|
if (m_Banned.size() > MAX_BANNED_CONNS) return true; // overload
|
||
8 years ago
|
auto end = std::end(m_Banned);
|
||
|
if ( std::find(std::begin(m_Banned), end, ih) != end) return true; // already banned
|
||
8 years ago
|
auto itr = m_Conns.find(ih);
|
||
|
if (itr == m_Conns.end())
|
||
|
m_Conns[ih] = 0;
|
||
|
|
||
8 years ago
|
m_Conns[ih] += 1;
|
||
8 years ago
|
|
||
8 years ago
|
bool ban = m_Conns[ih] >= m_ConnsPerMinute;
|
||
8 years ago
|
if (ban)
|
||
|
{
|
||
|
m_Banned.push_back(ih);
|
||
|
m_Conns.erase(ih);
|
||
8 years ago
|
LogPrint(eLogWarning, "Streaming: ban ", ih.ToBase32());
|
||
8 years ago
|
}
|
||
|
return ban;
|
||
|
}
|
||
|
|
||
|
void StreamingDestination::HandleConnTrack(const boost::system::error_code& ecode)
|
||
|
{
|
||
|
if (ecode != boost::asio::error::operation_aborted)
|
||
|
{
|
||
|
{ // acquire lock
|
||
|
std::lock_guard<std::mutex> lock(m_ConnsMutex);
|
||
|
// clear conn tracking
|
||
|
m_Conns.clear();
|
||
|
// check for ban clear
|
||
|
auto ts = i2p::util::GetMillisecondsSinceEpoch();
|
||
|
if (ts - m_LastBanClear >= DEFAULT_BAN_INTERVAL)
|
||
|
{
|
||
|
// clear bans
|
||
|
m_Banned.clear();
|
||
|
m_LastBanClear = ts;
|
||
|
}
|
||
|
}
|
||
|
// reschedule timer
|
||
|
ScheduleConnTrack();
|
||
8 years ago
|
}
|
||
8 years ago
|
}
|
||
|
|
||
|
void StreamingDestination::ScheduleConnTrack()
|
||
|
{
|
||
|
m_ConnTrackTimer.expires_from_now (boost::posix_time::seconds(60));
|
||
|
m_ConnTrackTimer.async_wait (
|
||
8 years ago
|
std::bind (&StreamingDestination::HandleConnTrack,
|
||
8 years ago
|
shared_from_this (), std::placeholders::_1));
|
||
|
}
|
||
8 years ago
|
}
|
||
|
}
|