From e76d09e1a157e54fbb5d082bdf48d04b4218fd55 Mon Sep 17 00:00:00 2001 From: orignal Date: Sun, 15 Dec 2024 18:03:31 -0500 Subject: [PATCH] send tunnel participant data to transport session directly. Implemented TunnelTransportSender --- libi2pd/NetDb.cpp | 2 +- libi2pd/TransitTunnel.cpp | 3 +- libi2pd/TransitTunnel.h | 1 + libi2pd/Transports.cpp | 7 ----- libi2pd/Transports.h | 1 - libi2pd/TunnelBase.cpp | 64 +++++++++++++++++++++++++++++++++++++++ libi2pd/TunnelBase.h | 25 ++++++++++++++- libi2pd/TunnelGateway.cpp | 35 ++------------------- libi2pd/TunnelGateway.h | 5 +-- 9 files changed, 95 insertions(+), 48 deletions(-) create mode 100644 libi2pd/TunnelBase.cpp diff --git a/libi2pd/NetDb.cpp b/libi2pd/NetDb.cpp index c712af8b..09182948 100644 --- a/libi2pd/NetDb.cpp +++ b/libi2pd/NetDb.cpp @@ -502,7 +502,7 @@ namespace data } // send them off - i2p::transport::transports.SendMessages(ih, requests); + i2p::transport::transports.SendMessages(ih, std::move (requests)); } bool NetDb::LoadRouterInfo (const std::string& path, uint64_t ts) diff --git a/libi2pd/TransitTunnel.cpp b/libi2pd/TransitTunnel.cpp index 883034fe..7dbed003 100644 --- a/libi2pd/TransitTunnel.cpp +++ b/libi2pd/TransitTunnel.cpp @@ -62,7 +62,8 @@ namespace tunnel auto num = m_TunnelDataMsgs.size (); if (num > 1) LogPrint (eLogDebug, "TransitTunnel: ", GetTunnelID (), "->", GetNextTunnelID (), " ", num); - i2p::transport::transports.SendMessages (GetNextIdentHash (), m_TunnelDataMsgs); // send and clear + if (!m_Sender) m_Sender = std::make_unique(); + m_Sender->SendMessagesTo (GetNextIdentHash (), m_TunnelDataMsgs); // send and clear } } diff --git a/libi2pd/TransitTunnel.h b/libi2pd/TransitTunnel.h index 14f9b997..89653931 100644 --- a/libi2pd/TransitTunnel.h +++ b/libi2pd/TransitTunnel.h @@ -63,6 +63,7 @@ namespace tunnel size_t m_NumTransmittedBytes; std::list > m_TunnelDataMsgs; + std::unique_ptr m_Sender; }; class TransitTunnelGateway: public TransitTunnel diff --git a/libi2pd/Transports.cpp b/libi2pd/Transports.cpp index 03416137..06bd724d 100644 --- a/libi2pd/Transports.cpp +++ b/libi2pd/Transports.cpp @@ -454,13 +454,6 @@ namespace transport return {}; // invalid future } - std::future > Transports::SendMessages (const i2p::data::IdentHash& ident, std::list >& msgs) - { - std::list > msgs1; - msgs.swap (msgs1); - return SendMessages (ident, std::move (msgs1)); - } - std::future > Transports::SendMessages (const i2p::data::IdentHash& ident, std::list >&& msgs) { return boost::asio::post (*m_Service, boost::asio::use_future ([this, ident, msgs = std::move(msgs)] () mutable diff --git a/libi2pd/Transports.h b/libi2pd/Transports.h index b72d5b70..85330162 100644 --- a/libi2pd/Transports.h +++ b/libi2pd/Transports.h @@ -145,7 +145,6 @@ namespace transport void ReuseX25519KeysPair (std::shared_ptr pair); std::future > SendMessage (const i2p::data::IdentHash& ident, std::shared_ptr msg); - std::future > SendMessages (const i2p::data::IdentHash& ident, std::list >& msgs); std::future > SendMessages (const i2p::data::IdentHash& ident, std::list >&& msgs); void PeerConnected (std::shared_ptr session); diff --git a/libi2pd/TunnelBase.cpp b/libi2pd/TunnelBase.cpp new file mode 100644 index 00000000..031988a6 --- /dev/null +++ b/libi2pd/TunnelBase.cpp @@ -0,0 +1,64 @@ +/* +* Copyright (c) 2024, The PurpleI2P Project +* +* This file is part of Purple i2pd project and licensed under BSD3 +* +* See full license text in LICENSE file at top of project tree +* +*/ + +#include "Transports.h" +#include "TunnelBase.h" + +namespace i2p +{ +namespace tunnel +{ + void TunnelTransportSender::SendMessagesTo (const i2p::data::IdentHash& to, + std::list >&& msgs) + { + if (msgs.empty ()) return; + auto currentTransport = m_CurrentTransport.lock (); + if (!currentTransport) + { + // try to obtain transport from pending request 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 transports 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 pending transport + i2p::transport::transports.SendMessages (to, std::move (msgs)); + return; + } + } + } + if (currentTransport) // session is good + // send to session directly + currentTransport->SendI2NPMessages (msgs); + else // no session yet + // send through transports + m_PendingTransport = i2p::transport::transports.SendMessages (to, std::move (msgs)); + + } + + void TunnelTransportSender::SendMessagesTo (const i2p::data::IdentHash& to, + std::list >& msgs) + { + std::list > msgs1; + msgs.swap (msgs1); + SendMessagesTo (to, std::move (msgs1)); + } +} +} diff --git a/libi2pd/TunnelBase.h b/libi2pd/TunnelBase.h index d58ec2d7..91d0fafc 100644 --- a/libi2pd/TunnelBase.h +++ b/libi2pd/TunnelBase.h @@ -1,5 +1,5 @@ /* -* Copyright (c) 2013-2022, The PurpleI2P Project +* Copyright (c) 2013-2024, The PurpleI2P Project * * This file is part of Purple i2pd project and licensed under BSD3 * @@ -11,12 +11,19 @@ #include #include +#include +#include #include "Timestamp.h" #include "I2NPProtocol.h" #include "Identity.h" namespace i2p { +namespace transport +{ + class TransportSession; +} + namespace tunnel { const size_t TUNNEL_DATA_MSG_SIZE = 1028; @@ -76,6 +83,22 @@ namespace tunnel return t1 < t2; } }; + + class TunnelTransportSender final + { + public: + + TunnelTransportSender () = default; + ~TunnelTransportSender () = default; + + void SendMessagesTo (const i2p::data::IdentHash& to, std::list >&& msgs); + void SendMessagesTo (const i2p::data::IdentHash& to, std::list >& msgs); // send and clear + + private: + + std::weak_ptr m_CurrentTransport; + std::future > m_PendingTransport; + }; } } diff --git a/libi2pd/TunnelGateway.cpp b/libi2pd/TunnelGateway.cpp index e2797f8c..9e27d207 100644 --- a/libi2pd/TunnelGateway.cpp +++ b/libi2pd/TunnelGateway.cpp @@ -235,40 +235,9 @@ namespace tunnel m_NumSentBytes += TUNNEL_DATA_MSG_SIZE; } m_Buffer.ClearTunnelDataMsgs (); - // send - auto currentTransport = m_CurrentTransport.lock (); - if (!currentTransport) - { - // try to obtain transport from pending request 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 transports 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 pending 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)); + if (!m_Sender) m_Sender = std::make_unique(); + m_Sender->SendMessagesTo (m_Tunnel.GetNextIdentHash (), std::move (newTunnelMsgs)); } } } diff --git a/libi2pd/TunnelGateway.h b/libi2pd/TunnelGateway.h index 4cfd5308..292392ae 100644 --- a/libi2pd/TunnelGateway.h +++ b/libi2pd/TunnelGateway.h @@ -12,9 +12,7 @@ #include #include #include -#include #include "I2NPProtocol.h" -#include "TransportSession.h" #include "TunnelBase.h" namespace i2p @@ -59,8 +57,7 @@ namespace tunnel TunnelBase& m_Tunnel; TunnelGatewayBuffer m_Buffer; size_t m_NumSentBytes; - std::weak_ptr m_CurrentTransport; - std::future > m_PendingTransport; + std::unique_ptr m_Sender; }; } }