diff --git a/libi2pd/I2NPProtocol.h b/libi2pd/I2NPProtocol.h index f38f79ee..f110100e 100644 --- a/libi2pd/I2NPProtocol.h +++ b/libi2pd/I2NPProtocol.h @@ -152,7 +152,9 @@ 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_LOCAL_EXPIRATION_TIMEOUT_FACTOR = 3; // multiples of RTT + const unsigned int I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_MIN = 200000; // in microseconds + const unsigned int I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_MAX = 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 @@ -162,10 +164,10 @@ namespace tunnel size_t len, offset, maxLen; std::shared_ptr from; std::function onDrop; - uint64_t localExpiration; // monotonic microseconds + uint64_t enqueueTime; // monotonic microseconds I2NPMessage (): buf (nullptr), len (I2NP_HEADER_SIZE + 2), - offset(2), maxLen (0), from (nullptr), localExpiration(0) {}; // reserve 2 bytes for NTCP header + offset(2), maxLen (0), from (nullptr), enqueueTime (0) {}; // reserve 2 bytes for NTCP header // header accessors uint8_t * GetHeader () { return GetBuffer (); }; @@ -175,8 +177,9 @@ 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; }; + void SetEnqueueTime (uint64_t mts) { enqueueTime = mts; }; uint64_t GetExpiration () const { return bufbe64toh (GetHeader () + I2NP_HEADER_EXPIRATION_OFFSET); }; + uint64_t GetEnqueueTime () const { return enqueueTime; }; void SetSize (uint16_t size) { htobe16buf (GetHeader () + I2NP_HEADER_SIZE_OFFSET, size); }; uint16_t GetSize () const { return bufbe16toh (GetHeader () + I2NP_HEADER_SIZE_OFFSET); }; void UpdateSize () { SetSize (GetPayloadLength ()); }; @@ -267,8 +270,6 @@ 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_LOCAL_EXPIRATION_TIMEOUT / 2; }; // monotonic microseconds void Drop () { if (onDrop) { onDrop (); onDrop = nullptr; }; } }; diff --git a/libi2pd/SSU2Session.cpp b/libi2pd/SSU2Session.cpp index 903b671a..42a26c5c 100644 --- a/libi2pd/SSU2Session.cpp +++ b/libi2pd/SSU2Session.cpp @@ -84,7 +84,10 @@ namespace transport m_Server (server), m_Address (addr), m_RemoteTransports (0), m_RemotePeerTestTransports (0), m_DestConnID (0), m_SourceConnID (0), m_State (eSSU2SessionStateUnknown), m_SendPacketNum (0), m_ReceivePacketNum (0), m_LastDatetimeSentPacketNum (0), - m_IsDataReceived (false), m_RTT (SSU2_UNKNOWN_RTT), m_WindowSize (SSU2_MIN_WINDOW_SIZE), + m_IsDataReceived (false), m_RTT (SSU2_UNKNOWN_RTT), + m_MsgLocalExpirationTimeout (I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_MAX), + m_MsgLocalSemiExpirationTimeout (I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_MAX / 2), + m_WindowSize (SSU2_MIN_WINDOW_SIZE), m_RTO (SSU2_INITIAL_RTO), m_RelayTag (0),m_ConnectTimer (server.GetService ()), m_TerminationReason (eSSU2TerminationReasonNormalClose), m_MaxPayloadSize (SSU2_MIN_PACKET_SIZE - IPV6_HEADER_SIZE - UDP_HEADER_SIZE - 32) // min size @@ -356,15 +359,16 @@ namespace transport { if (m_State == eSSU2SessionStateTerminated) return; 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); + int64_t queueLag = (int64_t)mts - (int64_t)m_SendQueue.front ()->GetEnqueueTime (); + isSemiFull = queueLag > m_MsgLocalSemiExpirationTimeout; if (isSemiFull) { LogPrint (eLogWarning, "SSU2: Outgoing messages queue to ", - GetIdentHashBase64 (), " is semi-full (", m_SendQueue.size (), ")"); + i2p::data::GetIdentHashAbbreviation (GetRemoteIdentity ()->GetIdentHash ()), + " is semi-full (size = ", m_SendQueue.size (), ", lag = ", queueLag / 1000, ", rtt = ", (int)m_RTT, ")"); } } for (auto it: msgs) @@ -373,7 +377,7 @@ namespace transport it->Drop (); // drop earlier because we can handle it else { - it->SetLocalExpiration (localExpiration); + it->SetEnqueueTime (mts); m_SendQueue.push_back (std::move (it)); } } @@ -397,7 +401,7 @@ namespace transport while (!m_SendQueue.empty () && m_SentPackets.size () <= m_WindowSize) { auto msg = m_SendQueue.front (); - if (!msg || msg->IsExpired (ts) || msg->IsLocalExpired (mts)) + if (!msg || msg->IsExpired (ts) || msg->GetEnqueueTime() + m_MsgLocalExpirationTimeout < mts) { // drop null or expired message if (msg) msg->Drop (); @@ -1758,6 +1762,10 @@ namespace transport else m_RTT = rtt; m_RTO = m_RTT*SSU2_kAPPA; + m_MsgLocalExpirationTimeout = std::max (I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_MIN, + std::min (I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_MAX, + (unsigned int)(m_RTT * 1000 * I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_FACTOR))); + m_MsgLocalSemiExpirationTimeout = m_MsgLocalExpirationTimeout / 2; if (m_RTO < SSU2_MIN_RTO) m_RTO = SSU2_MIN_RTO; if (m_RTO > SSU2_MAX_RTO) m_RTO = SSU2_MAX_RTO; } diff --git a/libi2pd/SSU2Session.h b/libi2pd/SSU2Session.h index 6b09f469..6505b233 100644 --- a/libi2pd/SSU2Session.h +++ b/libi2pd/SSU2Session.h @@ -359,6 +359,8 @@ namespace transport i2p::I2NPMessagesHandler m_Handler; bool m_IsDataReceived; double m_RTT; + int m_MsgLocalExpirationTimeout; + int m_MsgLocalSemiExpirationTimeout; size_t m_WindowSize, m_RTO; uint32_t m_RelayTag; // between Bob and Charlie OnEstablished m_OnEstablished; // callback from Established