dandri/i2pd
1
0
Fork 0
mirror of https://github.com/PurpleI2P/i2pd.git synced 2026-08-14 07:11:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

request next block

Browse Source
This commit is contained in:
orignal
2026-08-02 21:43:14 -04:00
parent ee50116dbc
commit 1a4e820379
2 changed files with 82 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libi2pd_client/Torrents.cpp
+66 -13
View File
@@ -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<std::vector<BlockStatus> >(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<size_t, size_t> 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<PeerConnection> connection)
{
m_Connections.emplace_back (connection);
@@ -370,6 +396,25 @@ namespace torrents
return ret;
}
std::tuple<uint32_t, uint32_t, uint32_t> Torrent::GetNextBlockToRequest (std::shared_ptr<PeerConnection> 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<i2p::stream::Stream> 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<uint8_t> 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<i2p::client::ClientDestination> localDestination,
std::string_view torrentsDir, std::string_view trackers):
i2p::client::I2PService (localDestination), m_TorrentsDir (torrentsDir),
libi2pd_client/Torrents.h
+16 -4
View File
@@ -22,6 +22,7 @@
#include <string>
#include <string_view>
#include <random>
#include <tuple>
#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<size_t, size_t> GetAvailableBuffer (size_t offset, size_t len) const; // return (offset, len) of available data
std::pair<size_t, size_t> GetNextBlockToRequest (); // return (offset, len) of next buffer, len = 0 if no next buffer
void AddConnection (std::shared_ptr<PeerConnection> connection);
void RemoveConnection (std::shared_ptr<PeerConnection> 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<std::vector<BlockStatus> > m_Blocks;
std::list<std::weak_ptr<PeerConnection> > m_Connections; // for incomplete pieces only
};
@@ -110,6 +120,7 @@ namespace torrents
Piece& GetPiece (int index) { return m_Pieces[index]; }
std::vector<uint8_t> CreateBitfield () const;
std::list<i2p::data::IdentHash> GetNonConnectedPeers () const;
std::tuple<uint32_t, uint32_t, uint32_t> GetNextBlockToRequest (std::shared_ptr<PeerConnection> 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: