PurpleI2P
/
i2pd
mirror of https://github.com/PurpleI2P/i2pd.git
1
0
Fork 0
Code Issues Releases Activity
Browse Source

Merge pull request #1703 from simonvetter/simon/memory-and-multithreading-fixes 

memory leak and concurrency fixes
pull/1705/head
orignal 3 years ago committed by GitHub
parent
4c5ec68ff1 58b7b7d731
commit
67863cfcf9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 78 additions and 19 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 33
      libi2pd/Tunnel.cpp
  2. 10
      libi2pd/Tunnel.h
  3. 2
      libi2pd/TunnelConfig.h
  4. 46
      libi2pd/TunnelPool.cpp
  5. 6
      libi2pd/TunnelPool.h

33
libi2pd/Tunnel.cpp
Unescape View File

@ -366,6 +366,7 @@ namespace tunnel
std::shared_ptr<TunnelBase> Tunnels::GetTunnel (uint32_t tunnelID) std::shared_ptr<TunnelBase> Tunnels::GetTunnel (uint32_t tunnelID)
{ {
std::unique_lock<std::recursive_mutex> l(m_TunnelsMutex);
auto it = m_Tunnels.find(tunnelID); auto it = m_Tunnels.find(tunnelID);
if (it != m_Tunnels.end ()) if (it != m_Tunnels.end ())
return it->second; return it->second;
@ -374,11 +375,13 @@ namespace tunnel
std::shared_ptr<InboundTunnel> Tunnels::GetPendingInboundTunnel (uint32_t replyMsgID) std::shared_ptr<InboundTunnel> Tunnels::GetPendingInboundTunnel (uint32_t replyMsgID)
{ {
std::unique_lock<std::mutex> l(m_PendingInboundTunnelsMutex);
return GetPendingTunnel (replyMsgID, m_PendingInboundTunnels); return GetPendingTunnel (replyMsgID, m_PendingInboundTunnels);
} }
std::shared_ptr<OutboundTunnel> Tunnels::GetPendingOutboundTunnel (uint32_t replyMsgID) std::shared_ptr<OutboundTunnel> Tunnels::GetPendingOutboundTunnel (uint32_t replyMsgID)
{ {
std::unique_lock<std::mutex> l(m_PendingOutboundTunnelsMutex);
return GetPendingTunnel (replyMsgID, m_PendingOutboundTunnels); return GetPendingTunnel (replyMsgID, m_PendingOutboundTunnels);
} }
@ -459,9 +462,11 @@ namespace tunnel
void Tunnels::AddTransitTunnel (std::shared_ptr<TransitTunnel> tunnel) void Tunnels::AddTransitTunnel (std::shared_ptr<TransitTunnel> tunnel)
{ {
if (m_Tunnels.emplace (tunnel->GetTunnelID (), tunnel).second) std::unique_lock<std::recursive_mutex> l(m_TunnelsMutex);
if (m_Tunnels.emplace (tunnel->GetTunnelID (), tunnel).second) {
std::unique_lock<std::mutex> l(m_TransitTunnelsMutex);
m_TransitTunnels.push_back (tunnel); m_TransitTunnels.push_back (tunnel);
else } else
LogPrint (eLogError, "Tunnel: tunnel with id ", tunnel->GetTunnelID (), " already exists"); LogPrint (eLogError, "Tunnel: tunnel with id ", tunnel->GetTunnelID (), " already exists");
} }
@ -616,7 +621,10 @@ namespace tunnel
void Tunnels::ManagePendingTunnels () void Tunnels::ManagePendingTunnels ()
{ {
std::unique_lock<std::mutex> li(m_PendingInboundTunnelsMutex);
ManagePendingTunnels (m_PendingInboundTunnels); ManagePendingTunnels (m_PendingInboundTunnels);
li.unlock();
std::unique_lock<std::mutex> lo(m_PendingOutboundTunnelsMutex);
ManagePendingTunnels (m_PendingOutboundTunnels); ManagePendingTunnels (m_PendingOutboundTunnels);
} }
@ -676,6 +684,7 @@ namespace tunnel
void Tunnels::ManageOutboundTunnels () void Tunnels::ManageOutboundTunnels ()
{ {
std::unique_lock<std::recursive_mutex> l(m_OutboundTunnelsMutex);
uint64_t ts = i2p::util::GetSecondsSinceEpoch (); uint64_t ts = i2p::util::GetSecondsSinceEpoch ();
{ {
for (auto it = m_OutboundTunnels.begin (); it != m_OutboundTunnels.end ();) for (auto it = m_OutboundTunnels.begin (); it != m_OutboundTunnels.end ();)
@ -730,6 +739,8 @@ namespace tunnel
void Tunnels::ManageInboundTunnels () void Tunnels::ManageInboundTunnels ()
{ {
std::unique_lock<std::recursive_mutex> lt(m_TunnelsMutex);
std::unique_lock<std::recursive_mutex> li(m_InboundTunnelsMutex);
uint64_t ts = i2p::util::GetSecondsSinceEpoch (); uint64_t ts = i2p::util::GetSecondsSinceEpoch ();
{ {
for (auto it = m_InboundTunnels.begin (); it != m_InboundTunnels.end ();) for (auto it = m_InboundTunnels.begin (); it != m_InboundTunnels.end ();)
@ -786,6 +797,7 @@ namespace tunnel
return; return;
} }
std::unique_lock<std::recursive_mutex> lo(m_OutboundTunnelsMutex);
if (m_OutboundTunnels.empty () || m_InboundTunnels.size () < 3) if (m_OutboundTunnels.empty () || m_InboundTunnels.size () < 3)
{ {
// trying to create one more inbound tunnel // trying to create one more inbound tunnel
@ -806,6 +818,8 @@ namespace tunnel
void Tunnels::ManageTransitTunnels () void Tunnels::ManageTransitTunnels ()
{ {
std::unique_lock<std::recursive_mutex> lt(m_TunnelsMutex);
std::unique_lock<std::mutex> l(m_TransitTunnelsMutex);
uint32_t ts = i2p::util::GetSecondsSinceEpoch (); uint32_t ts = i2p::util::GetSecondsSinceEpoch ();
for (auto it = m_TransitTunnels.begin (); it != m_TransitTunnels.end ();) for (auto it = m_TransitTunnels.begin (); it != m_TransitTunnels.end ();)
{ {
@ -876,17 +890,20 @@ namespace tunnel
void Tunnels::AddPendingTunnel (uint32_t replyMsgID, std::shared_ptr<InboundTunnel> tunnel) void Tunnels::AddPendingTunnel (uint32_t replyMsgID, std::shared_ptr<InboundTunnel> tunnel)
{ {
std::unique_lock<std::mutex> l(m_PendingInboundTunnelsMutex);
m_PendingInboundTunnels[replyMsgID] = tunnel; m_PendingInboundTunnels[replyMsgID] = tunnel;
} }
void Tunnels::AddPendingTunnel (uint32_t replyMsgID, std::shared_ptr<OutboundTunnel> tunnel) void Tunnels::AddPendingTunnel (uint32_t replyMsgID, std::shared_ptr<OutboundTunnel> tunnel)
{ {
std::unique_lock<std::mutex> l(m_PendingOutboundTunnelsMutex);
m_PendingOutboundTunnels[replyMsgID] = tunnel; m_PendingOutboundTunnels[replyMsgID] = tunnel;
} }
void Tunnels::AddOutboundTunnel (std::shared_ptr<OutboundTunnel> newTunnel) void Tunnels::AddOutboundTunnel (std::shared_ptr<OutboundTunnel> newTunnel)
{ {
// we don't need to insert it to m_Tunnels // we don't need to insert it to m_Tunnels
std::unique_lock<std::recursive_mutex> l(m_OutboundTunnelsMutex);
m_OutboundTunnels.push_back (newTunnel); m_OutboundTunnels.push_back (newTunnel);
auto pool = newTunnel->GetTunnelPool (); auto pool = newTunnel->GetTunnelPool ();
if (pool && pool->IsActive ()) if (pool && pool->IsActive ())
@ -897,8 +914,10 @@ namespace tunnel
void Tunnels::AddInboundTunnel (std::shared_ptr<InboundTunnel> newTunnel) void Tunnels::AddInboundTunnel (std::shared_ptr<InboundTunnel> newTunnel)
{ {
std::unique_lock<std::recursive_mutex> l(m_TunnelsMutex);
if (m_Tunnels.emplace (newTunnel->GetTunnelID (), newTunnel).second) if (m_Tunnels.emplace (newTunnel->GetTunnelID (), newTunnel).second)
{ {
std::unique_lock<std::recursive_mutex> l(m_InboundTunnelsMutex);
m_InboundTunnels.push_back (newTunnel); m_InboundTunnels.push_back (newTunnel);
auto pool = newTunnel->GetTunnelPool (); auto pool = newTunnel->GetTunnelPool ();
if (!pool) if (!pool)
@ -926,6 +945,8 @@ namespace tunnel
auto inboundTunnel = std::make_shared<ZeroHopsInboundTunnel> (); auto inboundTunnel = std::make_shared<ZeroHopsInboundTunnel> ();
inboundTunnel->SetTunnelPool (pool); inboundTunnel->SetTunnelPool (pool);
inboundTunnel->SetState (eTunnelStateEstablished); inboundTunnel->SetState (eTunnelStateEstablished);
std::unique_lock<std::recursive_mutex> lt(m_TunnelsMutex);
std::unique_lock<std::recursive_mutex> li(m_InboundTunnelsMutex);
m_InboundTunnels.push_back (inboundTunnel); m_InboundTunnels.push_back (inboundTunnel);
m_Tunnels[inboundTunnel->GetTunnelID ()] = inboundTunnel; m_Tunnels[inboundTunnel->GetTunnelID ()] = inboundTunnel;
return inboundTunnel; return inboundTunnel;
@ -936,6 +957,7 @@ namespace tunnel
auto outboundTunnel = std::make_shared<ZeroHopsOutboundTunnel> (); auto outboundTunnel = std::make_shared<ZeroHopsOutboundTunnel> ();
outboundTunnel->SetTunnelPool (pool); outboundTunnel->SetTunnelPool (pool);
outboundTunnel->SetState (eTunnelStateEstablished); outboundTunnel->SetState (eTunnelStateEstablished);
std::unique_lock<std::recursive_mutex> l(m_OutboundTunnelsMutex);
m_OutboundTunnels.push_back (outboundTunnel); m_OutboundTunnels.push_back (outboundTunnel);
// we don't insert into m_Tunnels // we don't insert into m_Tunnels
return outboundTunnel; return outboundTunnel;
@ -964,6 +986,7 @@ namespace tunnel
int timeout = 0; int timeout = 0;
uint32_t ts = i2p::util::GetSecondsSinceEpoch (); uint32_t ts = i2p::util::GetSecondsSinceEpoch ();
// TODO: possible race condition with I2PControl // TODO: possible race condition with I2PControl
std::unique_lock<std::mutex> l(m_TransitTunnelsMutex);
for (const auto& it : m_TransitTunnels) for (const auto& it : m_TransitTunnels)
{ {
int t = it->GetCreationTime () + TUNNEL_EXPIRATION_TIMEOUT - ts; int t = it->GetCreationTime () + TUNNEL_EXPIRATION_TIMEOUT - ts;
@ -974,19 +997,19 @@ namespace tunnel
size_t Tunnels::CountTransitTunnels() const size_t Tunnels::CountTransitTunnels() const
{ {
// TODO: locking std::unique_lock<std::mutex> l(m_TransitTunnelsMutex);
return m_TransitTunnels.size(); return m_TransitTunnels.size();
} }
size_t Tunnels::CountInboundTunnels() const size_t Tunnels::CountInboundTunnels() const
{ {
// TODO: locking std::unique_lock<std::recursive_mutex> l(m_InboundTunnelsMutex);
return m_InboundTunnels.size(); return m_InboundTunnels.size();
} }
size_t Tunnels::CountOutboundTunnels() const size_t Tunnels::CountOutboundTunnels() const
{ {
// TODO: locking std::unique_lock<std::recursive_mutex> l(m_OutboundTunnelsMutex);
return m_OutboundTunnels.size(); return m_OutboundTunnels.size();
} }
} }

10
libi2pd/Tunnel.h
Unescape View File

@ -68,7 +68,7 @@ namespace tunnel
public: public:
Tunnel (std::shared_ptr<const TunnelConfig> config); Tunnel (std::shared_ptr<const TunnelConfig> config);
~Tunnel (); virtual ~Tunnel ();
void Build (uint32_t replyMsgID, std::shared_ptr<OutboundTunnel> outboundTunnel = nullptr); void Build (uint32_t replyMsgID, std::shared_ptr<OutboundTunnel> outboundTunnel = nullptr);
@ -254,12 +254,18 @@ namespace tunnel
bool m_IsRunning; bool m_IsRunning;
std::thread * m_Thread; std::thread * m_Thread;
std::map<uint32_t, std::shared_ptr<InboundTunnel> > m_PendingInboundTunnels; // by replyMsgID std::map<uint32_t, std::shared_ptr<InboundTunnel> > m_PendingInboundTunnels; // by replyMsgID
mutable std::mutex m_PendingInboundTunnelsMutex;
std::map<uint32_t, std::shared_ptr<OutboundTunnel> > m_PendingOutboundTunnels; // by replyMsgID std::map<uint32_t, std::shared_ptr<OutboundTunnel> > m_PendingOutboundTunnels; // by replyMsgID
mutable std::mutex m_PendingOutboundTunnelsMutex;
std::list<std::shared_ptr<InboundTunnel> > m_InboundTunnels; std::list<std::shared_ptr<InboundTunnel> > m_InboundTunnels;
mutable std::recursive_mutex m_InboundTunnelsMutex;
std::list<std::shared_ptr<OutboundTunnel> > m_OutboundTunnels; std::list<std::shared_ptr<OutboundTunnel> > m_OutboundTunnels;
mutable std::recursive_mutex m_OutboundTunnelsMutex;
std::list<std::shared_ptr<TransitTunnel> > m_TransitTunnels; std::list<std::shared_ptr<TransitTunnel> > m_TransitTunnels;
mutable std::mutex m_TransitTunnelsMutex;
std::unordered_map<uint32_t, std::shared_ptr<TunnelBase> > m_Tunnels; // tunnelID->tunnel known by this id std::unordered_map<uint32_t, std::shared_ptr<TunnelBase> > m_Tunnels; // tunnelID->tunnel known by this id
std::mutex m_PoolsMutex; mutable std::recursive_mutex m_TunnelsMutex;
mutable std::mutex m_PoolsMutex;
std::list<std::shared_ptr<TunnelPool>> m_Pools; std::list<std::shared_ptr<TunnelPool>> m_Pools;
std::shared_ptr<TunnelPool> m_ExploratoryPool; std::shared_ptr<TunnelPool> m_ExploratoryPool;
i2p::util::Queue<std::shared_ptr<I2NPMessage> > m_Queue; i2p::util::Queue<std::shared_ptr<I2NPMessage> > m_Queue;

2
libi2pd/TunnelConfig.h
Unescape View File

@ -99,7 +99,7 @@ namespace tunnel
m_LastHop->SetReplyHop (replyTunnelID, replyIdent); m_LastHop->SetReplyHop (replyTunnelID, replyIdent);
} }
~TunnelConfig () virtual ~TunnelConfig ()
{ {
TunnelHopConfig * hop = m_FirstHop; TunnelHopConfig * hop = m_FirstHop;

46
libi2pd/TunnelPool.cpp
Unescape View File

@ -59,6 +59,7 @@ namespace tunnel
void TunnelPool::SetExplicitPeers (std::shared_ptr<std::vector<i2p::data::IdentHash> > explicitPeers) void TunnelPool::SetExplicitPeers (std::shared_ptr<std::vector<i2p::data::IdentHash> > explicitPeers)
{ {
std::unique_lock<std::mutex> l(m_ExplicitPeersMutex);
m_ExplicitPeers = explicitPeers; m_ExplicitPeers = explicitPeers;
if (m_ExplicitPeers) if (m_ExplicitPeers)
{ {
@ -92,6 +93,7 @@ namespace tunnel
it->SetTunnelPool (nullptr); it->SetTunnelPool (nullptr);
m_OutboundTunnels.clear (); m_OutboundTunnels.clear ();
} }
std::unique_lock<std::mutex> l(m_TestsMutex);
m_Tests.clear (); m_Tests.clear ();
} }
@ -126,6 +128,7 @@ namespace tunnel
} }
m_InboundTunnels.insert (createdTunnel); m_InboundTunnels.insert (createdTunnel);
} }
std::unique_lock<std::mutex> l(m_LocalDestinationMutex);
if (m_LocalDestination) if (m_LocalDestination)
m_LocalDestination->SetLeaseSetUpdated (); m_LocalDestination->SetLeaseSetUpdated ();
} }
@ -135,10 +138,11 @@ namespace tunnel
if (expiredTunnel) if (expiredTunnel)
{ {
expiredTunnel->SetTunnelPool (nullptr); expiredTunnel->SetTunnelPool (nullptr);
std::unique_lock<std::mutex> lt(m_TestsMutex);
for (auto& it: m_Tests) for (auto& it: m_Tests)
if (it.second.second == expiredTunnel) it.second.second = nullptr; if (it.second.second == expiredTunnel) it.second.second = nullptr;
std::unique_lock<std::mutex> l(m_InboundTunnelsMutex); std::unique_lock<std::mutex> li(m_InboundTunnelsMutex);
m_InboundTunnels.erase (expiredTunnel); m_InboundTunnels.erase (expiredTunnel);
} }
} }
@ -157,10 +161,11 @@ namespace tunnel
if (expiredTunnel) if (expiredTunnel)
{ {
expiredTunnel->SetTunnelPool (nullptr); expiredTunnel->SetTunnelPool (nullptr);
std::unique_lock<std::mutex> lt(m_TestsMutex);
for (auto& it: m_Tests) for (auto& it: m_Tests)
if (it.second.first == expiredTunnel) it.second.first = nullptr; if (it.second.first == expiredTunnel) it.second.first = nullptr;
std::unique_lock<std::mutex> l(m_OutboundTunnelsMutex); std::unique_lock<std::mutex> lo(m_OutboundTunnelsMutex);
m_OutboundTunnels.erase (expiredTunnel); m_OutboundTunnels.erase (expiredTunnel);
} }
} }
@ -261,11 +266,21 @@ namespace tunnel
return tunnel; return tunnel;
} }
std::shared_ptr<i2p::garlic::GarlicDestination> TunnelPool::GetLocalDestination () const {
std::unique_lock<std::mutex> l(m_LocalDestinationMutex);
return m_LocalDestination;
}
void TunnelPool::SetLocalDestination (std::shared_ptr<i2p::garlic::GarlicDestination> destination) {
std::unique_lock<std::mutex> l(m_LocalDestinationMutex);
m_LocalDestination = destination;
}
void TunnelPool::CreateTunnels () void TunnelPool::CreateTunnels ()
{ {
int num = 0; int num = 0;
{ {
std::unique_lock<std::mutex> l(m_OutboundTunnelsMutex); std::unique_lock<std::mutex> lo(m_OutboundTunnelsMutex);
for (const auto& it : m_OutboundTunnels) for (const auto& it : m_OutboundTunnels)
if (it->IsEstablished ()) num++; if (it->IsEstablished ()) num++;
} }
@ -274,10 +289,11 @@ namespace tunnel
num = 0; num = 0;
{ {
std::unique_lock<std::mutex> l(m_InboundTunnelsMutex); std::unique_lock<std::mutex> li(m_InboundTunnelsMutex);
for (const auto& it : m_InboundTunnels) for (const auto& it : m_InboundTunnels)
if (it->IsEstablished ()) num++; if (it->IsEstablished ()) num++;
} }
std::unique_lock<std::mutex> lo(m_OutboundTunnelsMutex);
if (!num && !m_OutboundTunnels.empty () && m_NumOutboundHops > 0) if (!num && !m_OutboundTunnels.empty () && m_NumOutboundHops > 0)
{ {
for (auto it: m_OutboundTunnels) for (auto it: m_OutboundTunnels)
@ -287,9 +303,11 @@ namespace tunnel
if (num >= m_NumInboundTunnels) break; if (num >= m_NumInboundTunnels) break;
} }
} }
lo.unlock();
for (int i = num; i < m_NumInboundTunnels; i++) for (int i = num; i < m_NumInboundTunnels; i++)
CreateInboundTunnel (); CreateInboundTunnel ();
std::unique_lock<std::mutex> l(m_LocalDestinationMutex);
if (num < m_NumInboundTunnels && m_NumInboundHops <= 0 && m_LocalDestination) // zero hops IB if (num < m_NumInboundTunnels && m_NumInboundHops <= 0 && m_LocalDestination) // zero hops IB
m_LocalDestination->SetLeaseSetUpdated (); // update LeaseSet immediately m_LocalDestination->SetLeaseSetUpdated (); // update LeaseSet immediately
} }
@ -311,7 +329,7 @@ namespace tunnel
if (it.second.first->GetState () == eTunnelStateTestFailed) if (it.second.first->GetState () == eTunnelStateTestFailed)
{ {
it.second.first->SetState (eTunnelStateFailed); it.second.first->SetState (eTunnelStateFailed);
std::unique_lock<std::mutex> l(m_OutboundTunnelsMutex); std::unique_lock<std::mutex> lo(m_OutboundTunnelsMutex);
m_OutboundTunnels.erase (it.second.first); m_OutboundTunnels.erase (it.second.first);
} }
else else
@ -323,9 +341,10 @@ namespace tunnel
{ {
it.second.second->SetState (eTunnelStateFailed); it.second.second->SetState (eTunnelStateFailed);
{ {
std::unique_lock<std::mutex> l(m_InboundTunnelsMutex); std::unique_lock<std::mutex> li(m_InboundTunnelsMutex);
m_InboundTunnels.erase (it.second.second); m_InboundTunnels.erase (it.second.second);
} }
std::unique_lock<std::mutex> ld(m_LocalDestinationMutex);
if (m_LocalDestination) if (m_LocalDestination)
m_LocalDestination->SetLeaseSetUpdated (); m_LocalDestination->SetLeaseSetUpdated ();
} }
@ -335,7 +354,10 @@ namespace tunnel
} }
// new tests // new tests
std::unique_lock<std::mutex> lt(m_TestsMutex);
std::unique_lock<std::mutex> lo(m_OutboundTunnelsMutex);
auto it1 = m_OutboundTunnels.begin (); auto it1 = m_OutboundTunnels.begin ();
std::unique_lock<std::mutex> li(m_InboundTunnelsMutex);
auto it2 = m_InboundTunnels.begin (); auto it2 = m_InboundTunnels.begin ();
while (it1 != m_OutboundTunnels.end () && it2 != m_InboundTunnels.end ()) while (it1 != m_OutboundTunnels.end () && it2 != m_InboundTunnels.end ())
{ {
@ -355,7 +377,6 @@ namespace tunnel
uint32_t msgID; uint32_t msgID;
RAND_bytes ((uint8_t *)&msgID, 4); RAND_bytes ((uint8_t *)&msgID, 4);
{ {
std::unique_lock<std::mutex> l(m_TestsMutex);
m_Tests[msgID] = std::make_pair (*it1, *it2); m_Tests[msgID] = std::make_pair (*it1, *it2);
} }
(*it1)->SendTunnelDataMsg ((*it2)->GetNextIdentHash (), (*it2)->GetNextTunnelID (), (*it1)->SendTunnelDataMsg ((*it2)->GetNextIdentHash (), (*it2)->GetNextTunnelID (),
@ -377,10 +398,11 @@ namespace tunnel
void TunnelPool::ProcessGarlicMessage (std::shared_ptr<I2NPMessage> msg) void TunnelPool::ProcessGarlicMessage (std::shared_ptr<I2NPMessage> msg)
{ {
std::unique_lock<std::mutex> l(m_LocalDestinationMutex);
if (m_LocalDestination) if (m_LocalDestination)
m_LocalDestination->ProcessGarlicMessage (msg); m_LocalDestination->ProcessGarlicMessage (msg);
else else
LogPrint (eLogWarning, "Tunnels: local destination doesn't exist, dropped"); LogPrint (eLogWarning, "Tunnels: local destination doesn't exist, garlic message dropped");
} }
void TunnelPool::ProcessDeliveryStatus (std::shared_ptr<I2NPMessage> msg) void TunnelPool::ProcessDeliveryStatus (std::shared_ptr<I2NPMessage> msg)
@ -425,10 +447,11 @@ namespace tunnel
} }
else else
{ {
std::unique_lock<std::mutex> l(m_LocalDestinationMutex);
if (m_LocalDestination) if (m_LocalDestination)
m_LocalDestination->ProcessDeliveryStatusMessage (msg); m_LocalDestination->ProcessDeliveryStatusMessage (msg);
else else
LogPrint (eLogWarning, "Tunnels: Local destination doesn't exist, dropped"); LogPrint (eLogWarning, "Tunnels: Local destination doesn't exist, delivery status message dropped");
} }
} }
@ -512,13 +535,16 @@ namespace tunnel
if (m_CustomPeerSelector) if (m_CustomPeerSelector)
return m_CustomPeerSelector->SelectPeers(path, numHops, isInbound); return m_CustomPeerSelector->SelectPeers(path, numHops, isInbound);
} }
// explicit peers in use // explicit peers in use
std::lock_guard<std::mutex> lock(m_ExplicitPeersMutex);
if (m_ExplicitPeers) return SelectExplicitPeers (path, isInbound); if (m_ExplicitPeers) return SelectExplicitPeers (path, isInbound);
return StandardSelectPeers(path, numHops, isInbound, std::bind(&TunnelPool::SelectNextHop, this, std::placeholders::_1, std::placeholders::_2)); return StandardSelectPeers(path, numHops, isInbound, std::bind(&TunnelPool::SelectNextHop, this, std::placeholders::_1, std::placeholders::_2));
} }
bool TunnelPool::SelectExplicitPeers (Path& path, bool isInbound) bool TunnelPool::SelectExplicitPeers (Path& path, bool isInbound)
{ {
std::unique_lock<std::mutex> l(m_ExplicitPeersMutex);
int numHops = isInbound ? m_NumInboundHops : m_NumOutboundHops; int numHops = isInbound ? m_NumInboundHops : m_NumOutboundHops;
if (numHops > (int)m_ExplicitPeers->size ()) numHops = m_ExplicitPeers->size (); if (numHops > (int)m_ExplicitPeers->size ()) numHops = m_ExplicitPeers->size ();
if (!numHops) return false; if (!numHops) return false;
@ -610,8 +636,10 @@ namespace tunnel
Path path; Path path;
if (SelectPeers (path, false)) if (SelectPeers (path, false))
{ {
std::unique_lock<std::mutex> ld(m_LocalDestinationMutex);
if (m_LocalDestination && !m_LocalDestination->SupportsEncryptionType (i2p::data::CRYPTO_KEY_TYPE_ECIES_X25519_AEAD)) if (m_LocalDestination && !m_LocalDestination->SupportsEncryptionType (i2p::data::CRYPTO_KEY_TYPE_ECIES_X25519_AEAD))
path.isShort = false; // because can't handle ECIES encrypted reply path.isShort = false; // because can't handle ECIES encrypted reply
ld.unlock();
std::shared_ptr<TunnelConfig> config; std::shared_ptr<TunnelConfig> config;
if (m_NumOutboundHops > 0) if (m_NumOutboundHops > 0)

6
libi2pd/TunnelPool.h
Unescape View File

@ -64,8 +64,8 @@ namespace tunnel
TunnelPool (int numInboundHops, int numOutboundHops, int numInboundTunnels, int numOutboundTunnels); TunnelPool (int numInboundHops, int numOutboundHops, int numInboundTunnels, int numOutboundTunnels);
~TunnelPool (); ~TunnelPool ();
std::shared_ptr<i2p::garlic::GarlicDestination> GetLocalDestination () const { return m_LocalDestination; }; std::shared_ptr<i2p::garlic::GarlicDestination> GetLocalDestination () const;
void SetLocalDestination (std::shared_ptr<i2p::garlic::GarlicDestination> destination) { m_LocalDestination = destination; }; void SetLocalDestination (std::shared_ptr<i2p::garlic::GarlicDestination> destination);
void SetExplicitPeers (std::shared_ptr<std::vector<i2p::data::IdentHash> > explicitPeers); void SetExplicitPeers (std::shared_ptr<std::vector<i2p::data::IdentHash> > explicitPeers);
void CreateTunnels (); void CreateTunnels ();
@ -126,8 +126,10 @@ namespace tunnel
private: private:
mutable std::mutex m_LocalDestinationMutex;
std::shared_ptr<i2p::garlic::GarlicDestination> m_LocalDestination; std::shared_ptr<i2p::garlic::GarlicDestination> m_LocalDestination;
int m_NumInboundHops, m_NumOutboundHops, m_NumInboundTunnels, m_NumOutboundTunnels; int m_NumInboundHops, m_NumOutboundHops, m_NumInboundTunnels, m_NumOutboundTunnels;
mutable std::mutex m_ExplicitPeersMutex;
std::shared_ptr<std::vector<i2p::data::IdentHash> > m_ExplicitPeers; std::shared_ptr<std::vector<i2p::data::IdentHash> > m_ExplicitPeers;
mutable std::mutex m_InboundTunnelsMutex; mutable std::mutex m_InboundTunnelsMutex;
std::set<std::shared_ptr<InboundTunnel>, TunnelCreationTimeCmp> m_InboundTunnels; // recent tunnel appears first std::set<std::shared_ptr<InboundTunnel>, TunnelCreationTimeCmp> m_InboundTunnels; // recent tunnel appears first

Write Preview
Loading…
Cancel
Save