memory pool for SSU messages and fragments

libi2pd/SSU.cpp

@@ -919,6 +919,11 @@ namespace transport
 			}
 			if (numDeleted > 0)
 				LogPrint (eLogDebug, "SSU: ", numDeleted, " peer tests have been expired");
+			// some cleaups. TODO: use separate timer
+			m_FragmentsPool.CleanUp ();
+			m_IncompleteMessagesPool.CleanUp ();
+			m_SentMessagesPool.CleanUp ();
+			
 			SchedulePeerTestsCleanupTimer ();
 		}
 	}

libi2pd/SSU.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
 *
@@ -18,6 +18,7 @@
 #include <mutex>
 #include <boost/asio.hpp>
 #include "Crypto.h"
+#include "util.h"
 #include "I2PEndian.h"
 #include "Identity.h"
 #include "RouterInfo.h"
@@ -63,6 +64,10 @@ namespace transport
 			void DeleteAllSessions ();
 
 			boost::asio::io_service& GetService () { return m_Service; };
+			i2p::util::MemoryPool<Fragment>& GetFragmentsPool () { return m_FragmentsPool; };
+			i2p::util::MemoryPool<IncompleteMessage>& GetIncompleteMessagesPool () { return m_IncompleteMessagesPool; };
+			i2p::util::MemoryPool<SentMessage>& GetSentMessagesPool () { return m_SentMessagesPool; };
+			
 			uint16_t GetPort () const { return m_Endpoint.port (); };
 			void SetLocalAddress (const boost::asio::ip::address& localAddress);
 			
@@ -136,6 +141,10 @@ namespace transport
 			std::map<uint32_t, std::shared_ptr<SSUSession> > m_Relays; // we are introducer
 			std::map<uint32_t, PeerTest> m_PeerTests; // nonce -> creation time in milliseconds
 
+			i2p::util::MemoryPool<Fragment> m_FragmentsPool;
+			i2p::util::MemoryPool<IncompleteMessage> m_IncompleteMessagesPool;
+			i2p::util::MemoryPool<SentMessage> m_SentMessagesPool;
+			
 		public:
 			// for HTTP only
 			const decltype(m_Sessions)& GetSessions () const { return m_Sessions; };

libi2pd/SSUData.cpp

@@ -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
 *
@@ -140,7 +140,7 @@ namespace transport
 							if (bitfield & mask)
 							{
 								if (fragment < numSentFragments)
-									it->second->fragments[fragment].reset (nullptr);
+									it->second->fragments[fragment] = nullptr;
 							}
 							fragment++;
 							mask <<= 1;
@@ -182,9 +182,9 @@ namespace transport
 				auto msg = NewI2NPShortMessage ();
 				msg->len -= I2NP_SHORT_HEADER_SIZE;
 				it = m_IncompleteMessages.insert (std::make_pair (msgID,
-					std::unique_ptr<IncompleteMessage>(new IncompleteMessage (msg)))).first;
+					m_Session.GetServer ().GetIncompleteMessagesPool ().AcquireShared (msg))).first;
 			}
-			std::unique_ptr<IncompleteMessage>& incompleteMessage = it->second;
+			auto& incompleteMessage = it->second;
 			// mark fragment as received
 			if (fragmentNum < 64)
 				incompleteMessage->receivedFragmentsBits |= (0x01 << fragmentNum);
