queue up and send piece messages one by one

2026-08-14 15:50:27 +00:00 · 2026-08-06 18:51:40 -04:00
parent bff47aff7c
commit 77f4f22df6
2 changed files with 110 additions and 53 deletions
@@ -130,7 +130,7 @@ namespace torrents
 //------------------------------------

 	Piece::Piece (size_t size, const uint8_t * hash):
-		m_Size (size), m_Data (nullptr)
+		m_Size (size), m_Data (nullptr), m_IsSending (false)
 	{
 		memcpy (m_Hash, hash, SHA_DIGEST_LENGTH);
 		m_Blocks = std::make_unique<std::vector<BlockStatus> >(GetNumBlocks (size), BlockStatus::Missing);
@@ -192,41 +192,34 @@ namespace torrents
 		{
 			f.seekp (offset, std::ios::beg);
 			f.write ((const char *)m_Data, m_Size);
-			delete[] m_Data; m_Data = nullptr;
+			if (!m_IsSending)
+			{
+				delete[] m_Data; m_Data = nullptr;
+			}
 			LogPrint (eLogDebug, "Torrents: Saved bytes ", offset, " - ", offset + m_Size - 1, " to ", fullPath);
 		}
 	}

-	void Piece::Load (const std::string& fullPath, size_t offset)
+	bool Piece::Load (const std::string& fullPath, size_t offset)
 	{
-		if (m_Data) return;
-		m_Data = new uint8_t[m_Size];
+		if (m_Data) return true;
 		std::ifstream f(fullPath, std::ifstream::binary);
-		if (f.is_open ())
+		if (f)
 		{
+			m_Data = new uint8_t[m_Size];
 			f.seekg (offset, std::ios::beg);
 			f.read ((char *)m_Data, m_Size);
+			LogPrint (eLogDebug, "Torrents: Loaded bytes ", offset, " - ", offset + m_Size - 1, " from ", fullPath);
 		}
+		else
+			return false;
+		return true;
 	}

-	std::pair<size_t, size_t> Piece::GetAvailableBuffer (size_t offset, size_t len) const
+	bool Piece::HasBlock (size_t offset) const
 	{
-		size_t block = offset/REQUEST_BLOCK_SIZE;
-		auto numBlocks = GetNumBlocks (len);
-		size_t ind = 0;
-		while (ind < numBlocks)
-		{
-			if (!IsAvailable (block + ind)) break;
-			ind++;
-		}
-		if (ind > 0)
-		{
-			if (offset + ind*REQUEST_BLOCK_SIZE > m_Size)
-				return { offset, m_Size - offset };
-			else
-				return { offset, ind*REQUEST_BLOCK_SIZE };
-		}
-		return { 0, 0 };
+		if (offset >= m_Size) return false;
+		return IsAvailable (offset/REQUEST_BLOCK_SIZE);
 	}

 	std::pair<size_t, size_t> Piece::GetNextBlockToRequest ()
@@ -440,7 +433,7 @@ namespace torrents
 				it.ClearAllRequests ();
 	}

