i2pd/NetDb.cpp

#include <string.h>
#include "I2PEndian.h"
#include <fstream>
#include <vector>
#include <boost/asio.hpp>
#include <cryptopp/gzip.h>
#include "base64.h"
#include "Log.h"
#include "Timestamp.h"
#include "I2NPProtocol.h"
#include "Tunnel.h"
#include "Transports.h"
#include "RouterContext.h"
#include "Garlic.h"
#include "NetDb.h"
#include "util.h"

using namespace i2p::transport;

namespace i2p
{
namespace data
{		
	const char NetDb::m_NetDbPath[] = "netDb";
	NetDb netdb;

	NetDb::NetDb (): m_IsRunning (false), m_Thread (nullptr), m_Reseeder (nullptr)
	{
	}
	
	NetDb::~NetDb ()
	{
		Stop ();	
		delete m_Reseeder;
	}	

	void NetDb::Start ()
	{	
		Load ();
		if (m_RouterInfos.size () < 25) // reseed if # of router less than 50
		{	
			// try SU3 first
			Reseed ();

			// deprecated
			if (m_Reseeder)
			{
				// if still not enough download .dat files
				int reseedRetries = 0;
				while (m_RouterInfos.size () < 25 && reseedRetries < 5)
				{
					m_Reseeder->reseedNow();
					reseedRetries++;
					Load ();
				}
			}
		}	
		m_IsRunning = true;
		m_Thread = new std::thread (std::bind (&NetDb::Run, this));
	}
	
	void NetDb::Stop ()
	{
		if (m_IsRunning)
		{	
			for (auto it: m_RouterInfos)
				it.second->SaveProfile ();
			DeleteObsoleteProfiles ();
			m_RouterInfos.clear ();
			m_Floodfills.clear ();
			if (m_Thread)
			{	
				m_IsRunning = false;
				m_Queue.WakeUp ();
				m_Thread->join (); 
				delete m_Thread;
				m_Thread = 0;
			}
			m_LeaseSets.clear();
			m_Requests.Stop ();
		}	
	}	
	
	void NetDb::Run ()
	{
		uint32_t lastSave = 0, lastPublish = 0, lastExploratory = 0, lastManageRequest = 0;
		while (m_IsRunning)
		{	
			try
			{	
				auto msg = m_Queue.GetNextWithTimeout (15000); // 15 sec
				if (msg)
				{	
					int numMsgs = 0;	
					while (msg)
					{
						switch (msg->GetTypeID ()) 
						{
							case eI2NPDatabaseStore:	
								LogPrint ("DatabaseStore");
								HandleDatabaseStoreMsg (msg);
							break;
							case eI2NPDatabaseSearchReply:
								LogPrint ("DatabaseSearchReply");
								HandleDatabaseSearchReplyMsg (msg);
							break;
							case eI2NPDatabaseLookup:
								LogPrint ("DatabaseLookup");
								HandleDatabaseLookupMsg (msg);
							break;	
							default: // WTF?
								LogPrint (eLogError, "NetDb: unexpected message type ", msg->GetTypeID ());
								//i2p::HandleI2NPMessage (msg);
						}	
						if (numMsgs > 100) break;
						msg = m_Queue.Get ();
						numMsgs++;
					}	
				}			
				if (!m_IsRunning) break;

				uint64_t ts = i2p::util::GetSecondsSinceEpoch ();
				if (ts - lastManageRequest >= 15) // manage requests every 15 seconds
				{
					m_Requests.ManageRequests ();
					lastManageRequest = ts;
				}	
				if (ts - lastSave >= 60) // save routers, manage leasesets and validate subscriptions every minute
				{
					if (lastSave)
					{
						SaveUpdated ();
						ManageLeaseSets ();
					}	
					lastSave = ts;
				}	
				if (ts - lastPublish >= 2400) // publish every 40 minutes
				{
					Publish ();
					lastPublish = ts;
				}	
				if (ts - lastExploratory >= 30) // exploratory every 30 seconds
				{	
					auto numRouters = m_RouterInfos.size ();
					if (numRouters < 2500 || ts - lastExploratory >= 90)
					{	
						numRouters = 800/numRouters;
						if (numRouters < 1) numRouters = 1;
						if (numRouters > 9) numRouters = 9;	
						m_Requests.ManageRequests ();					
						Explore (numRouters);
						lastExploratory = ts;
					}	
				}	
			}
			catch (std::exception& ex)
			{
				LogPrint ("NetDb: ", ex.what ());
			}	
		}	
	}	
	
