mirrors/i2pd
2
0
Fork 0
mirror of https://github.com/PurpleI2P/i2pd synced 2024-11-09 15:50:26 +03:00
Code Issues Packages Projects Releases Wiki Activity

NetDb/NetDbRequests split

Browse Source
This commit is contained in:
orignal 2015-04-09 12:45:00 -04:00
parent 01913d2b14
commit 950f250d66
8 changed files with 256 additions and 182 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

162
NetDb.cpp
View File

@ -21,49 +21,6 @@ namespace i2p
{
namespace data
{
I2NPMessage * RequestedDestination::CreateRequestMessage (std::shared_ptr<const RouterInfo> router,
std::shared_ptr<const i2p::tunnel::InboundTunnel> replyTunnel)
{
I2NPMessage * msg = i2p::CreateRouterInfoDatabaseLookupMsg (m_Destination,
replyTunnel->GetNextIdentHash (), replyTunnel->GetNextTunnelID (), m_IsExploratory,
&m_ExcludedPeers);
m_ExcludedPeers.insert (router->GetIdentHash ());
m_CreationTime = i2p::util::GetSecondsSinceEpoch ();
return msg;
}
I2NPMessage * RequestedDestination::CreateRequestMessage (const IdentHash& floodfill)
{
I2NPMessage * msg = i2p::CreateRouterInfoDatabaseLookupMsg (m_Destination,
i2p::context.GetRouterInfo ().GetIdentHash () , 0, false, &m_ExcludedPeers);
m_ExcludedPeers.insert (floodfill);
m_CreationTime = i2p::util::GetSecondsSinceEpoch ();
return msg;
}
void RequestedDestination::ClearExcludedPeers ()
{
m_ExcludedPeers.clear ();
}
void RequestedDestination::Success (std::shared_ptr<RouterInfo> r)
{
if (m_RequestComplete)
{
m_RequestComplete (r);
m_RequestComplete = nullptr;
}
}
void RequestedDestination::Fail ()
{
if (m_RequestComplete)
{
m_RequestComplete (nullptr);
m_RequestComplete = nullptr;
}
}
#ifndef _WIN32
const char NetDb::m_NetDbPath[] = "/netDb";
#else
@ -120,7 +77,7 @@ namespace data
m_Thread = 0;
}
m_LeaseSets.clear();
m_RequestedDestinations.clear ();
m_Requests.Stop ();
}
void NetDb::Run ()
@ -164,7 +121,7 @@ namespace data
uint64_t ts = i2p::util::GetSecondsSinceEpoch ();
if (ts - lastManageRequest >= 15) // manage requests every 15 seconds
{
ManageRequests ();
m_Requests.ManageRequests ();
lastManageRequest = ts;
}
if (ts - lastSave >= 60) // save routers, manage leasesets and validate subscriptions every minute
@ -189,7 +146,7 @@ namespace data
numRouters = 800/numRouters;
if (numRouters < 1) numRouters = 1;
if (numRouters > 9) numRouters = 9;
ManageRequests ();
m_Requests.ManageRequests ();
Explore (numRouters);
lastExploratory = ts;
}
@ -234,13 +191,7 @@ namespace data
}
}
// take care about requested destination
auto it = m_RequestedDestinations.find (ident);
if (it != m_RequestedDestinations.end ())
{
it->second->Success (r);
std::unique_lock<std::mutex> l(m_RequestedDestinationsMutex);
m_RequestedDestinations.erase (it);
}
m_Requests.RequestComplete (ident, r);
}
void NetDb::AddLeaseSet (const IdentHash& ident, const uint8_t * buf, int len,
@ -487,28 +438,20 @@ namespace data
void NetDb::RequestDestination (const IdentHash& destination, RequestedDestination::RequestComplete requestComplete)
{
// request RouterInfo directly
auto dest = new RequestedDestination (destination, false); // non-exploratory
dest->SetRequestComplete (requestComplete);
auto dest = m_Requests.CreateRequest (destination, false, requestComplete); // non-exploratory
if (!dest)
{
std::unique_lock<std::mutex> l(m_RequestedDestinationsMutex);
if (!m_RequestedDestinations.insert (std::make_pair (destination,
std::unique_ptr<RequestedDestination> (dest))).second) // not inserted
{
LogPrint (eLogWarning, "Destination ", destination.ToBase64(), " is requested already");
return;
}
LogPrint (eLogWarning, "Destination ", destination.ToBase64(), " is requested already");
return;
}
auto floodfill = GetClosestFloodfill (destination, dest->GetExcludedPeers ());
auto floodfill = netdb.GetClosestFloodfill (destination, dest->GetExcludedPeers ());
if (floodfill)
transports.SendMessage (floodfill->GetIdentHash (), dest->CreateRequestMessage (floodfill->GetIdentHash ()));
else
{
LogPrint (eLogError, "No floodfills found");
dest->Fail ();
std::unique_lock<std::mutex> l(m_RequestedDestinationsMutex);
m_RequestedDestinations.erase (destination);
m_Requests.RequestComplete (destination, nullptr);
}
}
@ -609,10 +552,10 @@ namespace data
key[l] = 0;
int num = buf[32]; // num
LogPrint ("DatabaseSearchReply for ", key, " num=", num);
auto it = m_RequestedDestinations.find (IdentHash (buf));
if (it != m_RequestedDestinations.end ())
IdentHash ident (buf);
auto dest = m_Requests.FindRequest (ident);
if (dest)
{
auto& dest = it->second;
bool deleteDest = true;
if (num > 0)
{
@ -659,18 +602,12 @@ namespace data
}
if (deleteDest)
{
// no more requests for the destinationation. delete it
it->second->Fail ();
m_RequestedDestinations.erase (it);
}
m_Requests.RequestComplete (ident, nullptr);
}
else
{
// no more requests for detination possible. delete it
it->second->Fail ();
m_RequestedDestinations.erase (it);
}
m_Requests.RequestComplete (ident, nullptr);
}
else
LogPrint ("Requested destination for ", key, " not found");
@ -832,15 +769,11 @@ namespace data
for (int i = 0; i < numDestinations; i++)
{
rnd.GenerateBlock (randomHash, 32);
auto dest = new RequestedDestination (randomHash, true); // exploratory
{
std::unique_lock<std::mutex> l(m_RequestedDestinationsMutex);
if (!m_RequestedDestinations.insert (std::make_pair (randomHash,
std::unique_ptr<RequestedDestination> (dest))).second) // not inserted
{
LogPrint (eLogWarning, "Exploratory destination is requested already");
return;
}
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
@ -867,10 +800,7 @@ namespace data
i2p::transport::transports.SendMessage (floodfill->GetIdentHash (), dest->CreateRequestMessage (floodfill->GetIdentHash ()));
}
else
{
std::unique_lock<std::mutex> l(m_RequestedDestinationsMutex);
m_RequestedDestinations.erase (dest->GetDestination ());
}
m_Requests.RequestComplete (randomHash, nullptr);
}
if (throughTunnels && msgs.size () > 0)
outbound->SendTunnelDataMsg (msgs);
@ -1028,53 +958,5 @@ namespace data
it++;
}
}
void NetDb::ManageRequests ()
{
uint64_t ts = i2p::util::GetSecondsSinceEpoch ();
std::unique_lock<std::mutex> l(m_RequestedDestinationsMutex);
for (auto it = m_RequestedDestinations.begin (); it != m_RequestedDestinations.end ();)
{
auto& dest = it->second;
bool done = false;
if (ts < dest->GetCreationTime () + 60) // request is worthless after 1 minute
{
if (ts > dest->GetCreationTime () + 5) // no response for 5 seconds
{
auto count = dest->GetExcludedPeers ().size ();
if (!dest->IsExploratory () && count < 7)
{
auto pool = i2p::tunnel::tunnels.GetExploratoryPool ();
auto outbound = pool->GetNextOutboundTunnel ();
auto inbound = pool->GetNextInboundTunnel ();
auto nextFloodfill = GetClosestFloodfill (dest->GetDestination (), dest->GetExcludedPeers ());
if (nextFloodfill && outbound && inbound)
outbound->SendTunnelDataMsg (nextFloodfill->GetIdentHash (), 0,
dest->CreateRequestMessage (nextFloodfill, inbound));
else
{
done = true;
if (!inbound) LogPrint (eLogWarning, "No inbound tunnels");
if (!outbound) LogPrint (eLogWarning, "No outbound tunnels");
if (!nextFloodfill) LogPrint (eLogWarning, "No more floodfills");
}
}
else
{
if (!dest->IsExploratory ())
LogPrint (eLogWarning, dest->GetDestination ().ToBase64 (), " not found after 7 attempts");
done = true;
}
}
}
else // delete obsolete request
done = true;
if (done)
it = m_RequestedDestinations.erase (it);
else
it++;
}
}
}
}

