diff --git a/libi2pd/I2NPProtocol.h b/libi2pd/I2NPProtocol.h index b1e2d170..5d5754c8 100644 --- a/libi2pd/I2NPProtocol.h +++ b/libi2pd/I2NPProtocol.h @@ -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 from; std::function 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; }; } }; diff --git a/libi2pd/SSU2Session.cpp b/libi2pd/SSU2Session.cpp index 2c499753..a51b54c4 100644 --- a/libi2pd/SSU2Session.cpp +++ b/libi2pd/SSU2Session.cpp @@ -353,25 +353,32 @@ namespace transport void SSU2Session::PostI2NPMessages (std::vector > 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 (); diff --git a/libi2pd/SSU2Session.h b/libi2pd/SSU2Session.h index ed17a6f4..347c3ef9 100644 --- a/libi2pd/SSU2Session.h +++ b/libi2pd/SSU2Session.h @@ -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