Compare commits

..

1 Commits

Author SHA1 Message Date
Pratik B.
78e27ea63a Merge 17399da399 into 8b9f427aa4 2024-11-29 01:17:35 +04:00
6 changed files with 29 additions and 74 deletions

View File

@@ -73,7 +73,7 @@ namespace tunnel
const i2p::data::IdentHash& nextIdent, uint32_t nextTunnelID,
const i2p::crypto::AESKey& layerKey, const i2p::crypto::AESKey& ivKey):
TransitTunnel (receiveTunnelID, nextIdent, nextTunnelID,
layerKey, ivKey), m_Gateway(*this) {};
layerKey, ivKey), m_Gateway(this) {};
void SendTunnelDataMsg (std::shared_ptr<i2p::I2NPMessage> msg) override;
void FlushTunnelDataMsgs () override;

View File

@@ -447,29 +447,28 @@ namespace transport
return std::max (bwCongestionLevel, tbwCongestionLevel);
}
std::future<std::shared_ptr<TransportSession> > Transports::SendMessage (const i2p::data::IdentHash& ident, std::shared_ptr<i2p::I2NPMessage> msg)
void Transports::SendMessage (const i2p::data::IdentHash& ident, std::shared_ptr<i2p::I2NPMessage> msg)
{
if (m_IsOnline)
return SendMessages (ident, { msg });
return {}; // invalid future
SendMessages (ident, { msg });
}
std::future<std::shared_ptr<TransportSession> > Transports::SendMessages (const i2p::data::IdentHash& ident, std::list<std::shared_ptr<i2p::I2NPMessage> >& msgs)
void Transports::SendMessages (const i2p::data::IdentHash& ident, std::list<std::shared_ptr<i2p::I2NPMessage> >& msgs)
{
std::list<std::shared_ptr<i2p::I2NPMessage> > msgs1;
msgs.swap (msgs1);
return SendMessages (ident, std::move (msgs1));
SendMessages (ident, std::move (msgs1));
}
std::future<std::shared_ptr<TransportSession> > Transports::SendMessages (const i2p::data::IdentHash& ident, std::list<std::shared_ptr<i2p::I2NPMessage> >&& msgs)
void Transports::SendMessages (const i2p::data::IdentHash& ident, std::list<std::shared_ptr<i2p::I2NPMessage> >&& msgs)
{
return boost::asio::post (*m_Service, boost::asio::use_future ([this, ident, msgs = std::move(msgs)] () mutable
boost::asio::post (*m_Service, [this, ident, msgs = std::move(msgs)] () mutable
{
return PostMessages (ident, msgs);
}));
PostMessages (ident, msgs);
});
}
std::shared_ptr<TransportSession> Transports::PostMessages (const i2p::data::IdentHash& ident, std::list<std::shared_ptr<i2p::I2NPMessage> >& msgs)
void Transports::PostMessages (const i2p::data::IdentHash& ident, std::list<std::shared_ptr<i2p::I2NPMessage> >& msgs)
{
if (ident == i2p::context.GetRouterInfo ().GetIdentHash ())
{
@@ -477,9 +476,9 @@ namespace transport
for (auto& it: msgs)
m_LoopbackHandler.PutNextMessage (std::move (it));
m_LoopbackHandler.Flush ();
return nullptr;
return;
}
if(RoutesRestricted() && !IsRestrictedPeer(ident)) return nullptr;
if(RoutesRestricted() && !IsRestrictedPeer(ident)) return;
std::shared_ptr<Peer> peer;
{
std::lock_guard<std::mutex> l(m_PeersMutex);
@@ -490,13 +489,13 @@ namespace transport
if (!peer)
{
// check if not banned
if (i2p::data::IsRouterBanned (ident)) return nullptr; // don't create peer to unreachable router
if (i2p::data::IsRouterBanned (ident)) return; // don't create peer to unreachable router
// try to connect
bool connected = false;
try
{
auto r = netdb.FindRouter (ident);
if (r && (r->IsUnreachable () || !r->IsReachableFrom (i2p::context.GetRouterInfo ()))) return nullptr; // router found but non-reachable
if (r && (r->IsUnreachable () || !r->IsReachableFrom (i2p::context.GetRouterInfo ()))) return; // router found but non-reachable
peer = std::make_shared<Peer>(r, i2p::util::GetSecondsSinceEpoch ());
{
@@ -510,16 +509,12 @@ namespace transport
{
LogPrint (eLogError, "Transports: PostMessages exception:", ex.what ());
}
if (!connected) return nullptr;
if (!connected) return;
}
if (!peer) return nullptr;
if (!peer) return;
if (peer->IsConnected ())
{
auto session = peer->sessions.front ();
if (session) session->SendI2NPMessages (msgs);
return session;
}
peer->sessions.front ()->SendI2NPMessages (msgs);
else
{
auto sz = peer->delayedMessages.size ();
@@ -532,7 +527,7 @@ namespace transport
LogPrint (eLogWarning, "Transports: Router ", ident.ToBase64 (), " is banned. Peer dropped");
std::lock_guard<std::mutex> l(m_PeersMutex);
m_Peers.erase (ident);
return nullptr;
return;
}
}
if (sz > MAX_NUM_DELAYED_MESSAGES/2)
@@ -554,7 +549,6 @@ namespace transport
m_Peers.erase (ident);
}
}
return nullptr;
}
bool Transports::ConnectToPeer (const i2p::data::IdentHash& ident, std::shared_ptr<Peer> peer)