	void NetDb::AddRouterInfo (const uint8_t * buf, int len)
	{
		IdentityEx identity;
		if (identity.FromBuffer (buf, len))
			AddRouterInfo (identity.GetIdentHash (), buf, len);	
	}

	void NetDb::AddRouterInfo (const IdentHash& ident, const uint8_t * buf, int len)
	{	
		auto r = FindRouter (ident);
		if (r)
		{
			auto ts = r->GetTimestamp ();
			r->Update (buf, len);
			if (r->GetTimestamp () > ts)
				LogPrint ("RouterInfo updated");
		}	
		else	
		{	
			LogPrint ("New RouterInfo added");
			r = std::make_shared<RouterInfo> (buf, len);
			{
				std::unique_lock<std::mutex> l(m_RouterInfosMutex);
				m_RouterInfos[r->GetIdentHash ()] = r;
			}
			if (r->IsFloodfill ())
			{
				std::unique_lock<std::mutex> l(m_FloodfillsMutex);
				m_Floodfills.push_back (r);
			}	
		}	
		// take care about requested destination
		m_Requests.RequestComplete (ident, r);
	}	

	void NetDb::AddLeaseSet (const IdentHash& ident, const uint8_t * buf, int len,
		std::shared_ptr<i2p::tunnel::InboundTunnel> from)
	{
		if (!from) // unsolicited LS must be received directly
		{	
			auto it = m_LeaseSets.find(ident);
			if (it != m_LeaseSets.end ())
			{
				it->second->Update (buf, len); 
				if (it->second->IsValid ())
					LogPrint (eLogInfo, "LeaseSet updated");
				else
				{
					LogPrint (eLogInfo, "LeaseSet update failed");
					m_LeaseSets.erase (it);
				}	
			}
			else
			{	
				auto leaseSet = std::make_shared<LeaseSet> (buf, len);
				if (leaseSet->IsValid ())
				{
					LogPrint (eLogInfo, "New LeaseSet added");
					m_LeaseSets[ident] = leaseSet;
				}
				else
					LogPrint (eLogError, "New LeaseSet validation failed");
			}	
		}	
	}	

	std::shared_ptr<RouterInfo> NetDb::FindRouter (const IdentHash& ident) const
	{
		std::unique_lock<std::mutex> l(m_RouterInfosMutex);
		auto it = m_RouterInfos.find (ident);
		if (it != m_RouterInfos.end ())
			return it->second;
		else
			return nullptr;
	}

	std::shared_ptr<LeaseSet> NetDb::FindLeaseSet (const IdentHash& destination) const
	{
		auto it = m_LeaseSets.find (destination);
		if (it != m_LeaseSets.end ())
			return it->second;
		else
			return nullptr;
	}

	void NetDb::SetUnreachable (const IdentHash& ident, bool unreachable)
	{
		auto it = m_RouterInfos.find (ident);
		if (it != m_RouterInfos.end ())
			return it->second->SetUnreachable (unreachable);
	}

	// TODO: Move to reseed and/or scheduled tasks. (In java version, scheduler fix this as well as sort RIs.)
	bool NetDb::CreateNetDb(boost::filesystem::path directory)
	{
		LogPrint (directory.string(), " doesn't exist, trying to create it.");
		if (!boost::filesystem::create_directory (directory))
		{
			LogPrint("Failed to create directory ", directory.string());
			return false;
		}

		// list of chars might appear in base64 string
		const char * chars = GetBase64SubstitutionTable (); // 64 bytes
		boost::filesystem::path suffix;
		for (int i = 0; i < 64; i++)
		{
#ifndef _WIN32
			suffix = std::string ("/r") + chars[i];
#else
			suffix = std::string ("\\r") + chars[i];
#endif
			if (!boost::filesystem::create_directory( boost::filesystem::path (directory / suffix) )) return false;
		}
		return true;
	}

	void NetDb::Reseed ()
	{
		if (!m_Reseeder)
		{		
			m_Reseeder = new Reseeder ();
			m_Reseeder->LoadCertificates (); // we need certificates for SU3 verification
		}
		int reseedRetries = 0;	
		while (reseedRetries < 10 && !m_Reseeder->ReseedNowSU3 ())
			reseedRetries++;
		if (reseedRetries >= 10)
			LogPrint (eLogWarning, "Failed to reseed after 10 attempts");
	}