39
NetDb.h
View File

@ -16,44 +16,12 @@
#include "Tunnel.h"
#include "TunnelPool.h"
#include "Reseed.h"
#include "NetDbRequests.h"
namespace i2p
{
namespace data
{
class RequestedDestination
{
public:
typedef std::function<void (std::shared_ptr<RouterInfo>)> RequestComplete;
RequestedDestination (const IdentHash& destination, bool isExploratory = false):
m_Destination (destination), m_IsExploratory (isExploratory), m_CreationTime (0) {};
~RequestedDestination () { if (m_RequestComplete) m_RequestComplete (nullptr); };
const IdentHash& GetDestination () const { return m_Destination; };
int GetNumExcludedPeers () const { return m_ExcludedPeers.size (); };
const std::set<IdentHash>& GetExcludedPeers () { return m_ExcludedPeers; };
void ClearExcludedPeers ();
bool IsExploratory () const { return m_IsExploratory; };
bool IsExcluded (const IdentHash& ident) const { return m_ExcludedPeers.count (ident); };
uint64_t GetCreationTime () const { return m_CreationTime; };
I2NPMessage * CreateRequestMessage (std::shared_ptr<const RouterInfo>, std::shared_ptr<const i2p::tunnel::InboundTunnel> replyTunnel);
I2NPMessage * CreateRequestMessage (const IdentHash& floodfill);
void SetRequestComplete (const RequestComplete& requestComplete) { m_RequestComplete = requestComplete; };
bool IsRequestComplete () const { return m_RequestComplete != nullptr; };
void Success (std::shared_ptr<RouterInfo> r);
void Fail ();
private:
IdentHash m_Destination;
bool m_IsExploratory;
std::set<IdentHash> m_ExcludedPeers;
uint64_t m_CreationTime;
RequestComplete m_RequestComplete;
};
class NetDb
{
@ -116,8 +84,6 @@ namespace data
std::map<IdentHash, std::shared_ptr<RouterInfo> > m_RouterInfos;
mutable std::mutex m_FloodfillsMutex;
std::list<std::shared_ptr<RouterInfo> > m_Floodfills;
std::mutex m_RequestedDestinationsMutex;
std::map<IdentHash, std::unique_ptr<RequestedDestination> > m_RequestedDestinations;
bool m_IsRunning;
std::thread * m_Thread;
@ -125,6 +91,9 @@ namespace data
Reseeder * m_Reseeder;
friend NetDbRequests;
NetDbRequests m_Requests;
static const char m_NetDbPath[];
};

149
NetDbRequests.cpp Normal file
View File

@ -0,0 +1,149 @@
#include "Log.h"
#include "I2NPProtocol.h"
#include "Transports.h"
#include "NetDb.h"
#include "NetDbRequests.h"
namespace i2p
{
namespace data
{
I2NPMessage * RequestedDestination::CreateRequestMessage (std::shared_ptr<const RouterInfo> router,
std::shared_ptr<const i2p::tunnel::InboundTunnel> replyTunnel)
{
I2NPMessage * msg = i2p::CreateRouterInfoDatabaseLookupMsg (m_Destination,
replyTunnel->GetNextIdentHash (), replyTunnel->GetNextTunnelID (), m_IsExploratory,
&m_ExcludedPeers);
m_ExcludedPeers.insert (router->GetIdentHash ());
m_CreationTime = i2p::util::GetSecondsSinceEpoch ();
return msg;
}
I2NPMessage * RequestedDestination::CreateRequestMessage (const IdentHash& floodfill)
{
I2NPMessage * msg = i2p::CreateRouterInfoDatabaseLookupMsg (m_Destination,
i2p::context.GetRouterInfo ().GetIdentHash () , 0, false, &m_ExcludedPeers);
m_ExcludedPeers.insert (floodfill);
m_CreationTime = i2p::util::GetSecondsSinceEpoch ();
return msg;
}
void RequestedDestination::ClearExcludedPeers ()
{
m_ExcludedPeers.clear ();
}
void RequestedDestination::Success (std::shared_ptr<RouterInfo> r)
{
if (m_RequestComplete)
{
m_RequestComplete (r);
m_RequestComplete = nullptr;
}
}
void RequestedDestination::Fail ()
{
if (m_RequestComplete)
{
m_RequestComplete (nullptr);
m_RequestComplete = nullptr;
}
}
void NetDbRequests::Start ()
{
}
void NetDbRequests::Stop ()
{
m_RequestedDestinations.clear ();
}
std::shared_ptr<RequestedDestination> NetDbRequests::CreateRequest (const IdentHash& destination, bool isExploratory, RequestedDestination::RequestComplete requestComplete)
{
// request RouterInfo directly
auto dest = std::make_shared<RequestedDestination> (destination, isExploratory);
dest->SetRequestComplete (requestComplete);
{
std::unique_lock<std::mutex> l(m_RequestedDestinationsMutex);
if (!m_RequestedDestinations.insert (std::make_pair (destination,
std::shared_ptr<RequestedDestination> (dest))).second) // not inserted
return nullptr;
}
return dest;
}
void NetDbRequests::RequestComplete (const IdentHash& ident, std::shared_ptr<RouterInfo> r)
{
auto it = m_RequestedDestinations.find (ident);
if (it != m_RequestedDestinations.end ())
{
if (r)
it->second->Success (r);
else
it->second->Fail ();
std::unique_lock<std::mutex> l(m_RequestedDestinationsMutex);
m_RequestedDestinations.erase (it);
}
}
std::shared_ptr<RequestedDestination> NetDbRequests::FindRequest (const IdentHash& ident) const
{
auto it = m_RequestedDestinations.find (ident);
if (it != m_RequestedDestinations.end ())
return it->second;
return nullptr;
}
void NetDbRequests::ManageRequests ()
{
uint64_t ts = i2p::util::GetSecondsSinceEpoch ();
std::unique_lock<std::mutex> l(m_RequestedDestinationsMutex);
for (auto it = m_RequestedDestinations.begin (); it != m_RequestedDestinations.end ();)
{
auto& dest = it->second;
bool done = false;
if (ts < dest->GetCreationTime () + 60) // request is worthless after 1 minute
{
if (ts > dest->GetCreationTime () + 5) // no response for 5 seconds
{
auto count = dest->GetExcludedPeers ().size ();
if (!dest->IsExploratory () && count < 7)
{
auto pool = i2p::tunnel::tunnels.GetExploratoryPool ();
auto outbound = pool->GetNextOutboundTunnel ();
auto inbound = pool->GetNextInboundTunnel ();
auto nextFloodfill = netdb.GetClosestFloodfill (dest->GetDestination (), dest->GetExcludedPeers ());
if (nextFloodfill && outbound && inbound)
outbound->SendTunnelDataMsg (nextFloodfill->GetIdentHash (), 0,
dest->CreateRequestMessage (nextFloodfill, inbound));
else
{
done = true;
if (!inbound) LogPrint (eLogWarning, "No inbound tunnels");
if (!outbound) LogPrint (eLogWarning, "No outbound tunnels");
if (!nextFloodfill) LogPrint (eLogWarning, "No more floodfills");
}
}
else
{
if (!dest->IsExploratory ())
LogPrint (eLogWarning, dest->GetDestination ().ToBase64 (), " not found after 7 attempts");
done = true;
}
}
}
else // delete obsolete request
done = true;
if (done)
it = m_RequestedDestinations.erase (it);
else
it++;
}
}
}
}

69
NetDbRequests.h Normal file
View File

@ -0,0 +1,69 @@
#ifndef NETDB_REQUESTS_H__
#define NETDB_REQUESTS_H__
#include <memory>
#include <set>
#include <map>
#include "Identity.h"
#include "RouterInfo.h"
namespace i2p
{
namespace data
{
class RequestedDestination
{
public:
typedef std::function<void (std::shared_ptr<RouterInfo>)> RequestComplete;
RequestedDestination (const IdentHash& destination, bool isExploratory = false):
m_Destination (destination), m_IsExploratory (isExploratory), m_CreationTime (0) {};
~RequestedDestination () { if (m_RequestComplete) m_RequestComplete (nullptr); };
const IdentHash& GetDestination () const { return m_Destination; };
int GetNumExcludedPeers () const { return m_ExcludedPeers.size (); };
const std::set<IdentHash>& GetExcludedPeers () { return m_ExcludedPeers; };
void ClearExcludedPeers ();
bool IsExploratory () const { return m_IsExploratory; };
bool IsExcluded (const IdentHash& ident) const { return m_ExcludedPeers.count (ident); };
uint64_t GetCreationTime () const { return m_CreationTime; };
I2NPMessage * CreateRequestMessage (std::shared_ptr<const RouterInfo>, std::shared_ptr<const i2p::tunnel::InboundTunnel> replyTunnel);
I2NPMessage * CreateRequestMessage (const IdentHash& floodfill);
void SetRequestComplete (const RequestComplete& requestComplete) { m_RequestComplete = requestComplete; };
bool IsRequestComplete () const { return m_RequestComplete != nullptr; };
void Success (std::shared_ptr<RouterInfo> r);
void Fail ();
private:
IdentHash m_Destination;
bool m_IsExploratory;
std::set<IdentHash> m_ExcludedPeers;
uint64_t m_CreationTime;
RequestComplete m_RequestComplete;
};
class NetDbRequests
{
public:
void Start ();
void Stop ();
std::shared_ptr<RequestedDestination> CreateRequest (const IdentHash& destination, bool isExploratory, RequestedDestination::RequestComplete requestComplete = nullptr);
void RequestComplete (const IdentHash& ident, std::shared_ptr<RouterInfo> r);
std::shared_ptr<RequestedDestination> FindRequest (const IdentHash& ident) const;
void ManageRequests ();
private:
std::mutex m_RequestedDestinationsMutex;
std::map<IdentHash, std::shared_ptr<RequestedDestination> > m_RequestedDestinations;
};
}
}
#endif

2
Win32/i2pd.vcxproj
View File

@ -37,6 +37,7 @@
<ClCompile Include="..\LeaseSet.cpp" />
<ClCompile Include="..\Log.cpp" />
<ClCompile Include="..\NetDb.cpp" />
<ClCompile Include="..\NetDbRequests.cpp" />
<ClCompile Include="..\NTCPSession.cpp" />
<ClCompile Include="..\Profiling.cpp" />
<ClCompile Include="..\Reseed.cpp" />
@ -81,6 +82,7 @@
<ClInclude Include="..\LeaseSet.h" />
<ClInclude Include="..\LittleBigEndian.h" />
<ClInclude Include="..\Log.h" />
<ClInclude Include="..\NetDbRequests.h" />
<ClInclude Include="..\NetDb.h" />
<ClInclude Include="..\NTCPSession.h" />
<ClInclude Include="..\Queue.h" />

1
build/CMakeLists.txt
View File

@ -21,6 +21,7 @@ set (COMMON_SRC
"${CMAKE_SOURCE_DIR}/LeaseSet.cpp"
"${CMAKE_SOURCE_DIR}/Log.cpp"
"${CMAKE_SOURCE_DIR}/NTCPSession.cpp"
"${CMAKE_SOURCE_DIR}/NetDbRequests.cpp"
"${CMAKE_SOURCE_DIR}/NetDb.cpp"
"${CMAKE_SOURCE_DIR}/Profiling.cpp"
"${CMAKE_SOURCE_DIR}/Reseed.cpp"

6
build/autotools/Makefile.in
View File

@ -328,7 +328,8 @@ i2p_SOURCES = AddressBook.cpp CryptoConst.cpp Daemon.cpp \
TunnelGateway.cpp TunnelPool.cpp UPnP.cpp aes.cpp \
base64.cpp i2p.cpp util.cpp SAM.cpp Destination.cpp \
ClientContext.cpp DataFram.cpp SSUSession.cpp BOB.cpp \
I2PControl.cpp Profiling.cpp Signature.cpp \
I2PControl.cpp Profiling.cpp Signature.cpp \
NetDbRequests.cpp \
\
AddressBook.h CryptoConst.h Daemon.h ElGamal.h \
Garlic.h HTTPProxy.h HTTPServer.h I2NPProtocol.h \
@ -341,7 +342,7 @@ i2p_SOURCES = AddressBook.cpp CryptoConst.cpp Daemon.cpp \
TunnelPool.h UPnP.h aes.h base64.h config.h hmac.h \
util.h version.h Destination.h ClientContext.h \
TransportSession.h Datagram.h SSUSession.h BOB.h \
I2PControl.h Profiling.h
I2PControl.h Profiling.h NetDbRequests.h
AM_LDFLAGS = @BOOST_DATE_TIME_LIB@ @BOOST_FILESYSTEM_LIB@ \
@BOOST_PROGRAM_OPTIONS_LIB@ @BOOST_REGEX_LIB@ \
@ -495,6 +496,7 @@ distclean-compile:
@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/SSUSession.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/Profiling.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/Signature.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/NetDbRequests.Po@am__quote@
.cpp.o:
@am__fastdepCXX_TRUE@ $(AM_V_CXX)$(CXXCOMPILE) -MT $@ -MD -MP -MF $(DEPDIR)/$*.Tpo -c -o $@ $<

10
filelist.mk
View File

@ -1,10 +1,10 @@
COMMON_SRC = \
CryptoConst.cpp Datagram.cpp Garlic.cpp I2NPProtocol.cpp LeaseSet.cpp \
Log.cpp NTCPSession.cpp NetDb.cpp Profiling.cpp Reseed.cpp RouterContext.cpp \
RouterInfo.cpp Signature.cpp SSU.cpp SSUSession.cpp SSUData.cpp Streaming.cpp \
Identity.cpp TransitTunnel.cpp Transports.cpp Tunnel.cpp TunnelEndpoint.cpp \
TunnelPool.cpp TunnelGateway.cpp Destination.cpp UPnP.cpp util.cpp aes.cpp \
base64.cpp
Log.cpp NTCPSession.cpp NetDb.cpp NetDbRequests.cpp Profiling.cpp \
Reseed.cpp RouterContext.cpp RouterInfo.cpp Signature.cpp SSU.cpp \
SSUSession.cpp SSUData.cpp Streaming.cpp Identity.cpp TransitTunnel.cpp \
Transports.cpp Tunnel.cpp TunnelEndpoint.cpp TunnelPool.cpp TunnelGateway.cpp \
Destination.cpp UPnP.cpp util.cpp aes.cpp base64.cpp
ifeq ($(UNAME),Darwin)