-	void Torrent::Complete ()
+	void Torrent::SetComplete ()
 	{
 		for (auto& it: m_Pieces)
 			if (!it.IsComplete ())
@@ -460,7 +453,7 @@ namespace torrents
 	PeerConnection::PeerConnection (i2p::client::I2PService * owner,  std::shared_ptr<i2p::stream::Stream> stream):
 		i2p::client::I2PServiceHandler (owner), m_Stream (stream), m_ReceiveBufferOffset (0),
 		m_IsHandshakeSent (false), m_IsEstablished (false), m_IsChoked (true),
-		m_LastReceiveTime (0), m_LastSendTime (0), m_NumRequests (0)
+		m_LastReceiveTime (0), m_LastSendTime (0), m_NumRequests (0), m_IsSendingPieceMsg (false)
 	{
 	}

@@ -647,7 +640,7 @@ namespace torrents
 					RequestNextBlocks ();
 				break;
 				case eMessageTypeInterested:
-					LogPrint (eLogInfo, "Torrents: Interested message is not implemented");
+					SendUnchokeMsg ();
 				break;
 				case eMessageTypeNotInterested:
 					LogPrint (eLogInfo, "Torrents: Not interested message is not implemented");
@@ -830,11 +823,29 @@ namespace torrents
 		htobe32buf (sendBuffer.data () + 5, index);
 		htobe32buf (sendBuffer.data () + 9, offset);
 		memcpy (sendBuffer.data () + 13, data, len);
-		WriteToStream (sendBuffer.data (), sendBuffer.size ());
+		LogPrint (eLogDebug, "Torrents: Sending piece of ", sendBuffer.size (), " bytes");
+		m_IsSendingPieceMsg = true;
+		m_Stream->AsyncSend (sendBuffer.data (), sendBuffer.size (),
+			[s = shared_from_this ()](const boost::system::error_code& ecode, size_t bytes_transferred)
+			{
+				s->m_IsSendingPieceMsg = false;
+				if (!ecode)
+				{
+					if (!s->m_IncomingRequestsQueue.empty ())
+					{
+						s->SendRequestedBlock (s->m_IncomingRequestsQueue.front ());
+						s->m_IncomingRequestsQueue.pop_front ();
+					}
+				}
+				else
+					s->Terminate ();
+			});
+		m_LastSendTime = i2p::util::GetMonotonicSeconds ();
 	}

 	void PeerConnection::HandleRequestMsg (const uint8_t * buf, size_t len)
 	{
+		if (!m_Torrent) return;
 		if (len != REQUEST_MSG_PAYLOAD_LENGTH)
 		{
 			LogPrint (eLogWarning, "Torrents: Unexpected length of request message ", len);
@@ -843,39 +854,60 @@ namespace torrents
 		uint32_t index = bufbe32toh (buf);
 		uint32_t offset = bufbe32toh (buf + 4);
 		uint32_t length = bufbe32toh (buf + 8);
+		if (length > REQUEST_BLOCK_SIZE)
+		{
+			LogPrint (eLogWarning, "Torrents: Requested length is too long ", length);
+			return;
+		}
 		if (index < m_Torrent->GetNumPieces ())
 		{
 			Piece& piece = m_Torrent->GetPiece (index);
-			auto [availableOffset, availableLen] = piece.GetAvailableBuffer (offset, length);
-			if (availableLen)
+			if (piece.HasBlock (offset))
 			{
-				auto data = piece.GetData ();
-				if (data)
-					SendPieceMsg (index, availableOffset, data + availableOffset, availableLen);
-				else
+				if (m_IsSendingPieceMsg)
+					 m_IncomingRequestsQueue.emplace_back (index, offset, length);
+				else if (!SendRequestedBlock ({index, offset, length})) // block was not sent
 				{
 					// try to load from file
-					auto path = GetTorrentsTunnel ()->GetTorrentFilePath (m_Torrent->GetName ());
-					auto offset = index*m_Torrent->GetPieceLength ();
+					auto path = GetTorrentsTunnel ()->GetTorrentFilePath (m_Torrent->GetName () + (m_Torrent->IsComplete () ? "" : ".part"));
 					boost::asio::post (GetTorrentsTunnel ()->GetDiskIOService (),
-						[&piece, path, offset, length, index, torrent = m_Torrent, this]() // piece belongs to torrent
+					[&piece, path, index, offset, length, torrent = m_Torrent, s = shared_from_this ()]() // piece belongs to torrent
+					{
+						piece.SetIsSending (true);
+						if (piece.Load (path, index*torrent->GetPieceLength ()))
 						{
-							piece.Load (path, offset);
-							// send after loading
-							boost::asio::post (GetTorrentsTunnel ()->GetService (),
-								[&piece, offset, length, index, torrent, this]()
+							boost::asio::post (s->GetTorrentsTunnel ()->GetService (),
+								[requestedBlock = RequestedBlock{index, offset, length}, s]()
 								{
-									auto data = piece.GetData ();
-									if (data)
-									{
-										auto [availableOffset, availableLen] = piece.GetAvailableBuffer (offset, length);
-										SendPieceMsg (index, availableOffset, data + availableOffset, availableLen);
-									}
+									if (s->m_IncomingRequestsQueue.empty ())
+										s->SendRequestedBlock (requestedBlock);
+									else
+										s->m_IncomingRequestsQueue.push_back (requestedBlock);
 								});
-						});
+						}
+						piece.SetIsSending (false);
+					});
 				}
 			}
+			else
+				LogPrint (eLogWarning, "Torrents: Requested block (", index, ",", offset, ") is not available");
 		}
+		else
+			LogPrint (eLogWarning, "Torrents: Requested index ", index, "exceeds number of pieces", m_Torrent->GetNumPieces ());
+	}
+
+	bool PeerConnection::SendRequestedBlock (const RequestedBlock& requestedBlock)
+	{
+		bool ret = true;
+		Piece& piece = m_Torrent->GetPiece (requestedBlock.index);
+		piece.SetIsSending (true);
+		auto data = piece.GetData ();
+		if (data && piece.HasBlock (requestedBlock.offset))
+			SendPieceMsg (requestedBlock.index, requestedBlock.offset, data + requestedBlock.offset, requestedBlock.length);
+		else
+			ret = false;
+		piece.SetIsSending (false);
+		return ret;
 	}

 	void PeerConnection::SendRequestMsg (uint32_t index, uint32_t offset, uint32_t len)
@@ -897,6 +929,14 @@ namespace torrents
 		WriteToStream (buf, INTERESTED_MSG_LENGTH);
 	}