	void NetDb::Load ()
	{
		boost::filesystem::path p(i2p::util::filesystem::GetDataDir() / m_NetDbPath);
		if (!boost::filesystem::exists (p))
		{
			// seems netDb doesn't exist yet
			if (!CreateNetDb(p)) return;
		}
		// make sure we cleanup netDb from previous attempts
		m_RouterInfos.clear ();	
		m_Floodfills.clear ();	

		// load routers now
		uint64_t ts = i2p::util::GetMillisecondsSinceEpoch ();	
		int numRouters = 0;
		boost::filesystem::directory_iterator end;
		for (boost::filesystem::directory_iterator it (p); it != end; ++it)
		{
			if (boost::filesystem::is_directory (it->status()))
			{
				for (boost::filesystem::directory_iterator it1 (it->path ()); it1 != end; ++it1)
				{
#if BOOST_VERSION > 10500
					const std::string& fullPath = it1->path().string();
#else
					const std::string& fullPath = it1->path();
#endif
					auto r = std::make_shared<RouterInfo>(fullPath);
					if (!r->IsUnreachable () && (!r->UsesIntroducer () || ts < r->GetTimestamp () + 3600*1000LL)) // 1 hour
					{	
						r->DeleteBuffer ();
						r->ClearProperties (); // properties are not used for regular routers
						m_RouterInfos[r->GetIdentHash ()] = r;
						if (r->IsFloodfill ())
							m_Floodfills.push_back (r);
						numRouters++;
					}	
					else
					{	
						if (boost::filesystem::exists (fullPath))  
							boost::filesystem::remove (fullPath);
					}	
				}	
			}	
		}
		LogPrint (numRouters, " routers loaded");
		LogPrint (m_Floodfills.size (), " floodfills loaded");	
	}	

	void NetDb::SaveUpdated ()
	{	
		auto GetFilePath = [](const boost::filesystem::path& directory, const RouterInfo * routerInfo)
		{
			std::string s(routerInfo->GetIdentHashBase64());
			return directory / (std::string("r") + s[0]) / ("routerInfo-" + s + ".dat");
		};	

		boost::filesystem::path fullDirectory (i2p::util::filesystem::GetDataDir() / m_NetDbPath);
		int count = 0, deletedCount = 0;
		auto total = m_RouterInfos.size ();
		uint64_t ts = i2p::util::GetMillisecondsSinceEpoch ();
		for (auto it: m_RouterInfos)
		{	
			if (it.second->IsUpdated ())
			{
				std::string f = GetFilePath(fullDirectory, it.second.get()).string();
				it.second->SaveToFile (f);
				it.second->SetUpdated (false);
				it.second->SetUnreachable (false);
				it.second->DeleteBuffer ();
				count++;
			}
			else 
			{
				// RouterInfo expires after 1 hour if uses introducer
				if (it.second->UsesIntroducer () && ts > it.second->GetTimestamp () + 3600*1000LL) // 1 hour
					it.second->SetUnreachable (true);
				else if (total > 75 && ts > (i2p::context.GetStartupTime () + 600)*1000LL) // routers don't expire if less than 25 or uptime is less than 10 minutes
				{
					if (i2p::context.IsFloodfill ())
					{
						if (ts > it.second->GetTimestamp () + 3600*1000LL) // 1 hours
						{	
							it.second->SetUnreachable (true);
							total--;
						}	
					}
					else if (total > 300)
					{
						if (ts > it.second->GetTimestamp () + 30*3600*1000LL) // 30 hours
						{	
							it.second->SetUnreachable (true);
							total--;
						}	
					}
					else if (total > 120)
					{
						if (ts > it.second->GetTimestamp () + 72*3600*1000LL) // 72 hours
						{	
							it.second->SetUnreachable (true);
							total--;
						}	
					}
				}
				
				if (it.second->IsUnreachable ())
				{	
					total--;
					// delete RI file
					if (boost::filesystem::exists (GetFilePath (fullDirectory, it.second.get ())))
					{    
						boost::filesystem::remove (GetFilePath (fullDirectory, it.second.get ()));
						deletedCount++;
					}	
					// delete from floodfills list
					if (it.second->IsFloodfill ())
					{
						std::unique_lock<std::mutex> l(m_FloodfillsMutex);
						m_Floodfills.remove (it.second);
					}
				}
			}	
		}	
		if (count > 0)
			LogPrint (count," new/updated routers saved");
		if (deletedCount > 0)
		{
			LogPrint (deletedCount," routers deleted");
			// clean up RouterInfos table
			std::unique_lock<std::mutex> l(m_RouterInfosMutex);
			for (auto it = m_RouterInfos.begin (); it != m_RouterInfos.end ();)
			{
				if (it->second->IsUnreachable ())
				{	
					it->second->SaveProfile ();
					it = m_RouterInfos.erase (it);
				}	
				else
					it++;
			}
		}
	}

