common termination timer for all NTCP sessions

NTCPSession.cpp

@@ -21,7 +21,8 @@ namespace transport
 	NTCPSession::NTCPSession (NTCPServer& server, std::shared_ptr<const i2p::data::RouterInfo> in_RemoteRouter): 
 		TransportSession (in_RemoteRouter, NTCP_TERMINATION_TIMEOUT),	
 		m_Server (server), m_Socket (m_Server.GetService ()), 
-		m_TerminationTimer (m_Server.GetService ()), m_IsEstablished (false), m_IsTerminated (false),
+		m_IsEstablished (false), m_IsTerminated (false),
+		m_LastActivityTimestamp (i2p::util::GetSecondsSinceEpoch ()),
 		m_ReceiveBufferOffset (0), m_NextMessage (nullptr), m_IsSending (false)
 	{		
 		m_Establisher = new Establisher;
@@ -78,7 +79,6 @@ namespace transport
 			m_Server.RemoveNTCPSession (shared_from_this ());
 			m_SendQueue.clear ();
 			m_NextMessage = nullptr;
-			m_TerminationTimer.cancel ();
 			LogPrint (eLogDebug, "NTCP: session terminated");
 		}	
 	}	
@@ -110,7 +110,6 @@ namespace transport
 		
 		boost::asio::async_write (m_Socket, boost::asio::buffer (&m_Establisher->phase1, sizeof (NTCPPhase1)), boost::asio::transfer_all (),
         	std::bind(&NTCPSession::HandlePhase1Sent, shared_from_this (), std::placeholders::_1, std::placeholders::_2));
-		ScheduleTermination ();
 	}	
 
 	void NTCPSession::ServerLogin ()
@@ -124,7 +123,6 @@ namespace transport
 			boost::asio::async_read (m_Socket, boost::asio::buffer(&m_Establisher->phase1, sizeof (NTCPPhase1)), boost::asio::transfer_all (),                    
 				std::bind(&NTCPSession::HandlePhase1Received, shared_from_this (), 
 					std::placeholders::_1, std::placeholders::_2));
-			ScheduleTermination ();	
 		}
 	}	
 		
@@ -558,7 +556,7 @@ namespace transport
 				m_Handler.Flush ();
 			}	
 			
-			ScheduleTermination (); // reset termination timer
+			m_LastActivityTimestamp = i2p::util::GetSecondsSinceEpoch ();
 			Receive ();
 		}	
 	}	
@@ -685,6 +683,7 @@ namespace transport
 		}
 		else
 		{	
+			m_LastActivityTimestamp = i2p::util::GetSecondsSinceEpoch ();
 			m_NumSentBytes += bytes_transferred;
 			i2p::transport::transports.UpdateSentBytes (bytes_transferred);
 			if (!m_SendQueue.empty())
@@ -692,8 +691,6 @@ namespace transport
 				Send (m_SendQueue);
 				m_SendQueue.clear ();
 			}	
-			else
-				ScheduleTermination (); // reset termination timer
 		}	
 	}	
 
@@ -729,28 +726,10 @@ namespace transport
 			Send (msgs);
 	}	
 
-	void NTCPSession::ScheduleTermination ()
-	{
-		m_TerminationTimer.cancel ();
-		m_TerminationTimer.expires_from_now (boost::posix_time::seconds(GetTerminationTimeout ()));
-		m_TerminationTimer.async_wait (std::bind (&NTCPSession::HandleTerminationTimer,
-			shared_from_this (), std::placeholders::_1));
-	}
-
-	void NTCPSession::HandleTerminationTimer (const boost::system::error_code& ecode)
-	{
-		if (ecode != boost::asio::error::operation_aborted)
-		{	
-			LogPrint (eLogDebug, "NTCP: No activity for ", GetTerminationTimeout (), " seconds");
-			//Terminate ();
-			m_Socket.close ();// invoke Terminate () from HandleReceive 
-		}	
-	}	
-
 //-----------------------------------------
 	NTCPServer::NTCPServer ():
 		m_IsRunning (false), m_Thread (nullptr), m_Work (m_Service), 
-		m_NTCPAcceptor (nullptr), m_NTCPV6Acceptor (nullptr)
+		m_TerminationTimer (m_Service), m_NTCPAcceptor (nullptr), m_NTCPV6Acceptor (nullptr)
 	{
 	}
 		
@@ -810,6 +789,7 @@ namespace transport
 					}	
 				}
 			}
+			ScheduleTermination ();	
 		}	
 	}
 		
