Browse Source

bypass slow transport sessions

pull/1909/head
orignal 1 year ago
parent
commit
cd1af85e39
  1. 1
      daemon/HTTPServer.cpp
  2. 4
      libi2pd/NTCP2.cpp
  3. 4
      libi2pd/SSU2Session.cpp
  4. 10
      libi2pd/TransportSession.h
  5. 5
      libi2pd/Transports.cpp
  6. 10
      libi2pd/Transports.h

1
daemon/HTTPServer.cpp

@ -838,6 +838,7 @@ namespace http {
tmp_s << " [itag:" << it->GetRelayTag () << "]"; tmp_s << " [itag:" << it->GetRelayTag () << "]";
if (it->GetSendQueueSize () > 0) if (it->GetSendQueueSize () > 0)
tmp_s << " [queue:" << it->GetSendQueueSize () << "]"; tmp_s << " [queue:" << it->GetSendQueueSize () << "]";
if (it->IsSlow ()) tmp_s << " [slow]";
tmp_s << "</div>\r\n" << std::endl; tmp_s << "</div>\r\n" << std::endl;
cnt++; cnt++;
} }

4
libi2pd/NTCP2.cpp

@ -452,6 +452,7 @@ namespace transport
{ {
m_Establisher->CreateSessionRequestMessage (); m_Establisher->CreateSessionRequestMessage ();
// send message // send message
m_HandshakeInterval = i2p::util::GetMillisecondsSinceEpoch ();
boost::asio::async_write (m_Socket, boost::asio::buffer (m_Establisher->m_SessionRequestBuffer, m_Establisher->m_SessionRequestBufferLen), boost::asio::transfer_all (), boost::asio::async_write (m_Socket, boost::asio::buffer (m_Establisher->m_SessionRequestBuffer, m_Establisher->m_SessionRequestBufferLen), boost::asio::transfer_all (),
std::bind(&NTCP2Session::HandleSessionRequestSent, shared_from_this (), std::placeholders::_1, std::placeholders::_2)); std::bind(&NTCP2Session::HandleSessionRequestSent, shared_from_this (), std::placeholders::_1, std::placeholders::_2));
} }
@ -529,6 +530,7 @@ namespace transport
{ {
m_Establisher->CreateSessionCreatedMessage (); m_Establisher->CreateSessionCreatedMessage ();
// send message // send message
m_HandshakeInterval = i2p::util::GetMillisecondsSinceEpoch ();
boost::asio::async_write (m_Socket, boost::asio::buffer (m_Establisher->m_SessionCreatedBuffer, m_Establisher->m_SessionCreatedBufferLen), boost::asio::transfer_all (), boost::asio::async_write (m_Socket, boost::asio::buffer (m_Establisher->m_SessionCreatedBuffer, m_Establisher->m_SessionCreatedBufferLen), boost::asio::transfer_all (),
std::bind(&NTCP2Session::HandleSessionCreatedSent, shared_from_this (), std::placeholders::_1, std::placeholders::_2)); std::bind(&NTCP2Session::HandleSessionCreatedSent, shared_from_this (), std::placeholders::_1, std::placeholders::_2));
} }
@ -542,6 +544,7 @@ namespace transport
} }
else else
{ {
m_HandshakeInterval = i2p::util::GetMillisecondsSinceEpoch () - m_HandshakeInterval;
LogPrint (eLogDebug, "NTCP2: SessionCreated received ", bytes_transferred); LogPrint (eLogDebug, "NTCP2: SessionCreated received ", bytes_transferred);
uint16_t paddingLen = 0; uint16_t paddingLen = 0;
if (m_Establisher->ProcessSessionCreatedMessage (paddingLen)) if (m_Establisher->ProcessSessionCreatedMessage (paddingLen))
@ -646,6 +649,7 @@ namespace transport
} }
else else
{ {
m_HandshakeInterval = i2p::util::GetMillisecondsSinceEpoch () - m_HandshakeInterval;
LogPrint (eLogDebug, "NTCP2: SessionConfirmed received"); LogPrint (eLogDebug, "NTCP2: SessionConfirmed received");
// part 1 // part 1
uint8_t nonce[12]; uint8_t nonce[12];

4
libi2pd/SSU2Session.cpp

@ -648,6 +648,7 @@ namespace transport
if (m_State == eSSU2SessionStateTokenReceived || m_Server.AddPendingOutgoingSession (shared_from_this ())) if (m_State == eSSU2SessionStateTokenReceived || m_Server.AddPendingOutgoingSession (shared_from_this ()))
{ {
m_State = eSSU2SessionStateSessionRequestSent; m_State = eSSU2SessionStateSessionRequestSent;
m_HandshakeInterval = ts;
m_Server.Send (header.buf, 16, headerX, 48, payload, payloadSize, m_RemoteEndpoint); m_Server.Send (header.buf, 16, headerX, 48, payload, payloadSize, m_RemoteEndpoint);
} }
else else
@ -770,6 +771,7 @@ namespace transport
m_State = eSSU2SessionStateSessionCreatedSent; m_State = eSSU2SessionStateSessionCreatedSent;
m_SentHandshakePacket->payloadSize = payloadSize; m_SentHandshakePacket->payloadSize = payloadSize;
// send // send
m_HandshakeInterval = ts;
m_Server.Send (header.buf, 16, headerX, 48, payload, payloadSize, m_RemoteEndpoint); m_Server.Send (header.buf, 16, headerX, 48, payload, payloadSize, m_RemoteEndpoint);
} }
@ -790,6 +792,7 @@ namespace transport
LogPrint (eLogWarning, "SSU2: SessionCreated message too short ", len); LogPrint (eLogWarning, "SSU2: SessionCreated message too short ", len);
return false; return false;
} }
m_HandshakeInterval = i2p::util::GetMillisecondsSinceEpoch () - m_HandshakeInterval;
const uint8_t nonce[12] = {0}; const uint8_t nonce[12] = {0};
uint8_t headerX[48]; uint8_t headerX[48];
i2p::crypto::ChaCha20 (buf + 16, 48, kh2, nonce, headerX); i2p::crypto::ChaCha20 (buf + 16, 48, kh2, nonce, headerX);
@ -995,6 +998,7 @@ namespace transport
if (m_SessionConfirmedFragment) m_SessionConfirmedFragment.reset (nullptr); if (m_SessionConfirmedFragment) m_SessionConfirmedFragment.reset (nullptr);
return false; return false;
} }
m_HandshakeInterval = i2p::util::GetMillisecondsSinceEpoch () - m_HandshakeInterval;
// KDF for Session Confirmed part 1 // KDF for Session Confirmed part 1
m_NoiseState->MixHash (header.buf, 16); // h = SHA256(h || header) m_NoiseState->MixHash (header.buf, 16); // h = SHA256(h || header)
// decrypt part1 // decrypt part1