	void NetDb::RequestDestination (const IdentHash& destination, RequestedDestination::RequestComplete requestComplete)
	{
		auto dest = m_Requests.CreateRequest (destination, false, requestComplete); // non-exploratory
		if (!dest)
		{
			LogPrint (eLogWarning, "Destination ", destination.ToBase64(), " is requested already"); 
			return;			
		}

		auto floodfill = GetClosestFloodfill (destination, dest->GetExcludedPeers ());
		if (floodfill)
			transports.SendMessage (floodfill->GetIdentHash (), dest->CreateRequestMessage (floodfill->GetIdentHash ()));	
		else
		{
			LogPrint (eLogError, "No floodfills found");
			m_Requests.RequestComplete (destination, nullptr);
		}	
	}	
	
	void NetDb::HandleDatabaseStoreMsg (std::shared_ptr<I2NPMessage> m)
	{	
		const uint8_t * buf = m->GetPayload ();
		size_t len = m->GetSize ();		
		IdentHash ident (buf + DATABASE_STORE_KEY_OFFSET);
		if (ident.IsZero ())
		{
			LogPrint (eLogError, "Database store with zero ident. Dropped");
			return;
		}	
		uint32_t replyToken = bufbe32toh (buf + DATABASE_STORE_REPLY_TOKEN_OFFSET);
		size_t offset = DATABASE_STORE_HEADER_SIZE;
		if (replyToken)
		{
			auto deliveryStatus = CreateDeliveryStatusMsg (replyToken);			
			uint32_t tunnelID = bufbe32toh (buf + offset);
			offset += 4;
			if (!tunnelID) // send response directly
				transports.SendMessage (buf + offset, deliveryStatus);
			else
			{
				auto pool = i2p::tunnel::tunnels.GetExploratoryPool ();
				auto outbound = pool ? pool->GetNextOutboundTunnel () : nullptr;
				if (outbound)
					outbound->SendTunnelDataMsg (buf + offset, tunnelID, deliveryStatus);
				else
					LogPrint (eLogError, "No outbound tunnels for DatabaseStore reply found");
			}		
			offset += 32;

			if (context.IsFloodfill ())
			{
				// flood it
				auto floodMsg = ToSharedI2NPMessage (NewI2NPShortMessage ());
				uint8_t * payload = floodMsg->GetPayload ();		
				memcpy (payload, buf, 33); // key + type
				htobe32buf (payload + DATABASE_STORE_REPLY_TOKEN_OFFSET, 0); // zero reply token
				memcpy (payload + DATABASE_STORE_HEADER_SIZE, buf + offset, len - offset);
				floodMsg->len += DATABASE_STORE_HEADER_SIZE + len -offset;
				FillI2NPMessageHeader (floodMsg.get (), eI2NPDatabaseStore); // TODO
				std::set<IdentHash> excluded;
				for (int i = 0; i < 3; i++)
				{
					auto floodfill = GetClosestFloodfill (ident, excluded);
					if (floodfill)
						transports.SendMessage (floodfill->GetIdentHash (), floodMsg);
				}	
			}	
		}
		
		if (buf[DATABASE_STORE_TYPE_OFFSET]) // type
		{
			LogPrint ("LeaseSet");
			AddLeaseSet (ident, buf + offset, len - offset, m->from);
		}	
		else
		{
			LogPrint ("RouterInfo");
			size_t size = bufbe16toh (buf + offset);
			offset += 2;
			if (size > 2048 || size > len - offset)
			{
				LogPrint ("Invalid RouterInfo length ", (int)size);
				return;
			}	
			try
			{	
				CryptoPP::Gunzip decompressor;
				decompressor.Put (buf + offset, size);
				decompressor.MessageEnd();
				uint8_t uncompressed[2048];
				size_t uncomressedSize = decompressor.MaxRetrievable ();
				if (uncomressedSize <= 2048)
				{
					decompressor.Get (uncompressed, uncomressedSize);
					AddRouterInfo (ident, uncompressed, uncomressedSize);
				}
				else
					LogPrint ("Invalid RouterInfo uncomressed length ", (int)uncomressedSize);
			}
			catch (CryptoPP::Exception& ex)
			{
				LogPrint (eLogError, "DatabaseStore: ", ex.what ());
			}
		}	
	}	

