mirrors/i2pd
2
0
Fork 0
mirror of https://github.com/PurpleI2P/i2pd synced 2024-11-10 00:00:29 +03:00
Code Issues Packages Projects Releases Wiki Activity

queue up out of sequence packets

Browse Source
This commit is contained in:
orignal 2016-02-08 14:42:20 -05:00
parent e90baf3ca6
commit 3d19e92059
2 changed files with 49 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
Streaming.cpp
View File

@ -787,6 +787,12 @@ namespace stream
StreamingDestination::~StreamingDestination ()
{
for (auto it: m_SavedPackets)
{
for (auto it1: it.second) delete it1;
it.second.clear ();
}
m_SavedPackets.clear ();
}
void StreamingDestination::Start ()
@ -822,7 +828,20 @@ namespace stream
if (packet->IsSYN () && !packet->GetSeqn ()) // new incoming stream
{
auto incomingStream = CreateNewIncomingStream ();
incomingStream->HandleNextPacket (packet);
uint32_t receiveStreamID = packet->GetReceiveStreamID ();
incomingStream->HandleNextPacket (packet); // SYN
// 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 receiveStreamID=", receiveStreamID);
for (auto it1: it->second)
incomingStream->HandleNextPacket (it1);
m_SavedPackets.erase (it);
}
}
// accept
if (m_Acceptor != nullptr)
m_Acceptor (incomingStream);
else
@ -834,7 +853,7 @@ namespace stream
m_PendingIncomingTimer.cancel ();
m_PendingIncomingTimer.expires_from_now (boost::posix_time::seconds(PENDING_INCOMING_TIMEOUT));
m_PendingIncomingTimer.async_wait (std::bind (&StreamingDestination::HandlePendingIncomingTimer,
this, std::placeholders::_1));
shared_from_this (), std::placeholders::_1));
LogPrint (eLogDebug, "Streaming: Pending incoming stream added");
}
else
@ -854,6 +873,31 @@ namespace stream
it.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.emplace (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) delete it1;
it->second.clear ();
s->m_SavedPackets.erase (it);
}
}
});
}
// TODO: should queue it up
LogPrint (eLogError, "Streaming: Unknown stream receiveStreamID=", receiveStreamID);
delete packet;

5
Streaming.h
View File

@ -189,7 +189,7 @@ namespace stream
SendHandler m_SendHandler;
};
class StreamingDestination
class StreamingDestination: public std::enable_shared_from_this<StreamingDestination>
{
public:
@ -222,10 +222,11 @@ namespace stream
std::shared_ptr<i2p::client::ClientDestination> m_Owner;
uint16_t m_LocalPort;
std::mutex m_StreamsMutex;
std::map<uint32_t, std::shared_ptr<Stream> > m_Streams;
std::map<uint32_t, std::shared_ptr<Stream> > m_Streams; // sendStreamID->stream
Acceptor m_Acceptor;
std::list<std::shared_ptr<Stream> > m_PendingIncomingStreams;
boost::asio::deadline_timer m_PendingIncomingTimer;
std::map<uint32_t, std::list<Packet *> > m_SavedPackets; // receiveStreamID->packets, arrived before SYN
public: