don't pick too active peer

libi2pd/NTCP2.cpp

@@ -374,7 +374,7 @@ namespace transport
 			transports.PeerDisconnected (shared_from_this ());
 			m_Server.RemoveNTCP2Session (shared_from_this ());
 			m_SendQueue.clear ();
-			m_SendQueueSize = 0;
+			SetSendQueueSize (0);
 			auto remoteIdentity = GetRemoteIdentity ();
 			if (remoteIdentity)
 			{
@@ -433,7 +433,7 @@ namespace transport
 	void NTCP2Session::DeleteNextReceiveBuffer (uint64_t ts)
 	{
 		if (m_NextReceivedBuffer && !m_IsReceiving &&
-			ts > m_LastActivityTimestamp + NTCP2_RECEIVE_BUFFER_DELETION_TIMEOUT)
+			ts > GetLastActivityTimestamp () + NTCP2_RECEIVE_BUFFER_DELETION_TIMEOUT)
 		{
 			delete[] m_NextReceivedBuffer;
 			m_NextReceivedBuffer = nullptr;
@@ -789,7 +789,7 @@ namespace transport
 	void NTCP2Session::ServerLogin ()
 	{
 		SetTerminationTimeout (NTCP2_ESTABLISH_TIMEOUT);
-		m_LastActivityTimestamp = i2p::util::GetSecondsSinceEpoch ();
+		SetLastActivityTimestamp (i2p::util::GetSecondsSinceEpoch ());
 		m_Establisher->CreateEphemeralKey ();
 		boost::asio::async_read (m_Socket, boost::asio::buffer(m_Establisher->m_SessionRequestBuffer, 64), boost::asio::transfer_all (),
 			std::bind(&NTCP2Session::HandleSessionRequestReceived, shared_from_this (),
@@ -872,9 +872,8 @@ namespace transport
 		}
 		else
 		{
-			m_LastActivityTimestamp = i2p::util::GetSecondsSinceEpoch ();
-			m_NumReceivedBytes += bytes_transferred + 2; // + length
-			i2p::transport::transports.UpdateReceivedBytes (bytes_transferred);
+			UpdateNumReceivedBytes (bytes_transferred + 2);
+			i2p::transport::transports.UpdateReceivedBytes (bytes_transferred + 2);
 			uint8_t nonce[12];
 			CreateNonce (m_ReceiveSequenceNumber, nonce); m_ReceiveSequenceNumber++;
 			if (i2p::crypto::AEADChaCha20Poly1305 (m_NextReceivedBuffer, m_NextReceivedLen-16, nullptr, 0, m_ReceiveKey, nonce, m_NextReceivedBuffer, m_NextReceivedLen, false))
@@ -1095,11 +1094,10 @@ namespace transport
 		}
 		else
 		{
-			m_LastActivityTimestamp = i2p::util::GetSecondsSinceEpoch ();
-			m_NumSentBytes += bytes_transferred;
+			UpdateNumSentBytes (bytes_transferred);
 			i2p::transport::transports.UpdateSentBytes (bytes_transferred);
 			LogPrint (eLogDebug, "NTCP2: Next frame sent ", bytes_transferred);
-			if (m_LastActivityTimestamp > m_NextRouterInfoResendTime)
+			if (GetLastActivityTimestamp () > m_NextRouterInfoResendTime)
 			{
 				m_NextRouterInfoResendTime += NTCP2_ROUTERINFO_RESEND_INTERVAL +
 					rand ()%NTCP2_ROUTERINFO_RESEND_INTERVAL_THRESHOLD;
@@ -1108,7 +1106,7 @@ namespace transport
 			else
 			{
 				SendQueue ();
-				m_SendQueueSize = m_SendQueue.size ();
+				SetSendQueueSize (m_SendQueue.size ());
 			}
 		}
 	}
@@ -1231,7 +1229,7 @@ namespace transport
 				GetIdentHashBase64(), " exceeds ", NTCP2_MAX_OUTGOING_QUEUE_SIZE);
 			Terminate ();
 		}
-		m_SendQueueSize = m_SendQueue.size ();
+		SetSendQueueSize (m_SendQueue.size ());
 	}
 
 	void NTCP2Session::SendLocalRouterInfo (bool update)

libi2pd/RouterContext.cpp

@@ -595,14 +595,14 @@ namespace i2p
 		switch (L)
 		{
 			case i2p::data::CAPS_FLAG_LOW_BANDWIDTH1   : limit = 12; type = low;   break;
-			case i2p::data::CAPS_FLAG_LOW_BANDWIDTH2   : limit =   48; type = low;   break;
+			case i2p::data::CAPS_FLAG_LOW_BANDWIDTH2   : limit = i2p::data::LOW_BANDWIDTH_LIMIT; type = low;   break; // 48
 			case i2p::data::CAPS_FLAG_HIGH_BANDWIDTH1  : limit = 64; type = high;  break;
 			case i2p::data::CAPS_FLAG_HIGH_BANDWIDTH2  : limit = 128; type = high;  break;
-			case i2p::data::CAPS_FLAG_HIGH_BANDWIDTH3  : limit =  256; type = high;  break;
-			case i2p::data::CAPS_FLAG_EXTRA_BANDWIDTH1 : limit = 2048; type = extra; break;
+			case i2p::data::CAPS_FLAG_HIGH_BANDWIDTH3  : limit = i2p::data::HIGH_BANDWIDTH_LIMIT; type = high;  break; // 256
+			case i2p::data::CAPS_FLAG_EXTRA_BANDWIDTH1 : limit = i2p::data::EXTRA_BANDWIDTH_LIMIT; type = extra; break; // 2048
 			case i2p::data::CAPS_FLAG_EXTRA_BANDWIDTH2 : limit = 1000000; type = unlim; break; // 1Gbyte/s
 			default:
-				limit = 48; type = low;
+				limit = i2p::data::LOW_BANDWIDTH_LIMIT; type = low; // 48
 		}
 		/* update caps & flags in RI */
 		auto caps = m_RouterInfo.GetCaps ();

libi2pd/RouterInfo.h

@@ -42,8 +42,12 @@ namespace data
 	const char CAPS_FLAG_HIGH_BANDWIDTH1  = 'M'; /*  48-64 KBps */
 	const char CAPS_FLAG_HIGH_BANDWIDTH2  = 'N'; /*  64-128 KBps */
 	const char CAPS_FLAG_HIGH_BANDWIDTH3  = 'O'; /* 128-256 KBps */
-	const char CAPS_FLAG_EXTRA_BANDWIDTH1 = 'P'; /* 256-2000 KBps */
-	const char CAPS_FLAG_EXTRA_BANDWIDTH2 = 'X'; /*   > 2000 KBps */
+	const char CAPS_FLAG_EXTRA_BANDWIDTH1 = 'P'; /* 256-2048 KBps */
+	const char CAPS_FLAG_EXTRA_BANDWIDTH2 = 'X'; /*   > 2048 KBps */
+	// bandwidth limits in kBps
+	const uint32_t LOW_BANDWIDTH_LIMIT = 48;
+	const uint32_t HIGH_BANDWIDTH_LIMIT = 256;
+	const uint32_t EXTRA_BANDWIDTH_LIMIT = 2048;	
 	// congesion flags
 	const char CAPS_FLAG_MEDIUM_CONGESTION = 'D';
 	const char CAPS_FLAG_HIGH_CONGESTION = 'E';

libi2pd/SSU2Session.cpp

@@ -263,7 +263,7 @@ namespace transport
 			m_SessionConfirmedFragment.reset (nullptr);
 			m_PathChallenge.reset (nullptr);
 			m_SendQueue.clear ();
-			m_SendQueueSize = 0;
+			SetSendQueueSize (0);
 			m_SentPackets.clear ();
 			m_IncompleteMessages.clear ();
 			m_RelaySessions.clear ();
@@ -364,7 +364,7 @@ namespace transport
 				RequestTermination (eSSU2TerminationReasonTimeout);
 			}
 		}
-		m_SendQueueSize = m_SendQueue.size ();
+		SetSendQueueSize (m_SendQueue.size ());
 	}
 
 	bool SSU2Session::SendQueue ()
@@ -524,7 +524,7 @@ namespace transport
 					LogPrint (eLogInfo, "SSU2: Packet was not Acked after ", it->second->numResends, " attempts. Terminate session");
 					m_SentPackets.clear ();
 					m_SendQueue.clear ();
-					m_SendQueueSize = 0;
+					SetSendQueueSize (0);
 					RequestTermination (eSSU2TerminationReasonTimeout);
 					return resentPackets.size ();
 				}
@@ -1452,8 +1452,7 @@ namespace transport
 		header.ll[1] ^= CreateHeaderMask (m_KeyDataSend + 32, payload + (len + 4));
 		m_Server.Send (header.buf, 16, payload, len + 16, m_RemoteEndpoint);
 		m_SendPacketNum++;
-		m_LastActivityTimestamp = i2p::util::GetSecondsSinceEpoch ();
-		m_NumSentBytes += len + 32;
+		UpdateNumSentBytes (len + 32);
 		return m_SendPacketNum - 1;
 	}
 
@@ -1494,8 +1493,7 @@ namespace transport
 			LogPrint (eLogWarning, "SSU2: Data AEAD verification failed ");
 			return;
 		}
-		m_LastActivityTimestamp = i2p::util::GetSecondsSinceEpoch ();
-		m_NumReceivedBytes += len;
+		UpdateNumReceivedBytes (len);
 		if (!packetNum || UpdateReceivePacketNum (packetNum))
 			HandlePayload (payload, payloadSize);
 	}
@@ -2357,7 +2355,7 @@ namespace transport
 		if (!msg->IsExpired ())
 		{
 			// m_LastActivityTimestamp is updated in ProcessData before
-			if (m_ReceivedI2NPMsgIDs.emplace (msgID, (uint32_t)m_LastActivityTimestamp).second)
+			if (m_ReceivedI2NPMsgIDs.emplace (msgID, (uint32_t)GetLastActivityTimestamp ()).second)
 				m_Handler.PutNextMessage (std::move (msg));
 			else
 				LogPrint (eLogDebug, "SSU2: Message ", msgID, " already received");
@@ -2943,7 +2941,7 @@ namespace transport
 			else
 				++it;
 		}
-		if (m_ReceivedI2NPMsgIDs.size () > SSU2_MAX_NUM_RECEIVED_I2NP_MSGIDS || ts > m_LastActivityTimestamp + SSU2_DECAY_INTERVAL)
+		if (m_ReceivedI2NPMsgIDs.size () > SSU2_MAX_NUM_RECEIVED_I2NP_MSGIDS || ts > GetLastActivityTimestamp () + SSU2_DECAY_INTERVAL)
 			// decay
 			m_ReceivedI2NPMsgIDs.clear ();
 		else