View File

@@ -11,7 +11,6 @@
#include <thread>
#include <mutex>
#include <future>
#include <condition_variable>
#include <functional>
#include <unordered_map>
@@ -144,9 +143,9 @@ namespace transport
std::shared_ptr<i2p::crypto::X25519Keys> GetNextX25519KeysPair ();
void ReuseX25519KeysPair (std::shared_ptr<i2p::crypto::X25519Keys> pair);
std::future<std::shared_ptr<TransportSession> > SendMessage (const i2p::data::IdentHash& ident, std::shared_ptr<i2p::I2NPMessage> msg);
std::future<std::shared_ptr<TransportSession> > SendMessages (const i2p::data::IdentHash& ident, std::list<std::shared_ptr<i2p::I2NPMessage> >& msgs);
std::future<std::shared_ptr<TransportSession> > SendMessages (const i2p::data::IdentHash& ident, std::list<std::shared_ptr<i2p::I2NPMessage> >&& msgs);
void SendMessage (const i2p::data::IdentHash& ident, std::shared_ptr<i2p::I2NPMessage> msg);
void SendMessages (const i2p::data::IdentHash& ident, std::list<std::shared_ptr<i2p::I2NPMessage> >& msgs);
void SendMessages (const i2p::data::IdentHash& ident, std::list<std::shared_ptr<i2p::I2NPMessage> >&& msgs);
void PeerConnected (std::shared_ptr<TransportSession> session);
void PeerDisconnected (std::shared_ptr<TransportSession> session);
@@ -190,7 +189,7 @@ namespace transport
void Run ();
void RequestComplete (std::shared_ptr<const i2p::data::RouterInfo> r, const i2p::data::IdentHash& ident);
void HandleRequestComplete (std::shared_ptr<const i2p::data::RouterInfo> r, i2p::data::IdentHash ident);
std::shared_ptr<TransportSession> PostMessages (const i2p::data::IdentHash& ident, std::list<std::shared_ptr<i2p::I2NPMessage> >& msgs);
void PostMessages (const i2p::data::IdentHash& ident, std::list<std::shared_ptr<i2p::I2NPMessage> >& msgs);
bool ConnectToPeer (const i2p::data::IdentHash& ident, std::shared_ptr<Peer> peer);
void SetPriority (std::shared_ptr<Peer> peer) const;
void HandlePeerCleanupTimer (const boost::system::error_code& ecode);

View File

