moved NTCP to separate thread

HTTPServer.cpp

@@ -708,9 +708,12 @@ namespace util
 	}	
 
 	void HTTPConnection::ShowTransports (std::stringstream& s)
+	{
+		auto ntcpServer = i2p::transport::transports.GetNTCPServer (); 
+		if (ntcpServer)
 		{	
 			s << "NTCP<br>";
-		for (auto it: i2p::transport::transports.GetNTCPSessions ())
+			for (auto it: ntcpServer->GetNTCPSessions ())
 			{
 				if (it.second && it.second->IsEstablished ())
 				{
@@ -725,6 +728,7 @@ namespace util
 				}
 				s << std::endl;
 			}
+		}	
 		auto ssuServer = i2p::transport::transports.GetSSUServer ();
 		if (ssuServer)
 		{

NTCPSession.cpp

@@ -18,9 +18,9 @@ namespace i2p
 {
 namespace transport
 {
-	NTCPSession::NTCPSession (boost::asio::io_service& service, std::shared_ptr<const i2p::data::RouterInfo> in_RemoteRouter): 
-		TransportSession (in_RemoteRouter),	m_Socket (service), 
-		m_TerminationTimer (service), m_IsEstablished (false), m_ReceiveBufferOffset (0), 
+	NTCPSession::NTCPSession (NTCPServer& server, std::shared_ptr<const i2p::data::RouterInfo> in_RemoteRouter): 
+		TransportSession (in_RemoteRouter),	m_Server (server), m_Socket (m_Server.GetService ()), 
+		m_TerminationTimer (m_Server.GetService ()), m_IsEstablished (false), m_ReceiveBufferOffset (0), 
 		m_NextMessage (nullptr), m_NumSentBytes (0), m_NumReceivedBytes (0)
 	{		
 		m_DHKeysPair = transports.GetNextDHKeysPair ();
@@ -89,7 +89,7 @@ namespace transport
 		if (numDelayed > 0)
 			LogPrint (eLogWarning, "NTCP session ", numDelayed, " not sent");
 		// TODO: notify tunnels
-		transports.RemoveNTCPSession (shared_from_this ());
+		m_Server.RemoveNTCPSession (shared_from_this ());
 		LogPrint ("NTCP session terminated");
 	}	
 
@@ -427,7 +427,7 @@ namespace transport
 		{	
 			LogPrint (eLogDebug, "Phase 4 sent: ", bytes_transferred);
 			LogPrint ("NTCP server session connected");
-			transports.AddNTCPSession (shared_from_this ());
+			m_Server.AddNTCPSession (shared_from_this ());
 
 			Connected ();
 			m_ReceiveBufferOffset = 0;
@@ -654,5 +654,177 @@ namespace transport
 			m_Socket.close ();// invoke Terminate () from HandleReceive 
 		}	
 	}	
+
+//-----------------------------------------
+	NTCPServer::NTCPServer (int port):
+		m_IsRunning (false), m_Thread (nullptr), m_Work (m_Service), 
+		m_NTCPAcceptor (nullptr), m_NTCPV6Acceptor (nullptr)
+	{
+	}
+		
+	NTCPServer::~NTCPServer ()
+	{
+		Stop ();
+	}	
+
+	void NTCPServer::Start ()
+	{
+		if (!m_IsRunning)
+		{	
+			m_IsRunning = true;
+			m_Thread = new std::thread (std::bind (&NTCPServer::Run, this));
+			// create acceptors
+			auto addresses = context.GetRouterInfo ().GetAddresses ();
+			for (auto& address : addresses)
+			{
+				if (address.transportStyle == i2p::data::RouterInfo::eTransportNTCP && address.host.is_v4 ())
+				{	
+					m_NTCPAcceptor = new boost::asio::ip::tcp::acceptor (m_Service,
+						boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), address.port));
+
+					LogPrint ("Start listening TCP port ", address.port);	
+					auto conn = std::make_shared<NTCPSession>(*this);
+					m_NTCPAcceptor->async_accept(conn->GetSocket (), std::bind (&NTCPServer::HandleAccept, this, 
+						conn, std::placeholders::_1));	
+				
+					if (context.SupportsV6 ())
+					{
+						m_NTCPV6Acceptor = new boost::asio::ip::tcp::acceptor (m_Service);
+						m_NTCPV6Acceptor->open (boost::asio::ip::tcp::v6());
+						m_NTCPV6Acceptor->set_option (boost::asio::ip::v6_only (true));
+						m_NTCPV6Acceptor->bind (boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v6(), address.port));
+						m_NTCPV6Acceptor->listen ();
+
+						LogPrint ("Start listening V6 TCP port ", address.port);	
+						auto conn = std::make_shared<NTCPSession> (*this);
+						m_NTCPV6Acceptor->async_accept(conn->GetSocket (), std::bind (&NTCPServer::HandleAcceptV6,
+							this, conn, std::placeholders::_1));
+					}	
+				}	
+			}	
+		}	
+	}
+		
+	void NTCPServer::Stop ()
+	{	
+		m_NTCPSessions.clear ();
+
+		if (m_IsRunning)
+		{	
+			m_IsRunning = false;
+			delete m_NTCPAcceptor;
+			m_NTCPAcceptor = nullptr;
+			delete m_NTCPV6Acceptor;
+			m_NTCPV6Acceptor = nullptr;
+
+			m_Service.stop ();
+			if (m_Thread)
+			{	
+				m_Thread->join (); 
+				delete m_Thread;
+				m_Thread = nullptr;
+			}	
+		}	
+	}	
+
+		
+	void NTCPServer::Run () 
+	{ 
+		while (m_IsRunning)
+		{
+			try
+			{	
+				m_Service.run ();
+			}
+			catch (std::exception& ex)
+			{
+				LogPrint ("NTCP server: ", ex.what ());
+			}	
+		}	
+	}	
+
+	void NTCPServer::AddNTCPSession (std::shared_ptr<NTCPSession> session)
+	{
+		if (session)
+			m_NTCPSessions[session->GetRemoteIdentity ().GetIdentHash ()] = session;
+	}	
+
+	void NTCPServer::RemoveNTCPSession (std::shared_ptr<NTCPSession> session)
+	{
+		if (session)
+			m_NTCPSessions.erase (session->GetRemoteIdentity ().GetIdentHash ());
+	}	
+
+	std::shared_ptr<NTCPSession> NTCPServer::FindNTCPSession (const i2p::data::IdentHash& ident)
+	{
+		auto it = m_NTCPSessions.find (ident);
+		if (it != m_NTCPSessions.end ())
+			return it->second;
+		return nullptr;
+	}	
+		
+	void NTCPServer::HandleAccept (std::shared_ptr<NTCPSession> conn, const boost::system::error_code& error)
+	{		
+		if (!error)
+		{
+			LogPrint ("Connected from ", conn->GetSocket ().remote_endpoint().address ().to_string ());
+			conn->ServerLogin ();
+		}
+		
+
+		if (error != boost::asio::error::operation_aborted)
+		{
+    		conn = std::make_shared<NTCPSession> (*this);
+			m_NTCPAcceptor->async_accept(conn->GetSocket (), std::bind (&NTCPServer::HandleAccept, this, 
+				conn, std::placeholders::_1));
+		}	
+	}
+
+	void NTCPServer::HandleAcceptV6 (std::shared_ptr<NTCPSession> conn, const boost::system::error_code& error)
+	{		
+		if (!error)
+		{
+			LogPrint ("Connected from ", conn->GetSocket ().remote_endpoint().address ().to_string ());
+			conn->ServerLogin ();
+		}
+
+		if (error != boost::asio::error::operation_aborted)
+		{
+    		conn = std::make_shared<NTCPSession> (*this);
+			m_NTCPV6Acceptor->async_accept(conn->GetSocket (), std::bind (&NTCPServer::HandleAcceptV6, this, 
+				conn, std::placeholders::_1));
+		}	
+	}	
+
+	void NTCPServer::Connect (const boost::asio::ip::address& address, int port, std::shared_ptr<NTCPSession> conn)
+	{
+		LogPrint ("Connecting to ", address ,":",  port);
+		m_Service.post([conn, this]()
+			{           
+				this->AddNTCPSession (conn);
+			});	
+		conn->GetSocket ().async_connect (boost::asio::ip::tcp::endpoint (address, port), 
+			std::bind (&NTCPServer::HandleConnect, this, std::placeholders::_1, conn));
+	}
+
+	void NTCPServer::HandleConnect (const boost::system::error_code& ecode, std::shared_ptr<NTCPSession> conn)
+	{
+		if (ecode)
+        {
+			LogPrint ("Connect error: ", ecode.message ());
+			if (ecode != boost::asio::error::operation_aborted)
+			{
+				i2p::data::netdb.SetUnreachable (conn->GetRemoteIdentity ().GetIdentHash (), true);
+				conn->Terminate ();
+			}
+		}
+		else
+		{
+			LogPrint ("Connected");
+			if (conn->GetSocket ().local_endpoint ().protocol () == boost::asio::ip::tcp::v6()) // ipv6
+				context.UpdateNTCPV6Address (conn->GetSocket ().local_endpoint ().address ());
+			conn->ClientLogin ();
+		}	
+	}	
 }	
 }	

NTCPSession.h

@@ -3,7 +3,10 @@
 
 #include <inttypes.h>
 #include <list>
+#include <map>
 #include <memory>
+#include <thread>
+#include <boost/asio.hpp>
 #include <cryptopp/modes.h>
 #include <cryptopp/aes.h>
 #include <cryptopp/adler32.h>
@@ -43,11 +46,12 @@ namespace transport
 	const int NTCP_TERMINATION_TIMEOUT = 120; // 2 minutes
 	const size_t NTCP_DEFAULT_PHASE3_SIZE = 2/*size*/ + i2p::data::DEFAULT_IDENTITY_SIZE/*387*/ + 4/*ts*/ + 15/*padding*/ + 40/*signature*/; // 448 	
 
+	class NTCPServer;
 	class NTCPSession: public TransportSession, public std::enable_shared_from_this<NTCPSession>
 	{
 		public:
 
-			NTCPSession (boost::asio::io_service& service, std::shared_ptr<const i2p::data::RouterInfo> in_RemoteRouter = nullptr);
+			NTCPSession (NTCPServer& server, std::shared_ptr<const i2p::data::RouterInfo> in_RemoteRouter = nullptr);
 			~NTCPSession ();
 			void Terminate ();
 
@@ -103,6 +107,7 @@ namespace transport
 			
 		private:
 
+			NTCPServer& m_Server;
 			boost::asio::ip::tcp::socket m_Socket;
 			boost::asio::deadline_timer m_TerminationTimer;
 			bool m_IsEstablished;
@@ -127,6 +132,47 @@ namespace transport
 
 			size_t m_NumSentBytes, m_NumReceivedBytes;
 	};	