10
libi2pd/TransportSession.h

@ -69,6 +69,8 @@ namespace transport
std::stringstream m_Stream; std::stringstream m_Stream;
}; };
const int64_t TRANSPORT_SESSION_SLOWNESS_THRESHOLD = 500; // in milliseconds
const int64_t TRANSPORT_SESSION_MAX_HANDSHAKE_INTERVAL = 10000; // in milliseconds
class TransportSession class TransportSession
{ {
public: public:
@ -76,7 +78,8 @@ namespace transport
TransportSession (std::shared_ptr<const i2p::data::RouterInfo> router, int terminationTimeout): TransportSession (std::shared_ptr<const i2p::data::RouterInfo> router, int terminationTimeout):
m_NumSentBytes (0), m_NumReceivedBytes (0), m_SendQueueSize (0), m_NumSentBytes (0), m_NumReceivedBytes (0), m_SendQueueSize (0),
m_IsOutgoing (router), m_TerminationTimeout (terminationTimeout), m_IsOutgoing (router), m_TerminationTimeout (terminationTimeout),
m_LastActivityTimestamp (i2p::util::GetSecondsSinceEpoch ()) m_LastActivityTimestamp (i2p::util::GetSecondsSinceEpoch ()),
m_HandshakeInterval (0)
{ {
if (router) if (router)
m_RemoteIdentity = router->GetRouterIdentity (); m_RemoteIdentity = router->GetRouterIdentity ();
@ -103,7 +106,9 @@ namespace transport
size_t GetNumReceivedBytes () const { return m_NumReceivedBytes; }; size_t GetNumReceivedBytes () const { return m_NumReceivedBytes; };
size_t GetSendQueueSize () const { return m_SendQueueSize; }; size_t GetSendQueueSize () const { return m_SendQueueSize; };
bool IsOutgoing () const { return m_IsOutgoing; }; bool IsOutgoing () const { return m_IsOutgoing; };
bool IsSlow () const { return m_HandshakeInterval > TRANSPORT_SESSION_SLOWNESS_THRESHOLD &&
m_HandshakeInterval < TRANSPORT_SESSION_MAX_HANDSHAKE_INTERVAL; };
int GetTerminationTimeout () const { return m_TerminationTimeout; }; int GetTerminationTimeout () const { return m_TerminationTimeout; };
void SetTerminationTimeout (int terminationTimeout) { m_TerminationTimeout = terminationTimeout; }; void SetTerminationTimeout (int terminationTimeout) { m_TerminationTimeout = terminationTimeout; };
bool IsTerminationTimeoutExpired (uint64_t ts) const bool IsTerminationTimeoutExpired (uint64_t ts) const
@ -129,6 +134,7 @@ namespace transport
int m_TerminationTimeout; int m_TerminationTimeout;
uint64_t m_LastActivityTimestamp; uint64_t m_LastActivityTimestamp;
uint32_t m_CreationTime; // seconds since epoch uint32_t m_CreationTime; // seconds since epoch
int64_t m_HandshakeInterval; // in milliseconds between SessionRequest->SessionCreated or SessionCreated->SessionConfirmed
}; };
// SOCKS5 proxy // SOCKS5 proxy

5
libi2pd/Transports.cpp

@ -906,9 +906,10 @@ namespace transport
return GetRandomPeer ( return GetRandomPeer (
[isHighBandwidth](const Peer& peer)->bool [isHighBandwidth](const Peer& peer)->bool
{ {
// connected and not overloaded // connected, not overloaded and not slow
return !peer.router && !peer.sessions.empty () && return !peer.router && !peer.sessions.empty () && peer.isReachable &&
peer.sessions.front ()->GetSendQueueSize () <= PEER_ROUTER_INFO_OVERLOAD_QUEUE_SIZE && peer.sessions.front ()->GetSendQueueSize () <= PEER_ROUTER_INFO_OVERLOAD_QUEUE_SIZE &&
!peer.sessions.front ()->IsSlow () &&
(!isHighBandwidth || peer.isHighBandwidth); (!isHighBandwidth || peer.isHighBandwidth);
}); });
} }