	void NetDb::HandleDatabaseSearchReplyMsg (std::shared_ptr<I2NPMessage> msg)
	{
		uint8_t * buf = msg->GetPayload ();
		char key[48];
		int l = i2p::data::ByteStreamToBase64 (buf, 32, key, 48);
		key[l] = 0;
		int num = buf[32]; // num
		LogPrint ("DatabaseSearchReply for ", key, " num=", num);
		IdentHash ident (buf);
		auto dest = m_Requests.FindRequest (ident); 
		if (dest)
		{	
			bool deleteDest = true;
			if (num > 0)
			{	
				auto pool = i2p::tunnel::tunnels.GetExploratoryPool ();
				auto outbound = pool ? pool->GetNextOutboundTunnel () : nullptr;
				auto inbound = pool ? pool->GetNextInboundTunnel () : nullptr;
				if (!dest->IsExploratory ())
				{
					// reply to our destination. Try other floodfills
					if (outbound && inbound )
					{
						std::vector<i2p::tunnel::TunnelMessageBlock> msgs;
						auto count = dest->GetExcludedPeers ().size ();
						if (count < 7)
						{	
							auto nextFloodfill = GetClosestFloodfill (dest->GetDestination (), dest->GetExcludedPeers ());
							if (nextFloodfill)
							{	
								// tell floodfill about us 
								msgs.push_back (i2p::tunnel::TunnelMessageBlock 
									{ 
										i2p::tunnel::eDeliveryTypeRouter,
										nextFloodfill->GetIdentHash (), 0,
										ToSharedI2NPMessage (CreateDatabaseStoreMsg ()) 
									});  
								
								// request destination
								LogPrint ("Try ", key, " at ", count, " floodfill ", nextFloodfill->GetIdentHash ().ToBase64 ()); 
								auto msg = dest->CreateRequestMessage (nextFloodfill, inbound);
								msgs.push_back (i2p::tunnel::TunnelMessageBlock 
									{ 
										i2p::tunnel::eDeliveryTypeRouter,
										nextFloodfill->GetIdentHash (), 0, msg
									});
								deleteDest = false;
							}	
						}
						else
							LogPrint (key, " was not found on 7 floodfills");

						if (msgs.size () > 0)
							outbound->SendTunnelDataMsg (msgs);	
					}	
				}	

				if (deleteDest)
					// no more requests for the destinationation. delete it
					m_Requests.RequestComplete (ident, nullptr);
			}
			else
				// no more requests for detination possible. delete it
				m_Requests.RequestComplete (ident, nullptr);
		}
		else	
			LogPrint ("Requested destination for ", key, " not found");

		// try responses
		for (int i = 0; i < num; i++)
		{
			uint8_t * router = buf + 33 + i*32;
			char peerHash[48];
			int l1 = i2p::data::ByteStreamToBase64 (router, 32, peerHash, 48);
			peerHash[l1] = 0;
			LogPrint (i,": ", peerHash);

			auto r = FindRouter (router); 
			if (!r || i2p::util::GetMillisecondsSinceEpoch () > r->GetTimestamp () + 3600*1000LL) 
			{	
				// router with ident not found or too old (1 hour)
				LogPrint ("Found new/outdated router. Requesting RouterInfo ...");
				RequestDestination (router);
			}
			else
				LogPrint ("Bayan");	
		}	
	}	
	