+
+	// TODO: move to NTCP.h/.cpp
+	class NTCPServer
+	{
+		public:
+
+			NTCPServer (int port);
+			~NTCPServer ();
+
+			void Start ();
+			void Stop ();
+
+			void AddNTCPSession (std::shared_ptr<NTCPSession> session);
+			void RemoveNTCPSession (std::shared_ptr<NTCPSession> session);
+			std::shared_ptr<NTCPSession> FindNTCPSession (const i2p::data::IdentHash& ident);
+			void Connect (const boost::asio::ip::address& address, int port, std::shared_ptr<NTCPSession> conn);
+			
+			boost::asio::io_service& GetService () { return m_Service; };
+			
+		private:
+
+			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);
+
+			void HandleConnect (const boost::system::error_code& ecode, std::shared_ptr<NTCPSession> conn);
+			
+		private:	
+
+			bool m_IsRunning;
+			std::thread * m_Thread;	
+			boost::asio::io_service m_Service;
+			boost::asio::io_service::work m_Work;
+			boost::asio::ip::tcp::acceptor * m_NTCPAcceptor, * m_NTCPV6Acceptor;
+			std::map<i2p::data::IdentHash, std::shared_ptr<NTCPSession> > m_NTCPSessions;
+
+		public:
+
+			// for HTTP/I2PControl
+			const decltype(m_NTCPSessions)& GetNTCPSessions () const { return m_NTCPSessions; };
+	};	
 }	
 }	
 

Transports.cpp

@@ -96,8 +96,9 @@ namespace transport
 	Transports transports;	
 	
 	Transports::Transports (): 
-		m_Thread (nullptr), m_Work (m_Service), m_NTCPAcceptor (nullptr), m_NTCPV6Acceptor (nullptr), 
-		m_SSUServer (nullptr), m_DHKeysPairSupplier (5) // 5 pre-generated keys
+		m_IsRunning (false), m_Thread (nullptr), m_Work (m_Service), 
+		m_NTCPServer (nullptr), m_SSUServer (nullptr), 
+		m_DHKeysPairSupplier (5) // 5 pre-generated keys
 	{		
 	}
 		
