Browse Source

save out-of-sequence fragments

pull/93/head
orignal 11 years ago
parent
commit
999abd517c
  1. 1
      SSU.h
  2. 79
      SSUData.cpp
  3. 37
      SSUData.h

1
SSU.h

@ -31,7 +31,6 @@ namespace ssu
}; };
#pragma pack() #pragma pack()
const size_t SSU_MTU = 1484;
const int SSU_CONNECT_TIMEOUT = 5; // 5 seconds const int SSU_CONNECT_TIMEOUT = 5; // 5 seconds
const int SSU_TERMINATION_TIMEOUT = 330; // 5.5 minutes const int SSU_TERMINATION_TIMEOUT = 330; // 5.5 minutes

79
SSUData.cpp

@ -82,55 +82,73 @@ namespace ssu
bool isLast = fragmentInfo & 0x010000; // bit 16 bool isLast = fragmentInfo & 0x010000; // bit 16
uint8_t fragmentNum = fragmentInfo >> 17; // bits 23 - 17 uint8_t fragmentNum = fragmentInfo >> 17; // bits 23 - 17
LogPrint ("SSU data fragment ", (int)fragmentNum, " of message ", msgID, " size=", (int)fragmentSize, isLast ? " last" : " non-last"); LogPrint ("SSU data fragment ", (int)fragmentNum, " of message ", msgID, " size=", (int)fragmentSize, isLast ? " last" : " non-last");
// find message with msgID
I2NPMessage * msg = nullptr; I2NPMessage * msg = nullptr;
if (fragmentNum > 0) // follow-up fragment IncompleteMessage * incompleteMessage = nullptr;
{
auto it = m_IncomleteMessages.find (msgID); auto it = m_IncomleteMessages.find (msgID);
if (it != m_IncomleteMessages.end ()) if (it != m_IncomleteMessages.end ())
{ {
if (fragmentNum == it->second->nextFragmentNum) // message exists
incompleteMessage = it->second;
msg = incompleteMessage->msg;
}
else
{
// create new message
msg = NewI2NPMessage ();
msg->len -= sizeof (I2NPHeaderShort);
incompleteMessage = new IncompleteMessage (msg);
m_IncomleteMessages[msgID] = incompleteMessage;
}
// handle current fragment
if (fragmentNum == incompleteMessage->nextFragmentNum)
{ {
// expected fragment // expected fragment
msg = it->second->msg;
memcpy (msg->buf + msg->len, buf, fragmentSize); memcpy (msg->buf + msg->len, buf, fragmentSize);
msg->len += fragmentSize; msg->len += fragmentSize;
it->second->nextFragmentNum++; incompleteMessage->nextFragmentNum++;
if (!isLast && !incompleteMessage->savedFragments.empty ())
{
// try saved fragments
for (auto it1 = incompleteMessage->savedFragments.begin (); it1 != incompleteMessage->savedFragments.end ();)
{
auto savedFragment = *it1;
if (savedFragment->fragmentNum == incompleteMessage->nextFragmentNum)
{
memcpy (msg->buf + msg->len, savedFragment->buf, savedFragment->len);
msg->len += savedFragment->len;
isLast = savedFragment->isLast;
incompleteMessage->nextFragmentNum++;
incompleteMessage->savedFragments.erase (it1++);
delete savedFragment;
} }
else if (fragmentNum < it->second->nextFragmentNum)
// duplicate fragment
LogPrint ("Duplicate fragment ", (int)fragmentNum, " of message ", msgID, ". Ignored");
else else
{ break;
// missing fragment
LogPrint ("Missing fragments from ", it->second->nextFragmentNum, " to ", fragmentNum - 1, " of message ", msgID);
//TODO
} }
if (isLast)
{
if (!msg)
DeleteI2NPMessage (it->second->msg);
delete it->second;
m_IncomleteMessages.erase (it);
} }
} }
else else
// TODO:
LogPrint ("Unexpected follow-on fragment ", (int)fragmentNum, " of message ", msgID);
}
else // first fragment
{ {
msg = NewI2NPMessage (); if (fragmentNum < incompleteMessage->nextFragmentNum)
memcpy (msg->GetSSUHeader (), buf, fragmentSize); // duplicate fragment
msg->len += fragmentSize - sizeof (I2NPHeaderShort); LogPrint ("Duplicate fragment ", (int)fragmentNum, " of message ", msgID, ". Ignored");
else
{
// missing fragment
LogPrint ("Missing fragments from ", (int)incompleteMessage->nextFragmentNum, " to ", fragmentNum - 1, " of message ", msgID);
incompleteMessage->savedFragments.insert (new Fragment (fragmentNum, buf, fragmentSize, isLast));
}
isLast = false;
} }
if (msg)
{
if (!fragmentNum && !isLast)
m_IncomleteMessages[msgID] = new IncompleteMessage (msg);
if (isLast) if (isLast)
{ {
// delete incomplete message
delete incompleteMessage;
m_IncomleteMessages.erase (msgID);
// process message
SendMsgAck (msgID); SendMsgAck (msgID);
msg->FromSSU (msgID); msg->FromSSU (msgID);
if (m_Session.GetState () == eSessionStateEstablished) if (m_Session.GetState () == eSessionStateEstablished)
@ -148,7 +166,6 @@ namespace ssu
DeleteI2NPMessage (msg); DeleteI2NPMessage (msg);
} }
} }
}
buf += fragmentSize; buf += fragmentSize;
} }
} }

