mirror of
https://github.com/PurpleI2P/i2pd.git
synced 2026-10-20 14:32:47 +00:00
1253 lines
38 KiB
C++
1253 lines
38 KiB
C++
/*
|
|
* Copyright (c) 2026, The PurpleI2P Project
|
|
*
|
|
* This file is part of Purple i2pd project and licensed under BSD3
|
|
*
|
|
* See full license text in LICENSE file at top of project tree
|
|
*/
|
|
|
|
#ifndef NO_TORRENTS
|
|
|
|
#include <openssl/rand.h>
|
|
#include <string.h>
|
|
#include <vector>
|
|
#include <fstream>
|
|
#include "I2PEndian.h"
|
|
#include "Timestamp.h"
|
|
#include "FS.h"
|
|
#include "TorrentsTunnel.h"
|
|
#include "TorrentsDHT.h"
|
|
|
|
namespace i2p
|
|
{
|
|
namespace torrents
|
|
{
|
|
Node::Node (const NodeInfo& nodeInfo):
|
|
lastUpdateTime (i2p::util::GetMonotonicSeconds ())
|
|
{
|
|
memcpy (id.data (), nodeInfo.data (), id.size ());
|
|
memcpy ((uint8_t *)peer, nodeInfo.data () + id.size (), peer.len);
|
|
port = bufbe16toh (nodeInfo.data () + nodeInfo.size () - 2);
|
|
}
|
|
|
|
NodeInfo Node::GetNodeInfo () const
|
|
{
|
|
NodeInfo nodeInfo;
|
|
memcpy (nodeInfo.data (), id.data (), id.size ());
|
|
memcpy (nodeInfo.data () + id.size (), peer, peer.len);
|
|
htobe16buf (nodeInfo.data () + nodeInfo.size () - 2, port);
|
|
return nodeInfo;
|
|
}
|
|
|
|
bool Node::VerifyID () const
|
|
{
|
|
return !memcmp (id.data (), peer, 4) && (((uint16_t)(id[4] ^ peer[4]) << 8) | (id[5] ^ peer[5])) == port;
|
|
}
|
|
|
|
std::optional<NodeID> Bucket::GetMiddleID () const
|
|
{
|
|
uint8_t bit = std::max (start.FindLowestBit (), next ? next->start.FindLowestBit () : -1) + 1;
|
|
if (bit >= NodeID::len * 8) return {};
|
|
NodeID middleID = start;
|
|
middleID[bit >> 3] |= (0x80 >> (bit & 0x07));
|
|
return { middleID };
|
|
}
|
|
|
|
NodeID Bucket::GetRandomID (std::mt19937& rng) const
|
|
{
|
|
uint8_t bit = std::max (start.FindLowestBit (), next ? next->start.FindLowestBit () : -1) + 1;
|
|
if (bit >= NodeID::len * 8) return start;
|
|
|
|
NodeID randomID;
|
|
auto d = div (bit, 8);
|
|
memcpy (randomID.data (), start.data (), d.quot);
|
|
randomID[d.quot] = (start[d.quot] & (0xFF00 >> d.rem)) | (rng () & (0xFF >> d.rem));
|
|
for (size_t i = d.quot + 1; i < NodeID::len; i += 4)
|
|
{
|
|
uint32_t r = rng ();
|
|
if (i + 4 <= NodeID::len)
|
|
memcpy (randomID.data () + i, &r, 4);
|
|
else
|
|
memcpy (randomID.data () + i, &r, NodeID::len - i);
|
|
}
|
|
return randomID;
|
|
}
|
|
|
|
bool Bucket::Split ()
|
|
{
|
|
auto middleID = GetMiddleID ();
|
|
if (!middleID) return false;
|
|
auto newBucket = new Bucket (*middleID);
|
|
newBucket->next = next;
|
|
next = newBucket;
|
|
// move some nodes
|
|
auto it = nodes.begin ();
|
|
while (it != nodes.end ())
|
|
{
|
|
if (it->first < *middleID)
|
|
it++; // stay in old bucket
|
|
else
|
|
// move to new bucket
|
|
newBucket->nodes.insert (nodes.extract (it++));
|
|
}
|
|
return true;
|
|
}
|
|
|
|
RoutingTable::RoutingTable (const NodeID& ourNode):
|
|
m_OurNode (ourNode)
|
|
{
|
|
m_Buckets = new Bucket;
|
|
}
|
|
|
|
RoutingTable::~RoutingTable ()
|
|
{
|
|
CleanUp ();
|
|
delete m_Buckets;
|
|
}
|
|
|
|
void RoutingTable::CleanUp ()
|
|
{
|
|
if (!m_Buckets) return;
|
|
auto bucket = m_Buckets->next;
|
|
while (bucket)
|
|
{
|
|
auto tmp = bucket;
|
|
bucket = bucket->next;
|
|
delete tmp;
|
|
}
|
|
m_Buckets->next = nullptr;
|
|
m_Buckets->nodes.clear ();
|
|
}
|
|
|
|
size_t RoutingTable::GetNumBuckets () const
|
|
{
|
|
size_t num = 0;
|
|
auto bucket = m_Buckets;
|
|
while (bucket)
|
|
{
|
|
num++;
|
|
bucket = bucket->next;
|
|
}
|
|
return num;
|
|
}
|
|
|
|
size_t RoutingTable::GetNumNodes () const
|
|
{
|
|
size_t num = 0;
|
|
auto bucket = m_Buckets;
|
|
while (bucket)
|
|
{
|
|
num += bucket->nodes.size ();
|
|
bucket = bucket->next;
|
|
}
|
|
return num;
|
|
}
|
|
|
|
std::list<std::shared_ptr<Node> > RoutingTable::GetNodes ()
|
|
{
|
|
std::list<std::shared_ptr<Node> > nodes;
|
|
auto bucket = m_Buckets;
|
|
while (bucket)
|
|
{
|
|
for (auto it: bucket->nodes)
|
|
nodes.emplace_back (it.second);
|
|
bucket = bucket->next;
|
|
}
|
|
return nodes;
|
|
}
|
|
|
|
Bucket * RoutingTable::FindBucket (const Torrent::InfoHash& id) const
|
|
{
|
|
if (!m_Buckets) return nullptr;
|
|
auto bucket = m_Buckets;
|
|
while (bucket->next)
|
|
{
|
|
if (id < bucket->next->start)
|
|
break;
|
|
bucket = bucket->next;
|
|
}
|
|
return bucket;
|
|
}
|
|
|
|
size_t RoutingTable::DeleteExpiredNodes (uint64_t ts)
|
|
{
|
|
size_t numDeleted = 0;
|
|
auto bucket = m_Buckets;
|
|
while (bucket)
|
|
{
|
|
if (ts > bucket->lastUpdateTime + DHT_BUCKET_EXPIRATION_THRESHOLD)
|
|
{
|
|
auto it = bucket->nodes.begin ();
|
|
while (it != bucket->nodes.end ())
|
|
{
|
|
if (ts > it->second->lastUpdateTime + DHT_NODE_EXPIRATION_TIME)
|
|
{
|
|
numDeleted++;
|
|
it = bucket->nodes.erase (it);
|
|
}
|
|
else
|
|
it++;
|
|
}
|
|
}
|
|
bucket = bucket->next;
|
|
}
|
|
return numDeleted;
|
|
}
|
|
|
|
std::list<std::shared_ptr<Node> > RoutingTable::GetNodesToPing (uint64_t ts)
|
|
{
|
|
std::list<std::shared_ptr<Node> > toPing;
|
|
auto bucket = m_Buckets;
|
|
while (bucket)
|
|
{
|
|
if (ts > bucket->lastUpdateTime + DHT_BUCKET_EXPIRATION_THRESHOLD)
|
|
{
|
|
for (const auto& it: bucket->nodes)
|
|
if (ts > it.second->lastUpdateTime + DHT_NODE_SEND_PING_TIME)
|
|
toPing.push_back (it.second);
|
|
}
|
|
bucket = bucket->next;
|
|
}
|
|
return toPing;
|
|
}
|
|
|
|
std::shared_ptr<Node> RoutingTable::AddNode (std::shared_ptr<Node> node)
|
|
{
|
|
if (!node) return nullptr;
|
|
if (node->id == m_OurNode) return nullptr;
|
|
auto bucket = FindBucket (node->id);
|
|
if (!bucket) return nullptr;
|
|
auto it = bucket->nodes.find (node->id);
|
|
if (it != bucket->nodes.end ())
|
|
{
|
|
auto ts = i2p::util::GetMonotonicSeconds ();
|
|
it->second->lastUpdateTime = ts;
|
|
bucket->lastUpdateTime = ts;
|
|
return it->second;
|
|
}
|
|
if (bucket->IsFull ())
|
|
{
|
|
do
|
|
{
|
|
if (!bucket->IsInBucket (m_OurNode)) return nullptr;
|
|
if (!bucket->Split ()) return nullptr;
|
|
bucket = FindBucket (node->id);
|
|
}
|
|
while (bucket->IsFull ());
|
|
}
|
|
if (bucket)
|
|
{
|
|
bucket->nodes.emplace (node->id, node);
|
|
bucket->lastUpdateTime = i2p::util::GetMonotonicSeconds ();
|
|
}
|
|
return node;
|
|
}
|
|
|
|
void RoutingTable::RemoveNode (const NodeID& id)
|
|
{
|
|
auto bucket = FindBucket (id);
|
|
if (!bucket) return;
|
|
bucket->nodes.erase (id);
|
|
}
|
|
|
|
std::list<std::pair<NodeID, Bucket * > > RoutingTable::GetExploratoryTargets (std::mt19937& rng) const
|
|
{
|
|
std::list<std::pair<NodeID, Bucket * > > ret;
|
|
auto bucket = m_Buckets;
|
|
while (bucket)
|
|
{
|
|
if (!bucket->IsFull () || bucket->IsInBucket (m_OurNode))
|
|
ret.emplace_back (std::make_pair (bucket->GetRandomID (rng), bucket));
|
|
bucket = bucket->next;
|
|
}
|
|
return ret;
|
|
}
|
|
|
|
std::string DHTTorrent::GetBEncodedPeers () const
|
|
{
|
|
std::vector<std::string> peers;
|
|
for (const auto& [peer, ts]: m_Peers)
|
|
peers.emplace_back (CreateByteString (std::string_view ((const char *)peer.data (), peer.len)));
|
|
return CreateList (peers);
|
|
}
|
|
|
|
bool DHTTorrent::AddPeer (const i2p::data::IdentHash& peer)
|
|
{
|
|
auto ts = i2p::util::GetMonotonicSeconds ();
|
|
auto [it, inserted] = m_Peers.emplace (peer, ts);
|
|
if (!inserted)
|
|
it->second = ts;
|
|
return inserted;
|
|
}
|
|
|
|
bool DHTTorrent::CleanUp (uint64_t ts)
|
|
{
|
|
auto it = m_Peers.begin ();
|
|
while (it != m_Peers.end ())
|
|
{
|
|
if (ts > it->second + DHT_TORRENT_PEER_EXPIRATION_TIME)
|
|
it = m_Peers.erase (it);
|
|
else
|
|
it++;
|
|
}
|
|
return m_Peers.empty ();
|
|
}
|
|
|
|
bool RequestInfo::AddNode (std::shared_ptr<Node> node)
|
|
{
|
|
if (!node || tried.contains (node->id)) return false;
|
|
return nodesToRequest.emplace (torrent ? node->id ^ torrent->GetInfoHash () : node->id ^ target, node).second;
|
|
}
|
|
|
|
bool RequestInfo::AddNodeToken (std::shared_ptr<Node> node, uint64_t token)
|
|
{
|
|
if (!node || !token || !torrent) return false;
|
|
return tokens.emplace (node->id ^ torrent->GetInfoHash (), node, token).second;
|
|
}
|
|
|
|
std::shared_ptr<Node> RequestInfo::GetNextNode ()
|
|
{
|
|
if (nodesToRequest.empty ()) return nullptr;
|
|
auto node = nodesToRequest.begin ()->second;
|
|
nodesToRequest.erase (nodesToRequest.begin ());
|
|
numAttempts++;
|
|
tried.emplace (node->id);
|
|
lastNode = node;
|
|
return node;
|
|
}
|
|
|
|
TorrentsDHT::TorrentsDHT (TorrentsTunnel& tunnel, uint16_t port, std::string_view dhtFileName):
|
|
m_Tunnel (tunnel), m_Port (port), m_DHTFileName (dhtFileName),
|
|
m_DHTUpdateCheckTimer (tunnel.GetService ()), m_DHTExpirationCheckTimer (tunnel.GetService ()),
|
|
m_DHTSendPingCheckTimer (tunnel.GetService ()), m_DHTQueryExpirationCheckTimer (tunnel.GetService ()),
|
|
m_NextDHTExploratoryTime (i2p::util::GetMonotonicSeconds () + DHT_INITIAL_EXPLORATORY_INTERVAL)
|
|
{
|
|
auto dest = tunnel.GetLocalDestination ();
|
|
if (dest)
|
|
{
|
|
memcpy (m_NodeID.data (), dest->GetIdentHash (), 6);
|
|
m_NodeID[4] ^= (port >> 8);
|
|
m_NodeID[5] ^= (port & 0xFF);
|
|
RAND_bytes (m_NodeID.data () + 6, m_NodeID.size () - 6);
|
|
m_RoutingTable = std::make_unique<RoutingTable> (m_NodeID);
|
|
}
|
|
else
|
|
m_NodeID.fill (0);
|
|
}
|
|
|
|
void TorrentsDHT::Start ()
|
|
{
|
|
auto dest = m_Tunnel.GetLocalDestination ();
|
|
if (dest)
|
|
{
|
|
auto dgramDest = dest->GetDatagramDestination ();
|
|
if (dgramDest)
|
|
dgramDest->SetReceiver (std::bind_front (&TorrentsDHT::HandleDatagram, this));
|
|
if (m_DHTFileName.empty ())
|
|
m_DHTFileName = dest->GetIdentHash ().ToBase32 () + ".dht";
|
|
}
|
|
if (m_DHTFileName.empty ()) m_DHTFileName = "nodest.dht";
|
|
Load (GetDHTFilePath (m_DHTFileName));
|
|
ScheduleDHTUpdateCheck ();
|
|
ScheduleDHTExpirationCheck ();
|
|
ScheduleDHTSendPingCheck ();
|
|
ScheduleDHTQueryExpirationCheck ();
|
|
}
|
|
|
|
void TorrentsDHT::Stop ()
|
|
{
|
|
m_DHTUpdateCheckTimer.cancel ();
|
|
m_DHTExpirationCheckTimer.cancel ();
|
|
m_DHTSendPingCheckTimer.cancel ();
|
|
m_DHTQueryExpirationCheckTimer.cancel ();
|
|
Save (GetDHTFilePath (m_DHTFileName));
|
|
}
|
|
|
|
std::filesystem::path TorrentsDHT::GetDHTFilePath (std::string_view filename) const
|
|
{
|
|
std::filesystem::path dhtFilePath (i2p::fs::GetDataDir()); dhtFilePath /= "torrents";
|
|
if (!std::filesystem::exists (dhtFilePath))
|
|
std::filesystem::create_directories (dhtFilePath);
|
|
dhtFilePath /= filename;
|
|
return dhtFilePath;
|
|
}
|
|
|
|
void TorrentsDHT::Save (const std::filesystem::path& file)
|
|
{
|
|
if (m_RoutingTable)
|
|
{
|
|
std::ofstream f(file, std::ofstream::binary);
|
|
if (f.is_open ())
|
|
{
|
|
auto dest = m_Tunnel.GetLocalDestination ();
|
|
if (dest)
|
|
{
|
|
// save our nodeInfo first
|
|
NodeInfo nodeInfo = Node (m_NodeID, dest->GetIdentHash (), m_Port).GetNodeInfo ();
|
|
f.write ((const char *)nodeInfo.data (), nodeInfo.size ());
|
|
}
|
|
int numSaved = 0;
|
|
for (auto it: m_RoutingTable->GetNodes ())
|
|
{
|
|
auto nodeInfo = it->GetNodeInfo ();
|
|
if (f.write ((const char *)nodeInfo.data (), nodeInfo.size ()))
|
|
numSaved++;
|
|
}
|
|
if (numSaved > 0)
|
|
LogPrint (eLogInfo, "TorrentsDHT: ", numSaved, " DHT nodes saved");
|
|
}
|
|
}
|
|
}
|
|
|
|
void TorrentsDHT::Load (const std::filesystem::path& file)
|
|
{
|
|
std::ifstream f (file, std::ifstream::in | std::ifstream::binary);
|
|
if (f.is_open ())
|
|
{
|
|
NodeInfo nodeInfo;
|
|
if (f.read ((char *)nodeInfo.data (), nodeInfo.size ()))
|
|
{
|
|
auto dest = m_Tunnel.GetLocalDestination ();
|
|
if (dest)
|
|
{
|
|
Node node (nodeInfo);
|
|
if (Node (nodeInfo).peer == dest->GetIdentHash ()) // first nodeInfo is ours
|
|
{
|
|
m_NodeID = node.id;
|
|
m_Port = node.port;
|
|
}
|
|
else
|
|
f.seekg (0, std::ios::beg);
|
|
}
|
|
}
|
|
else
|
|
return;
|
|
|
|
m_RoutingTable = std::make_unique<RoutingTable> (m_NodeID);
|
|
while (f.read ((char *)nodeInfo.data (), nodeInfo.size ()))
|
|
{
|
|
auto bytesRead = f.gcount();
|
|
if (bytesRead == nodeInfo.size ())
|
|
{
|
|
auto node = std::make_shared<Node>(nodeInfo);
|
|
if (node->VerifyID ())
|
|
m_RoutingTable->AddNode (node);
|
|
}
|
|
}
|
|
LogPrint (eLogInfo, "TorrentsDHT: ", m_RoutingTable->GetNumNodes (),
|
|
" DHT nodes loaded to ", m_RoutingTable->GetNumBuckets (), " buckets");
|
|
}
|
|
}
|
|
|
|
void TorrentsDHT::HandleRawDatagram (const uint8_t * buf, size_t len)
|
|
{
|
|
// response or error
|
|
char type = 0; uint64_t token = 0;
|
|
bool isMalformed = false;
|
|
NodeID id; Torrent::InfoHash infoHash;
|
|
std::string_view nodes;
|
|
std::string transactionID, query;
|
|
std::vector<std::string_view> values{""};
|
|
ParseDictionary (std::string_view ((const char *)buf, len),
|
|
[&type, &transactionID, &id, &values, &token, &infoHash, &query, &isMalformed, &nodes]
|
|
(std::string_view key, std::string_view buf)->size_t
|
|
{
|
|
if (key == "y")
|
|
{
|
|
auto [value, l] = ExtractByteString (buf);
|
|
if (l && !value.empty ()) type = value[0];
|
|
return l;
|
|
}
|
|
else if (key == "t")
|
|
{
|
|
auto [value, l] = ExtractByteString (buf);
|
|
if (l) transactionID = value;
|
|
return l;
|
|
}
|
|
else if (key == "q")
|
|
{
|
|
auto [value, l] = ExtractByteString (buf);
|
|
if (l) query = value;
|
|
return l;
|
|
}
|
|
else if (key == "r" || key == "a")
|
|
{
|
|
return ParseDictionary (buf,
|
|
[&id, &values, &token, &infoHash, &isMalformed, &nodes](std::string_view key, std::string_view buf)->size_t
|
|
{
|
|
if (key == "id")
|
|
{
|
|
auto [l, success] = ParseByteArray (buf, id);
|
|
if (!success) isMalformed = true;
|
|
return l;
|
|
}
|
|
else if (key == "values")
|
|
{
|
|
auto [v, l] = ParseStringList (buf);
|
|
if (l) values = v;
|
|
return l;
|
|
}
|
|
else if (key == "nodes")
|
|
{
|
|
auto [value, l] = ExtractByteString (buf);
|
|
if (l) nodes = value;
|
|
return l;
|
|
}
|
|
else if (key == "token")
|
|
{
|
|
auto [value, l] = ExtractByteString (buf);
|
|
if (l && value.size () >= 8)
|
|
memcpy (&token, value.data (), 8);
|
|
return l;
|
|
}
|
|
else if (key == "info_hash")
|
|
{
|
|
auto [l, success] = ParseByteArray (buf, infoHash);
|
|
if (!success) isMalformed = true;
|
|
return l;
|
|
}
|
|
return 0;
|
|
});
|
|
}
|
|
return 0;
|
|
});
|
|
if (isMalformed)
|
|
{
|
|
LogPrint (eLogInfo, "TorrentsDHT: Malformed raw datagram received");
|
|
return;
|
|
}
|
|
if (type)
|
|
{
|
|
UpdateHeardFrom (id);
|
|
switch (type)
|
|
{
|
|
case 'r':
|
|
HandleResponse (transactionID, id, token, values, nodes);
|
|
break;
|
|
case 'e':
|
|
LogPrint (eLogDebug, "TorrentsDHT: Error msg received");
|
|
break;
|
|
case 'q':
|
|
if (query == "announce_peer")
|
|
HandleAnnouncePeer (transactionID, id, infoHash, token);
|
|
else
|
|
LogPrint (eLogError, "TorrentsDHT: Query can't come as raw datagram");
|
|
break;
|
|
default:
|
|
LogPrint (eLogInfo, "TorrentsDHT: Unxpected msg type ", (int)type);
|
|
}
|
|
}
|
|
}
|
|
|
|
void TorrentsDHT::HandleDatagram (const i2p::data::IdentityEx& from, uint16_t fromPort, uint16_t toPort,
|
|
const uint8_t * buf, size_t len, const i2p::util::Mapping * options)
|
|
{
|
|
// query
|
|
char type = 0;
|
|
bool isMalformed = false;
|
|
std::string transactionID, query;
|
|
NodeID id, target; Torrent::InfoHash infoHash;
|
|
ParseDictionary (std::string_view ((const char *)buf, len),
|
|
[&type, &transactionID, &query, &id, &infoHash, &isMalformed, &target]
|
|
(std::string_view key, std::string_view buf)->size_t
|
|
{
|
|
if (key == "y")
|
|
{
|
|
auto [value, l] = ExtractByteString (buf);
|
|
if (l && !value.empty ()) type = value[0];
|
|
return l;
|
|
}
|
|
else if (key == "t")
|
|
{
|
|
auto [value, l] = ExtractByteString (buf);
|
|
if (l) transactionID = value;
|
|
return l;
|
|
}
|
|
else if (key == "q")
|
|
{
|
|
auto [value, l] = ExtractByteString (buf);
|
|
if (l) query = value;
|
|
return l;
|
|
}
|
|
else if (key == "a")
|
|
{
|
|
return ParseDictionary (buf,
|
|
[&id, &infoHash, &isMalformed, &target](std::string_view key, std::string_view buf)->size_t
|
|
{
|
|
if (key == "id")
|
|
{
|
|
auto [l, success] = ParseByteArray (buf, id);
|
|
if (!success) isMalformed = true;
|
|
return l;
|
|
}
|
|
else if (key == "info_hash")
|
|
{
|
|
auto [l, success] = ParseByteArray (buf, infoHash);
|
|
if (!success) isMalformed = true;
|
|
return l;
|
|
}
|
|
else if (key == "target")
|
|
{
|
|
auto [l, success] = ParseByteArray (buf, target);
|
|
if (!success) isMalformed = true;
|
|
return l;
|
|
}
|
|
return 0;
|
|
});
|
|
}
|
|
return 0;
|
|
});
|
|
if (isMalformed)
|
|
{
|
|
LogPrint (eLogInfo, "TorrentsDHT: Malformed datagram received");
|
|
return;
|
|
}
|
|
if (type == 'q')
|
|
{
|
|
UpdateHeardFrom (id);
|
|
auto node = UpdateNode (std::make_shared<Node> (id, from.GetIdentHash (), fromPort));
|
|
if (!node) return;
|
|
if (query == "ping")
|
|
HandlePingQuery (from.GetIdentHash (), fromPort, transactionID, id);
|
|
else if (query == "get_peers")
|
|
HandleGetPeersQuery (transactionID, node, infoHash);
|
|
else if (query == "find_node")
|
|
HandleFindNodeQuery (from.GetIdentHash (), fromPort, transactionID, target);
|
|
else
|
|
LogPrint (eLogDebug, "TorrentsDHT: Unexpected query ", query);
|
|
}
|
|
else if (type)
|
|
LogPrint (eLogInfo, "TorrentsDHT: Unxpected msg type ", (int)type);
|
|
}
|
|
|
|
void TorrentsDHT::HandlePingQuery (const i2p::data::IdentHash& fromIdent, uint16_t fromPort,
|
|
std::string_view transactionID, const NodeID& nodeID)
|
|
{
|
|
LogPrint (eLogDebug, "TorrentsDHT: Ping query msg received from ", nodeID.ToBase64 ());
|
|
SendPingResponse (transactionID, fromIdent, fromPort + 1); // to rport
|
|
}
|
|
|
|
void TorrentsDHT::HandleGetPeersQuery (std::string_view transactionID, std::shared_ptr<Node> from, const Torrent::InfoHash& infoHash)
|
|
{
|
|
LogPrint (eLogDebug, "TorrentsDHT: Get peers query msg received from ", from->id.ToBase64 ());
|
|
|
|
uint64_t token = m_Tunnel.GetLocalDestination () ? m_Tunnel.GetLocalDestination ()->GetRng ()() : 1;
|
|
m_IncomingTokens.emplace (token, std::make_pair(from, i2p::util::GetMonotonicSeconds ()));
|
|
|
|
auto it = m_Torrents.find (infoHash);
|
|
if (it != m_Torrents.end () && it->second->HasPeers ())
|
|
SendGetPeersResponse (transactionID, it->second, token, from->peer, from->port + 1); // to rport
|
|
else if (m_RoutingTable)
|
|
{
|
|
std::vector<uint8_t> nodes;
|
|
auto bucket = m_RoutingTable->FindBucket (infoHash);
|
|
if (bucket && bucket->IsEmpty () && bucket->next && !bucket->next->IsEmpty ())
|
|
bucket = bucket->next;
|
|
|
|
if (bucket && !bucket->IsEmpty ())
|
|
for (auto it: bucket->nodes)
|
|
{
|
|
auto nodeInfo = it.second->GetNodeInfo ();
|
|
nodes.insert (nodes.end(), nodeInfo.data (), nodeInfo.data () + nodeInfo.size ());
|
|
}
|
|
|
|
SendGetPeersResponse (transactionID, std::string_view ((const char *)nodes.data (), nodes.size ()),
|
|
token, from->peer, from->port + 1); // to rport
|
|
}
|
|
}
|
|
|
|
void TorrentsDHT::HandleFindNodeQuery (const i2p::data::IdentHash& fromIdent, uint16_t fromPort,
|
|
std::string_view transactionID, const NodeID& target)
|
|
{
|
|
LogPrint (eLogDebug, "TorrentsDHT: Find node query received");
|
|
if (m_RoutingTable)
|
|
{
|
|
auto bucket = m_RoutingTable->FindBucket (target);
|
|
if (bucket && bucket->IsEmpty () && bucket->next && !bucket->next->IsEmpty ())
|
|
bucket = bucket->next;
|
|
if (bucket && !bucket->IsEmpty ())
|
|
{
|
|
std::vector<uint8_t> nodes;
|
|
for (auto it: bucket->nodes)
|
|
{
|
|
auto nodeInfo = it.second->GetNodeInfo ();
|
|
nodes.insert (nodes.end(), nodeInfo.data (), nodeInfo.data () + nodeInfo.size ());
|
|
}
|
|
SendFindNodeResponse (transactionID, std::string_view ((const char *)nodes.data (), nodes.size ()), fromIdent, fromPort + 1); // to rport
|
|
}
|
|
}
|
|
}
|
|
|
|
void TorrentsDHT::HandleAnnouncePeer (std::string_view transactionID, const NodeID& nodeID,
|
|
const Torrent::InfoHash& infoHash, uint64_t token)
|
|
{
|
|
LogPrint (eLogDebug, "TorrentsDHT: Announce peer received from ", nodeID.ToBase64 ());
|
|
auto it = m_IncomingTokens.find (token);
|
|
if (it != m_IncomingTokens.end ())
|
|
{
|
|
auto node = it->second.first;
|
|
auto torrent = m_Torrents.emplace (infoHash, std::make_shared<DHTTorrent>()).first->second;
|
|
torrent->AddPeer (node->peer);
|
|
SendResponseMsg (CreateDictionary ({
|
|
{ "id", CreateByteString (std::string_view ((const char *)m_NodeID.data (), m_NodeID.size ())) },
|
|
}),
|
|
transactionID, node->peer, node->port + 1); // to rport
|
|
}
|
|
else
|
|
LogPrint (eLogInfo, "TorrentsDHT: Token for announce not found");
|
|
}
|
|
|
|
void TorrentsDHT::HandleResponse (std::string_view transactionID, const NodeID& nodeID,
|
|
uint64_t token, const std::vector<std::string_view>& values, std::string_view nodes)
|
|
{
|
|
uint16_t t = 0;
|
|
if (transactionID.size () >= 2)
|
|
memcpy (&t, transactionID.data (), 2);
|
|
auto it = m_Queries.find (t);
|
|
if (t && it != m_Queries.end ())
|
|
{
|
|
if (m_RoutingTable)
|
|
{
|
|
const auto& [ident, port, query, info, time] = it->second;
|
|
if (query != eKRPCQueryAnnouncePeer) // port is rport there
|
|
UpdateNode (std::make_shared<Node> (nodeID, ident, port));
|
|
switch (query)
|
|
{
|
|
case eKRPCQueryPing:
|
|
{
|
|
LogPrint (eLogDebug, "TorrentsDHT: Ping response received");
|
|
break;
|
|
}
|
|
case eKRPCQueryGetPeers:
|
|
{
|
|
LogPrint (eLogDebug, "TorrentsDHT: get_peers response received from peer ", ident.ToBase64 (), "after attempt #", info->numAttempts);
|
|
if (!values.empty () && values[0].empty ()) // nodes, because values not set
|
|
{
|
|
if (info && token)
|
|
info->AddNodeToken (std::make_shared<Node>(nodeID, ident, port), token);
|
|
HandleGetPeersResponseNodes (info, nodes);
|
|
}
|
|
else //values
|
|
HandleGetPeersResponsePeersAndAnnounce (info, values, token, ident, port + 1); // to rport
|
|
break;
|
|
}
|
|
case eKRPCQueryFindNode:
|
|
{
|
|
LogPrint (eLogDebug, "TorrentsDHT: find_node response received after attempt #", info->numAttempts);
|
|
HandleFindNodeResponse (info, nodes);
|
|
break;
|
|
}
|
|
case eKRPCQueryAnnouncePeer:
|
|
LogPrint (eLogDebug, "TorrentsDHT: Announce peer response received from ", ident.ToBase64 ());
|
|
HandleAnnouncePeerResponse (info);
|
|
break;
|
|
default:
|
|
LogPrint (eLogInfo, "TorrentsDHT: Response to unknown KRPC query ", (int)query);
|
|
}
|
|
}
|
|
m_Queries.erase (it);
|
|
}
|
|
else
|
|
LogPrint (eLogInfo, "TorrentsDHT: Query not found");
|
|
}
|
|
|
|
void TorrentsDHT::HandleGetPeersResponseNodes (std::shared_ptr<RequestInfo> info, std::string_view nodes)
|
|
{
|
|
NodeInfo nodeInfo;
|
|
while (nodes.size () >= nodeInfo.size ())
|
|
{
|
|
memcpy (nodeInfo.data (), nodes.data (), nodeInfo.size ());
|
|
auto node = std::make_shared<Node>(nodeInfo);
|
|
if (info && node->id != m_NodeID && node->VerifyID () && info->AddNode (node))
|
|
{
|
|
if (m_HeardFrom.contains (node->id))
|
|
UpdateNode (node);
|
|
else if (!m_FailedPeers.contains (node->peer))
|
|
SendPingQuery (node->peer, node->port);
|
|
}
|
|
nodes = nodes.substr (nodeInfo.size ());
|
|
}
|
|
if (info)
|
|
SendNextGetPeersQuery (info);
|
|
}
|
|
|
|
void TorrentsDHT::HandleGetPeersResponsePeersAndAnnounce (std::shared_ptr<RequestInfo> info,
|
|
const std::vector<std::string_view>& peers, uint64_t token, const i2p::data::IdentHash& toIdent, uint16_t toPort)
|
|
{
|
|
if (!info->torrent->IsComplete ())
|
|
{
|
|
std::unordered_set<i2p::data::IdentHash> newPeers;
|
|
for (auto it: peers)
|
|
if (it.size () == i2p::data::IdentHash::len)
|
|
newPeers.emplace ((const uint8_t *)it.data ());
|
|
if (!newPeers.empty ())
|
|
{
|
|
LogPrint (eLogDebug, "TorrentsDHT: ", newPeers.size (), " new peers received");
|
|
m_Tunnel.ConnectToNewPeers (info->torrent, newPeers, ePeerConnectionOriginDHT);
|
|
}
|
|
}
|
|
LogPrint (eLogDebug, "TorrentsDHT: Send announce to ", toIdent.ToBase64 ());
|
|
SendAnnouncePeerQuery (info, token, toIdent, toPort);
|
|
}
|
|
|
|
void TorrentsDHT::HandleFindNodeResponse (std::shared_ptr<RequestInfo> info, std::string_view nodes)
|
|
{
|
|
NodeInfo nodeInfo;
|
|
while (nodes.size () >= nodeInfo.size ())
|
|
{
|
|
memcpy (nodeInfo.data (), nodes.data (), nodeInfo.size ());
|
|
auto node = std::make_shared<Node>(nodeInfo);
|
|
if (info && node->id != m_NodeID && node->VerifyID () && info->AddNode (node))
|
|
{
|
|
if (m_HeardFrom.contains (node->id))
|
|
UpdateNode (node);
|
|
else if (!m_FailedPeers.contains (node->peer))
|
|
SendPingQuery (node->peer, node->port);
|
|
}
|
|
nodes = nodes.substr (nodeInfo.size ());
|
|
}
|
|
if (info)
|
|
SendNextFindNodeQuery (info);
|
|
}
|
|
|
|
void TorrentsDHT::HandleAnnouncePeerResponse (std::shared_ptr<RequestInfo> info)
|
|
{
|
|
if (info && info->torrent)
|
|
info->torrent->SetLastDHTAnnounceTime (i2p::util::GetMonotonicSeconds ());
|
|
else
|
|
LogPrint (eLogError, "TorrentsDHT: Announced torrent not found");
|
|
}
|
|
|
|
void TorrentsDHT::SendDatagram (std::string_view msg, const i2p::data::IdentHash& toIdent, uint16_t toPort)
|
|
{
|
|
auto dest = m_Tunnel.GetLocalDestination ();
|
|
if (dest)
|
|
{
|
|
if (toIdent != dest->GetIdentHash ())
|
|
{
|
|
auto dgramDest = dest->GetDatagramDestination ();
|
|
if (dgramDest)
|
|
dgramDest->SendDatagramTo ((const uint8_t *)msg.data (), msg.size (), toIdent, m_Port, toPort);
|
|
}
|
|
else
|
|
LogPrint (eLogWarning, "TorrentsDHT: Can't send datagram to self");
|
|
}
|
|
}
|
|
|
|
void TorrentsDHT::SendRawDatagram (std::string_view msg, const i2p::data::IdentHash& toIdent, uint16_t toPort)
|
|
{
|
|
auto dest = m_Tunnel.GetLocalDestination ();
|
|
if (dest)
|
|
{
|
|
if (toIdent != dest->GetIdentHash ())
|
|
{
|
|
auto dgramDest = dest->GetDatagramDestination ();
|
|
if (dgramDest)
|
|
dgramDest->SendRawDatagramTo ((const uint8_t *)msg.data (), msg.size (), toIdent, m_Port, toPort);
|
|
}
|
|
else
|
|
LogPrint (eLogWarning, "TorrentsDHT: Can't send raw datagram to self");
|
|
}
|
|
}
|
|
|
|
void TorrentsDHT::SendQueryMsg (KRPCQuery query, std::string_view arguments,
|
|
const i2p::data::IdentHash& toIdent, uint16_t toPort, bool isRaw, std::shared_ptr<RequestInfo> info)
|
|
{
|
|
uint16_t transactionID = m_Tunnel.GetLocalDestination () ? static_cast<uint16_t>(m_Tunnel.GetLocalDestination ()->GetRng ()()) : 1;
|
|
auto msg = CreateDictionary ({
|
|
{ "a", arguments },
|
|
{ "q", CreateByteString (KRPCQueryStr[query]) },
|
|
{ "t", CreateByteString (std::string_view ((const char *)&transactionID, 2)) },
|
|
{ "y", CreateByteString ("q") }
|
|
});
|
|
m_Queries.insert_or_assign (transactionID, std::make_tuple (toIdent, toPort,
|
|
query, info, i2p::util::GetMonotonicSeconds ()));
|
|
if (isRaw)
|
|
SendRawDatagram (msg, toIdent, toPort);
|
|
else
|
|
SendDatagram (msg, toIdent, toPort);
|
|
}
|
|
|
|
void TorrentsDHT::SendResponseMsg (std::string_view response, std::string_view transactionID,
|
|
const i2p::data::IdentHash& toIdent, uint16_t toPort)
|
|
{
|
|
auto msg = CreateDictionary ({
|
|
{ "r", response },
|
|
{ "t", CreateByteString (transactionID) },
|
|
{ "y", CreateByteString ("r") }
|
|
});
|
|
SendRawDatagram (msg, toIdent, toPort);
|
|
}
|
|
|
|
void TorrentsDHT::SendPingQuery (const i2p::data::IdentHash& toIdent, uint16_t toPort)
|
|
{
|
|
SendQueryMsg (eKRPCQueryPing, CreateDictionary ({
|
|
{ "id", CreateByteString (std::string_view ((const char *)m_NodeID.data (), m_NodeID.size ())) }
|
|
}),
|
|
toIdent, toPort);
|
|
}
|
|
|
|
void TorrentsDHT::SendPingResponse (std::string_view transactionID, const i2p::data::IdentHash& toIdent, uint16_t toPort)
|
|
{
|
|
SendResponseMsg (CreateDictionary ({
|
|
{ "id", CreateByteString (std::string_view ((const char *)m_NodeID.data (), m_NodeID.size ())) }
|
|
}),
|
|
transactionID, toIdent, toPort);
|
|
}
|
|
|
|
void TorrentsDHT::SendGetPeersQuery (std::shared_ptr<RequestInfo> info, const i2p::data::IdentHash& toIdent, uint16_t toPort)
|
|
{
|
|
if (!info || !info->torrent) return;
|
|
const auto& infoHash = info->torrent->GetInfoHash ();
|
|
SendQueryMsg (eKRPCQueryGetPeers, CreateDictionary ({
|
|
{ "id", CreateByteString (std::string_view ((const char *)m_NodeID.data (), m_NodeID.size ())) },
|
|
{ "info_hash", CreateByteString (std::string_view ((const char *)infoHash.data (), infoHash.size ())) }
|
|
}),
|
|
toIdent, toPort, false, info);
|
|
}
|
|
|
|
void TorrentsDHT::SendNextGetPeersQuery (std::shared_ptr<RequestInfo> info)
|
|
{
|
|
if (!info) return;
|
|
bool sent = false;
|
|
if (!info->IsDone () && m_RoutingTable)
|
|
{
|
|
auto nextNode = info->GetNextNode ();
|
|
while (nextNode && m_FailedPeers.contains (nextNode->peer) && !m_HeardFrom.contains (nextNode->id)) // skip failed peers
|
|
nextNode = info->GetNextNode ();
|
|
if (nextNode)
|
|
{
|
|
sent = true;
|
|
SendGetPeersQuery (info, nextNode->peer, nextNode->port);
|
|
}
|
|
else
|
|
LogPrint (eLogDebug, "TorrentsDHT: No more nodes to send get_peers");
|
|
}
|
|
else
|
|
LogPrint (eLogDebug, "TorrentsDHT: Closest node not found after ", DHT_MAX_NUM_GET_PEERS_ATTEMPTS, " get_peers attempts");
|
|
if (!sent && !info->tokens.empty () && info->torrent)
|
|
{
|
|
// announce
|
|
int numAnnounces = 0;
|
|
for (const auto& [distance, node, token]: info->tokens)
|
|
{
|
|
SendAnnouncePeerQuery (info, token, node->peer, node->port + 1); // to rport
|
|
numAnnounces++;
|
|
if (numAnnounces >= DHT_MAX_NUM_CLOSEST_NODES_TO_ANNOUNCE) break;
|
|
}
|
|
if (numAnnounces > 0)
|
|
LogPrint (eLogDebug, "TorrentsDHT: Announce sent to ", numAnnounces, " closest nodes");
|
|
}
|
|
}
|
|
|
|
void TorrentsDHT::SendGetPeersResponse (std::string_view transactionID, std::shared_ptr<DHTTorrent> torrent,
|
|
uint64_t token, const i2p::data::IdentHash& toIdent, uint16_t toPort)
|
|
{
|
|
if (!torrent) return;
|
|
SendResponseMsg (CreateDictionary ({
|
|
{ "id", CreateByteString (std::string_view ((const char *)m_NodeID.data (), m_NodeID.size ())) },
|
|
{ "token", CreateByteString (std::string_view ((const char *)&token, 8)) },
|
|
{ "values", torrent->GetBEncodedPeers () }
|
|
}),
|
|
transactionID, toIdent, toPort);
|
|
}
|
|
|
|
void TorrentsDHT::SendGetPeersResponse (std::string_view transactionID, std::string_view nodes,
|
|
uint64_t token, const i2p::data::IdentHash& toIdent, uint16_t toPort)
|
|
{
|
|
SendResponseMsg (CreateDictionary ({
|
|
{ "id", CreateByteString (std::string_view ((const char *)m_NodeID.data (), m_NodeID.size ())) },
|
|
{ "token", CreateByteString (std::string_view ((const char *)&token, 8)) },
|
|
{ "nodes", CreateByteString (nodes) }
|
|
}),
|
|
transactionID, toIdent, toPort);
|
|
}
|
|
|
|
void TorrentsDHT::SendFindNodeQuery (std::shared_ptr<RequestInfo> info, const i2p::data::IdentHash& toIdent, uint16_t toPort)
|
|
{
|
|
if (!info) return;
|
|
SendQueryMsg (eKRPCQueryFindNode, CreateDictionary ({
|
|
{ "id", CreateByteString (std::string_view ((const char *)m_NodeID.data (), m_NodeID.size ())) },
|
|
{ "target", CreateByteString (std::string_view ((const char *)info->target.data (), info->target.size ())) }
|
|
}),
|
|
toIdent, toPort, false, info);
|
|
}
|
|
|
|
void TorrentsDHT::SendNextFindNodeQuery (std::shared_ptr<RequestInfo> info)
|
|
{
|
|
if (!info) return;
|
|
if (!info->IsDone () && m_RoutingTable)
|
|
{
|
|
auto nextNode = info->GetNextNode ();
|
|
auto numAttempts = info->numAttempts;
|
|
while (nextNode && m_FailedPeers.contains (nextNode->peer) && !m_HeardFrom.contains (nextNode->id)) // skip failed peers
|
|
nextNode = info->GetNextNode ();
|
|
info->numAttempts = numAttempts;
|
|
if (nextNode)
|
|
SendFindNodeQuery (info, nextNode->peer, nextNode->port);
|
|
else
|
|
LogPrint (eLogDebug, "TorrentsDHT: No more nodes to send find_node");
|
|
}
|
|
}
|
|
|
|
void TorrentsDHT::SendFindNodeResponse (std::string_view transactionID, std::string_view nodes,
|
|
const i2p::data::IdentHash& toIdent, uint16_t toPort)
|
|
{
|
|
SendResponseMsg (CreateDictionary ({
|
|
{ "id", CreateByteString (std::string_view ((const char *)m_NodeID.data (), m_NodeID.size ())) },
|
|
{ "nodes", CreateByteString (nodes) }
|
|
}),
|
|
transactionID, toIdent, toPort);
|
|
}
|
|
|
|
void TorrentsDHT::SendAnnouncePeerQuery (std::shared_ptr<RequestInfo> info,
|
|
uint64_t token, const i2p::data::IdentHash& toIdent, uint16_t toPort)
|
|
{
|
|
if (!info || !info->torrent) return;
|
|
auto infoHash = info->torrent->GetInfoHash ();
|
|
SendQueryMsg (eKRPCQueryAnnouncePeer, CreateDictionary ({
|
|
{ "id", CreateByteString (std::string_view ((const char *)m_NodeID.data (), m_NodeID.size ())) },
|
|
{ "info_hash", CreateByteString (std::string_view ((const char *)infoHash.data (), infoHash.size ())) },
|
|
{ "token", CreateByteString (std::string_view ((const char *)&token, 8)) }
|
|
}),
|
|
toIdent, toPort, true, info); // raw
|
|
}
|
|
|
|
void TorrentsDHT::Explore ()
|
|
{
|
|
if (m_RoutingTable && m_Tunnel.GetLocalDestination ())
|
|
{
|
|
auto targets = m_RoutingTable->GetExploratoryTargets (m_Tunnel.GetLocalDestination ()->GetRng ());
|
|
for (auto [target, bucket]: targets)
|
|
{
|
|
auto request = std::make_shared<RequestInfo>(nullptr, DHT_MAX_NUM_FIND_NODE_ATTEMPTS);
|
|
request->target = target;
|
|
if (bucket->IsEmpty ())
|
|
{
|
|
if (bucket->next && !bucket->next->IsEmpty ())
|
|
bucket = bucket->next;
|
|
else
|
|
continue;
|
|
}
|
|
// fill initial list of nodes to request from bucket
|
|
for (auto it: bucket->nodes)
|
|
request->AddNode (it.second);
|
|
SendNextFindNodeQuery (request);
|
|
}
|
|
}
|
|
}
|
|
|
|
std::shared_ptr<Node> TorrentsDHT::UpdateNode (std::shared_ptr<Node> node)
|
|
{
|
|
if (!node) return nullptr;
|
|
if (!node->VerifyID ())
|
|
{
|
|
LogPrint (eLogInfo, "TorrentsDHT: Invalid node ID received ", node->id.ToBase64 ());
|
|
return nullptr;
|
|
}
|
|
if (m_RoutingTable)
|
|
{
|
|
auto addedNode = m_RoutingTable->AddNode (node);
|
|
if (addedNode != node)
|
|
node = addedNode;
|
|
else
|
|
LogPrint (eLogDebug, "TorrentsDHT: Node ", node->id.ToBase64 (), " added");
|
|
}
|
|
else
|
|
LogPrint (eLogError, "TorrentsDHT: No routing table");
|
|
return node;
|
|
}
|
|
|
|
void TorrentsDHT::UpdateHeardFrom (const NodeID& nodeID)
|
|
{
|
|
m_HeardFrom[nodeID] = i2p::util::GetMonotonicSeconds ();
|
|
}
|
|
|
|
void TorrentsDHT::ScheduleDHTUpdateCheck ()
|
|
{
|
|
m_DHTUpdateCheckTimer.cancel ();
|
|
m_DHTUpdateCheckTimer.expires_after (std::chrono::seconds (DHT_UPDATE_CHECK_INTERVAL));
|
|
m_DHTUpdateCheckTimer.async_wait (std::bind_front(&TorrentsDHT::HandleDHTUpdateCheckTimer, this));
|
|
}
|
|
|
|
void TorrentsDHT::HandleDHTUpdateCheckTimer (const boost::system::error_code& ecode)
|
|
{
|
|
if (ecode != boost::asio::error::operation_aborted)
|
|
{
|
|
auto ts = i2p::util::GetMonotonicSeconds ();
|
|
if (ts > m_NextDHTExploratoryTime)
|
|
{
|
|
Explore ();
|
|
m_NextDHTExploratoryTime = ts + DHT_EXPLORATORY_INTERVAL + (m_Tunnel.GetLocalDestination () ?
|
|
m_Tunnel.GetLocalDestination ()->GetRng ()() % DHT_EXPLORATORY_INTERVAL_VARIANCE : 0);
|
|
}
|
|
ScheduleDHTUpdateCheck ();
|
|
}
|
|
}
|
|
|
|
void TorrentsDHT::ScheduleDHTExpirationCheck ()
|
|
{
|
|
m_DHTExpirationCheckTimer.cancel ();
|
|
m_DHTExpirationCheckTimer.expires_after (std::chrono::seconds (DHT_EXPIRATION_CHECK_INTERVAL));
|
|
m_DHTExpirationCheckTimer.async_wait (std::bind_front(&TorrentsDHT::HandleDHTExpirationCheckTimer, this));
|
|
}
|
|
|
|
void TorrentsDHT::HandleDHTExpirationCheckTimer (const boost::system::error_code& ecode)
|
|
{
|
|
if (ecode != boost::asio::error::operation_aborted)
|
|
{
|
|
auto ts = i2p::util::GetMonotonicSeconds ();
|
|
if (m_RoutingTable)
|
|
{
|
|
m_RoutingTable->DeleteExpiredNodes (ts);
|
|
LogPrint (eLogDebug, "TorrentsDHT: Stats ", m_RoutingTable->GetNumNodes (), " nodes in ", m_RoutingTable->GetNumBuckets (), " buckets ");
|
|
}
|
|
{
|
|
auto it = m_Torrents.begin ();
|
|
while (it != m_Torrents.end ())
|
|
{
|
|
if (it->second->CleanUp (ts))
|
|
it = m_Torrents.erase (it);
|
|
else
|
|
it++;
|
|
}
|
|
}
|
|
{
|
|
auto it = m_HeardFrom.begin ();
|
|
while (it != m_HeardFrom.end ())
|
|
{
|
|
if (ts > it->second + DHT_HEARD_FROM_EXPIRATION_TIME)
|
|
it = m_HeardFrom.erase (it);
|
|
else
|
|
it++;
|
|
}
|
|
}
|
|
{
|
|
auto it = m_FailedPeers.begin ();
|
|
while (it != m_FailedPeers.end ())
|
|
{
|
|
if (ts > it->second + DHT_PEER_FAILURE_INTERVAL)
|
|
it = m_FailedPeers.erase (it);
|
|
else
|
|
it++;
|
|
}
|
|
}
|
|
{
|
|
auto it = m_IncomingTokens.begin ();
|
|
while (it != m_IncomingTokens.end ())
|
|
{
|
|
if (ts > it->second.second + DHT_INCOMING_GET_PEERS_TOKEN_EXPIRATION_TIME)
|
|
it = m_IncomingTokens.erase (it);
|
|
else
|
|
it++;
|
|
}
|
|
}
|
|
ScheduleDHTExpirationCheck ();
|
|
}
|
|
}
|
|
|
|
void TorrentsDHT::ScheduleDHTSendPingCheck ()
|
|
{
|
|
m_DHTSendPingCheckTimer.cancel ();
|
|
m_DHTSendPingCheckTimer.expires_after (std::chrono::seconds (DHT_SEND_PING_CHECK_INTERVAL));
|
|
m_DHTSendPingCheckTimer.async_wait (std::bind_front(&TorrentsDHT::HandleDHTSendPingCheckTimer, this));
|
|
}
|
|
|
|
void TorrentsDHT::HandleDHTSendPingCheckTimer (const boost::system::error_code& ecode)
|
|
{
|
|
if (ecode != boost::asio::error::operation_aborted)
|
|
{
|
|
auto ts = i2p::util::GetMonotonicSeconds ();
|
|
if (m_RoutingTable)
|
|
{
|
|
auto toPing = m_RoutingTable->GetNodesToPing (ts);
|
|
for (const auto& it: toPing)
|
|
SendPingQuery (it->peer, it->port);
|
|
}
|
|
ScheduleDHTSendPingCheck ();
|
|
}
|
|
}
|
|
|
|
void TorrentsDHT::ScheduleDHTQueryExpirationCheck ()
|
|
{
|
|
m_DHTQueryExpirationCheckTimer.cancel ();
|
|
m_DHTQueryExpirationCheckTimer.expires_after (std::chrono::seconds (DHT_QUERY_EXPIRATION_CHECK_INTERVAL));
|
|
m_DHTQueryExpirationCheckTimer.async_wait (std::bind_front(&TorrentsDHT::DHTQueryExpirationCheckTimer, this));
|
|
}
|
|
|
|
void TorrentsDHT::DHTQueryExpirationCheckTimer (const boost::system::error_code& ecode)
|
|
{
|
|
if (ecode != boost::asio::error::operation_aborted)
|
|
{
|
|
std::list<std::shared_ptr<RequestInfo> > getpeers, findnode;
|
|
auto ts = i2p::util::GetMonotonicSeconds ();
|
|
auto it = m_Queries.begin ();
|
|
while (it != m_Queries.end ())
|
|
{
|
|
if (ts > std::get<4>(it->second) + DHT_QUERY_EXPIRATION_TIME)
|
|
{
|
|
m_FailedPeers.emplace (std::get<0>(it->second), std::get<4>(it->second));
|
|
switch (std::get<2>(it->second))
|
|
{
|
|
case eKRPCQueryGetPeers:
|
|
getpeers.push_back (std::get<3>(it->second));
|
|
break;
|
|
case eKRPCQueryFindNode:
|
|
findnode.push_back (std::get<3>(it->second));
|
|
break;
|
|
default: ;
|
|
}
|
|
it = m_Queries.erase (it);
|
|
}
|
|
else
|
|
it++;
|
|
}
|
|
for (auto it1: getpeers)
|
|
{
|
|
LogPrint (eLogDebug, "TorrentsDHT: get_peers response timeout after attempt #", it1->numAttempts);
|
|
if (m_RoutingTable && it1->lastNode && !m_HeardFrom.contains (it1->lastNode->id))
|
|
{
|
|
auto bucket = m_RoutingTable->FindBucket (it1->lastNode->id);
|
|
if (bucket && bucket->nodes.size () > MAX_BUCKET_CAPACITY/2)
|
|
bucket->nodes.erase (it1->lastNode->id);
|
|
}
|
|
SendNextGetPeersQuery (it1);
|
|
}
|
|
for (auto it1: findnode)
|
|
{
|
|
LogPrint (eLogDebug, "TorrentsDHT: find_node response timeout after attempt #", it1->numAttempts);
|
|
if (m_RoutingTable && it1->lastNode && !m_HeardFrom.contains (it1->lastNode->id))
|
|
{
|
|
auto bucket = m_RoutingTable->FindBucket (it1->lastNode->id);
|
|
if (bucket && bucket->nodes.size () > MAX_BUCKET_CAPACITY/2)
|
|
bucket->nodes.erase (it1->lastNode->id);
|
|
}
|
|
SendNextFindNodeQuery (it1);
|
|
}
|
|
ScheduleDHTQueryExpirationCheck ();
|
|
}
|
|
}
|
|
|
|
void TorrentsDHT::GetPeersAndAnnounce (std::shared_ptr<Torrent> torrent)
|
|
{
|
|
if (!torrent || !m_RoutingTable) return;
|
|
auto request = std::make_shared<RequestInfo>(torrent, DHT_MAX_NUM_GET_PEERS_ATTEMPTS);
|
|
auto bucket = m_RoutingTable->FindBucket (torrent->GetInfoHash ());
|
|
if (!bucket) return; // DHT is empty
|
|
if (bucket->IsEmpty ())
|
|
{
|
|
if (bucket->next && !bucket->next->IsEmpty ())
|
|
bucket = bucket->next;
|
|
else
|
|
return;
|
|
}
|
|
// fill initial list of nodes to request from bucket
|
|
for (auto it: bucket->nodes)
|
|
request->AddNode (it.second);
|
|
SendNextGetPeersQuery (request);
|
|
}
|
|
}
|
|
}
|
|
|
|
#endif
|