From 1a4e82037963215fa32901389628fdfad8b6025b Mon Sep 17 00:00:00 2001 From: orignal Date: Sun, 2 Aug 2026 21:43:14 -0400 Subject: [PATCH] request next block --- libi2pd_client/Torrents.cpp | 79 +++++++++++++++++++++++++++++++------ libi2pd_client/Torrents.h | 20 ++++++++-- 2 files changed, 82 insertions(+), 17 deletions(-) diff --git a/libi2pd_client/Torrents.cpp b/libi2pd_client/Torrents.cpp index cd1a62f7..55998fbb 100644 --- a/libi2pd_client/Torrents.cpp +++ b/libi2pd_client/Torrents.cpp @@ -130,9 +130,10 @@ namespace torrents //------------------------------------ Piece::Piece (size_t size, const uint8_t * hash): - m_Size (size), m_Data (nullptr), m_Blocks (GetNumBlocks (size)) + m_Size (size), m_Data (nullptr) { memcpy (m_Hash, hash, SHA_DIGEST_LENGTH); + m_Blocks = std::make_unique >(GetNumBlocks (size), BlockStatus::Missing); } Piece::~Piece () @@ -150,9 +151,9 @@ namespace torrents bool Piece::IsAvailable (int block) const { - if (m_Blocks.empty ()) return true; - if (block < 0 || block >= (int)m_Blocks.size ()) return false; - return m_Blocks.test (block); + if (!m_Blocks) return true; + if (block < 0 || block >= (int)m_Blocks->size ()) return false; + return (*m_Blocks)[block] == BlockStatus::Available; } size_t Piece::GetNumBlocks (size_t len) const @@ -165,13 +166,19 @@ namespace torrents void Piece::BlockReceived (const uint8_t * block, size_t len, size_t offset) { - if (offset + len >= m_Size) return; + if (offset + len >= m_Size || !m_Blocks) return; if (!m_Data) m_Data = new uint8_t[m_Size]; memcpy (m_Data + offset, block, len); size_t startBlock = offset/REQUEST_BLOCK_SIZE; auto numBlocks = GetNumBlocks (len); - for (size_t i = 0; i < numBlocks; i++) - m_Blocks.set (startBlock + i); + if (numBlocks > 0) + { + std::fill_n (m_Blocks->begin () + startBlock, numBlocks, BlockStatus::Available); + if (std::find_if (m_Blocks->begin (), m_Blocks->end (), + [](BlockStatus status) { return status != BlockStatus::Available; }) == m_Blocks->end ()) + // all blocks are available + m_Blocks = nullptr; + } } void Piece::Dump (const std::string& fullPath, size_t offset) @@ -183,7 +190,7 @@ namespace torrents f.seekp (offset, std::ios::beg); f.write ((const char *)m_Data, m_Size); delete[] m_Data; m_Data = nullptr; - m_Blocks.resize (0); + m_Blocks = nullptr; m_Connections.clear (); } } @@ -220,6 +227,25 @@ namespace torrents return { 0, 0 }; } + std::pair Piece::GetNextBlockToRequest () + { + if (m_Blocks) + { + size_t ind = 0; + for (auto& it: *m_Blocks) + { + if (it == BlockStatus::Missing) + { + it = BlockStatus::Requested; + auto offset = ind*REQUEST_BLOCK_SIZE; + return { offset, (offset + REQUEST_BLOCK_SIZE <= m_Size) ? REQUEST_BLOCK_SIZE : m_Size - offset }; + } + ind++; + } + } + return { 0, 0 }; + } + void Piece::AddConnection (std::shared_ptr connection) { m_Connections.emplace_back (connection); @@ -370,6 +396,25 @@ namespace torrents return ret; } + std::tuple Torrent::GetNextBlockToRequest (std::shared_ptr conn) + { + if (conn) + { + uint32_t ind = 0; + for (auto& it: m_Pieces) + { + if (!it.IsComplete ()) + { + auto [offset, len] = it.GetNextBlockToRequest (); + if (len > 0) + return { ind, offset, len }; + } + ind++; + } + } + return { 0, 0, 0 }; + } + PeerConnection::PeerConnection (i2p::client::I2PService * owner, std::shared_ptr stream): i2p::client::I2PServiceHandler (owner), m_Stream (stream), m_ReceiveBufferOffset (0), m_IsHandshakeSent (false), m_IsEstablished (false), m_LastReceiveTime (0), m_LastSendTime (0) @@ -523,7 +568,7 @@ namespace torrents offset += 4; if (msgLen >= 1) { - LogPrint (eLogDebug, "Torrents: Received msg type ", (int)m_ReceiveBuffer[offset]); + LogPrint (eLogDebug, "Torrents: Received msg type ", (int)m_ReceiveBuffer[offset], " len ", msgLen); switch (m_ReceiveBuffer[offset]) { case eMessageTypeHave: @@ -584,6 +629,7 @@ namespace torrents if (bitfield.size ()) SendBitfieldMsg (bitfield.data (), bitfield.size ()); m_IsEstablished = true; + RequestNextBlock (); // TODO: remove later return HANDSHAKE_MSG_LENGTH; } @@ -752,18 +798,25 @@ namespace torrents } } - void PeerConnection::RequestPiece (uint32_t index) + void PeerConnection::SendRequestMsg (uint32_t index, uint32_t offset, uint32_t len) { - if (!m_Torrent) return; std::vector sendBuffer (REQUEST_MSG_PAYLOAD_LENGTH + 5); htobe32buf (sendBuffer.data (), REQUEST_MSG_PAYLOAD_LENGTH + 1); // msg length sendBuffer[4] = eMessageTypeRequest; // msg ID htobe32buf (sendBuffer.data () + 5, index); // index - memset (sendBuffer.data () + 9, 0, 4); // offset - htobe32buf (sendBuffer.data () + 13, m_Torrent->GetPieceLength ()); // length + memset (sendBuffer.data () + 9, offset, 4); // offset + htobe32buf (sendBuffer.data () + 13, len); // length WriteToStream (sendBuffer.data (), sendBuffer.size ()); } + void PeerConnection::RequestNextBlock () + { + if (!m_Torrent) return; + auto [index, offset, len] = m_Torrent->GetNextBlockToRequest (shared_from_this ()); + if (len > 0) + SendRequestMsg (index, offset, len); + } + TorrentsTunnel::TorrentsTunnel (std::shared_ptr localDestination, std::string_view torrentsDir, std::string_view trackers): i2p::client::I2PService (localDestination), m_TorrentsDir (torrentsDir), diff --git a/libi2pd_client/Torrents.h b/libi2pd_client/Torrents.h index 59b734d9..0550bbf7 100644 --- a/libi2pd_client/Torrents.h +++ b/libi2pd_client/Torrents.h @@ -22,6 +22,7 @@ #include #include #include +#include #include "util.h" #include "Streaming.h" #include "HTTP.h" @@ -59,12 +60,20 @@ namespace torrents class PeerConnection; class Piece final { + enum class BlockStatus + { + Missing, + Available, + Requested + }; + public: Piece (size_t size, const uint8_t * hash); + Piece (Piece&& ) = default; ~Piece (); - bool IsComplete () const { return m_Blocks.all (); } + bool IsComplete () const { return !m_Blocks; } bool VerifyHash () const; void BlockReceived (const uint8_t * block, size_t len, size_t offset); @@ -73,6 +82,7 @@ namespace torrents const uint8_t * GetData () const { return m_Data; } size_t GetSize () const { return m_Size; } std::pair GetAvailableBuffer (size_t offset, size_t len) const; // return (offset, len) of available data + std::pair GetNextBlockToRequest (); // return (offset, len) of next buffer, len = 0 if no next buffer void AddConnection (std::shared_ptr connection); void RemoveConnection (std::shared_ptr connection); @@ -86,7 +96,7 @@ namespace torrents size_t m_Size; uint8_t * m_Data, m_Hash[SHA_DIGEST_LENGTH]; - boost::dynamic_bitset<> m_Blocks; + std::unique_ptr > m_Blocks; std::list > m_Connections; // for incomplete pieces only }; @@ -110,6 +120,7 @@ namespace torrents Piece& GetPiece (int index) { return m_Pieces[index]; } std::vector CreateBitfield () const; std::list GetNonConnectedPeers () const; + std::tuple GetNextBlockToRequest (std::shared_ptr conn); // return (index, offest, len) uint64_t GetNextTrackerRequestTime () const { return m_NextTrackerRequestTime; } void SetNextTrackerRequestTime (uint64_t ts) { m_NextTrackerRequestTime = ts; } @@ -145,8 +156,6 @@ namespace torrents void ReceiveHandshake (); void CheckKeepAlive (uint64_t ts); - void RequestPiece (uint32_t index); - private: void Terminate (); @@ -169,6 +178,9 @@ namespace torrents void HandlePieceMsg (const uint8_t * buf, size_t len); void SendPieceMsg (uint32_t index, uint32_t offset, const uint8_t * data, size_t len); void HandleRequestMsg (const uint8_t * buf, size_t len); + void SendRequestMsg (uint32_t index, uint32_t offset, uint32_t len); + + void RequestNextBlock (); private: