PurpleI2P
/
i2pd
mirror of https://github.com/PurpleI2P/i2pd.git
1
0
Fork 0
Code Issues Releases Activity
Browse Source

Merge pull request #2047 from Vort/ssu2_lag 

select maximum SSU2 queue size depending on RTT value
pull/2050/head
orignal 2 months ago committed by GitHub
parent
a30d1972e5 ac9d92c681
commit
835c480269
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 23 additions and 12 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 13
      libi2pd/I2NPProtocol.h
  2. 20
      libi2pd/SSU2Session.cpp
  3. 2
      libi2pd/SSU2Session.h

13
libi2pd/I2NPProtocol.h
Unescape View File

@ -152,7 +152,9 @@ namespace tunnel
const size_t I2NP_MAX_MESSAGE_SIZE = 62708; const size_t I2NP_MAX_MESSAGE_SIZE = 62708;
const size_t I2NP_MAX_SHORT_MESSAGE_SIZE = 4096; const size_t I2NP_MAX_SHORT_MESSAGE_SIZE = 4096;
const size_t I2NP_MAX_MEDIUM_MESSAGE_SIZE = 16384; 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_EXPIRATION_TIMEOUT = 8000; // in milliseconds (as initial RTT)
const unsigned int I2NP_MESSAGE_CLOCK_SKEW = 60*1000; // 1 minute in milliseconds const unsigned int I2NP_MESSAGE_CLOCK_SKEW = 60*1000; // 1 minute in milliseconds
@ -162,10 +164,10 @@ namespace tunnel
size_t len, offset, maxLen; size_t len, offset, maxLen;
std::shared_ptr<i2p::tunnel::InboundTunnel> from; std::shared_ptr<i2p::tunnel::InboundTunnel> from;
std::function<void ()> onDrop; std::function<void ()> onDrop;
uint64_t localExpiration; // monotonic microseconds uint64_t enqueueTime; // monotonic microseconds
I2NPMessage (): buf (nullptr), len (I2NP_HEADER_SIZE + 2), 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 // header accessors
uint8_t * GetHeader () { return GetBuffer (); }; uint8_t * GetHeader () { return GetBuffer (); };
@ -175,8 +177,9 @@ namespace tunnel
void SetMsgID (uint32_t msgID) { htobe32buf (GetHeader () + I2NP_HEADER_MSGID_OFFSET, msgID); }; void SetMsgID (uint32_t msgID) { htobe32buf (GetHeader () + I2NP_HEADER_MSGID_OFFSET, msgID); };
uint32_t GetMsgID () const { return bufbe32toh (GetHeader () + I2NP_HEADER_MSGID_OFFSET); }; uint32_t GetMsgID () const { return bufbe32toh (GetHeader () + I2NP_HEADER_MSGID_OFFSET); };
void SetExpiration (uint64_t expiration) { htobe64buf (GetHeader () + I2NP_HEADER_EXPIRATION_OFFSET, expiration); }; 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 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); }; void SetSize (uint16_t size) { htobe16buf (GetHeader () + I2NP_HEADER_SIZE_OFFSET, size); };
uint16_t GetSize () const { return bufbe16toh (GetHeader () + I2NP_HEADER_SIZE_OFFSET); }; uint16_t GetSize () const { return bufbe16toh (GetHeader () + I2NP_HEADER_SIZE_OFFSET); };
void UpdateSize () { SetSize (GetPayloadLength ()); }; void UpdateSize () { SetSize (GetPayloadLength ()); };
@ -267,8 +270,6 @@ namespace tunnel
void RenewI2NPMessageHeader (); void RenewI2NPMessageHeader ();
bool IsExpired () const; bool IsExpired () const;
bool IsExpired (uint64_t ts) const; // in milliseconds 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; }; } void Drop () { if (onDrop) { onDrop (); onDrop = nullptr; }; }
}; };

20
libi2pd/SSU2Session.cpp
Unescape View File

@ -84,7 +84,10 @@ namespace transport
m_Server (server), m_Address (addr), m_RemoteTransports (0), m_RemotePeerTestTransports (0), m_Server (server), m_Address (addr), m_RemoteTransports (0), m_RemotePeerTestTransports (0),
m_DestConnID (0), m_SourceConnID (0), m_State (eSSU2SessionStateUnknown), m_DestConnID (0), m_SourceConnID (0), m_State (eSSU2SessionStateUnknown),
m_SendPacketNum (0), m_ReceivePacketNum (0), m_LastDatetimeSentPacketNum (0), 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_RTO (SSU2_INITIAL_RTO), m_RelayTag (0),m_ConnectTimer (server.GetService ()),
m_TerminationReason (eSSU2TerminationReasonNormalClose), m_TerminationReason (eSSU2TerminationReasonNormalClose),
m_MaxPayloadSize (SSU2_MIN_PACKET_SIZE - IPV6_HEADER_SIZE - UDP_HEADER_SIZE - 32) // min size 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; if (m_State == eSSU2SessionStateTerminated) return;
uint64_t mts = i2p::util::GetMonotonicMicroseconds (); uint64_t mts = i2p::util::GetMonotonicMicroseconds ();
uint64_t localExpiration = mts + I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT;
bool isSemiFull = false; bool isSemiFull = false;
if (m_SendQueue.size ()) 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) if (isSemiFull)
{ {
LogPrint (eLogWarning, "SSU2: Outgoing messages queue to ", 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) for (auto it: msgs)
@ -373,7 +377,7 @@ namespace transport
it->Drop (); // drop earlier because we can handle it it->Drop (); // drop earlier because we can handle it
else else
{ {
it->SetLocalExpiration (localExpiration); it->SetEnqueueTime (mts);
m_SendQueue.push_back (std::move (it)); m_SendQueue.push_back (std::move (it));
} }
} }
@ -397,7 +401,7 @@ namespace transport
while (!m_SendQueue.empty () && m_SentPackets.size () <= m_WindowSize) while (!m_SendQueue.empty () && m_SentPackets.size () <= m_WindowSize)
{ {
auto msg = m_SendQueue.front (); 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 // drop null or expired message
if (msg) msg->Drop (); if (msg) msg->Drop ();
@ -1758,6 +1762,10 @@ namespace transport
else else
m_RTT = rtt; m_RTT = rtt;
m_RTO = m_RTT*SSU2_kAPPA; 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_MIN_RTO) m_RTO = SSU2_MIN_RTO;
if (m_RTO > SSU2_MAX_RTO) m_RTO = SSU2_MAX_RTO; if (m_RTO > SSU2_MAX_RTO) m_RTO = SSU2_MAX_RTO;
} }

2
libi2pd/SSU2Session.h
Unescape View File

@ -359,6 +359,8 @@ namespace transport
i2p::I2NPMessagesHandler m_Handler; i2p::I2NPMessagesHandler m_Handler;
bool m_IsDataReceived; bool m_IsDataReceived;
double m_RTT; double m_RTT;
int m_MsgLocalExpirationTimeout;
int m_MsgLocalSemiExpirationTimeout;
size_t m_WindowSize, m_RTO; size_t m_WindowSize, m_RTO;
uint32_t m_RelayTag; // between Bob and Charlie uint32_t m_RelayTag; // between Bob and Charlie
OnEstablished m_OnEstablished; // callback from Established OnEstablished m_OnEstablished; // callback from Established

Write Preview
Loading…
Cancel
Save