diff --git a/libi2pd/Streaming.cpp b/libi2pd/Streaming.cpp index 612b1058..3269e040 100644 --- a/libi2pd/Streaming.cpp +++ b/libi2pd/Streaming.cpp @@ -1050,6 +1050,7 @@ namespace stream it.second->Terminate (false); // we delete here m_Streams.clear (); m_IncomingStreams.clear (); + m_LastStream = nullptr; } } @@ -1058,9 +1059,16 @@ namespace stream uint32_t sendStreamID = packet->GetSendStreamID (); if (sendStreamID) { - auto it = m_Streams.find (sendStreamID); - if (it != m_Streams.end ()) - it->second->HandleNextPacket (packet); + if (!m_LastStream || sendStreamID != m_LastStream->GetRecvStreamID ()) + { + auto it = m_Streams.find (sendStreamID); + if (it != m_Streams.end ()) + m_LastStream = it->second; + else + m_LastStream = nullptr; + } + if (m_LastStream) + m_LastStream->HandleNextPacket (packet); else if (packet->IsEcho () && m_Owner->IsStreamingAnswerPings ()) { // ping @@ -1166,7 +1174,7 @@ namespace stream { auto s = std::make_shared (m_Owner->GetService (), *this, remote, port); std::unique_lock l(m_StreamsMutex); - m_Streams[s->GetRecvStreamID ()] = s; + m_Streams.emplace (s->GetRecvStreamID (), s); return s; } @@ -1174,8 +1182,8 @@ namespace stream { auto s = std::make_shared (m_Owner->GetService (), *this); std::unique_lock l(m_StreamsMutex); - m_Streams[s->GetRecvStreamID ()] = s; - m_IncomingStreams[receiveStreamID] = s; + m_Streams.emplace (s->GetRecvStreamID (), s); + m_IncomingStreams.emplace (receiveStreamID, s); return s; } @@ -1186,6 +1194,7 @@ namespace stream std::unique_lock l(m_StreamsMutex); m_Streams.erase (stream->GetRecvStreamID ()); m_IncomingStreams.erase (stream->GetSendStreamID ()); + if (m_LastStream == stream) m_LastStream = nullptr; } } diff --git a/libi2pd/Streaming.h b/libi2pd/Streaming.h index fe035136..c40c49f5 100644 --- a/libi2pd/Streaming.h +++ b/libi2pd/Streaming.h @@ -1,5 +1,5 @@ /* -* Copyright (c) 2013-2020, The PurpleI2P Project +* Copyright (c) 2013-2021, The PurpleI2P Project * * This file is part of Purple i2pd project and licensed under BSD3 * @@ -11,7 +11,7 @@ #include #include -#include +#include #include #include #include @@ -297,12 +297,13 @@ namespace stream uint16_t m_LocalPort; bool m_Gzip; // gzip compression of data messages std::mutex m_StreamsMutex; - std::map > m_Streams; // sendStreamID->stream - std::map > m_IncomingStreams; // receiveStreamID->stream + std::unordered_map > m_Streams; // sendStreamID->stream + std::unordered_map > m_IncomingStreams; // receiveStreamID->stream + std::shared_ptr m_LastStream; Acceptor m_Acceptor; std::list > m_PendingIncomingStreams; boost::asio::deadline_timer m_PendingIncomingTimer; - std::map > m_SavedPackets; // receiveStreamID->packets, arrived before SYN + std::unordered_map > m_SavedPackets; // receiveStreamID->packets, arrived before SYN i2p::util::MemoryPool m_PacketsPool; i2p::util::MemoryPool > m_I2NPMsgsPool;