	void NetDb::HandleDatabaseLookupMsg (std::shared_ptr<I2NPMessage> msg)
	{
		uint8_t * buf = msg->GetPayload ();
		IdentHash ident (buf);
		if (ident.IsZero ())
		{
			LogPrint (eLogError, "DatabaseLookup for zero ident. Ignored");
			return;
		}	
		char key[48];
		int l = i2p::data::ByteStreamToBase64 (buf, 32, key, 48);
		key[l] = 0;
		uint8_t flag = buf[64];
		LogPrint ("DatabaseLookup for ", key, " recieved flags=", (int)flag);
		uint8_t lookupType = flag & DATABASE_LOOKUP_TYPE_FLAGS_MASK;
		uint8_t * excluded = buf + 65;		
		uint32_t replyTunnelID = 0;
		if (flag & DATABASE_LOOKUP_DELIVERY_FLAG) //reply to tunnel
		{
			replyTunnelID = bufbe32toh (buf + 64);
			excluded += 4;
		}
		uint16_t numExcluded = bufbe16toh (excluded);	
		excluded += 2;
		if (numExcluded > 512)
		{
			LogPrint ("Number of excluded peers", numExcluded, " exceeds 512");
			numExcluded = 0; // TODO:
		} 
		
		std::shared_ptr<I2NPMessage> replyMsg;
		if (lookupType == DATABASE_LOOKUP_TYPE_EXPLORATORY_LOOKUP)
		{
			LogPrint ("Exploratory close to  ", key, " ", numExcluded, " excluded");
			std::set<IdentHash> excludedRouters;
			for (int i = 0; i < numExcluded; i++)
			{
				excludedRouters.insert (excluded);
				excluded += 32;
			}	
			std::vector<IdentHash> routers;
			for (int i = 0; i < 3; i++)
			{
				auto r = GetClosestNonFloodfill (ident, excludedRouters);
				if (r)
				{	
					routers.push_back (r->GetIdentHash ());
					excludedRouters.insert (r->GetIdentHash ());
				}	
			}	
			replyMsg = ToSharedI2NPMessage (CreateDatabaseSearchReply (ident, routers));
		}	
		else
		{	
			if (lookupType == DATABASE_LOOKUP_TYPE_ROUTERINFO_LOOKUP  ||
			    lookupType == DATABASE_LOOKUP_TYPE_NORMAL_LOOKUP)
			{	
				auto router = FindRouter (ident);
				if (router)
				{
					LogPrint ("Requested RouterInfo ", key, " found");
					router->LoadBuffer ();
					if (router->GetBuffer ()) 
						replyMsg = ToSharedI2NPMessage (CreateDatabaseStoreMsg (router));
				}
			}
			
			if (!replyMsg && (lookupType == DATABASE_LOOKUP_TYPE_LEASESET_LOOKUP  ||
			    lookupType == DATABASE_LOOKUP_TYPE_NORMAL_LOOKUP))
			{
				auto leaseSet = FindLeaseSet (ident);
				if (leaseSet) // we don't send back our LeaseSets
				{
					LogPrint ("Requested LeaseSet ", key, " found");
					replyMsg = ToSharedI2NPMessage (CreateDatabaseStoreMsg (leaseSet));
				}
			}
			
			if (!replyMsg)
			{
				LogPrint ("Requested ", key, " not found. ", numExcluded, " excluded");
				std::set<IdentHash> excludedRouters;	
				for (int i = 0; i < numExcluded; i++)
				{
					excludedRouters.insert (excluded);
					excluded += 32;
				}
				replyMsg = ToSharedI2NPMessage (CreateDatabaseSearchReply (ident, GetClosestFloodfills (ident, 3, excludedRouters)));
			}
		}
		
		if (replyMsg)
		{	
			if (replyTunnelID)
			{
				// encryption might be used though tunnel only
				if (flag & DATABASE_LOOKUP_ENCYPTION_FLAG) // encrypted reply requested
				{
					uint8_t * sessionKey = excluded;
					uint8_t numTags = sessionKey[32];
					if (numTags > 0) 
					{
						uint8_t * sessionTag = sessionKey + 33; // take first tag
						i2p::garlic::GarlicRoutingSession garlic (sessionKey, sessionTag);
						replyMsg = garlic.WrapSingleMessage (replyMsg);
					}
				}	
				auto exploratoryPool = i2p::tunnel::tunnels.GetExploratoryPool ();
				auto outbound = exploratoryPool ? exploratoryPool->GetNextOutboundTunnel () : nullptr;
				if (outbound)
					outbound->SendTunnelDataMsg (buf+32, replyTunnelID, replyMsg);
				else
					transports.SendMessage (buf+32, i2p::CreateTunnelGatewayMsg (replyTunnelID, replyMsg));
			}
			else
				transports.SendMessage (buf+32, replyMsg);
		}
	}	

