Merge pull request #982 from majestrate/i2np-transport-limits

I2np transport limits

libi2pd/Config.cpp

@@ -71,6 +71,8 @@ namespace config {
 			("limits.coresize", value<uint32_t>()->default_value(0),          "Maximum size of corefile in Kb (0 - use system limit)")
 			("limits.openfiles", value<uint16_t>()->default_value(0),         "Maximum number of open files (0 - use system default)")
 			("limits.transittunnels", value<uint16_t>()->default_value(2500), "Maximum active transit sessions (default:2500)")
+			("limits.ntcpsoft", value<uint16_t>()->default_value(0),          "Threshold to start probabalistic backoff with ntcp sessions (default: use system limit)")
+			("limits.ntcphard", value<uint16_t>()->default_value(0),          "Maximum number of ntcp sessions (default: use system limit)")
 		;
 
 		options_description httpserver("HTTP Server options");

libi2pd/NTCPSession.cpp

@@ -791,7 +791,8 @@ namespace transport
 	NTCPServer::NTCPServer ():
 		m_IsRunning (false), m_Thread (nullptr), m_Work (m_Service),
 		m_TerminationTimer (m_Service), m_NTCPAcceptor (nullptr), m_NTCPV6Acceptor (nullptr),
-		m_ProxyType(eNoProxy), m_Resolver(m_Service), m_ProxyEndpoint(nullptr)
+		m_ProxyType(eNoProxy), m_Resolver(m_Service), m_ProxyEndpoint(nullptr),
+		m_SoftLimit(0), m_HardLimit(0)
 	{
 	}
 
@@ -965,6 +966,13 @@ namespace transport
 			auto ep = conn->GetSocket ().remote_endpoint(ec);
 			if (!ec)
 			{
+				if(ShouldLimit())
+				{
+					// hit limit, close premature
+					LogPrint(eLogWarning, "NTCP: limiting with backoff session from ", ep);
+					conn->Terminate();
+					return;
+				}
 				LogPrint (eLogDebug, "NTCP: Connected from ", ep);
 				if (conn)
 				{
@@ -993,6 +1001,14 @@ namespace transport
 			auto ep = conn->GetSocket ().remote_endpoint(ec);
 			if (!ec)
 			{
+				if(ShouldLimit())
+				{
+					// hit limit, close premature
+					LogPrint(eLogWarning, "NTCP: limiting with backoff on session from ", ep);
+					conn->Terminate();
+					return;
+				}
+
 				LogPrint (eLogDebug, "NTCP: Connected from ", ep);
 				if (conn)
 				{

libi2pd/NTCPSession.h

@@ -167,8 +167,19 @@ namespace transport
 
 			boost::asio::io_service& GetService () { return m_Service; };
 
+			void SetSessionLimits(uint16_t softLimit, uint16_t hardLimit) { m_SoftLimit = softLimit; m_HardLimit = hardLimit; }
+			bool ShouldLimit() const { return ShouldHardLimit() || ShouldSoftLimit(); }
 		private:
 
+			/** @brief return true for hard limit */
+			bool ShouldHardLimit() const { return m_HardLimit && m_NTCPSessions.size() >= m_HardLimit; }
+
+			/** @brief return true for probabalistic soft backoff */
+			bool ShouldSoftLimit() const
+			{
+				auto sessions = m_NTCPSessions.size();
+				return sessions && m_SoftLimit && m_SoftLimit < sessions && ( rand() % sessions ) <= m_SoftLimit;
+			}
 			void Run ();
 			void HandleAccept (std::shared_ptr<NTCPSession> conn, const boost::system::error_code& error);
 			void HandleAcceptV6 (std::shared_ptr<NTCPSession> conn, const boost::system::error_code& error);
@@ -198,6 +209,8 @@ namespace transport
 			uint16_t m_ProxyPort;
 			boost::asio::ip::tcp::resolver m_Resolver;
 			boost::asio::ip::tcp::endpoint * m_ProxyEndpoint;
+
+			uint16_t m_SoftLimit, m_HardLimit;
 		public:
 
 			// for HTTP/I2PControl

libi2pd/Transports.cpp

@@ -148,12 +148,19 @@ namespace transport
 		}
 
 		i2p::config::GetOption("nat", m_IsNAT);
-
 		m_DHKeysPairSupplier.Start ();
 		m_IsRunning = true;
 		m_Thread = new std::thread (std::bind (&Transports::Run, this));
 		std::string ntcpproxy; i2p::config::GetOption("ntcpproxy", ntcpproxy);
 		i2p::http::URL proxyurl;
+		uint16_t softLimit, hardLimit;
+		i2p::config::GetOption("limits.ntcpsoft", softLimit);
+		i2p::config::GetOption("limits.ntcphard", hardLimit);
+		if(softLimit >= hardLimit)
+		{
+			LogPrint(eLogError, "ntcp soft limit must be less than ntcp hard limit");
+			return;
+		}
 		if(ntcpproxy.size() && enableNTCP)
 		{
 			if(proxyurl.parse(ntcpproxy))
@@ -161,12 +168,11 @@ namespace transport
 				if(proxyurl.schema == "socks" || proxyurl.schema == "http")
 				{
 					m_NTCPServer = new NTCPServer();
-
+					m_NTCPServer->SetSessionLimits(softLimit, hardLimit);
 					NTCPServer::ProxyType proxytype = NTCPServer::eSocksProxy;
 
 					if (proxyurl.schema == "http")
 						proxytype = NTCPServer::eHTTPProxy;
-
 					m_NTCPServer->UseProxy(proxytype, proxyurl.host, proxyurl.port) ;
 					m_NTCPServer->Start();
 					if(!m_NTCPServer->NetworkIsReady())
@@ -193,6 +199,7 @@ namespace transport
 			if (m_NTCPServer == nullptr && enableNTCP)
 			{
 				m_NTCPServer = new NTCPServer ();
+				m_NTCPServer->SetSessionLimits(softLimit, hardLimit);
 				m_NTCPServer->Start ();
 				if (!(m_NTCPServer->IsBoundV6() || m_NTCPServer->IsBoundV4())) {
 					/** failed to bind to NTCP */
@@ -393,6 +400,8 @@ namespace transport
 #endif
 					{
 						if (!peer.router->UsesIntroducer () && !peer.router->IsUnreachable ())
+						{
+							if(!m_NTCPServer->ShouldLimit())
 							{
 								auto s = std::make_shared<NTCPSession> (*m_NTCPServer, peer.router);
 								if(m_NTCPServer->UsingProxy())
@@ -409,6 +418,11 @@ namespace transport
 									m_NTCPServer->Connect (address->host, address->port, s);
 								return true;
 							}
+							else
+							{
+								LogPrint(eLogWarning, "Transports: NTCP Limit hit falling back to SSU");
+							}
+						}
 					}
 					else // we don't have address
 					{