diff --git a/libi2pd/Streaming.cpp b/libi2pd/Streaming.cpp index b9782005..c59946a3 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_PacketsPool (local.GetPacketsPool ()), m_RemoteLeaseSet (remote), m_ReceiveTimer (m_Service), + m_LocalDestination (local), m_RemoteLeaseSet (remote), m_SizeToReceive (0), 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_PacketsPool (local.GetPacketsPool ()), m_ReceiveTimer (m_Service), m_SendTimer (m_Service), + m_LocalDestination (local), m_SizeToReceive (0), 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_PacketsPool->Release (packet); + delete packet; } m_NACKedPackets.clear (); for (auto it: m_SentPackets) - m_PacketsPool->Release (it); + delete it; m_SentPackets.clear (); for (auto it: m_SavedPackets) - m_PacketsPool->Release (it); + delete it; m_SavedPackets.clear (); } @@ -422,10 +422,19 @@ namespace stream } packet->offset = packet->GetPayload () - packet->buf; - if (packet->GetLength () > 0) + size_t packetLength = packet->GetLength (); + if (packetLength > 0) { m_ReceiveQueue.push (packet); - m_ReceiveTimer.cancel (); + if (m_SizeToReceive) + { + if (packetLength < m_SizeToReceive) + m_SizeToReceive -= packetLength; + else + m_SizeToReceive = 0; + } + if (!m_SizeToReceive) + m_ReceiveTimer.cancel (); } else m_LocalDestination.DeletePacket (packet); @@ -2141,7 +2150,6 @@ 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 ()) @@ -2366,7 +2374,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 8ba64004..85616c9f 100644 --- a/libi2pd/Streaming.h +++ b/libi2pd/Streaming.h @@ -215,7 +215,7 @@ namespace stream void SendPing (); template - void AsyncReceive (const Buffer& buffer, ReceiveHandler&& handler, int timeout = 0); + void AsyncReceive (const Buffer& buffer, ReceiveHandler&& handler, int timeout = 0, size_t minSize = 0); size_t ReadSome (uint8_t * buf, size_t len) { return ConcatenatePackets (buf, len); }; size_t Receive (uint8_t * buf, size_t len, int timeout); @@ -286,7 +286,6 @@ namespace stream 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; @@ -295,6 +294,7 @@ namespace stream std::shared_ptr m_NextRemoteLease; std::shared_ptr m_CurrentOutboundTunnel; std::queue m_ReceiveQueue; + size_t m_SizeToReceive; // before invoke AsyncReceive handler std::set m_SavedPackets; std::list m_SentPackets; std::list m_NACKedPackets; @@ -350,10 +350,8 @@ 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); } - // streams keep the pool alive: pending handlers may outlive this destination - std::shared_ptr > GetPacketsPool () const { return m_PacketsPool; } + Packet * NewPacket () { return m_PacketsPool.Acquire(); } + void DeletePacket (Packet * p) { return m_PacketsPool.Release(p); } uint32_t GetRandom (); private: @@ -366,7 +364,7 @@ namespace stream private: - std::shared_ptr > m_PacketsPool; + i2p::util::MemoryPool m_PacketsPool; i2p::util::MemoryPool > m_I2NPMsgsPool; std::shared_ptr m_Owner; @@ -400,15 +398,16 @@ namespace stream //------------------------------------------------- template - void Stream::AsyncReceive (const Buffer& buffer, ReceiveHandler&& handler, int timeout) + void Stream::AsyncReceive (const Buffer& buffer, ReceiveHandler&& handler, int timeout, size_t minSize) { auto s = shared_from_this(); - boost::asio::post (m_Service, [s, buffer, handler = std::move (handler), timeout](void) mutable + boost::asio::post (m_Service, [s, buffer, handler = std::move (handler), timeout, minSize](void) mutable { if (!s->m_ReceiveQueue.empty () || (s->m_Status != eStreamStatusNew && s->m_Status != eStreamStatusOpen)) s->HandleReceiveTimer (boost::asio::error::make_error_code (boost::asio::error::operation_aborted), buffer, handler, 0); else { + s->m_SizeToReceive = minSize; int t = (timeout > MAX_RECEIVE_TIMEOUT) ? MAX_RECEIVE_TIMEOUT : timeout; s->m_ReceiveTimer.expires_after (std::chrono::seconds(t)); int left = timeout - t; @@ -456,6 +455,7 @@ namespace stream // timeout expired if (remainingTimeout <= 0) { + m_SizeToReceive = 0; if (m_Status != eStreamStatusTerminated) handler (boost::asio::error::make_error_code (boost::asio::error::timed_out), 0); } @@ -463,7 +463,7 @@ namespace stream { // intermediate interrupt SendUpdatedLeaseSet (); // send our leaseset if applicable - AsyncReceive (buffer, handler, remainingTimeout); + AsyncReceive (buffer, handler, remainingTimeout, m_SizeToReceive); } } } diff --git a/libi2pd_client/Torrents.cpp b/libi2pd_client/Torrents.cpp index f8359a88..f4e96b06 100644 --- a/libi2pd_client/Torrents.cpp +++ b/libi2pd_client/Torrents.cpp @@ -820,7 +820,8 @@ namespace torrents PEER_CONNECTION_RECEIVE_BUFFER_SIZE - m_ReceiveBufferOffset), std::bind (&PeerConnection::HandleStreamReceive, shared_from_this (), std::placeholders::_1, std::placeholders::_2), - PEER_CONNECTION_MAX_IDLE); + PEER_CONNECTION_MAX_IDLE, + m_NextMsgLength > m_ReceiveBufferOffset ? m_NextMsgLength - m_ReceiveBufferOffset : 0); } else // closed by peer {