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

add expiration for messages in SSU2 send queue

Browse Source
This commit is contained in:
Vort 2024-03-09 19:05:02 +02:00
parent 66d0b7aec4
commit 2d06c0cbe6
3 changed files with 28 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
libi2pd/I2NPProtocol.h
View File

@ -152,6 +152,7 @@ namespace tunnel
const size_t I2NP_MAX_MESSAGE_SIZE = 62708;
const size_t I2NP_MAX_SHORT_MESSAGE_SIZE = 4096;
const size_t I2NP_MAX_MEDIUM_MESSAGE_SIZE = 16384;
const unsigned int I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT = 2000000; // in microseconds
const unsigned int I2NP_MESSAGE_EXPIRATION_TIMEOUT = 8000; // in milliseconds (as initial RTT)
const unsigned int I2NP_MESSAGE_CLOCK_SKEW = 60*1000; // 1 minute in milliseconds
@ -161,9 +162,10 @@ namespace tunnel
size_t len, offset, maxLen;
std::shared_ptr<i2p::tunnel::InboundTunnel> from;
std::function<void ()> onDrop;
I2NPMessage (): buf (nullptr),len (I2NP_HEADER_SIZE + 2),
offset(2), maxLen (0), from (nullptr) {}; // reserve 2 bytes for NTCP header
uint64_t localExpiration; // monotonic microseconds
I2NPMessage (): buf (nullptr), len (I2NP_HEADER_SIZE + 2),
offset(2), maxLen (0), from (nullptr), localExpiration(0) {}; // reserve 2 bytes for NTCP header
// header accessors
uint8_t * GetHeader () { return GetBuffer (); };
@ -173,6 +175,7 @@ namespace tunnel
void SetMsgID (uint32_t msgID) { htobe32buf (GetHeader () + I2NP_HEADER_MSGID_OFFSET, msgID); };
uint32_t GetMsgID () const { return bufbe32toh (GetHeader () + I2NP_HEADER_MSGID_OFFSET); };
void SetExpiration (uint64_t expiration) { htobe64buf (GetHeader () + I2NP_HEADER_EXPIRATION_OFFSET, expiration); };
void SetLocalExpiration (uint64_t expiration) { localExpiration = expiration; };
uint64_t GetExpiration () const { return bufbe64toh (GetHeader () + I2NP_HEADER_EXPIRATION_OFFSET); };
void SetSize (uint16_t size) { htobe16buf (GetHeader () + I2NP_HEADER_SIZE_OFFSET, size); };
uint16_t GetSize () const { return bufbe16toh (GetHeader () + I2NP_HEADER_SIZE_OFFSET); };
@ -264,6 +267,8 @@ namespace tunnel
void RenewI2NPMessageHeader ();
bool IsExpired () const;
bool IsExpired (uint64_t ts) const; // in milliseconds
bool IsLocalExpired (uint64_t mts) const { return mts > localExpiration; }; // monotonic microseconds
bool IsLocalSemiExpired (uint64_t mts) const { return mts > localExpiration - I2NP_MESSAGE_EXPIRATION_TIMEOUT / 2; }; // monotonic microseconds
void Drop () { if (onDrop) { onDrop (); onDrop = nullptr; }; }
};

32
libi2pd/SSU2Session.cpp
View File

@ -353,25 +353,32 @@ namespace transport
void SSU2Session::PostI2NPMessages (std::vector<std::shared_ptr<I2NPMessage> > msgs)
{
if (m_State == eSSU2SessionStateTerminated) return;
bool isSemiFull = m_SendQueue.size () > SSU2_MAX_OUTGOING_QUEUE_SIZE/2;
uint64_t mts = i2p::util::GetMonotonicMicroseconds ();
uint64_t localExpiration = mts + I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT;
bool isSemiFull = false;
if (m_SendQueue.size ())
{
isSemiFull = m_SendQueue.front ()->IsLocalSemiExpired (mts);
if (isSemiFull)
{
LogPrint (eLogWarning, "SSU2: Outgoing messages queue to ",
GetIdentHashBase64 (), " is semi-full (", m_SendQueue.size (), ")");
}
}
for (auto it: msgs)
{
if (isSemiFull && it->onDrop)
it->Drop (); // drop earlier because we can handle it
else
{
it->SetLocalExpiration (localExpiration);
m_SendQueue.push_back (std::move (it));
}
}
SendQueue ();
if (m_SendQueue.size () > 0) // windows is full
{
if (m_SendQueue.size () <= SSU2_MAX_OUTGOING_QUEUE_SIZE)
Resend (i2p::util::GetMillisecondsSinceEpoch ());
else
{
LogPrint (eLogWarning, "SSU2: Outgoing messages queue size to ",
GetIdentHashBase64(), " exceeds ", SSU2_MAX_OUTGOING_QUEUE_SIZE);
RequestTermination (eSSU2TerminationReasonTimeout);
}
}
Resend (i2p::util::GetMillisecondsSinceEpoch ());
SetSendQueueSize (m_SendQueue.size ());
}
@ -380,6 +387,7 @@ namespace transport
if (!m_SendQueue.empty () && m_SentPackets.size () <= m_WindowSize)
{
auto ts = i2p::util::GetMillisecondsSinceEpoch ();
uint64_t mts = i2p::util::GetMonotonicMicroseconds ();
auto packet = m_Server.GetSentPacketsPool ().AcquireShared ();
size_t ackBlockSize = CreateAckBlock (packet->payload, m_MaxPayloadSize);
bool ackBlockSent = false;
@ -387,7 +395,7 @@ namespace transport
while (!m_SendQueue.empty () && m_SentPackets.size () <= m_WindowSize)
{
auto msg = m_SendQueue.front ();
if (!msg || msg->IsExpired (ts))
if (!msg || msg->IsExpired (ts) || msg->IsLocalExpired (mts))
{
// drop null or expired message
if (msg) msg->Drop ();

1
libi2pd/SSU2Session.h
View File

@ -47,7 +47,6 @@ namespace transport
const size_t SSU2_MIN_RTO = 100; // in milliseconds
const size_t SSU2_MAX_RTO = 2500; // in milliseconds
const float SSU2_kAPPA = 1.8;
const size_t SSU2_MAX_OUTGOING_QUEUE_SIZE = 500; // in messages
const int SSU2_MAX_NUM_ACNT = 255; // acnt, acks or nacks
const int SSU2_MAX_NUM_ACK_PACKETS = 511; // ackthrough + acnt + 1 range
const int SSU2_MAX_NUM_ACK_RANGES = 32; // to send