@@ -115,31 +116,13 @@ namespace transport
 		auto addresses = context.GetRouterInfo ().GetAddresses ();
 		for (auto& address : addresses)
 		{
-			if (address.transportStyle == RouterInfo::eTransportNTCP && address.host.is_v4 ())
+			if (!m_NTCPServer)
 			{	
-				m_NTCPAcceptor = new boost::asio::ip::tcp::acceptor (m_Service,
-					boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), address.port));
-
-				LogPrint ("Start listening TCP port ", address.port);	
-				auto conn = std::make_shared<NTCPSession>(m_Service);
-				m_NTCPAcceptor->async_accept(conn->GetSocket (), boost::bind (&Transports::HandleAccept, this, 
-					conn, boost::asio::placeholders::error));	
-				
-				if (context.SupportsV6 ())
-				{
-					m_NTCPV6Acceptor = new boost::asio::ip::tcp::acceptor (m_Service);
-					m_NTCPV6Acceptor->open (boost::asio::ip::tcp::v6());
-					m_NTCPV6Acceptor->set_option (boost::asio::ip::v6_only (true));
-					m_NTCPV6Acceptor->bind (boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v6(), address.port));
-					m_NTCPV6Acceptor->listen ();
-
-					LogPrint ("Start listening V6 TCP port ", address.port);	
-					auto conn = std::make_shared<NTCPSession> (m_Service);
-					m_NTCPV6Acceptor->async_accept(conn->GetSocket (), boost::bind (&Transports::HandleAcceptV6,
-						this, conn, boost::asio::placeholders::error));
+				m_NTCPServer = new NTCPServer (address.port);
+				m_NTCPServer->Start ();
 			}	
-			}	
-			else if (address.transportStyle == RouterInfo::eTransportSSU && address.host.is_v4 ())
+			
+			if (address.transportStyle == RouterInfo::eTransportSSU && address.host.is_v4 ())
 			{
 				if (!m_SSUServer)
 				{	
@@ -162,12 +145,12 @@ namespace transport
 			delete m_SSUServer;
 			m_SSUServer = nullptr;
 		}	
-		m_NTCPSessions.clear ();
-		
-		delete m_NTCPAcceptor;
-		m_NTCPAcceptor = nullptr;
-		delete m_NTCPV6Acceptor;
-		m_NTCPV6Acceptor = nullptr;
+		if (m_NTCPServer)
+		{
+			m_NTCPServer->Stop ();
+			delete m_NTCPServer;
+			m_NTCPServer = nullptr;
+		}	
 
 		m_DHKeysPairSupplier.Stop ();
 		m_IsRunning = false;
@@ -195,93 +178,6 @@ namespace transport
 		}	
 	}
 		
-	void Transports::AddNTCPSession (std::shared_ptr<NTCPSession> session)
-	{
-		if (session)
-			m_NTCPSessions[session->GetRemoteIdentity ().GetIdentHash ()] = session;
-	}	
-
-	void Transports::RemoveNTCPSession (std::shared_ptr<NTCPSession> session)
-	{
-		if (session)
-			m_NTCPSessions.erase (session->GetRemoteIdentity ().GetIdentHash ());
-	}	
-		
-	void Transports::HandleAccept (std::shared_ptr<NTCPSession> conn, const boost::system::error_code& error)
-	{		
-		if (!error)
-		{
-			LogPrint ("Connected from ", conn->GetSocket ().remote_endpoint().address ().to_string ());
-			conn->ServerLogin ();
-		}
-		
-
-		if (error != boost::asio::error::operation_aborted)
-		{
-    		conn = std::make_shared<NTCPSession> (m_Service);
-			m_NTCPAcceptor->async_accept(conn->GetSocket (), boost::bind (&Transports::HandleAccept, this, 
-				conn, boost::asio::placeholders::error));
-		}	
-	}
-
-	void Transports::HandleAcceptV6 (std::shared_ptr<NTCPSession> conn, const boost::system::error_code& error)
-	{		
-		if (!error)
-		{
-			LogPrint ("Connected from ", conn->GetSocket ().remote_endpoint().address ().to_string ());
-			conn->ServerLogin ();
-		}
-
-		if (error != boost::asio::error::operation_aborted)
-		{
-    		conn = std::make_shared<NTCPSession> (m_Service);
-			m_NTCPV6Acceptor->async_accept(conn->GetSocket (), boost::bind (&Transports::HandleAcceptV6, this, 
-				conn, boost::asio::placeholders::error));
-		}	
-	}
-
-	void Transports::Connect (const boost::asio::ip::address& address, int port, std::shared_ptr<NTCPSession> conn)
-	{
-		LogPrint ("Connecting to ", address ,":",  port);
-		conn->GetSocket ().async_connect (boost::asio::ip::tcp::endpoint (address, port), 
-			boost::bind (&Transports::HandleConnect, this, boost::asio::placeholders::error, conn));
-	}
-
-	void Transports::HandleConnect (const boost::system::error_code& ecode, std::shared_ptr<NTCPSession> conn)
-	{
-		if (ecode)
-        {
-			LogPrint ("Connect error: ", ecode.message ());
-			if (ecode != boost::asio::error::operation_aborted)
-			{
-				i2p::data::netdb.SetUnreachable (conn->GetRemoteIdentity ().GetIdentHash (), true);
-				conn->Terminate ();
-			}
-		}
-		else
-		{
-			LogPrint ("Connected");
-			if (conn->GetSocket ().local_endpoint ().protocol () == boost::asio::ip::tcp::v6()) // ipv6
-				context.UpdateNTCPV6Address (conn->GetSocket ().local_endpoint ().address ());
-			conn->ClientLogin ();
-		}	
-	}	
-
-	std::shared_ptr<NTCPSession> Transports::GetNextNTCPSession ()
-	{
-		for (auto session: m_NTCPSessions)
-			if (session.second->IsEstablished ())
-				return session.second;
-		return 0;
-	}	
-
-	std::shared_ptr<NTCPSession> Transports::FindNTCPSession (const i2p::data::IdentHash& ident)
-	{
-		auto it = m_NTCPSessions.find (ident);
-		if (it != m_NTCPSessions.end ())
-			return it->second;
-		return 0;
-	}	
 
 	void Transports::SendMessage (const i2p::data::IdentHash& ident, i2p::I2NPMessage * msg)
 	{
@@ -296,7 +192,7 @@ namespace transport
 			i2p::HandleI2NPMessage (msg);
 			return;
 		}	
