select sessions for introducers

libi2pd/SSU2.cpp

@@ -84,11 +84,17 @@ namespace transport
 	void SSU2Server::Stop ()
 	{
 		for (auto& it: m_Sessions)
+		{	
+			it.second->RequestTermination (eSSU2TerminationReasonRouterShutdown);
 			it.second->Done ();
+		}	
 		m_Sessions.clear ();
 		m_SessionsByRouterHash.clear ();
 		m_PendingOutgoingSessions.clear ();
-
+		m_Relays.clear ();
+		m_Introducers.clear ();
+		m_IntroducersV6.clear ();
+		
 		if (context.SupportsV4 () || context.SupportsV6 ())
 			m_ReceiveService.Stop ();
 
@@ -715,5 +721,58 @@ namespace transport
 		m_IncomingTokens.emplace (ep, ret);
 		return ret;
 	}	
+
+	std::list<std::shared_ptr<SSU2Session> > SSU2Server::FindIntroducers (int maxNumIntroducers, 
+		bool v4, const std::set<i2p::data::IdentHash>& excluded) const
+	{
+		std::list<std::shared_ptr<SSU2Session> > ret;
+		for (const auto& s : m_Sessions)
+		{
+			if (s.second->IsEstablished () && (s.second->GetRelayTag () && !s.second->IsOutgoing ()) &&	
+			    !excluded.count (s.second->GetRemoteIdentity ()->GetIdentHash ()) &&
+			    ((v4 && (s.second->GetRemoteTransports () | i2p::data::RouterInfo::eSSU2V4)) || 
+			     (!v4 && (s.second->GetRemoteTransports () | i2p::data::RouterInfo::eSSU2V6))))
+				ret.push_back (s.second);
+		}	
+		if ((int)ret.size () > maxNumIntroducers)
+		{
+			// shink ret randomly
+			int sz = ret.size () - maxNumIntroducers;
+			for (int i = 0; i < sz; i++)
+			{
+				auto ind = rand () % ret.size ();
+				auto it = ret.begin ();
+				std::advance (it, ind);
+				ret.erase (it);
+			}
+		}
+		return ret;
+	}	
+
+	void SSU2Server::UpdateIntroducers (bool v4)
+	{
+		std::list<std::shared_ptr<SSU2Session>> newList;
+		auto& introducers = v4 ? m_Introducers : m_IntroducersV6;
+		for (const auto& it : introducers)
+		{
+			if (it->IsEstablished ())
+			{
+				it->SendKeepAlive ();
+				newList.push_back (it);
+			}	
+		}	
+		if (newList.size () < SSU2_MAX_NUM_INTRODUCERS)
+		{
+			std::set<i2p::data::IdentHash> excluded;
+			for (auto& it1: newList)
+				excluded.insert (it1->GetRemoteIdentity ()->GetIdentHash ());
+			auto sessions = FindIntroducers (SSU_MAX_NUM_INTRODUCERS - newList.size (), v4, excluded);
+			for (const auto& it : sessions)
+			{
+				newList.push_back (it);
+			}	
+		}	
+		introducers = newList;
+	}	
 }
 }

libi2pd/SSU2.h

@@ -20,6 +20,9 @@ namespace transport
 	const int SSU2_TERMINATION_CHECK_TIMEOUT = 30; // 30 seconds
 	const size_t SSU2_SOCKET_RECEIVE_BUFFER_SIZE = 0x1FFFF; // 128K
 	const size_t SSU2_SOCKET_SEND_BUFFER_SIZE = 0x1FFFF; // 128K
+	const size_t SSU2_MAX_NUM_INTRODUCERS = 3;
+	const int SSU2_TO_INTRODUCER_SESSION_DURATION = 3600; // 1 hour
+	const int SSU2_TO_INTRODUCER_SESSION_EXPIRATION = 4800; // 80 minutes
 		
 	class SSU2Server: private i2p::util::RunnableServiceWithWork
 	{
@@ -97,7 +100,10 @@ namespace transport
 			void HandleResendTimer (const boost::system::error_code& ecode);
 
 			void ConnectThroughIntroducer (std::shared_ptr<SSU2Session> session);
-
+			std::list<std::shared_ptr<SSU2Session> > FindIntroducers (int maxNumIntroducers, 
+				bool v4, const std::set<i2p::data::IdentHash>& excluded) const;
+			void UpdateIntroducers (bool v4);
+		
 		private:
 
 			ReceiveService m_ReceiveService;
@@ -108,10 +114,11 @@ namespace transport
 			std::map<boost::asio::ip::udp::endpoint, std::shared_ptr<SSU2Session> > m_PendingOutgoingSessions;
 			std::map<boost::asio::ip::udp::endpoint, std::pair<uint64_t, uint32_t> > m_IncomingTokens, m_OutgoingTokens; // remote endpoint -> (token, expires in seconds)
 			std::map<uint32_t, std::shared_ptr<SSU2Session> > m_Relays; // we are introducer, relay tag -> session
+			std::list<std::shared_ptr<SSU2Session> > m_Introducers, m_IntroducersV6; // introducers we are connected to
 			i2p::util::MemoryPoolMt<Packet> m_PacketsPool;
 			boost::asio::deadline_timer m_TerminationTimer, m_ResendTimer;
 			std::shared_ptr<SSU2Session> m_LastSession;
-
+			
 		public:
 
 			// for HTTP/I2PControl

libi2pd/SSU2Session.cpp

@@ -150,6 +150,16 @@ namespace transport
 		payloadSize += CreatePaddingBlock (payload + payloadSize, SSU2_MAX_PAYLOAD_SIZE - payloadSize);
 		SendData (payload, payloadSize);
 	}	
+
+	void SSU2Session::SendKeepAlive ()
+	{
+		if (IsEstablished ())
+		{	
+			uint8_t payload[20];
+			size_t payloadSize = CreatePaddingBlock (payload, 20, 5);
+			SendData (payload, payloadSize);
+		}	
+	}	
 		
 	void SSU2Session::Terminate ()
 	{
@@ -416,7 +426,7 @@ namespace transport
 		htobe16buf (payload + 1, 4);
 		htobe32buf (payload + 3, ts);
 		size_t payloadSize = 7;
-		if (GetRouterStatus () == eRouterStatusFirewalled)
+		if (GetRouterStatus () == eRouterStatusFirewalled && m_Address->IsIntroducer ())
 		{
 			// relay tag request
 			payload[payloadSize] = eSSU2BlkRelayTagRequest;

libi2pd/SSU2Session.h

@@ -215,6 +215,7 @@ namespace transport
 			bool Introduce (std::shared_ptr<SSU2Session> session, uint32_t relayTag);
 			void WaitForIntroduction ();
 			void SendPeerTest (); // Alice, Data message
+			void SendKeepAlive ();
 			void Terminate ();
 			void RequestTermination (SSU2TerminationReason reason);
 			void CleanUp (uint64_t ts);