37
SSUData.h

@ -2,8 +2,10 @@
#define SSU_DATA_H__ #define SSU_DATA_H__
#include <inttypes.h> #include <inttypes.h>
#include <string.h>
#include <map> #include <map>
#include <vector> #include <vector>
#include <set>
#include "I2NPProtocol.h" #include "I2NPProtocol.h"
namespace i2p namespace i2p
@ -11,6 +13,7 @@ namespace i2p
namespace ssu namespace ssu
{ {
const size_t SSU_MTU = 1484;
// data flags // data flags
const uint8_t DATA_FLAG_EXTENDED_DATA_INCLUDED = 0x02; const uint8_t DATA_FLAG_EXTENDED_DATA_INCLUDED = 0x02;
const uint8_t DATA_FLAG_WANT_REPLY = 0x04; const uint8_t DATA_FLAG_WANT_REPLY = 0x04;
@ -19,6 +22,34 @@ namespace ssu
const uint8_t DATA_FLAG_ACK_BITFIELDS_INCLUDED = 0x40; const uint8_t DATA_FLAG_ACK_BITFIELDS_INCLUDED = 0x40;
const uint8_t DATA_FLAG_EXPLICIT_ACKS_INCLUDED = 0x80; const uint8_t DATA_FLAG_EXPLICIT_ACKS_INCLUDED = 0x80;
struct Fragment
{
int fragmentNum, len;
bool isLast;
uint8_t buf[SSU_MTU];
Fragment (int n, const uint8_t * b, int l, bool last):
fragmentNum (n), len (l), isLast (last) { memcpy (buf, b, len); };
};
struct FragmentCmp
{
bool operator() (const Fragment * f1, const Fragment * f2) const
{
return f1->fragmentNum < f2->fragmentNum;
};
};
struct IncompleteMessage
{
I2NPMessage * msg;
int nextFragmentNum;
std::set<Fragment *, FragmentCmp> savedFragments;
IncompleteMessage (I2NPMessage * m): msg (m), nextFragmentNum (0) {};
~IncompleteMessage () { for (auto it: savedFragments) { delete it; }; };
};
class SSUSession; class SSUSession;
class SSUData class SSUData
{ {
@ -37,13 +68,7 @@ namespace ssu
private: private:
struct IncompleteMessage
{
I2NPMessage * msg;
uint8_t nextFragmentNum;
IncompleteMessage (I2NPMessage * m): msg (m), nextFragmentNum (1) {};
};
SSUSession& m_Session; SSUSession& m_Session;
std::map<uint32_t, IncompleteMessage *> m_IncomleteMessages; std::map<uint32_t, IncompleteMessage *> m_IncomleteMessages;

Loading…
Cancel
Save