@@ -820,13 +800,17 @@ namespace transport
 		if (m_IsRunning)
 		{	
 			m_IsRunning = false;
+			m_TerminationTimer.cancel ();	
 		  	if (m_NTCPAcceptor)
+			{	
 		    	delete m_NTCPAcceptor;
 				m_NTCPAcceptor = nullptr;
+			}
 		  	if (m_NTCPV6Acceptor)
+			{
 		    	delete m_NTCPV6Acceptor;
 				m_NTCPV6Acceptor = nullptr;
-
+			}
 			m_Service.stop ();
 			if (m_Thread)
 			{	
@@ -989,5 +973,31 @@ namespace transport
 		m_BanList[addr] = ts + NTCP_BAN_EXPIRATION_TIMEOUT;
 		LogPrint (eLogWarning, "NTCP: ", addr, " has been banned for ", NTCP_BAN_EXPIRATION_TIMEOUT, " seconds");
 	}
+
+	void NTCPServer::ScheduleTermination ()
+	{
+		m_TerminationTimer.expires_from_now (boost::posix_time::seconds(NTCP_TERMINATION_CHECK_TIMEOUT));
+		m_TerminationTimer.async_wait (std::bind (&NTCPServer::HandleTerminationTimer,
+			this, std::placeholders::_1));
+	}
+
+	void NTCPServer::HandleTerminationTimer (const boost::system::error_code& ecode)
+	{
+		if (ecode != boost::asio::error::operation_aborted)
+		{	
+			auto ts = i2p::util::GetSecondsSinceEpoch ();
+			for (auto& it: m_NTCPSessions)
+ 				if (it.second->IsTerminationTimeoutExpired (ts))
+				{
+					auto session = it.second;
+					m_Service.post ([session] 
+						{ 
+							LogPrint (eLogDebug, "NTCP: No activity for ", session->GetTerminationTimeout (), " seconds");
+							session->Terminate ();
+						});	
+				}
+			ScheduleTermination ();	
+		}	
+	}	
 }	
 }	

NTCPSession.h

@@ -37,6 +37,7 @@ namespace transport
 	const size_t NTCP_MAX_MESSAGE_SIZE = 16384; 
 	const size_t NTCP_BUFFER_SIZE = 4160; // fits 4 tunnel messages (4*1028)
 	const int NTCP_TERMINATION_TIMEOUT = 120; // 2 minutes
+	const int NTCP_TERMINATION_CHECK_TIMEOUT = 30; // 30 seconds	
 	const size_t NTCP_DEFAULT_PHASE3_SIZE = 2/*size*/ + i2p::data::DEFAULT_IDENTITY_SIZE/*387*/ + 4/*ts*/ + 15/*padding*/ + 40/*signature*/; // 448 	
 	const int NTCP_BAN_EXPIRATION_TIMEOUT = 70; // in second
 	const int NTCP_CLOCK_SKEW = 60; // in seconds 
@@ -54,6 +55,8 @@ namespace transport
 
 			boost::asio::ip::tcp::socket& GetSocket () { return m_Socket; };
 			bool IsEstablished () const { return m_IsEstablished; };
+			bool IsTerminationTimeoutExpired (uint64_t ts) const 
+			{ return ts >= m_LastActivityTimestamp + GetTerminationTimeout (); };		
 
 			void ClientLogin ();
 			void ServerLogin ();
@@ -95,17 +98,12 @@ namespace transport
 			void Send (const std::vector<std::shared_ptr<I2NPMessage> >& msgs);
 			void HandleSent (const boost::system::error_code& ecode, std::size_t bytes_transferred, std::vector<std::shared_ptr<I2NPMessage> > msgs);
 			
-			
-			// timer
-			void ScheduleTermination ();
-			void HandleTerminationTimer (const boost::system::error_code& ecode);
-			
 		private:
 
 			NTCPServer& m_Server;
 			boost::asio::ip::tcp::socket m_Socket;
-			boost::asio::deadline_timer m_TerminationTimer;
 			bool m_IsEstablished, m_IsTerminated;
+			uint64_t m_LastActivityTimestamp;
 			
 			i2p::crypto::CBCDecryption m_Decryption;
 			i2p::crypto::CBCEncryption m_Encryption;
@@ -160,12 +158,17 @@ namespace transport
 
 			void HandleConnect (const boost::system::error_code& ecode, std::shared_ptr<NTCPSession> conn);
 
+			// timer
+			void ScheduleTermination ();
+			void HandleTerminationTimer (const boost::system::error_code& ecode);
+			
 		private:	
 
 			bool m_IsRunning;
 			std::thread * m_Thread;	
 			boost::asio::io_service m_Service;
 			boost::asio::io_service::work m_Work;
+			boost::asio::deadline_timer m_TerminationTimer;
 			boost::asio::ip::tcp::acceptor * m_NTCPAcceptor, * m_NTCPV6Acceptor;
 			std::map<i2p::data::IdentHash, std::shared_ptr<NTCPSession> > m_NTCPSessions; // access from m_Thread only
 			std::map<boost::asio::ip::address, uint32_t> m_BanList; // IP -> ban expiration time in seconds