@@ -3015,7 +3013,7 @@ namespace transport
 	{
 		bool sent = SendQueue (); // if we have something to send
 		if (sent)
-			m_SendQueueSize = m_SendQueue.size ();
+			SetSendQueueSize (m_SendQueue.size ());
 		if (m_IsDataReceived)
 		{
 			if (!sent) SendQuickAck ();

libi2pd/TransportSession.h

@@ -71,15 +71,17 @@ namespace transport
 
 	const int64_t TRANSPORT_SESSION_SLOWNESS_THRESHOLD = 500; // in milliseconds
 	const int64_t TRANSPORT_SESSION_MAX_HANDSHAKE_INTERVAL = 10000; // in milliseconds
+	const uint64_t TRANSPORT_SESSION_BANDWIDTH_UPDATE_MIN_INTERVAL = 5; // in seconds
 	class TransportSession
 	{
 		public:
 
 			TransportSession (std::shared_ptr<const i2p::data::RouterInfo> router, int terminationTimeout):
-				m_NumSentBytes (0), m_NumReceivedBytes (0), m_SendQueueSize (0),
-				m_IsOutgoing (router), m_TerminationTimeout (terminationTimeout),
+				m_IsOutgoing (router), m_TerminationTimeout (terminationTimeout), m_HandshakeInterval (0), 
+				m_SendQueueSize (0), m_NumSentBytes (0), m_NumReceivedBytes (0),
+				m_LastBandWidthUpdateNumSentBytes (0), m_LastBandWidthUpdateNumReceivedBytes (0),
 				m_LastActivityTimestamp (i2p::util::GetSecondsSinceEpoch ()), 
-				m_HandshakeInterval (0)
+				m_LastBandwidthUpdateTimestamp (m_LastActivityTimestamp), m_InBandwidth (0), m_OutBandwidth (0)
 			{
 				if (router)
 					m_RemoteIdentity = router->GetRouterIdentity ();
@@ -103,11 +105,29 @@ namespace transport
 			}
 
 			size_t GetNumSentBytes () const { return m_NumSentBytes; };
+			void UpdateNumSentBytes (size_t len)
+			{
+				m_LastActivityTimestamp = i2p::util::GetSecondsSinceEpoch ();
+				m_NumSentBytes += len;
+				UpdateBandwidth ();
+			}	
 			size_t GetNumReceivedBytes () const { return m_NumReceivedBytes; };
+			void UpdateNumReceivedBytes (size_t len)
+			{
+				m_LastActivityTimestamp = i2p::util::GetSecondsSinceEpoch ();
+				m_NumReceivedBytes += len;
+				UpdateBandwidth ();
+			}		
 			size_t GetSendQueueSize () const { return m_SendQueueSize; };
+			void SetSendQueueSize (size_t s) { m_SendQueueSize = s; };
 			bool IsOutgoing () const { return m_IsOutgoing; };
 			bool IsSlow () const { return m_HandshakeInterval > TRANSPORT_SESSION_SLOWNESS_THRESHOLD &&
 				m_HandshakeInterval < TRANSPORT_SESSION_MAX_HANDSHAKE_INTERVAL; };
+			bool IsBandwidthExceeded (bool isHighBandwidth) const
+			{
+				auto limit = isHighBandwidth ? i2p::data::HIGH_BANDWIDTH_LIMIT*1024 : i2p::data::LOW_BANDWIDTH_LIMIT*1024; // convert to bytes
+				return std::max (m_InBandwidth, m_OutBandwidth) > limit;
+			}	
 			
 			int GetTerminationTimeout () const { return m_TerminationTimeout; };
 			void SetTerminationTimeout (int terminationTimeout) { m_TerminationTimeout = terminationTimeout; };
@@ -120,21 +140,44 @@ namespace transport
 			uint32_t GetCreationTime () const { return m_CreationTime; };
 			void SetCreationTime (uint32_t ts) { m_CreationTime = ts; }; // for introducers
 
+			uint64_t GetLastActivityTimestamp () const { return m_LastActivityTimestamp; };
+			void SetLastActivityTimestamp (uint64_t ts) { m_LastActivityTimestamp = ts; };
+			
 			virtual uint32_t GetRelayTag () const { return 0; };
 			virtual void SendLocalRouterInfo (bool update = false) { SendI2NPMessages ({ CreateDatabaseStoreMsg () }); };
 			virtual void SendI2NPMessages (const std::vector<std::shared_ptr<I2NPMessage> >& msgs) = 0;
 			virtual bool IsEstablished () const = 0;
 
+		private:
+
+			void UpdateBandwidth ()
+			{
+				uint64_t interval = m_LastActivityTimestamp - m_LastBandwidthUpdateTimestamp;
+				if (interval > TRANSPORT_SESSION_BANDWIDTH_UPDATE_MIN_INTERVAL)
+				{	
+					m_OutBandwidth = (m_NumSentBytes - m_LastBandWidthUpdateNumSentBytes)/interval;
+					m_LastBandWidthUpdateNumSentBytes = m_NumSentBytes;
+					m_InBandwidth = (m_NumReceivedBytes - m_LastBandWidthUpdateNumReceivedBytes)/interval;
+					m_LastBandWidthUpdateNumReceivedBytes = m_NumReceivedBytes;
+					m_LastBandwidthUpdateTimestamp = m_LastActivityTimestamp;
+				}	
+			}	
+			
 		protected:
 
 			std::shared_ptr<const i2p::data::IdentityEx> m_RemoteIdentity;
 			mutable std::mutex m_RemoteIdentityMutex;
-			size_t m_NumSentBytes, m_NumReceivedBytes, m_SendQueueSize;
 			bool m_IsOutgoing;
 			int m_TerminationTimeout;
-			uint64_t m_LastActivityTimestamp;
 			uint32_t m_CreationTime; // seconds since epoch
 			int64_t m_HandshakeInterval; // in milliseconds between SessionRequest->SessionCreated or SessionCreated->SessionConfirmed
+
+		private:
+
+			size_t m_SendQueueSize, m_NumSentBytes, m_NumReceivedBytes, 
+				m_LastBandWidthUpdateNumSentBytes, m_LastBandWidthUpdateNumReceivedBytes;
+			uint64_t m_LastActivityTimestamp, m_LastBandwidthUpdateTimestamp;	
+			uint32_t m_InBandwidth, m_OutBandwidth;
 	};
 
 	// SOCKS5 proxy

libi2pd/Transports.cpp

@@ -951,7 +951,7 @@ namespace transport
 				// connected, not overloaded and not slow
 				return !peer.router && !peer.sessions.empty () && peer.isReachable &&
 					peer.sessions.front ()->GetSendQueueSize () <= PEER_ROUTER_INFO_OVERLOAD_QUEUE_SIZE &&
-					!peer.sessions.front ()->IsSlow () &&
+					!peer.sessions.front ()->IsSlow () && !peer.sessions.front ()->IsBandwidthExceeded (peer.isHighBandwidth) &&
 					(!isHighBandwidth || peer.isHighBandwidth);
 			});
 	}