From ae0cf2e8311cd2b8cdaa5a0c9f33907a6b82bfeb Mon Sep 17 00:00:00 2001 From: orignal Date: Wed, 20 Oct 2021 21:05:22 -0400 Subject: [PATCH] use memory pool for tunnel messages --- libi2pd/I2NPProtocol.cpp | 12 +++--------- libi2pd/Tunnel.cpp | 16 +++++++++++++++- libi2pd/Tunnel.h | 6 ++++++ libi2pd/util.h | 38 ++++++++++++++++++++++++++++++++------ 4 files changed, 56 insertions(+), 16 deletions(-) diff --git a/libi2pd/I2NPProtocol.cpp b/libi2pd/I2NPProtocol.cpp index 0bc721ec..fa4fa7a6 100644 --- a/libi2pd/I2NPProtocol.cpp +++ b/libi2pd/I2NPProtocol.cpp @@ -38,20 +38,14 @@ namespace i2p std::shared_ptr NewI2NPTunnelMessage (bool endpoint) { - I2NPMessage * msg = nullptr; if (endpoint) - { - // should fit two tunnel message + tunnel gateway header, enough for one garlic encrypted streaming packet - msg = new I2NPMessageBuffer<2*i2p::tunnel::TUNNEL_DATA_MSG_SIZE + I2NP_HEADER_SIZE + TUNNEL_GATEWAY_HEADER_SIZE + 28>(); // reserved for alignment and NTCP 16 + 6 + 6 - msg->Align (6); - msg->offset += TUNNEL_GATEWAY_HEADER_SIZE; // reserve room for TunnelGateway header - } + return i2p::tunnel::tunnels.NewI2NPTunnelMessage (); else { - msg = new I2NPMessageBuffer(); // reserved for alignment and NTCP 16 + 6 + 12 + auto msg = new I2NPMessageBuffer(); // reserved for alignment and NTCP 16 + 6 + 12 msg->Align (12); + return std::shared_ptr(msg); } - return std::shared_ptr(msg); } std::shared_ptr NewI2NPMessage (size_t len) diff --git a/libi2pd/Tunnel.cpp b/libi2pd/Tunnel.cpp index c8a1cc99..3de91940 100644 --- a/libi2pd/Tunnel.cpp +++ b/libi2pd/Tunnel.cpp @@ -488,7 +488,7 @@ namespace tunnel i2p::util::SetThreadName("Tunnels"); std::this_thread::sleep_for (std::chrono::seconds(1)); // wait for other parts are ready - uint64_t lastTs = 0, lastPoolsTs = 0; + uint64_t lastTs = 0, lastPoolsTs = 0, lastMemoryPoolTs = 0; while (m_IsRunning) { try @@ -564,6 +564,11 @@ namespace tunnel ManageTunnelPools (ts); lastPoolsTs = ts; } + if (ts - lastMemoryPoolTs >= 120) // manage memory pool every 2 minutes + { + m_I2NPTunnelMessagesMemoryPool.CleanUpMt (); + lastMemoryPoolTs = ts; + } } } catch (std::exception& ex) @@ -935,6 +940,15 @@ namespace tunnel return outboundTunnel; } + std::shared_ptr Tunnels::NewI2NPTunnelMessage () + { + // should fit two tunnel message + tunnel gateway header, enough for one garlic encrypted streaming packet + auto msg = m_I2NPTunnelMessagesMemoryPool.AcquireSharedMt (); + msg->Align (6); + msg->offset += TUNNEL_GATEWAY_HEADER_SIZE; // reserve room for TunnelGateway header + return msg; + } + int Tunnels::GetTransitTunnelsExpirationTimeout () { int timeout = 0; diff --git a/libi2pd/Tunnel.h b/libi2pd/Tunnel.h index 930d0c58..30d9d565 100644 --- a/libi2pd/Tunnel.h +++ b/libi2pd/Tunnel.h @@ -18,6 +18,7 @@ #include #include #include +#include "util.h" #include "Queue.h" #include "Crypto.h" #include "TunnelConfig.h" @@ -40,6 +41,8 @@ namespace tunnel const int MAX_NUM_RECORDS = 8; const int HIGH_LATENCY_PER_HOP = 250; // in milliseconds + const size_t I2NP_TUNNEL_ENPOINT_MESSAGE_SIZE = 2*TUNNEL_DATA_MSG_SIZE + I2NP_HEADER_SIZE + TUNNEL_GATEWAY_HEADER_SIZE + 28; // reserved for alignment and NTCP 16 + 6 + 6 + enum TunnelState { eTunnelStatePending, @@ -219,6 +222,8 @@ namespace tunnel void DeleteTunnelPool (std::shared_ptr pool); void StopTunnelPool (std::shared_ptr pool); + std::shared_ptr NewI2NPTunnelMessage (); + private: template @@ -257,6 +262,7 @@ namespace tunnel std::list> m_Pools; std::shared_ptr m_ExploratoryPool; i2p::util::Queue > m_Queue; + i2p::util::MemoryPoolMt > m_I2NPTunnelMessagesMemoryPool; // some stats int m_NumSuccesiveTunnelCreations, m_NumFailedTunnelCreations; diff --git a/libi2pd/util.h b/libi2pd/util.h index 282ce7aa..d681829c 100644 --- a/libi2pd/util.h +++ b/libi2pd/util.h @@ -56,12 +56,8 @@ namespace util void CleanUp () { - while (m_Head) - { - auto tmp = m_Head; - m_Head = static_cast(*(void * *)m_Head); // next - ::operator delete ((void *)tmp); - } + CleanUp (m_Head); + m_Head = nullptr; } template @@ -98,6 +94,18 @@ namespace util std::bind (&MemoryPool::Release, this, std::placeholders::_1)); } + protected: + + void CleanUp (T * head) + { + while (head) + { + auto tmp = head; + head = static_cast(*(void * *)head); // next + ::operator delete ((void *)tmp); + } + } + protected: T * m_Head; @@ -131,6 +139,24 @@ namespace util this->Release (it); } + template + std::shared_ptr AcquireSharedMt (TArgs&&... args) + { + return std::shared_ptr(AcquireMt (std::forward(args)...), + std::bind::*)(T *)> (&MemoryPoolMt::ReleaseMt, this, std::placeholders::_1)); + } + + void CleanUpMt () + { + T * head; + { + std::lock_guard l(m_Mutex); + head = this->m_Head; + this->m_Head = nullptr; + } + if (head) this->CleanUp (head); + } + private: std::mutex m_Mutex;