-		std::shared_ptr<TransportSession> session = FindNTCPSession (ident);
+		std::shared_ptr<TransportSession> session = m_NTCPServer->FindNTCPSession (ident);
 		if (!session)
 		{
 			auto r = netdb.FindRouter (ident);
@@ -311,10 +207,9 @@ namespace transport
 					auto address = r->GetNTCPAddress (!context.SupportsV6 ()); 
 					if (address && !r->UsesIntroducer () && !r->IsUnreachable () && msg->GetLength () < NTCP_MAX_MESSAGE_SIZE)
 					{	
-						auto s = std::make_shared<NTCPSession> (m_Service, r);
-						AddNTCPSession (s);
+						auto s = std::make_shared<NTCPSession> (*m_NTCPServer, r);
 						session = s;
-						Connect (address->host, address->port, s);
+						m_NTCPServer->Connect (address->host, address->port, s);
 					}	
 					else
 					{	

Transports.h

@@ -64,28 +64,17 @@ namespace transport
 			i2p::transport::DHKeysPair * GetNextDHKeysPair ();	
 			void ReuseDHKeysPair (DHKeysPair * pair);
 
-			void AddNTCPSession (std::shared_ptr<NTCPSession> session);
-			void RemoveNTCPSession (std::shared_ptr<NTCPSession> session);
-			
-			std::shared_ptr<NTCPSession> GetNextNTCPSession ();
-			std::shared_ptr<NTCPSession> FindNTCPSession (const i2p::data::IdentHash& ident);
-
 			void SendMessage (const i2p::data::IdentHash& ident, i2p::I2NPMessage * msg);
 			void CloseSession (std::shared_ptr<const i2p::data::RouterInfo> router);
 			
 		private:
 
 			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);
 			void HandleResendTimer (const boost::system::error_code& ecode, boost::asio::deadline_timer * timer,
 				const i2p::data::IdentHash& ident, i2p::I2NPMessage * msg);
 			void PostMessage (const i2p::data::IdentHash& ident, i2p::I2NPMessage * msg);
 			void PostCloseSession (std::shared_ptr<const i2p::data::RouterInfo> router);
 
-			void Connect (const boost::asio::ip::address& address, int port, std::shared_ptr<NTCPSession> conn);
-			void HandleConnect (const boost::system::error_code& ecode, std::shared_ptr<NTCPSession> conn);
-
 			void DetectExternalIP ();
 			
 		private:
@@ -94,9 +83,8 @@ namespace transport
 			std::thread * m_Thread;	
 			boost::asio::io_service m_Service;
 			boost::asio::io_service::work m_Work;
-			boost::asio::ip::tcp::acceptor * m_NTCPAcceptor, * m_NTCPV6Acceptor;
 
-			std::map<i2p::data::IdentHash, std::shared_ptr<NTCPSession> > m_NTCPSessions;
+			NTCPServer * m_NTCPServer;
 			SSUServer * m_SSUServer;
 
 			DHKeysPairSupplier m_DHKeysPairSupplier;
@@ -104,7 +92,7 @@ namespace transport
 		public:
 
 			// for HTTP only
-			const decltype(m_NTCPSessions)& GetNTCPSessions () const { return m_NTCPSessions; };
+			const NTCPServer * GetNTCPServer () const { return m_NTCPServer; };
 			const SSUServer * GetSSUServer () const { return m_SSUServer; };
 	};