	void NetDb::Explore (int numDestinations)
	{	
		// new requests
		auto exploratoryPool = i2p::tunnel::tunnels.GetExploratoryPool ();
		auto outbound = exploratoryPool ? exploratoryPool->GetNextOutboundTunnel () : nullptr;
		auto inbound = exploratoryPool ? exploratoryPool->GetNextInboundTunnel () : nullptr;
		bool throughTunnels = outbound && inbound;
		
		CryptoPP::RandomNumberGenerator& rnd = i2p::context.GetRandomNumberGenerator ();
		uint8_t randomHash[32];
		std::vector<i2p::tunnel::TunnelMessageBlock> msgs;
		std::set<const RouterInfo *> floodfills;
		LogPrint ("Exploring new ", numDestinations, " routers ...");
		for (int i = 0; i < numDestinations; i++)
		{	
			rnd.GenerateBlock (randomHash, 32);
			auto dest = m_Requests.CreateRequest (randomHash, true); // exploratory
			if (!dest)
			{	
				LogPrint (eLogWarning, "Exploratory destination is requested already");
				return; 	
			}	
			auto floodfill = GetClosestFloodfill (randomHash, dest->GetExcludedPeers ());
			if (floodfill && !floodfills.count (floodfill.get ())) // request floodfill only once
			{	
				floodfills.insert (floodfill.get ());
				if (i2p::transport::transports.IsConnected (floodfill->GetIdentHash ()))
					throughTunnels = false;
				if (throughTunnels)
				{	
					msgs.push_back (i2p::tunnel::TunnelMessageBlock 
						{ 
							i2p::tunnel::eDeliveryTypeRouter,
							floodfill->GetIdentHash (), 0,
							ToSharedI2NPMessage (CreateDatabaseStoreMsg ()) // tell floodfill about us 
						});  
					msgs.push_back (i2p::tunnel::TunnelMessageBlock 
						{ 
							i2p::tunnel::eDeliveryTypeRouter,
							floodfill->GetIdentHash (), 0, 
							dest->CreateRequestMessage (floodfill, inbound) // explore
						}); 
				}	
				else
					i2p::transport::transports.SendMessage (floodfill->GetIdentHash (), dest->CreateRequestMessage (floodfill->GetIdentHash ()));
			}	
			else
				m_Requests.RequestComplete (randomHash, nullptr);
		}	
		if (throughTunnels && msgs.size () > 0)
			outbound->SendTunnelDataMsg (msgs);		
	}	

	void NetDb::Publish ()
	{
		std::set<IdentHash> excluded; // TODO: fill up later
		for (int i = 0; i < 2; i++)
		{	
			auto floodfill = GetClosestFloodfill (i2p::context.GetRouterInfo ().GetIdentHash (), excluded);
			if (floodfill)
			{
				uint32_t replyToken = i2p::context.GetRandomNumberGenerator ().GenerateWord32 ();
				LogPrint ("Publishing our RouterInfo to ", floodfill->GetIdentHashAbbreviation (), ". reply token=", replyToken);
				transports.SendMessage (floodfill->GetIdentHash (), ToSharedI2NPMessage (CreateDatabaseStoreMsg (i2p::context.GetSharedRouterInfo (), replyToken)));	
				excluded.insert (floodfill->GetIdentHash ());
			}
		}	
	}		

	std::shared_ptr<const RouterInfo> NetDb::GetRandomRouter () const
	{
		return GetRandomRouter (
			[](std::shared_ptr<const RouterInfo> router)->bool 
			{ 
				return !router->IsHidden (); 
			});
	}	
	
	std::shared_ptr<const RouterInfo> NetDb::GetRandomRouter (std::shared_ptr<const RouterInfo> compatibleWith) const
	{
		return GetRandomRouter (
			[compatibleWith](std::shared_ptr<const RouterInfo> router)->bool 
			{ 
				return !router->IsHidden () && router != compatibleWith && 
					router->IsCompatible (*compatibleWith); 
			});
	}	

	std::shared_ptr<const RouterInfo> NetDb::GetRandomPeerTestRouter () const
	{
		return GetRandomRouter (
			[](std::shared_ptr<const RouterInfo> router)->bool 
			{ 
				return !router->IsHidden () && router->IsPeerTesting (); 
			});
	}

	std::shared_ptr<const RouterInfo> NetDb::GetRandomIntroducer () const
	{
		return GetRandomRouter (
			[](std::shared_ptr<const RouterInfo> router)->bool 
			{ 
				return !router->IsHidden () && router->IsIntroducer (); 
			});
	}	
	
	std::shared_ptr<const RouterInfo> NetDb::GetHighBandwidthRandomRouter (std::shared_ptr<const RouterInfo> compatibleWith) const
	{
		return GetRandomRouter (
			[compatibleWith](std::shared_ptr<const RouterInfo> router)->bool 
			{ 
				return !router->IsHidden () && router != compatibleWith &&
					router->IsCompatible (*compatibleWith) && 
					(router->GetCaps () & RouterInfo::eHighBandwidth);
			});
	}	
	