@@ -139,7 +139,7 @@ namespace tunnel
public:
OutboundTunnel (std::shared_ptr<const TunnelConfig> config):
Tunnel (config), m_Gateway (*this), m_EndpointIdentHash (config->GetLastIdentHash ()) {};
Tunnel (config), m_Gateway (this), m_EndpointIdentHash (config->GetLastIdentHash ()) {};
void SendTunnelDataMsgTo (const uint8_t * gwHash, uint32_t gwTunnel, std::shared_ptr<i2p::I2NPMessage> msg);
virtual void SendTunnelDataMsgs (const std::vector<TunnelMessageBlock>& msgs); // multiple messages

View File

@@ -220,55 +220,21 @@ namespace tunnel
void TunnelGateway::SendBuffer ()
{
// create list or tunnel messages
m_Buffer.CompleteCurrentTunnelDataMessage ();
std::list<std::shared_ptr<I2NPMessage> > newTunnelMsgs;
const auto& tunnelDataMsgs = m_Buffer.GetTunnelDataMsgs ();
for (auto& tunnelMsg : tunnelDataMsgs)
{
auto newMsg = CreateEmptyTunnelDataMsg (false);
m_Tunnel.EncryptTunnelMsg (tunnelMsg, newMsg);
htobe32buf (newMsg->GetPayload (), m_Tunnel.GetNextTunnelID ());
m_Tunnel->EncryptTunnelMsg (tunnelMsg, newMsg);
htobe32buf (newMsg->GetPayload (), m_Tunnel->GetNextTunnelID ());
newMsg->FillI2NPMessageHeader (eI2NPTunnelData);
if (tunnelMsg->onDrop) newMsg->onDrop = tunnelMsg->onDrop;
newTunnelMsgs.push_back (newMsg);
m_NumSentBytes += TUNNEL_DATA_MSG_SIZE;
}
m_Buffer.ClearTunnelDataMsgs ();
// send
auto currentTransport = m_CurrentTransport.lock ();
if (!currentTransport)
{
// try to obtain transport from peding reequest or send thought transport is not complete
if (m_PendingTransport.valid ()) // pending request?
{
if (m_PendingTransport.wait_for(std::chrono::seconds(0)) == std::future_status::ready)
{
// pending request complete
currentTransport = m_PendingTransport.get (); // take tarnsports used in pending request
if (currentTransport)
{
if (currentTransport->IsEstablished ())
m_CurrentTransport = currentTransport;
else
currentTransport = nullptr;
}
}
else // still pending
{
// send through transports, but don't update pedning transport
i2p::transport::transports.SendMessages (m_Tunnel.GetNextIdentHash (), std::move (newTunnelMsgs));
return;
}
}
}
if (currentTransport) // session is good
// send to session directly
currentTransport->SendI2NPMessages (newTunnelMsgs);
else // no session yet
// send through transports
m_PendingTransport = i2p::transport::transports.SendMessages (m_Tunnel.GetNextIdentHash (), std::move (newTunnelMsgs));
i2p::transport::transports.SendMessages (m_Tunnel->GetNextIdentHash (), std::move (newTunnelMsgs));
}
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright (c) 2013-2024, The PurpleI2P Project
* Copyright (c) 2013-2021, The PurpleI2P Project
*
* This file is part of Purple i2pd project and licensed under BSD3
*
@@ -12,9 +12,7 @@
#include <inttypes.h>
#include <vector>
#include <memory>
#include <future>
#include "I2NPProtocol.h"
#include "TransportSession.h"
#include "TunnelBase.h"
namespace i2p
@@ -47,7 +45,7 @@ namespace tunnel
{
public:
TunnelGateway (TunnelBase& tunnel):
TunnelGateway (TunnelBase * tunnel):
m_Tunnel (tunnel), m_NumSentBytes (0) {};
void SendTunnelDataMsg (const TunnelMessageBlock& block);
void PutTunnelDataMsg (const TunnelMessageBlock& block);
@@ -56,11 +54,9 @@ namespace tunnel
private:
TunnelBase& m_Tunnel;
TunnelBase * m_Tunnel;
TunnelGatewayBuffer m_Buffer;
size_t m_NumSentBytes;
std::weak_ptr<i2p::transport::TransportSession> m_CurrentTransport;
std::future<std::shared_ptr<i2p::transport::TransportSession> > m_PendingTransport;
};
}
}