@@ -224,8 +224,8 @@ namespace transport
 				{
 					// missing fragment
 					LogPrint (eLogWarning, "SSU: Missing fragments from ", (int)incompleteMessage->nextFragmentNum, " to ", fragmentNum - 1, " of message ", msgID);
-					auto savedFragment = new Fragment (fragmentNum, buf, fragmentSize, isLast);
-					if (incompleteMessage->savedFragments.insert (std::unique_ptr<Fragment>(savedFragment)).second)	
+					auto savedFragment = m_Session.GetServer ().GetFragmentsPool ().AcquireShared (fragmentNum, buf, fragmentSize, isLast);
+					if (incompleteMessage->savedFragments.insert (savedFragment).second)	
 						incompleteMessage->lastFragmentInsertTime = i2p::util::GetSecondsSinceEpoch ();
 					else
 						LogPrint (eLogWarning, "SSU: Fragment ", (int)fragmentNum, " of message ", msgID, " already saved");
@@ -313,8 +313,8 @@ namespace transport
 		if (m_SentMessages.empty ()) // schedule resend at first message only
 			ScheduleResend ();
 
-		auto ret = m_SentMessages.insert (std::make_pair (msgID, std::unique_ptr<SentMessage>(new SentMessage)));
-		std::unique_ptr<SentMessage>& sentMessage = ret.first->second;
+		auto ret = m_SentMessages.insert (std::make_pair (msgID, m_Session.GetServer ().GetSentMessagesPool ().AcquireShared ()));
+		auto& sentMessage = ret.first->second;
 		if (ret.second)
 		{
 			sentMessage->nextResendTime = i2p::util::GetSecondsSinceEpoch () + RESEND_INTERVAL;
@@ -328,7 +328,7 @@ namespace transport
 		uint32_t fragmentNum = 0;
 		while (len > 0 && fragmentNum <= 127)
 		{
-			Fragment * fragment = new Fragment;
+			auto fragment = m_Session.GetServer ().GetFragmentsPool ().AcquireShared ();
 			fragment->fragmentNum = fragmentNum;
 			uint8_t	* payload = fragment->buf + sizeof (SSUHeader);
 			*payload = DATA_FLAG_WANT_REPLY; // for compatibility
@@ -358,7 +358,7 @@ namespace transport
 				size += padding;
 			}	
 			fragment->len = size;
-			fragments.push_back (std::unique_ptr<Fragment> (fragment));
+			fragments.push_back (fragment);
 
 			// encrypt message with session key
 			uint8_t buf[SSU_V4_MAX_PACKET_SIZE + 18];

libi2pd/SSUData.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
 *
@@ -64,7 +64,7 @@ namespace transport
 
 	struct FragmentCmp
 	{
-		bool operator() (const std::unique_ptr<Fragment>& f1, const std::unique_ptr<Fragment>& f2) const
+		bool operator() (const std::shared_ptr<Fragment>& f1, const std::shared_ptr<Fragment>& f2) const
 		{
 			return f1->fragmentNum < f2->fragmentNum;
 		};
@@ -76,7 +76,7 @@ namespace transport
 		int nextFragmentNum;
 		uint32_t lastFragmentInsertTime; // in seconds
 		uint64_t receivedFragmentsBits;
-		std::set<std::unique_ptr<Fragment>, FragmentCmp> savedFragments;
+		std::set<std::shared_ptr<Fragment>, FragmentCmp> savedFragments;
 
 		IncompleteMessage (std::shared_ptr<I2NPMessage> m): msg (m), nextFragmentNum (0), 
 			lastFragmentInsertTime (0), receivedFragmentsBits (0) {};
@@ -85,7 +85,7 @@ namespace transport
 
 	struct SentMessage
 	{
-		std::vector<std::unique_ptr<Fragment> > fragments;
+		std::vector<std::shared_ptr<Fragment> > fragments;
 		uint32_t nextResendTime; // in seconds
 		int numResends;
 	};
@@ -126,8 +126,8 @@ namespace transport
 		private:
 
 			SSUSession& m_Session;
-			std::unordered_map<uint32_t, std::unique_ptr<IncompleteMessage> > m_IncompleteMessages;
-			std::unordered_map<uint32_t, std::unique_ptr<SentMessage> > m_SentMessages;
+			std::unordered_map<uint32_t, std::shared_ptr<IncompleteMessage> > m_IncompleteMessages;
+			std::unordered_map<uint32_t, std::shared_ptr<SentMessage> > m_SentMessages;
 			std::unordered_set<uint32_t> m_ReceivedMessages;
 			boost::asio::deadline_timer m_ResendTimer, m_IncompleteMessagesCleanupTimer;
 			int m_MaxPacketSize, m_PacketSize;

libi2pd/SSUSession.h

@@ -89,7 +89,8 @@ namespace transport
 			void Done ();
 			void Failed ();
 			const boost::asio::ip::udp::endpoint& GetRemoteEndpoint () { return m_RemoteEndpoint; };
-
+			SSUServer& GetServer () { return m_Server; };
+			
 			bool IsV6 () const { return m_RemoteEndpoint.address ().is_v6 (); };
 			void SendI2NPMessages (const std::vector<std::shared_ptr<I2NPMessage> >& msgs);
 			void SendPeerTest (); // Alice

libi2pd/util.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
 *
@@ -50,6 +50,11 @@ namespace util
 
 			MemoryPool (): m_Head (nullptr) {}
 			~MemoryPool ()
+			{
+				CleanUp ();
+			}
+
+			void CleanUp ()
 			{
 				while (m_Head)
 				{
@@ -57,8 +62,8 @@ namespace util
 					m_Head = static_cast<T*>(*(void * *)m_Head); // next
 					::operator delete ((void *)tmp);
 				}
-			}
-
+			}		
+			
 			template<typename... TArgs>
 			T * Acquire (TArgs&&... args)
 			{