10
libi2pd/Transports.h

@ -72,15 +72,18 @@ namespace transport
uint64_t creationTime, nextRouterInfoUpdateTime; uint64_t creationTime, nextRouterInfoUpdateTime;
std::vector<std::shared_ptr<i2p::I2NPMessage> > delayedMessages; std::vector<std::shared_ptr<i2p::I2NPMessage> > delayedMessages;
std::vector<i2p::data::RouterInfo::SupportedTransports> priority; std::vector<i2p::data::RouterInfo::SupportedTransports> priority;
bool isHighBandwidth; bool isHighBandwidth, isReachable;
Peer (std::shared_ptr<const i2p::data::RouterInfo> r, uint64_t ts): Peer (std::shared_ptr<const i2p::data::RouterInfo> r, uint64_t ts):
numAttempts (0), router (r), creationTime (ts), numAttempts (0), router (r), creationTime (ts),
nextRouterInfoUpdateTime (ts + PEER_ROUTER_INFO_UPDATE_INTERVAL), nextRouterInfoUpdateTime (ts + PEER_ROUTER_INFO_UPDATE_INTERVAL),
isHighBandwidth (false) isHighBandwidth (false), isReachable (false)
{ {
if (router) if (router)
{
isHighBandwidth = router->IsHighBandwidth (); isHighBandwidth = router->IsHighBandwidth ();
isReachable = (bool)router->GetCompatibleTransports (true);
}
} }
void Done () void Done ()
@ -93,7 +96,10 @@ namespace transport
{ {
router = r; router = r;
if (router) if (router)
{
isHighBandwidth = router->IsHighBandwidth (); isHighBandwidth = router->IsHighBandwidth ();
isReachable = (bool)router->GetCompatibleTransports (true);
}
} }
}; };

Loading…
Cancel
Save