	template<typename Filter>
	std::shared_ptr<const RouterInfo> NetDb::GetRandomRouter (Filter filter) const
	{
		CryptoPP::RandomNumberGenerator& rnd = i2p::context.GetRandomNumberGenerator ();
		uint32_t ind = rnd.GenerateWord32 (0, m_RouterInfos.size () - 1);	
		for (int j = 0; j < 2; j++)
		{	
			uint32_t i = 0;
			std::unique_lock<std::mutex> l(m_RouterInfosMutex);
			for (auto it: m_RouterInfos)
			{	
				if (i >= ind)
				{	
					if (!it.second->IsUnreachable () && filter (it.second))
						return it.second;
				}	
				else 
					i++;
			}
			// we couldn't find anything, try second pass
			ind = 0;
		}	
		return nullptr; // seems we have too few routers
	}	
	
	void NetDb::PostI2NPMsg (std::shared_ptr<I2NPMessage> msg)
	{
		if (msg) m_Queue.Put (msg);	
	}	

	std::shared_ptr<const RouterInfo> NetDb::GetClosestFloodfill (const IdentHash& destination, 
		const std::set<IdentHash>& excluded) const
	{
		std::shared_ptr<const RouterInfo> r;
		XORMetric minMetric;
		IdentHash destKey = CreateRoutingKey (destination);
		minMetric.SetMax ();
		std::unique_lock<std::mutex> l(m_FloodfillsMutex);
		for (auto it: m_Floodfills)
		{	
			if (!it->IsUnreachable ())
			{	
				XORMetric m = destKey ^ it->GetIdentHash ();
				if (m < minMetric && !excluded.count (it->GetIdentHash ()))
				{
					minMetric = m;
					r = it;
				}
			}	
		}	
		return r;
	}	

	std::vector<IdentHash> NetDb::GetClosestFloodfills (const IdentHash& destination, size_t num,
		std::set<IdentHash>& excluded) const
	{
		struct Sorted
		{
			std::shared_ptr<const RouterInfo> r;
			XORMetric metric;
			bool operator< (const Sorted& other) const { return metric < other.metric; };
		};

		std::set<Sorted> sorted;
		IdentHash destKey = CreateRoutingKey (destination);
		{
			std::unique_lock<std::mutex> l(m_FloodfillsMutex);
			for (auto it: m_Floodfills)
			{
				if (!it->IsUnreachable ())
				{	
					XORMetric m = destKey ^ it->GetIdentHash ();
					if (sorted.size () < num)
						sorted.insert ({it, m});
					else if (m < sorted.rbegin ()->metric)
					{
						sorted.insert ({it, m});
						sorted.erase (std::prev (sorted.end ()));
					}
				}
			}
		}

		std::vector<IdentHash> res;	
		size_t i = 0;			
		for (auto it: sorted)
		{
			if (i < num)
			{
				auto& ident = it.r->GetIdentHash ();
				if (!excluded.count (ident))
				{	
					res.push_back (ident);
					i++;
				}
			}
			else
				break;
		}	
		return res;
	}

	std::shared_ptr<const RouterInfo> NetDb::GetClosestNonFloodfill (const IdentHash& destination, 
		const std::set<IdentHash>& excluded) const
	{
		std::shared_ptr<const RouterInfo> r;
		XORMetric minMetric;
		IdentHash destKey = CreateRoutingKey (destination);
		minMetric.SetMax ();
		// must be called from NetDb thread only
		for (auto it: m_RouterInfos)
		{	
			if (!it.second->IsFloodfill ())
			{	
				XORMetric m = destKey ^ it.first;
				if (m < minMetric && !excluded.count (it.first))
				{
					minMetric = m;
					r = it.second;
				}
			}	
		}	
		return r;
	}	
	
	void NetDb::ManageLeaseSets ()
	{
		for (auto it = m_LeaseSets.begin (); it != m_LeaseSets.end ();)
		{
			if (!it->second->HasNonExpiredLeases ()) // all leases expired
			{
				LogPrint ("LeaseSet ", it->second->GetIdentHash ().ToBase64 (), " expired");
				it = m_LeaseSets.erase (it);
			}	
			else 
				it++;
		}
	}
}
}