+	void PeerConnection::SendUnchokeMsg ()
+	{
+		uint8_t buf[UNCHOKE_MSG_LENGTH];
+		htobe32buf (buf, 1);
+		buf[4] = eMessageTypeUnchoke;
+		WriteToStream (buf, UNCHOKE_MSG_LENGTH);
+	}
+
 	bool PeerConnection::RequestNextBlock ()
 	{
 		if (!m_Torrent) return false;
@@ -960,6 +1000,9 @@ namespace torrents

 	void TorrentsTunnel::Stop ()
 	{
+		auto localDestination = GetLocalDestination ();
+		if (localDestination)
+			localDestination->StopAcceptingStreams ();
 		m_TrackerRequestsCheckTimer.cancel ();
 		m_KeepAliveCheckTimer.cancel ();
 		m_ReconnectCheckTimer.cancel ();
@@ -993,7 +1036,7 @@ namespace torrents
 				m_Torrents.emplace (torrent->GetInfoHash (), torrent);
 				auto filePath = GetTorrentFilePath (torrent->GetName ());
 				if (i2p::fs::Exists (filePath))
-					torrent->Complete ();
+					torrent->SetComplete ();
 				else
 				{
 					auto partFilePath = GetTorrentFilePath (torrent->GetName () + ".part");
@@ -50,6 +50,7 @@ namespace torrents

 	constexpr size_t HANDSHAKE_MSG_LENGTH = 68;
 	constexpr size_t INTERESTED_MSG_LENGTH = 5;
+	constexpr size_t UNCHOKE_MSG_LENGTH = 5;
 	constexpr size_t REQUEST_MSG_PAYLOAD_LENGTH = 12;
 	constexpr size_t HAVE_MSG_PAYLOAD_LENGTH = 4;

@@ -86,13 +87,14 @@ namespace torrents
 			bool IsComplete () const { return !m_Blocks; }
 			void Complete () { m_Blocks = nullptr; }
 			bool VerifyHash () const;
+			void SetIsSending (bool isSending) { m_IsSending = isSending; };

 			void BlockReceived (const uint8_t * block, size_t len, size_t offset);
 			void Dump (const std::string& fullPath, size_t offset);
-			void Load (const std::string& fullPath, size_t offset);
+			bool Load (const std::string& fullPath, size_t offset);
 			const uint8_t * GetData () const { return m_Data; }
 			size_t GetSize () const { return m_Size; }
-			std::pair<size_t, size_t> GetAvailableBuffer (size_t offset, size_t len) const; // return (offset, len) of available data
+			bool HasBlock (size_t offset) const;
 			std::pair<size_t, size_t> GetNextBlockToRequest (); // return (offset, len) of next buffer, len = 0 if no next buffer
 			void ClearAllRequests ();

@@ -106,6 +108,7 @@ namespace torrents
 			size_t m_Size;
 			uint8_t * m_Data, m_Hash[SHA_DIGEST_LENGTH];
 			std::unique_ptr<std::vector<BlockStatus> > m_Blocks;
+			bool m_IsSending;
 	};

 	class Torrent final
@@ -117,6 +120,9 @@ namespace torrents
 			Torrent (std::string_view buf);
 			void ParseTrackerResponse (std::string_view buf);

+			bool IsComplete () const { return m_IsComplete; }
+			void SetComplete ();
+
 			const std::string& GetAnnounce () const { return m_Announce; }
 			const std::string& GetName () const { return m_Name; }
 			size_t GetLength () const { return m_Length; }
@@ -131,8 +137,6 @@ namespace torrents
 			const std::unordered_set<i2p::data::IdentHash>&  GetPeers () const { return m_Peers; }
 			std::tuple<uint32_t, uint32_t, uint32_t> GetNextBlockToRequest (std::shared_ptr<PeerConnection> conn); // return (index, offset, len)
 			void ClearAllRequests ();
-			void Complete ();
-			bool IsComplete () const { return m_IsComplete; }

 			uint64_t GetNextTrackerRequestTime () const { return m_NextTrackerRequestTime; }
 			void SetNextTrackerRequestTime (uint64_t ts) { m_NextTrackerRequestTime = ts; }
@@ -160,6 +164,11 @@ namespace torrents
 	class TorrentsTunnel;
 	class PeerConnection: public i2p::client::I2PServiceHandler, public std::enable_shared_from_this<PeerConnection>
 	{
+		struct RequestedBlock
+		{
+			uint32_t index, offset, length;
+		};
+
 		public:

 			PeerConnection (i2p::client::I2PService * owner,  std::shared_ptr<i2p::stream::Stream> stream); // incoming
@@ -200,10 +209,13 @@ namespace torrents
 			void HandleRequestMsg (const uint8_t * buf, size_t len);
 			void SendRequestMsg (uint32_t index, uint32_t offset, uint32_t len);
 			void SendInterestedMsg ();
+			void SendUnchokeMsg ();

 			bool RequestNextBlock ();
 			void RequestNextBlocks ();

+			bool SendRequestedBlock (const RequestedBlock& requestedBlock);
+
 		private:

 			std::shared_ptr<i2p::stream::Stream> m_Stream;
@@ -214,7 +226,9 @@ namespace torrents
 			boost::dynamic_bitset<> m_RemoteBitfield;
 			bool m_IsHandshakeSent, m_IsEstablished, m_IsChoked;
 			uint64_t m_LastReceiveTime, m_LastSendTime; // monotonic seconds
-			size_t m_NumRequests;
+			size_t m_NumRequests; // outgoing
+			std::list<RequestedBlock> m_IncomingRequestsQueue;
+			bool m_IsSendingPieceMsg;
 	};

 	class TorrentsTunnel final: public i2p::client::I2PService