i2pd/Streaming.cpp

780 lines
22 KiB
C++
Raw Normal View History

#include <cryptopp/gzip.h>
2013-12-13 02:36:24 +00:00
#include "Log.h"
#include "RouterInfo.h"
#include "RouterContext.h"
2013-12-31 01:46:33 +00:00
#include "Tunnel.h"
#include "Timestamp.h"
#include "Destination.h"
2013-12-13 02:36:24 +00:00
#include "Streaming.h"
namespace i2p
{
namespace stream
{
2014-10-07 14:44:42 +00:00
Stream::Stream (boost::asio::io_service& service, StreamingDestination& local,
2015-01-27 16:27:58 +00:00
std::shared_ptr<const i2p::data::LeaseSet> remote, int port): m_Service (service), m_SendStreamID (0),
2014-10-10 15:53:27 +00:00
m_SequenceNumber (0), m_LastReceivedSequenceNumber (-1), m_IsOpen (false),
m_IsReset (false), m_IsAckSendScheduled (false), m_LocalDestination (local),
2015-02-03 18:46:44 +00:00
m_RemoteLeaseSet (remote), m_ReceiveTimer (m_Service), m_ResendTimer (m_Service),
m_AckSendTimer (m_Service), m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (port),
m_WindowSize (MIN_WINDOW_SIZE), m_RTT (INITIAL_RTT), m_LastWindowSizeIncreaseTime (0)
{
m_RecvStreamID = i2p::context.GetRandomNumberGenerator ().GenerateWord32 ();
2014-03-23 13:25:16 +00:00
UpdateCurrentRemoteLease ();
}
2014-10-07 14:44:42 +00:00
Stream::Stream (boost::asio::io_service& service, StreamingDestination& local):
2014-08-06 23:19:59 +00:00
m_Service (service), m_SendStreamID (0), m_SequenceNumber (0), m_LastReceivedSequenceNumber (-1),
2014-10-10 15:53:27 +00:00
m_IsOpen (false), m_IsReset (false), m_IsAckSendScheduled (false), m_LocalDestination (local),
2015-01-27 19:55:46 +00:00
m_ReceiveTimer (m_Service), m_ResendTimer (m_Service), m_AckSendTimer (m_Service),
2015-02-03 18:46:44 +00:00
m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (0), m_WindowSize (MIN_WINDOW_SIZE),
m_RTT (INITIAL_RTT), m_LastWindowSizeIncreaseTime (0)
2014-08-01 18:54:14 +00:00
{
m_RecvStreamID = i2p::context.GetRandomNumberGenerator ().GenerateWord32 ();
}
2014-01-11 01:21:38 +00:00
Stream::~Stream ()
2014-10-04 20:27:21 +00:00
{
2014-10-10 15:53:27 +00:00
m_AckSendTimer.cancel ();
2014-04-12 01:13:52 +00:00
while (!m_ReceiveQueue.empty ())
{
auto packet = m_ReceiveQueue.front ();
m_ReceiveQueue.pop ();
2014-01-11 01:21:38 +00:00
delete packet;
2014-10-04 20:27:21 +00:00
}
m_ReceiveTimer.cancel ();
2014-08-10 22:27:23 +00:00
for (auto it: m_SentPackets)
delete it;
2014-10-04 20:27:21 +00:00
m_SentPackets.clear ();
m_ResendTimer.cancel ();
for (auto it: m_SavedPackets)
delete it;
m_SavedPackets.clear ();
LogPrint (eLogDebug, "Stream deleted");
2014-01-11 01:21:38 +00:00
}
void Stream::HandleNextPacket (Packet * packet)
{
2014-10-13 21:03:27 +00:00
m_NumReceivedBytes += packet->GetLength ();
2014-08-01 18:54:14 +00:00
if (!m_SendStreamID)
m_SendStreamID = packet->GetReceiveStreamID ();
2014-08-10 22:27:23 +00:00
if (!packet->IsNoAck ()) // ack received
ProcessAck (packet);
2014-08-06 23:19:59 +00:00
int32_t receivedSeqn = packet->GetSeqn ();
2014-08-07 02:08:57 +00:00
bool isSyn = packet->IsSYN ();
if (!receivedSeqn && !isSyn)
2014-08-06 19:44:00 +00:00
{
// plain ack
LogPrint (eLogDebug, "Plain ACK received");
2014-08-06 19:44:00 +00:00
delete packet;
return;
}
LogPrint (eLogDebug, "Received seqn=", receivedSeqn);
2014-08-07 02:08:57 +00:00
if (isSyn || receivedSeqn == m_LastReceivedSequenceNumber + 1)
{
2014-02-02 03:20:41 +00:00
// we have received next in sequence message
ProcessPacket (packet);
2014-08-07 02:08:57 +00:00
// we should also try stored messages if any
for (auto it = m_SavedPackets.begin (); it != m_SavedPackets.end ();)
{
2014-08-06 23:19:59 +00:00
if ((*it)->GetSeqn () == (uint32_t)(m_LastReceivedSequenceNumber + 1))
{
2014-02-02 03:20:41 +00:00
Packet * savedPacket = *it;
m_SavedPackets.erase (it++);
2014-02-02 03:20:41 +00:00
ProcessPacket (savedPacket);
}
else
break;
2014-03-25 21:43:36 +00:00
}
2014-10-10 15:53:27 +00:00
// schedule ack for last message
2014-03-30 13:02:43 +00:00
if (m_IsOpen)
2014-10-10 15:53:27 +00:00
{
if (!m_IsAckSendScheduled)
{
m_IsAckSendScheduled = true;
m_AckSendTimer.expires_from_now (boost::posix_time::milliseconds(ACK_SEND_TIMEOUT));
2014-11-23 22:00:45 +00:00
m_AckSendTimer.async_wait (std::bind (&Stream::HandleAckSendTimer,
shared_from_this (), std::placeholders::_1));
2014-10-10 15:53:27 +00:00
}
}
2014-08-07 02:08:57 +00:00
else if (isSyn)
// we have to send SYN back to incoming connection
2014-10-02 01:18:41 +00:00
Send (nullptr, 0); // also sets m_IsOpen
}
else
{
if (receivedSeqn <= m_LastReceivedSequenceNumber)
{
// we have received duplicate
LogPrint (eLogWarning, "Duplicate message ", receivedSeqn, " received");
SendQuickAck (); // resend ack for previous message again
delete packet; // packet dropped
}
else
{
LogPrint (eLogWarning, "Missing messages from ", m_LastReceivedSequenceNumber + 1, " to ", receivedSeqn - 1);
// save message and wait for missing message again
SavePacket (packet);
2014-11-29 21:21:19 +00:00
// send NACKs for missing messages ASAP
if (m_IsAckSendScheduled)
{
m_IsAckSendScheduled = false;
m_AckSendTimer.cancel ();
}
SendQuickAck ();
}
}
2014-02-02 03:20:41 +00:00
}
void Stream::SavePacket (Packet * packet)
{
m_SavedPackets.insert (packet);
}
void Stream::ProcessPacket (Packet * packet)
{
// process flags
uint32_t receivedSeqn = packet->GetSeqn ();
uint16_t flags = packet->GetFlags ();
LogPrint (eLogDebug, "Process seqn=", receivedSeqn, ", flags=", flags);
2014-02-02 03:20:41 +00:00
const uint8_t * optionData = packet->GetOptionData ();
if (flags & PACKET_FLAG_SYNCHRONIZE)
LogPrint (eLogDebug, "Synchronize");
2014-02-02 03:20:41 +00:00
2014-08-06 23:19:59 +00:00
if (flags & PACKET_FLAG_DELAY_REQUESTED)
{
optionData += 2;
}
2014-02-02 03:20:41 +00:00
if (flags & PACKET_FLAG_FROM_INCLUDED)
{
2014-08-22 23:47:29 +00:00
optionData += m_RemoteIdentity.FromBuffer (optionData, packet->GetOptionSize ());
LogPrint (eLogInfo, "From identity ", m_RemoteIdentity.GetIdentHash ().ToBase64 ());
2014-08-06 23:19:59 +00:00
if (!m_RemoteLeaseSet)
LogPrint (eLogDebug, "Incoming stream from ", m_RemoteIdentity.GetIdentHash ().ToBase64 ());
2014-02-02 03:20:41 +00:00
}
2014-08-04 22:22:54 +00:00
if (flags & PACKET_FLAG_MAX_PACKET_SIZE_INCLUDED)
{
uint16_t maxPacketSize = bufbe16toh (optionData);
LogPrint (eLogDebug, "Max packet size ", maxPacketSize);
2014-08-04 22:22:54 +00:00
optionData += 2;
}
if (flags & PACKET_FLAG_SIGNATURE_INCLUDED)
{
LogPrint (eLogDebug, "Signature");
2014-08-22 23:47:29 +00:00
uint8_t signature[256];
auto signatureLen = m_RemoteIdentity.GetSignatureLen ();
memcpy (signature, optionData, signatureLen);
memset (const_cast<uint8_t *>(optionData), 0, signatureLen);
if (!m_RemoteIdentity.Verify (packet->GetBuffer (), packet->GetLength (), signature))
{
LogPrint (eLogError, "Signature verification failed");
2014-08-22 23:47:29 +00:00
Close ();
flags |= PACKET_FLAG_CLOSE;
}
memcpy (const_cast<uint8_t *>(optionData), signature, signatureLen);
optionData += signatureLen;
2014-08-04 22:22:54 +00:00
}
2014-02-02 03:20:41 +00:00
packet->offset = packet->GetPayload () - packet->buf;
if (packet->GetLength () > 0)
2014-03-29 12:11:00 +00:00
{
2014-04-12 01:13:52 +00:00
m_ReceiveQueue.push (packet);
2014-03-29 12:11:00 +00:00
m_ReceiveTimer.cancel ();
}
2014-02-02 03:20:41 +00:00
else
delete packet;
m_LastReceivedSequenceNumber = receivedSeqn;
2014-01-12 20:57:10 +00:00
if (flags & PACKET_FLAG_CLOSE)
{
LogPrint (eLogInfo, "Closed");
Close ();
2014-01-12 20:57:10 +00:00
m_IsOpen = false;
2014-10-09 17:03:59 +00:00
m_IsReset = true;
2014-01-15 00:00:12 +00:00
}
2013-12-13 02:36:24 +00:00
}
2014-08-10 22:27:23 +00:00
void Stream::ProcessAck (Packet * packet)
{
2015-01-25 22:43:34 +00:00
bool acknowledged = false;
2015-02-03 18:46:44 +00:00
auto ts = i2p::util::GetMillisecondsSinceEpoch ();
2014-08-10 22:27:23 +00:00
uint32_t ackThrough = packet->GetAckThrough ();
2014-08-12 19:57:23 +00:00
int nackCount = packet->GetNACKCount ();
2014-08-10 22:27:23 +00:00
for (auto it = m_SentPackets.begin (); it != m_SentPackets.end ();)
{
2014-08-12 19:57:23 +00:00
auto seqn = (*it)->GetSeqn ();
if (seqn <= ackThrough)
2014-08-10 22:27:23 +00:00
{
2014-08-12 19:57:23 +00:00
if (nackCount > 0)
{
bool nacked = false;
for (int i = 0; i < nackCount; i++)
if (seqn == packet->GetNACK (i))
{
nacked = true;
break;
}
if (nacked)
{
LogPrint (eLogDebug, "Packet ", seqn, " NACK");
2014-08-14 01:27:12 +00:00
it++;
2014-08-12 19:57:23 +00:00
continue;
}
}
2014-08-10 22:27:23 +00:00
auto sentPacket = *it;
2015-02-03 18:46:44 +00:00
uint64_t rtt = ts - sentPacket->sendTime;
m_RTT = (m_RTT*seqn + rtt)/(seqn + 1);
LogPrint (eLogDebug, "Packet ", seqn, " acknowledged rtt=", rtt);
2014-08-10 22:27:23 +00:00
m_SentPackets.erase (it++);
delete sentPacket;
2015-01-25 22:43:34 +00:00
acknowledged = true;
2015-01-29 20:34:43 +00:00
if (m_WindowSize < WINDOW_SIZE)
m_WindowSize++; // slow start
else
{
// linear growth
2015-02-03 18:46:44 +00:00
if (ts > m_LastWindowSizeIncreaseTime + m_RTT)
2015-01-29 20:34:43 +00:00
{
m_WindowSize++;
if (m_WindowSize > MAX_WINDOW_SIZE) m_WindowSize = MAX_WINDOW_SIZE;
m_LastWindowSizeIncreaseTime = ts;
}
}
2014-08-10 22:27:23 +00:00
}
else
break;
}
if (m_SentPackets.empty ())
m_ResendTimer.cancel ();
2015-01-25 22:43:34 +00:00
if (acknowledged)
SendBuffer ();
2014-08-10 22:27:23 +00:00
}
2014-10-02 01:18:41 +00:00
size_t Stream::Send (const uint8_t * buf, size_t len)
2014-01-01 23:19:03 +00:00
{
2015-01-25 21:18:26 +00:00
if (len > 0 && buf)
{
std::unique_lock<std::mutex> l(m_SendBufferMutex);
m_SendBuffer.clear ();
m_SendBuffer.write ((const char *)buf, len);
}
m_Service.post (std::bind (&Stream::SendBuffer, shared_from_this ()));
2014-04-18 23:27:39 +00:00
return len;
2014-01-01 23:19:03 +00:00
}
2014-04-18 23:27:39 +00:00
2015-01-25 21:18:26 +00:00
void Stream::SendBuffer ()
2015-01-25 22:43:34 +00:00
{
2015-01-29 20:34:43 +00:00
int numMsgs = m_WindowSize - m_SentPackets.size ();
2015-01-25 22:43:34 +00:00
if (numMsgs <= 0) return; // window is full
2015-01-25 21:18:26 +00:00
bool isNoAck = m_LastReceivedSequenceNumber < 0; // first packet
std::vector<Packet *> packets;
{
std::unique_lock<std::mutex> l(m_SendBufferMutex);
2015-01-25 22:43:34 +00:00
while (!m_IsOpen || (IsEstablished () && !m_SendBuffer.eof () && numMsgs > 0))
2015-01-25 21:18:26 +00:00
{
Packet * p = new Packet ();
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)
htobe32buf (packet + size, m_LastReceivedSequenceNumber);
else
htobuf32 (packet + size, 0);
size += 4; // ack Through
packet[size] = 0;
size++; // NACK count
packet[size] = RESEND_TIMEOUT;
size++; // resend delay
if (!m_IsOpen)
{
// initial packet
m_IsOpen = true; m_IsReset = false;
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;
htobe16buf (packet + size, flags);
size += 2; // flags
size_t identityLen = m_LocalDestination.GetOwner ().GetIdentity ().GetFullLen ();
size_t signatureLen = m_LocalDestination.GetOwner ().GetIdentity ().GetSignatureLen ();
htobe16buf (packet + size, identityLen + signatureLen + 2); // identity + signature + packet size
size += 2; // options size
m_LocalDestination.GetOwner ().GetIdentity ().ToBuffer (packet + size, identityLen);
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
m_SendBuffer.read ((char *)(packet + size), STREAMING_MTU - size);
size += m_SendBuffer.gcount (); // 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
m_SendBuffer.read((char *)(packet + size), STREAMING_MTU - size);
size += m_SendBuffer.gcount (); // payload
}
p->len = size;
packets.push_back (p);
2015-01-25 22:43:34 +00:00
numMsgs--;
2015-01-25 21:18:26 +00:00
}
}
if (packets.size () > 0)
{
m_IsAckSendScheduled = false;
m_AckSendTimer.cancel ();
bool isEmpty = m_SentPackets.empty ();
2015-02-03 18:46:44 +00:00
auto ts = i2p::util::GetMillisecondsSinceEpoch ();
2015-01-25 21:18:26 +00:00
for (auto it: packets)
2015-02-03 18:46:44 +00:00
{
it->sendTime = ts;
2015-01-25 21:18:26 +00:00
m_SentPackets.insert (it);
2015-02-03 18:46:44 +00:00
}
2015-01-25 21:18:26 +00:00
SendPackets (packets);
if (isEmpty)
ScheduleResend ();
}
}
2014-01-12 20:57:10 +00:00
2014-08-08 02:03:25 +00:00
void Stream::SendQuickAck ()
2014-01-11 03:23:17 +00:00
{
2014-11-29 21:21:19 +00:00
int32_t lastReceivedSeqn = m_LastReceivedSequenceNumber;
if (!m_SavedPackets.empty ())
{
int32_t seqn = (*m_SavedPackets.rbegin ())->GetSeqn ();
if (seqn > lastReceivedSeqn) lastReceivedSeqn = seqn;
}
if (lastReceivedSeqn < 0)
{
LogPrint (eLogError, "No packets have been received yet");
return;
}
2014-08-12 20:52:04 +00:00
Packet p;
uint8_t * packet = p.GetBuffer ();
2014-01-11 03:23:17 +00:00
size_t size = 0;
htobe32buf (packet + size, m_SendStreamID);
2014-01-11 03:23:17 +00:00
size += 4; // sendStreamID
htobe32buf (packet + size, m_RecvStreamID);
2014-01-11 03:23:17 +00:00
size += 4; // receiveStreamID
htobuf32 (packet + size, 0); // this is plain Ack message
2014-01-11 03:23:17 +00:00
size += 4; // sequenceNum
htobe32buf (packet + size, lastReceivedSeqn);
2014-01-11 03:23:17 +00:00
size += 4; // ack Through
2015-01-01 14:53:30 +00:00
uint8_t numNacks = 0;
2014-11-29 21:21:19 +00:00
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 ();
2015-01-26 03:01:09 +00:00
if (numNacks + (seqn - nextSeqn) >= 256)
{
LogPrint (eLogError, "Number of NACKs exceeds 256");
htobe32buf (packet + 12, nextSeqn); // change ack Through
break;
}
2014-11-29 21:21:19 +00:00
for (uint32_t i = nextSeqn; i < seqn; i++)
{
htobe32buf (nacks, i);
2014-11-29 21:21:19 +00:00
nacks += 4;
numNacks++;
}
nextSeqn = seqn + 1;
}
packet[size] = numNacks;
size++; // NACK count
2014-11-30 00:15:41 +00:00
size += numNacks*4; // NACKs
2014-11-29 21:21:19 +00:00
}
else
{
// No NACKs
packet[size] = 0;
size++; // NACK count
}
2014-01-11 03:23:17 +00:00
size++; // resend delay
htobuf16 (packet + size, 0); // nof flags set
2014-01-11 03:23:17 +00:00
size += 2; // flags
htobuf16 (packet + size, 0); // no options
2014-01-11 03:23:17 +00:00
size += 2; // options size
2014-08-12 20:52:04 +00:00
p.len = size;
2014-08-12 22:42:53 +00:00
SendPackets (std::vector<Packet *> { &p });
2015-01-01 14:53:30 +00:00
LogPrint ("Quick Ack sent. ", (int)numNacks, " NACKs");
2014-01-11 03:23:17 +00:00
}
2014-01-12 20:57:10 +00:00
void Stream::Close ()
{
if (m_IsOpen)
{
m_IsOpen = false;
2014-08-06 20:00:34 +00:00
Packet * p = new Packet ();
uint8_t * packet = p->GetBuffer ();
2014-01-12 20:57:10 +00:00
size_t size = 0;
htobe32buf (packet + size, m_SendStreamID);
2014-01-12 20:57:10 +00:00
size += 4; // sendStreamID
htobe32buf (packet + size, m_RecvStreamID);
2014-01-12 20:57:10 +00:00
size += 4; // receiveStreamID
htobe32buf (packet + size, m_SequenceNumber++);
2014-01-12 20:57:10 +00:00
size += 4; // sequenceNum
htobe32buf (packet + size, m_LastReceivedSequenceNumber);
2014-01-12 20:57:10 +00:00
size += 4; // ack Through
packet[size] = 0;
size++; // NACK count
size++; // resend delay
htobe16buf (packet + size, PACKET_FLAG_CLOSE | PACKET_FLAG_SIGNATURE_INCLUDED);
2014-01-12 20:57:10 +00:00
size += 2; // flags
size_t signatureLen = m_LocalDestination.GetOwner ().GetIdentity ().GetSignatureLen ();
htobe16buf (packet + size, signatureLen); // signature only
2014-01-12 20:57:10 +00:00
size += 2; // options size
uint8_t * signature = packet + size;
2014-08-24 14:36:59 +00:00
memset (packet + size, 0, signatureLen);
size += signatureLen; // signature
m_LocalDestination.GetOwner ().Sign (packet, size, signature);
2014-03-24 23:27:20 +00:00
2014-08-06 20:00:34 +00:00
p->len = size;
m_Service.post (std::bind (&Stream::SendPacket, shared_from_this (), p));
2014-08-06 20:00:34 +00:00
LogPrint ("FIN sent");
2014-01-12 20:57:10 +00:00
}
2015-01-23 00:02:28 +00:00
m_ReceiveTimer.cancel ();
m_LocalDestination.DeleteStream (shared_from_this ());
2014-01-12 20:57:10 +00:00
}
2014-03-26 19:06:27 +00:00
size_t Stream::ConcatenatePackets (uint8_t * buf, size_t len)
{
2014-01-11 01:21:38 +00:00
size_t pos = 0;
2014-04-12 01:13:52 +00:00
while (pos < len && !m_ReceiveQueue.empty ())
2014-01-11 01:21:38 +00:00
{
2014-04-12 01:13:52 +00:00
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 ())
2014-01-11 01:21:38 +00:00
{
2014-04-12 01:13:52 +00:00
m_ReceiveQueue.pop ();
delete packet;
}
2014-01-11 01:21:38 +00:00
}
return pos;
2014-03-26 19:06:27 +00:00
}
2014-03-18 02:55:02 +00:00
2014-03-24 23:27:20 +00:00
bool Stream::SendPacket (Packet * packet)
{
if (packet)
{
if (m_IsAckSendScheduled)
{
m_IsAckSendScheduled = false;
m_AckSendTimer.cancel ();
}
2014-08-12 20:52:04 +00:00
SendPackets (std::vector<Packet *> { packet });
2014-10-04 20:27:21 +00:00
if (m_IsOpen)
{
bool isEmpty = m_SentPackets.empty ();
m_SentPackets.insert (packet);
if (isEmpty)
ScheduleResend ();
}
2014-10-18 19:50:34 +00:00
else
delete packet;
2014-10-04 20:27:21 +00:00
return true;
2014-03-24 23:27:20 +00:00
}
else
return false;
}
2014-08-12 20:35:35 +00:00
void Stream::SendPackets (const std::vector<Packet *>& packets)
{
if (!m_RemoteLeaseSet)
{
UpdateCurrentRemoteLease ();
if (!m_RemoteLeaseSet)
{
LogPrint ("Can't send packets. Missing remote LeaseSet");
return;
}
}
if (!m_CurrentOutboundTunnel || !m_CurrentOutboundTunnel->IsEstablished ())
m_CurrentOutboundTunnel = m_LocalDestination.GetOwner ().GetTunnelPool ()->GetNextOutboundTunnel ();
if (!m_CurrentOutboundTunnel)
{
LogPrint ("No outbound tunnels in the pool");
return;
}
2014-08-12 20:35:35 +00:00
auto ts = i2p::util::GetMillisecondsSinceEpoch ();
if (ts >= m_CurrentRemoteLease.endDate)
UpdateCurrentRemoteLease ();
if (ts < m_CurrentRemoteLease.endDate)
{
std::vector<i2p::tunnel::TunnelMessageBlock> msgs;
for (auto it: packets)
{
2014-10-23 01:36:11 +00:00
auto msg = m_RoutingSession->WrapSingleMessage (CreateDataMessage (it->GetBuffer (), it->GetLength ()));
msgs.push_back (i2p::tunnel::TunnelMessageBlock
{
i2p::tunnel::eDeliveryTypeTunnel,
m_CurrentRemoteLease.tunnelGateway, m_CurrentRemoteLease.tunnelID,
msg
});
2014-10-13 21:03:27 +00:00
m_NumSentBytes += it->GetLength ();
}
m_CurrentOutboundTunnel->SendTunnelDataMsg (msgs);
2014-08-12 20:35:35 +00:00
}
else
LogPrint ("All leases are expired");
2014-08-12 20:35:35 +00:00
}
2014-08-10 22:27:23 +00:00
void Stream::ScheduleResend ()
{
m_ResendTimer.cancel ();
m_ResendTimer.expires_from_now (boost::posix_time::seconds(RESEND_TIMEOUT));
2014-11-23 22:00:45 +00:00
m_ResendTimer.async_wait (std::bind (&Stream::HandleResendTimer,
shared_from_this (), std::placeholders::_1));
2014-08-10 22:27:23 +00:00
}
void Stream::HandleResendTimer (const boost::system::error_code& ecode)
{
if (ecode != boost::asio::error::operation_aborted)
{
2015-02-03 18:46:44 +00:00
auto ts = i2p::util::GetMillisecondsSinceEpoch ();
bool congesion = false, first = true;
2014-08-12 20:35:35 +00:00
std::vector<Packet *> packets;
2014-08-10 22:27:23 +00:00
for (auto it : m_SentPackets)
2014-08-11 23:32:07 +00:00
{
it->numResendAttempts++;
if (first && it->numResendAttempts == 1) // detect congesion at first attempt of first packet only
congesion = true;
first = false;
2014-08-11 23:32:07 +00:00
if (it->numResendAttempts <= MAX_NUM_RESEND_ATTEMPTS)
2015-02-03 18:46:44 +00:00
{
it->sendTime = ts;
2014-08-12 20:35:35 +00:00
packets.push_back (it);
2015-02-03 18:46:44 +00:00
}
2014-08-11 23:32:07 +00:00
else
{
2015-01-25 22:43:34 +00:00
LogPrint (eLogWarning, "Packet ", it->GetSeqn (), " was not ACKed after ", MAX_NUM_RESEND_ATTEMPTS, " attempts. Terminate");
2014-12-02 02:26:51 +00:00
m_IsOpen = false;
2014-10-09 17:03:59 +00:00
m_IsReset = true;
m_ReceiveTimer.cancel ();
2014-08-11 23:32:07 +00:00
return;
}
}
2014-08-12 20:35:35 +00:00
if (packets.size () > 0)
2014-08-15 01:38:46 +00:00
{
if (congesion)
{
// congesion avoidance
m_WindowSize /= 2;
if (m_WindowSize < MIN_WINDOW_SIZE) m_WindowSize = MIN_WINDOW_SIZE;
}
else
{
// congesion avoidance didn't help
m_CurrentOutboundTunnel = nullptr; // pick another outbound tunnel
UpdateCurrentRemoteLease (); // pick another lease
}
2014-08-12 20:35:35 +00:00
SendPackets (packets);
2014-08-15 01:38:46 +00:00
}
2014-08-10 22:27:23 +00:00
ScheduleResend ();
}
}
2014-10-10 15:53:27 +00:00
void Stream::HandleAckSendTimer (const boost::system::error_code& ecode)
{
if (m_IsAckSendScheduled)
2014-10-10 15:53:27 +00:00
{
if (m_IsOpen)
SendQuickAck ();
m_IsAckSendScheduled = false;
}
2014-10-10 15:53:27 +00:00
}
2014-03-23 13:25:16 +00:00
void Stream::UpdateCurrentRemoteLease ()
{
2014-08-06 15:09:06 +00:00
if (!m_RemoteLeaseSet)
{
m_RemoteLeaseSet = m_LocalDestination.GetOwner ().FindLeaseSet (m_RemoteIdentity.GetIdentHash ());
2014-08-30 02:10:00 +00:00
if (!m_RemoteLeaseSet)
2014-08-22 23:47:29 +00:00
LogPrint ("LeaseSet ", m_RemoteIdentity.GetIdentHash ().ToBase64 (), " not found");
2014-08-06 15:09:06 +00:00
}
2014-08-01 18:54:14 +00:00
if (m_RemoteLeaseSet)
{
2014-08-30 02:10:00 +00:00
if (!m_RoutingSession)
2015-01-29 02:37:08 +00:00
m_RoutingSession = m_LocalDestination.GetOwner ().GetRoutingSession (m_RemoteLeaseSet, 32);
2014-08-01 18:54:14 +00:00
auto leases = m_RemoteLeaseSet->GetNonExpiredLeases ();
if (!leases.empty ())
{
uint32_t i = i2p::context.GetRandomNumberGenerator ().GenerateWord32 (0, leases.size () - 1);
m_CurrentRemoteLease = leases[i];
}
else
2014-10-17 01:09:59 +00:00
{
m_RemoteLeaseSet = m_LocalDestination.GetOwner ().FindLeaseSet (m_RemoteIdentity.GetIdentHash ()); // re-request expired
2014-08-01 18:54:14 +00:00
m_CurrentRemoteLease.endDate = 0;
2014-10-17 01:09:59 +00:00
}
2014-08-01 18:54:14 +00:00
}
2014-03-23 13:25:16 +00:00
else
m_CurrentRemoteLease.endDate = 0;
}
2014-10-23 01:36:11 +00:00
I2NPMessage * Stream::CreateDataMessage (const uint8_t * payload, size_t len)
{
I2NPMessage * msg = NewI2NPShortMessage ();
CryptoPP::Gzip compressor;
if (len <= i2p::stream::COMPRESSION_THRESHOLD_SIZE)
compressor.SetDeflateLevel (CryptoPP::Gzip::MIN_DEFLATE_LEVEL);
else
compressor.SetDeflateLevel (CryptoPP::Gzip::DEFAULT_DEFLATE_LEVEL);
compressor.Put (payload, len);
compressor.MessageEnd();
int size = compressor.MaxRetrievable ();
uint8_t * buf = msg->GetPayload ();
htobe32buf (buf, size); // length
2014-10-23 01:36:11 +00:00
buf += 4;
compressor.Get (buf, size);
htobuf16(buf + 4, 0); // source port
htobe16buf (buf + 6, m_Port); // destination port
2014-10-23 01:36:11 +00:00
buf[9] = i2p::client::PROTOCOL_TYPE_STREAMING; // streaming protocol
msg->len += size + 4;
FillI2NPMessageHeader (msg, eI2NPData);
return msg;
}
void StreamingDestination::Start ()
{
}
void StreamingDestination::Stop ()
{
ResetAcceptor ();
{
std::unique_lock<std::mutex> l(m_StreamsMutex);
m_Streams.clear ();
}
}
void StreamingDestination::HandleNextPacket (Packet * packet)
{
uint32_t sendStreamID = packet->GetSendStreamID ();
if (sendStreamID)
{
auto it = m_Streams.find (sendStreamID);
if (it != m_Streams.end ())
it->second->HandleNextPacket (packet);
else
{
LogPrint ("Unknown stream sendStreamID=", sendStreamID);
delete packet;
}
}
else
{
if (packet->IsSYN () && !packet->GetSeqn ()) // new incoming stream
{
auto incomingStream = CreateNewIncomingStream ();
incomingStream->HandleNextPacket (packet);
if (m_Acceptor != nullptr)
m_Acceptor (incomingStream);
else
{
LogPrint ("Acceptor for incoming stream is not set");
DeleteStream (incomingStream);
}
}
else // follow on packet without SYN
{
uint32_t receiveStreamID = packet->GetReceiveStreamID ();
for (auto it: m_Streams)
if (it.second->GetSendStreamID () == receiveStreamID)
{
// found
it.second->HandleNextPacket (packet);
return;
}
// TODO: should queue it up
LogPrint ("Unknown stream receiveStreamID=", receiveStreamID);
delete packet;
}
}
}
2015-01-27 16:27:58 +00:00
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[s->GetRecvStreamID ()] = s;
return s;
}
2014-11-23 16:33:58 +00:00
std::shared_ptr<Stream> StreamingDestination::CreateNewIncomingStream ()
{
auto s = std::make_shared<Stream> (m_Owner.GetService (), *this);
std::unique_lock<std::mutex> l(m_StreamsMutex);
m_Streams[s->GetRecvStreamID ()] = s;
return s;
}
2014-11-23 16:33:58 +00:00
void StreamingDestination::DeleteStream (std::shared_ptr<Stream> stream)
{
if (stream)
{
std::unique_lock<std::mutex> l(m_StreamsMutex);
auto it = m_Streams.find (stream->GetRecvStreamID ());
if (it != m_Streams.end ())
m_Streams.erase (it);
}
2014-08-26 18:56:00 +00:00
}
void StreamingDestination::HandleDataMessagePayload (const uint8_t * buf, size_t len)
{
// unzip it
CryptoPP::Gunzip decompressor;
decompressor.Put (buf, len);
decompressor.MessageEnd();
Packet * uncompressed = new Packet;
uncompressed->offset = 0;
uncompressed->len = decompressor.MaxRetrievable ();
if (uncompressed->len <= MAX_PACKET_SIZE)
{
decompressor.Get (uncompressed->buf, uncompressed->len);
HandleNextPacket (uncompressed);
}
else
{
LogPrint ("Received packet size ", uncompressed->len, " exceeds max packet size. Skipped");
delete uncompressed;
}
}
2013-12-13 02:36:24 +00:00
}
}