From 6be42e68f147afe76f29c556ae1bcab4532968c2 Mon Sep 17 00:00:00 2001 From: PobreGato <315121269+pobregat0@users.noreply.github.com> Date: Thu, 13 Aug 2026 02:53:00 +0300 Subject: [PATCH] keep streaming packets pool alive for streams --- libi2pd/Streaming.cpp | 13 +++++++------ libi2pd/Streaming.h | 10 +++++++--- 2 files changed, 14 insertions(+), 9 deletions(-) diff --git a/libi2pd/Streaming.cpp b/libi2pd/Streaming.cpp index 4fb55163..497f4848 100644 --- a/libi2pd/Streaming.cpp +++ b/libi2pd/Streaming.cpp @@ -112,7 +112,7 @@ namespace stream m_IsWinDropped (true), m_IsChoking2 (false), m_IsChoking3 (false), m_IsClientChoked (false), m_IsClientChoked2 (false), m_IsTimeOutResend (false), m_IsImmediateAckRequested (false), m_IsRemoteLeaseChangeInProgress (false), m_IsBufferEmpty (false), m_IsJavaClient (false), m_DontSign (local.GetOwner ()->IsStreamingDontSign ()), - m_LocalDestination (local), m_RemoteLeaseSet (remote), m_ReceiveTimer (m_Service), + m_LocalDestination (local), m_PacketsPool (local.GetPacketsPool ()), m_RemoteLeaseSet (remote), m_ReceiveTimer (m_Service), m_SendTimer (m_Service), m_ResendTimer (m_Service), m_AckSendTimer (m_Service), m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (port), m_RTT (INITIAL_RTT), m_MinRTT (INITIAL_RTT), m_SlowRTT (INITIAL_RTT), m_FastRTT (INITIAL_RTT), m_WindowSize (INITIAL_WINDOW_SIZE), @@ -146,7 +146,7 @@ namespace stream m_IsWinDropped (true), m_IsChoking2 (false), m_IsChoking3 (false), m_IsClientChoked (false), m_IsClientChoked2 (false), m_IsTimeOutResend (false), m_IsImmediateAckRequested (false), m_IsRemoteLeaseChangeInProgress (false), m_IsBufferEmpty (false), m_IsJavaClient (false), m_DontSign (local.GetOwner ()->IsStreamingDontSign ()), - m_LocalDestination (local),m_ReceiveTimer (m_Service), m_SendTimer (m_Service), + m_LocalDestination (local), m_PacketsPool (local.GetPacketsPool ()), m_ReceiveTimer (m_Service), m_SendTimer (m_Service), m_ResendTimer (m_Service), m_AckSendTimer (m_Service),m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (0), m_RTT (INITIAL_RTT), m_MinRTT (INITIAL_RTT), m_SlowRTT (INITIAL_RTT), m_FastRTT (INITIAL_RTT), m_WindowSize (INITIAL_WINDOW_SIZE), m_MaxWindowSize (local.GetOwner ()->GetStreamingMaxWindowSize ()), @@ -204,17 +204,17 @@ namespace stream { auto packet = m_ReceiveQueue.front (); m_ReceiveQueue.pop (); - m_LocalDestination.DeletePacket (packet); + m_PacketsPool->Release (packet); } m_NACKedPackets.clear (); for (auto it: m_SentPackets) - m_LocalDestination.DeletePacket (it); + m_PacketsPool->Release (it); m_SentPackets.clear (); for (auto it: m_SavedPackets) - m_LocalDestination.DeletePacket (it); + m_PacketsPool->Release (it); m_SavedPackets.clear (); } @@ -2134,6 +2134,7 @@ namespace stream } StreamingDestination::StreamingDestination (std::shared_ptr owner, uint16_t localPort, bool gzip): + m_PacketsPool (std::make_shared >()), m_Owner (owner), m_LocalPort (localPort), m_Gzip (gzip), m_PendingIncomingTimer (m_Owner->GetService ()), m_LastCleanupTime (i2p::util::GetSecondsSinceEpoch ()) @@ -2358,7 +2359,7 @@ namespace stream auto ts = i2p::util::GetSecondsSinceEpoch (); if (m_Streams.empty () || ts > m_LastCleanupTime + STREAMING_DESTINATION_POOLS_CLEANUP_INTERVAL) { - m_PacketsPool.CleanUp (); + m_PacketsPool->CleanUp (); m_I2NPMsgsPool.CleanUp (); if (!m_NumIncomingConnectionsPerSecond.empty ()) { diff --git a/libi2pd/Streaming.h b/libi2pd/Streaming.h index 8852ea93..47b8431e 100644 --- a/libi2pd/Streaming.h +++ b/libi2pd/Streaming.h @@ -284,6 +284,8 @@ namespace stream m_IsClientChoked2, m_IsTimeOutResend, m_IsImmediateAckRequested, m_IsRemoteLeaseChangeInProgress, m_IsBufferEmpty, m_IsJavaClient, m_DontSign; StreamingDestination& m_LocalDestination; + // own reference to the pool: the destination can be gone by the time we are destroyed + std::shared_ptr > m_PacketsPool; std::shared_ptr m_RemoteIdentity; std::shared_ptr m_TransientVerifier; // in case of offline key std::shared_ptr m_RemoteLeaseSet; @@ -347,8 +349,10 @@ namespace stream void HandleDataMessagePayload (const uint8_t * buf, size_t len, i2p::garlic::ECIESX25519AEADRatchetSession * from); std::shared_ptr CreateDataMessage (const uint8_t * payload, size_t len, uint16_t toPort, bool checksum = true, bool gzip = false); - Packet * NewPacket () { return m_PacketsPool.Acquire(); } - void DeletePacket (Packet * p) { return m_PacketsPool.Release(p); } + Packet * NewPacket () { return m_PacketsPool->Acquire(); } + void DeletePacket (Packet * p) { return m_PacketsPool->Release(p); } + // streams keep the pool alive: pending handlers may outlive this destination + std::shared_ptr > GetPacketsPool () const { return m_PacketsPool; } uint32_t GetRandom (); private: @@ -361,7 +365,7 @@ namespace stream private: - i2p::util::MemoryPool m_PacketsPool; + std::shared_ptr > m_PacketsPool; i2p::util::MemoryPool > m_I2NPMsgsPool; std::shared_ptr m_Owner;