From 80691ba64d8e2ff04880b7cf0a6036b8f0b3a49f Mon Sep 17 00:00:00 2001 From: orignal Date: Wed, 26 Aug 2026 16:12:08 -0400 Subject: [PATCH] Torrents/TorrentsTunnel split --- libi2pd_client/ClientContext.h | 2 +- libi2pd_client/Torrents.cpp | 750 +---------------------------- libi2pd_client/Torrents.h | 97 ---- libi2pd_client/TorrentsRPC.cpp | 1 + libi2pd_client/TorrentsTunnel.cpp | 773 ++++++++++++++++++++++++++++++ libi2pd_client/TorrentsTunnel.h | 126 +++++ 6 files changed, 902 insertions(+), 847 deletions(-) create mode 100644 libi2pd_client/TorrentsTunnel.cpp create mode 100644 libi2pd_client/TorrentsTunnel.h diff --git a/libi2pd_client/ClientContext.h b/libi2pd_client/ClientContext.h index 13345352..663a3276 100644 --- a/libi2pd_client/ClientContext.h +++ b/libi2pd_client/ClientContext.h @@ -25,7 +25,7 @@ #include "BOB.h" #include "I2CP.h" #include "AddressBook.h" -#include "Torrents.h" +#include "TorrentsTunnel.h" #include "TorrentsRPC.h" #include "I18N_langs.h" diff --git a/libi2pd_client/Torrents.cpp b/libi2pd_client/Torrents.cpp index 1a183eeb..32627599 100644 --- a/libi2pd_client/Torrents.cpp +++ b/libi2pd_client/Torrents.cpp @@ -18,8 +18,8 @@ #include #include "Log.h" #include "I2PEndian.h" -#include "ClientContext.h" #include "Timestamp.h" +#include "TorrentsTunnel.h" #include "Torrents.h" namespace i2p @@ -1580,753 +1580,5 @@ namespace torrents WriteToStream (buf.data (), bufOffset); return bufOffset > 0; } - - TorrentsTunnel::TorrentsTunnel (std::string_view name, std::shared_ptr localDestination, - std::string_view torrentsDir, std::string_view trackers): - i2p::client::I2PService (localDestination), m_Name (name), m_PeerID ("-I2PD-"), - m_TorrentsDir (torrentsDir), m_TrackerRequestsCheckTimer (GetService ()), - m_KeepAliveCheckTimer (GetService ()), m_ReconnectCheckTimer (GetService ()), - m_TorrentsStatusUpdateTimer (GetService ()) - { - if (localDestination) - m_PeerID += localDestination->GetIdentHash ().ToBase64 (); - m_PeerID.resize (20, '0'); - if (!trackers.empty ()) - boost::split(m_Trackers, trackers, boost::is_any_of(","), boost::token_compress_on); - } - - void TorrentsTunnel::Start () - { - i2p::client::I2PService::Start (); - m_DiskIOService.Start (); - - auto dgramDest = GetLocalDestination ()->CreateDatagramDestination (false, i2p::datagram::eDatagramV3); - if (dgramDest) - dgramDest->SetRawReceiver (std::bind (&TorrentsTunnel::HandleRecvFromI2PRaw, - std::static_pointer_cast(shared_from_this ()), - std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4)); - - Accept (); - - if (!m_TorrentsDir.empty() && std::filesystem::exists (m_TorrentsDir) && - std::filesystem::is_directory (m_TorrentsDir)) - { - for (const auto& it: std::filesystem::directory_iterator (m_TorrentsDir)) - if (std::filesystem::is_regular_file (it.status()) && it.path ().extension () == ".torrent") - ReadTorrentFile (it.path ()); - } - - ScheduleTrackerRequestsCheck (); - ScheduleKeepAliveCheck (); - ScheduleStatusUpdate (); - } - - void TorrentsTunnel::Stop () - { - auto localDestination = GetLocalDestination (); - if (localDestination) - { - localDestination->StopAcceptingStreams (); - auto dgramDest = localDestination->GetDatagramDestination (); - if (dgramDest) - dgramDest->ResetRawReceiver (); - } - m_TrackerRequestsCheckTimer.cancel (); - m_KeepAliveCheckTimer.cancel (); - m_ReconnectCheckTimer.cancel (); - m_TorrentsStatusUpdateTimer.cancel (); - m_Torrents.clear (); - for (auto it: m_Torrents) - { - auto fullPath = it.second->GetFullPath (); fullPath += ".resume"; - boost::asio::post (m_DiskIOService.GetService (), [torrent = it.second, fullPath]() - { - torrent->SaveTorrentResumeFile (fullPath); - }); - } - m_DiskIOService.Stop (); - i2p::client::I2PService::Stop (); - } - - void TorrentsTunnel::ReadTorrentFile (const std::filesystem::path& torrentFilePath) - { - std::shared_ptr torrent; - std::ifstream s(torrentFilePath, std::ifstream::binary); - if (s) - { - s.seekg (0,std::ios::end); - size_t len = s.tellg (); - if (len > 0) - { - s.seekg(0, std::ios::beg); - char * buf = new char[len]; - s.read(buf, len); - torrent = std::make_shared(std::string_view{buf, len}); - delete[] buf; - } - else - LogPrint (eLogError, "Torrents: Empty file ", torrentFilePath); - } - else - LogPrint (eLogError, "Torrents: Can't open file ", torrentFilePath); - - if (torrent && !torrent->IsValid ()) - { - // a torrent whose parsing stopped, an unsafe name among the rest, must - // not be used even in part: it would leave stray files behind - LogPrint (eLogError, "Torrents: Invalid torrent file ", torrentFilePath, ". Skipped"); - torrent = nullptr; - } - if (torrent) - { - torrent->SetFullPath (m_TorrentsDir/std::filesystem::path (torrent->GetName ())); - InitTorrentFiles (torrent); - InsertTorrent (torrent); - } - } - - void TorrentsTunnel::InitTorrentFiles (std::shared_ptr torrent) - { - if (!torrent) return; - if (torrent->GetFiles ().empty ()) - { - if (std::filesystem::exists (torrent->GetFullPath ())) - torrent->SetComplete (); - else - { - auto partFilePath = torrent->GetFullPath (); partFilePath += ".part"; - if (!std::filesystem::exists (partFilePath)) - CreateAndReserveFile (partFilePath, torrent->GetLength ()); - } - } - else - { - bool completed = true; - for (auto& [filePath, fileLength]: torrent->GetFiles ()) - { - filePath = torrent->GetFullPath ()/filePath; - if (!std::filesystem::exists (filePath)) - { - auto partFilePath = filePath; partFilePath += ".part"; - if (!std::filesystem::exists (partFilePath)) - CreateAndReserveFile (partFilePath, fileLength); - completed = false; - } - } - if (completed) torrent->SetComplete (); - } - auto resumeFilePath = torrent->GetFullPath (); resumeFilePath += ".resume"; - if (std::filesystem::exists (resumeFilePath)) - { - if (!torrent->IsComplete ()) - { - std::ifstream rs(resumeFilePath, std::ifstream::binary); - if (rs) - { - rs.seekg (0,std::ios::end); - size_t l = rs.tellg (); - if (l > 0) - { - rs.seekg(0, std::ios::beg); - std::vector bitfield(l); - rs.read((char *)bitfield.data (), l); - if (torrent->ApplyBitfield (bitfield)) - CompleteTorrent (torrent); - } - } - } - else - std::filesystem::remove (resumeFilePath); - } - } - - bool TorrentsTunnel::CreateAndReserveFile (const std::filesystem::path& filePath, size_t reserve) - { - if (std::filesystem::exists (filePath)) return false; - auto subdirs = filePath.parent_path (); - if (!subdirs.empty ()) - { - // try to create all subdirs - try - { - std::filesystem::create_directories (subdirs); - } - catch (std::exception& ex) - { - LogPrint (eLogError, "Torrents: Can't create subdirs ", subdirs, " : ", ex.what()); - return false; - } - } - // create file - std::ofstream f(filePath, std::ios::binary); - if (!f) return false; - f.close (); - if (reserve > 0) - { - // resize - try - { - std::filesystem::resize_file (filePath, reserve); - } - catch (std::exception& ex) - { - LogPrint (eLogError, "Torrents: Can't resize file ", filePath, " to ", reserve, " : ", ex.what()); - return false; - } - } - return true; - } - - void TorrentsTunnel::CompleteTorrent (std::shared_ptr torrent) - { - boost::asio::post (GetDiskIOService (), [this, torrent]() - { - bool completed = false; - if (torrent->GetFiles ().empty ()) - { - auto partFilePath = torrent->GetFullPath (); partFilePath += ".part"; - std::error_code ec; - std::filesystem::rename (partFilePath, torrent->GetFullPath (), ec); - if (!ec) - completed = true; - else - LogPrint (eLogError, "Torrents: Can't rename ", partFilePath); - } - else - { - completed = true; - for (const auto& [filePath, fileSize]: torrent->GetFiles ()) - { - auto partFilePath = filePath; partFilePath += ".part"; - std::error_code ec; - std::filesystem::rename (partFilePath, filePath, ec); - if (ec) - { - completed = false; - LogPrint (eLogError, "Torrents: Can't rename ", partFilePath); - } - } - } - if (completed) - { - torrent->SetComplete (); - auto resumeFilePath = torrent->GetFullPath (); resumeFilePath += ".resume"; - if (!std::filesystem::remove (resumeFilePath)) - LogPrint (eLogError, "Torrents: Can't delete resume file ", resumeFilePath); - LogPrint (eLogInfo, "Torrents: Download complete ", torrent->GetFullPath ()); - - boost::asio::post (GetService (), [this, torrent]() - { - // inform tracker that we are done - RequestTorrentTrackers (torrent, "completed"); - // close connections with seeds and reset stats for remaining - auto conns = GetTorrentConnections (torrent); - for (auto it: conns) - { - if (it->GetRemoteBitfield ().all ()) // seed - it->Close (); - else - it->ResetStats (); - } - }); - } - }); - } - - std::shared_ptr TorrentsTunnel::FindTorrent (const Torrent::InfoHash& infoHash) const - { - std::lock_guard l(m_TorrentsMutex); - auto it = m_Torrents.find (infoHash); - if (it != m_Torrents.end ()) - return it->second; - return nullptr; - } - - std::shared_ptr TorrentsTunnel::FindTorrentByID (int id) const - { - std::lock_guard l(m_TorrentsMutex); - auto it = m_TorrentsByID.find (id); - if (it != m_TorrentsByID.end ()) - return it->second.lock (); - return nullptr; - } - - std::vector TorrentsTunnel::GetTorrentIDs () const - { - std::vector ids; - std::lock_guard l(m_TorrentsMutex); - for (const auto& it: m_TorrentsByID) - if (!it.second.expired ()) ids.push_back (it.first); - return ids; - } - - std::pair, int> TorrentsTunnel::AddTorrent (std::string_view torrentFileContent) - { - auto torrent = std::make_shared (torrentFileContent); - if (m_Torrents.find (torrent->GetInfoHash ()) == m_Torrents.end ()) - { - torrent->SetFullPath (m_TorrentsDir/std::filesystem::path (torrent->GetName ())); - { - auto torrentFilePath = torrent->GetFullPath (); torrentFilePath += ".torrent"; - std::ofstream f(torrentFilePath, std::ofstream::binary); - if (f) - f.write (torrentFileContent.data (), torrentFileContent.size ()); - else - return { torrent, 0 }; - } - InitTorrentFiles (torrent); - return { torrent, InsertTorrent (torrent) }; - } - return { torrent, 0 }; - } - - int TorrentsTunnel::InsertTorrent (std::shared_ptr torrent) - { - if (!torrent) return 0; - std::lock_guard l(m_TorrentsMutex); - if (m_Torrents.emplace (torrent->GetInfoHash (), torrent).second) - { - int id = 1; - if (!m_TorrentsByID.empty ()) - id = m_TorrentsByID.rbegin ()->first + 1; - m_TorrentsByID.emplace (id, torrent); - return id; - } - return 0; - } - - bool TorrentsTunnel::RemoveTorrent (int id, bool deleteFiles) - { - std::shared_ptr torrent; - { - std::lock_guard l(m_TorrentsMutex); - auto it = m_TorrentsByID.find (id); - if (it == m_TorrentsByID.end ()) return false; - torrent = it->second.lock (); - m_TorrentsByID.erase (it); - if (!torrent) return false; - m_Torrents.erase (torrent->GetInfoHash ()); - } - boost::asio::post (GetService (), [this, torrent, deleteFiles]() - { - RemoveTorrent (torrent, deleteFiles); - }); - return true; - } - - void TorrentsTunnel::RemoveTorrent (std::shared_ptr torrent, bool deleteFiles) - { - if (!torrent) return; - auto connections = GetTorrentConnections (torrent); - // close connections - for (auto it: connections) - it->Close (); - if (deleteFiles) - boost::asio::post (GetDiskIOService (), [torrent]() - { - auto fullPath = torrent->GetFullPath (); - auto torrentFilePath = fullPath; torrentFilePath += ".torrent"; - std::error_code ec; - std::filesystem::remove (torrentFilePath, ec); - if (ec) - LogPrint (eLogError, "Torrents: Can't delete ", torrentFilePath); - auto resumeFilePath = fullPath; resumeFilePath += ".resume"; - if (std::filesystem::exists (resumeFilePath)) - { - std::filesystem::remove (resumeFilePath, ec); - if (ec) - LogPrint (eLogError, "Torrents: Can't delete ", resumeFilePath); - } - if (torrent->IsComplete () || !torrent->GetFiles ().empty ()) - { - std::filesystem::remove_all (fullPath, ec); - if (ec) - LogPrint (eLogError, "Torrents: Can't delete ", fullPath); - } - else - { - auto partFilePath = fullPath; partFilePath += ".part"; - std::filesystem::remove (partFilePath, ec); - if (ec) - LogPrint (eLogError, "Torrents: Can't delete ", partFilePath); - } - }); - } - - void TorrentsTunnel::Accept () - { - auto localDestination = GetLocalDestination (); - if (localDestination) - { - if (!localDestination->IsAcceptingStreams ()) // set it as default if not set yet - localDestination->AcceptStreams ([this](std::shared_ptr stream) - { - if (stream) - { - auto conn = std::make_shared (shared_from_this (), stream); - AddHandler (conn); - conn->ReceiveHandshake (); - } - }); - } - else - LogPrint (eLogError, "Torrents: Local destination not set"); - } - - void TorrentsTunnel::RequestTorrentTrackers (std::shared_ptr torrent, std::string_view event) - { - if (!m_Trackers.empty ()) - for (size_t i = 0; i < m_Trackers.size (); i++) - RequestTracker (i, torrent, event); - else - RequestTracker (0, torrent, event); // from announce - } - - void TorrentsTunnel::RequestTracker (size_t trackerID, std::shared_ptr torrent, std::string_view event) - { - if (!torrent) return; - i2p::http::URL reqURL; - if (trackerID < m_Trackers.size()) - reqURL.parse (m_Trackers[trackerID]); - else - reqURL.parse (torrent->GetAnnounce ()); -#if __cplusplus >= 202002L // C++20 - if (!reqURL.host.ends_with (".i2p")) -#else - if (reqURL.host.find(".i2p") == reqURL.host.npos) -#endif - { - LogPrint (eLogWarning, "Torrents: Non-I2P address ", reqURL.host, " for torrent ", torrent->GetName ()); - return; - } - if (reqURL.schema == "udp") - { - ConnectToDatagramTracker (reqURL.host, reqURL.port); - return; - } - std::map params; - params.emplace ("info_hash", torrent->GetHexStringInfoHash ()); - params.emplace ("peer_id", m_PeerID); - params.emplace ("ip", GetLocalDestination ()->GetIdentity ()->ToBase64 () + ".i2p"); - params.emplace ("port", std::to_string (TORRENT_PORT)); // 6881 - params.emplace ("compact", "1"); - params.emplace ("uploaded", std::to_string (torrent->GetUploaded ())); - params.emplace ("downloaded", std::to_string (torrent->GetLength () - torrent->GetLeft ())); - params.emplace ("left", std::to_string (torrent->GetLeft ())); - params.emplace ("numwant", torrent->IsComplete () ? "0" : "25"); // max num of peers, 0 if seeding - if (!event.empty ()) - params.emplace ("event", event); - reqURL.create_query (params); - - auto req = std::make_shared >(boost::beast::http::verb::get, reqURL.to_string (true), 11); // HTTP 1.1 - req->set (boost::beast::http::field::host, reqURL.host); - req->set (boost::beast::http::field::user_agent, "I2PSocketEepGet"); - req->keep_alive (false); // Connection: close - CreateStream ([this, req, torrent, trackerID](std::shared_ptr stream) - { - if (stream) - { - auto httpStream = std::make_shared(stream); - boost::beast::http::async_write (*httpStream, *req, - std::bind (&TorrentsTunnel::TrackerRequestSent, this, std::placeholders::_1, - std::placeholders::_2, httpStream, torrent, req, trackerID)); - } - }, reqURL.host, reqURL.port); - } - - void TorrentsTunnel::TrackerRequestSent (const boost::beast::error_code& ecode, size_t bytes_transferred, - std::shared_ptr httpStream, std::shared_ptr torrent, - std::shared_ptr > req, size_t trackerID) - { - if (!ecode) - { - // receive - auto buf = std::make_shared (); - auto res = std::make_shared >(); - boost::beast::http::async_read (*httpStream, *buf, *res, - [this, httpStream, torrent, buf, res, trackerID](const boost::beast::error_code& ecode, size_t bytes_transferred) - { - httpStream->GetStream ()->AsyncClose (); - if (!ecode) - { - if (res->result () == boost::beast::http::status::ok) - { - torrent->ParseTrackerResponse (trackerID, res->body ()); - ConnectToPeers (torrent); - ScheduleReconnectCheck (); - } - else - LogPrint (eLogWarning, "Torrents: Tracker ", trackerID, " response code ", res->result_int()); - } - }); - } - } - - void TorrentsTunnel::ConnectToPeer (std::shared_ptr torrent, const i2p::data::IdentHash& peer) - { - if (!torrent) return; - LogPrint (eLogDebug, "Torrents: Connecting to peer ", peer.ToBase32 () + ".b32.i2p"); - if (peer == GetLocalDestination ()->GetIdentHash ()) - { - LogPrint (eLogInfo, "Torrents: Can't connect to self"); - return; - } - CreateStream ([this, torrent, peer](std::shared_ptr stream) - { - if (stream) - { - LogPrint (eLogDebug, "Torrents: Connected to peer ", peer.ToBase32 () + ".b32.i2p"); - auto connection = std::make_shared(shared_from_this (), stream, torrent); - AddHandler (connection); - connection->Connect (); - } - else - LogPrint (eLogInfo, "Torrents: Can't connect to peer ", peer.ToBase32 () + ".b32.i2p"); - }, std::make_shared(peer), TORRENT_PORT); - } - - size_t TorrentsTunnel::ConnectToPeers (std::shared_ptr torrent) - { - if (!torrent) return 0; - auto peersToConnect = GetNonConnectedPeers (torrent); - if (!peersToConnect.empty ()) - { - for (const auto& it: peersToConnect) - ConnectToPeer (torrent, it); - } - return peersToConnect.size (); - } - - void TorrentsTunnel::ScheduleTrackerRequestsCheck () - { - m_TrackerRequestsCheckTimer.expires_after (std::chrono::milliseconds(TRACKER_REQUESTS_CHECK_TIMEOUT)); - m_TrackerRequestsCheckTimer.async_wait (std::bind (&TorrentsTunnel::HandleTrackerRequestsCheckTimer, - this, std::placeholders::_1)); - } - - void TorrentsTunnel::HandleTrackerRequestsCheckTimer (const boost::system::error_code& ecode) - { - if (ecode != boost::asio::error::operation_aborted) - { - auto ts = i2p::util::GetMonotonicMilliseconds (); - for (auto it: m_Torrents) - for (size_t i = 0; i < m_Trackers.size (); i++) - if (ts > it.second->GetNextTrackerRequestTime (i)) - { - auto nextInterval = it.second->GetInterval (i) + GetLocalDestination ()->GetRng()() % TRACKER_REQUESTS_INTERVAL_VARIANCE; - it.second->SetNextTrackerRequestTime (i, ts + nextInterval); - RequestTracker (i, it.second); - } - ScheduleTrackerRequestsCheck (); - } - } - - void TorrentsTunnel::ScheduleKeepAliveCheck () - { - m_KeepAliveCheckTimer.expires_after (std::chrono::seconds(PEER_KEEP_ALIVE_CHECK_INTERVAL)); - m_KeepAliveCheckTimer.async_wait (std::bind (&TorrentsTunnel::HandleKeepAliveCheckTimer, - this, std::placeholders::_1)); - } - - void TorrentsTunnel::HandleKeepAliveCheckTimer (const boost::system::error_code& ecode) - { - if (ecode != boost::asio::error::operation_aborted) - { - auto ts = i2p::util::GetMonotonicSeconds (); - IterateHandlers ([ts](std::shared_ptr handler) - { - if (handler) - std::static_pointer_cast(handler)->CheckKeepAlive (ts); - }); - ScheduleKeepAliveCheck (); - } - } - - void TorrentsTunnel::ScheduleReconnectCheck () - { - m_ReconnectCheckTimer.cancel (); - m_ReconnectCheckTimer.expires_after (std::chrono::seconds(RECONNECT_CHECK_INTERVAL)); - m_ReconnectCheckTimer.async_wait (std::bind (&TorrentsTunnel::HandleReconnectCheckTimer, - this, std::placeholders::_1)); - } - - void TorrentsTunnel::HandleReconnectCheckTimer (const boost::system::error_code& ecode) - { - if (ecode != boost::asio::error::operation_aborted) - { - for (auto it: m_Torrents) - { - if (!it.second->IsComplete ()) - { - auto numPeers = ConnectToPeers (it.second); - if (numPeers) - LogPrint (eLogDebug, "Torrents: Reconnecting to ", numPeers, " peers"); - } - } - ScheduleReconnectCheck (); - } - } - - void TorrentsTunnel::ScheduleStatusUpdate () - { - m_TorrentsStatusUpdateTimer.cancel (); - m_TorrentsStatusUpdateTimer.expires_after (std::chrono::seconds(TORRENTS_STATUS_UPDATE_INTERVAL)); - m_TorrentsStatusUpdateTimer.async_wait (std::bind (&TorrentsTunnel::HandleTorrentsStatusUpdateTimer, - this, std::placeholders::_1)); - } - - void TorrentsTunnel::HandleTorrentsStatusUpdateTimer (const boost::system::error_code& ecode) - { - if (ecode != boost::asio::error::operation_aborted) - { - auto ts = i2p::util::GetMonotonicSeconds (); - for (auto it: m_Torrents) - { - if (!it.second->IsComplete ()) - { - if (it.second->UpdateStatus (ts)) - CompleteTorrent (it.second); - else - UpdatePeersPerPiece (it.second); - } - } - UpdateStats (); - ScheduleStatusUpdate (); - } - } - - std::list > TorrentsTunnel::GetTorrentConnections (std::shared_ptr torrent) - { - std::list > ret; - if (torrent) - { - IterateHandlers ([&ret, torrent](std::shared_ptr handler) - { - if (handler) - { - auto conn = std::static_pointer_cast(handler); - if (conn->GetTorrent () == torrent && conn->GetStream ()) - ret.emplace_back (conn); - } - }); - } - return ret; - } - - std::unordered_set TorrentsTunnel::GetNonConnectedPeers (std::shared_ptr torrent) - { - std::unordered_set ret; - if (torrent) - { - ret = torrent->GetPeers (); - if(!ret.empty ()) - { - IterateHandlers ([&ret, torrent](std::shared_ptr handler) - { - if (handler) - { - auto conn = std::static_pointer_cast(handler); - if (conn->GetTorrent () == torrent && conn->GetStream ()) - { - auto ident = conn->GetStream ()->GetRemoteIdentity (); - if (ident) - ret.erase (ident->GetIdentHash ()); - } - } - }); - } - } - return ret; - } - - void TorrentsTunnel::UpdatePeersPerPiece (std::shared_ptr torrent) - { - if (!torrent) return; - torrent->StartCountingPeers (); - IterateHandlers ([torrent](std::shared_ptr handler) - { - if (handler) - { - auto conn = std::static_pointer_cast(handler); - if (conn->GetTorrent () == torrent) - torrent->ApplyPeerRemoteBitfield (conn->GetRemoteBitfield ()); - } - }); - } - - void TorrentsTunnel::UpdateStats () - { - for (auto it: m_Torrents) - it.second->ResetStats (); - IterateHandlers ([](std::shared_ptr handler) mutable - { - if (handler) - { - auto conn = std::static_pointer_cast(handler); - auto torrent = conn->GetTorrent (); - if (torrent) - { - torrent->SetDownloadRate (torrent->GetDownloadRate () + conn->GetDownloadRate ()); - torrent->SetUploadRate (torrent->GetUploadRate () + conn->GetUploadRate ()); - if (conn->IsDownloading ()) - torrent->SetNumDownloadingFromPeers (torrent->GetNumDownloadingFromPeers () + 1); - if (conn->IsUploading ()) - torrent->SetNumUploadingToPeers (torrent->GetNumUploadingToPeers () + 1); - } - } - }); - } - - void TorrentsTunnel::HandleRecvFromI2PRaw (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len) - { - // response from tracker - if (len < 8) return; - uint32_t action = bufbe32toh (buf); - switch (action) - { - case eDatagramTrackerActionConnect: - LogPrint (eLogDebug, "Torrents: action connect"); - break; - case eDatagramTrackerActionAnnounce: - LogPrint (eLogDebug, "Torrents: action announce"); - break; - case eDatagramTrackerActionError: - LogPrint (eLogDebug, "Torrents: action error"); - break; - default: - LogPrint (eLogInfo, "Torrents: Unexpected action ", action, " from tracker"); - } - } - - void TorrentsTunnel::ConnectToDatagramTracker (std::string_view dest, uint16_t port) - { - LogPrint (eLogDebug, "Torrents: Connecting to datagram tracker ", dest, ":", port); - auto address = i2p::client::context.GetAddressBook ().GetAddress (dest); - if (address && address->IsIdentHash ()) - { - auto localDestination = GetLocalDestination (); - auto dgramDest = localDestination->GetDatagramDestination (); - if (dgramDest) - { - uint8_t connectRequest[16]; - htobe64buf (connectRequest, 0x41727101980); // protocol_id - htobe32buf (connectRequest + 8, eDatagramTrackerActionConnect); // action - htobe32buf (connectRequest + 12, localDestination->GetRng()()); // transactionID - uint16_t fromPort = localDestination->GetRng()() % 1000 + 6000; - auto session = dgramDest->GetSession (address->identHash); - if (session) - { - session->SetVersion (i2p::datagram::eDatagramV2); // send datagram2 - dgramDest->SendDatagram (session, connectRequest, 16, fromPort, port); - } - else - LogPrint (eLogInfo, "Torrents: Can't obtain datagram session to ", dest); - } - else - LogPrint (eLogError, "Torrents: Datagram destination is not avaliable"); - } - else - LogPrint (eLogInfo, "Torrents: Tracker not found: ", dest); - } } } diff --git a/libi2pd_client/Torrents.h b/libi2pd_client/Torrents.h index c4b6bec1..a151c3ed 100644 --- a/libi2pd_client/Torrents.h +++ b/libi2pd_client/Torrents.h @@ -12,26 +12,20 @@ #include #include #include -#include #include #include #include #include #include -#include #include #include #include #include #include #include -#include #include "util.h" #include "Streaming.h" -#include "HTTP.h" #include "I2PService.h" -#include "AddressBook.h" -#include "BoostStream.h" namespace i2p { @@ -339,97 +333,6 @@ namespace torrents size_t m_ReceivedSinceLastTimestamp, m_SentSinceLastTimestamp; // bytes size_t m_Downloaded, m_Uploaded; // bytes }; - - enum DatagramTrackerAction - { - eDatagramTrackerActionConnect = 0, - eDatagramTrackerActionAnnounce = 1, - eDatagramTrackerActionError = 3 - }; - - class TorrentsTunnel final: public i2p::client::I2PService - { - private: - - class DiskIOService: private i2p::util::RunnableServiceWithWork - { - public: - - DiskIOService (): RunnableServiceWithWork ("TDiskIO") {} - auto& GetService () { return GetIOService (); } - void Start () { StartIOService (); } - void Stop () { StopWorkAndFinishTasks (); } - }; - - public: - - TorrentsTunnel (std::string_view name, std::shared_ptr localDestination, - std::string_view torrentsDir, std::string_view trackers = ""); - - void Start () override; - void Stop () override; - auto& GetDiskIOService () { return m_DiskIOService.GetService (); }; - - const std::string& GetPeerID () const { return m_PeerID; } - const std::vector& GetTrackers () const { return m_Trackers; } - std::shared_ptr FindTorrent (const Torrent::InfoHash& infoHash) const; - std::shared_ptr FindTorrentByID (int id) const; - std::vector GetTorrentIDs () const; - std::pair, int> AddTorrent (std::string_view torrentFileContent); // (tunnel, id) - bool RemoveTorrent (int id, bool deleteFiles); - std::list > GetTorrentConnections (std::shared_ptr torrent); - - const char* GetName() const override { return m_Name.c_str (); } - - private: - - - void Accept (); - void ReadTorrentFile (const std::filesystem::path& torrentFilePath); - void InitTorrentFiles (std::shared_ptr torrent); - int InsertTorrent (std::shared_ptr torrent); // returns id > 0 if success and 0 if failed - void RemoveTorrent (std::shared_ptr torrent, bool deleteFiles); - bool CreateAndReserveFile (const std::filesystem::path& filePath, size_t reserve); - void CompleteTorrent (std::shared_ptr torrent); - void RequestTracker (size_t trackerID, std::shared_ptr torrent, std::string_view event = ""); - void RequestTorrentTrackers (std::shared_ptr torrent, std::string_view event = ""); - void TrackerRequestSent (const boost::beast::error_code& ecode, size_t bytes_transferred, - std::shared_ptr httpStream, std::shared_ptr torrent, - std::shared_ptr > req, size_t trackerID); - - void ScheduleTrackerRequestsCheck (); - void HandleTrackerRequestsCheckTimer (const boost::system::error_code& ecode); - - void ScheduleKeepAliveCheck (); - void HandleKeepAliveCheckTimer (const boost::system::error_code& ecode); - - void ScheduleReconnectCheck (); - void HandleReconnectCheckTimer (const boost::system::error_code& ecode); - - void ScheduleStatusUpdate (); - void HandleTorrentsStatusUpdateTimer (const boost::system::error_code& ecode); - - std::unordered_set GetNonConnectedPeers (std::shared_ptr torrent); - void ConnectToPeer (std::shared_ptr torrent, const i2p::data::IdentHash& peer); - size_t ConnectToPeers (std::shared_ptr torrent); - void UpdatePeersPerPiece (std::shared_ptr torrent); - void UpdateStats (); - - void HandleRecvFromI2PRaw (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len); - void ConnectToDatagramTracker (std::string_view dest, uint16_t port); - - private: - - std::string m_Name, m_PeerID; // 20 characters - std::filesystem::path m_TorrentsDir; - std::vector m_Trackers; - std::map > m_Torrents; - std::map > m_TorrentsByID; - mutable std::mutex m_TorrentsMutex; - boost::asio::steady_timer m_TrackerRequestsCheckTimer, m_KeepAliveCheckTimer, - m_ReconnectCheckTimer, m_TorrentsStatusUpdateTimer; - DiskIOService m_DiskIOService; - }; } } #endif diff --git a/libi2pd_client/TorrentsRPC.cpp b/libi2pd_client/TorrentsRPC.cpp index 8e48b669..b82aa09d 100644 --- a/libi2pd_client/TorrentsRPC.cpp +++ b/libi2pd_client/TorrentsRPC.cpp @@ -17,6 +17,7 @@ #include #include "Log.h" #include "Torrents.h" +#include "TorrentsTunnel.h" #include "TorrentsRPC.h" namespace i2p diff --git a/libi2pd_client/TorrentsTunnel.cpp b/libi2pd_client/TorrentsTunnel.cpp new file mode 100644 index 00000000..fdec5d69 --- /dev/null +++ b/libi2pd_client/TorrentsTunnel.cpp @@ -0,0 +1,773 @@ +/* +* 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 +*/ + +#include +#include +#include +#include +#include "Log.h" +#include "Timestamp.h" +#include "I2PEndian.h" +#include "HTTP.h" +#include "AddressBook.h" +#include "ClientContext.h" +#include "TorrentsTunnel.h" + +namespace i2p +{ +namespace torrents +{ + TorrentsTunnel::TorrentsTunnel (std::string_view name, std::shared_ptr localDestination, + std::string_view torrentsDir, std::string_view trackers): + i2p::client::I2PService (localDestination), m_Name (name), m_PeerID ("-I2PD-"), + m_TorrentsDir (torrentsDir), m_TrackerRequestsCheckTimer (GetService ()), + m_KeepAliveCheckTimer (GetService ()), m_ReconnectCheckTimer (GetService ()), + m_TorrentsStatusUpdateTimer (GetService ()) + { + if (localDestination) + m_PeerID += localDestination->GetIdentHash ().ToBase64 (); + m_PeerID.resize (20, '0'); + if (!trackers.empty ()) + boost::split(m_Trackers, trackers, boost::is_any_of(","), boost::token_compress_on); + } + + void TorrentsTunnel::Start () + { + i2p::client::I2PService::Start (); + m_DiskIOService.Start (); + + auto dgramDest = GetLocalDestination ()->CreateDatagramDestination (false, i2p::datagram::eDatagramV3); + if (dgramDest) + dgramDest->SetRawReceiver (std::bind (&TorrentsTunnel::HandleRecvFromI2PRaw, + std::static_pointer_cast(shared_from_this ()), + std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4)); + + Accept (); + + if (!m_TorrentsDir.empty() && std::filesystem::exists (m_TorrentsDir) && + std::filesystem::is_directory (m_TorrentsDir)) + { + for (const auto& it: std::filesystem::directory_iterator (m_TorrentsDir)) + if (std::filesystem::is_regular_file (it.status()) && it.path ().extension () == ".torrent") + ReadTorrentFile (it.path ()); + } + + ScheduleTrackerRequestsCheck (); + ScheduleKeepAliveCheck (); + ScheduleStatusUpdate (); + } + + void TorrentsTunnel::Stop () + { + auto localDestination = GetLocalDestination (); + if (localDestination) + { + localDestination->StopAcceptingStreams (); + auto dgramDest = localDestination->GetDatagramDestination (); + if (dgramDest) + dgramDest->ResetRawReceiver (); + } + m_TrackerRequestsCheckTimer.cancel (); + m_KeepAliveCheckTimer.cancel (); + m_ReconnectCheckTimer.cancel (); + m_TorrentsStatusUpdateTimer.cancel (); + m_Torrents.clear (); + for (auto it: m_Torrents) + { + auto fullPath = it.second->GetFullPath (); fullPath += ".resume"; + boost::asio::post (m_DiskIOService.GetService (), [torrent = it.second, fullPath]() + { + torrent->SaveTorrentResumeFile (fullPath); + }); + } + m_DiskIOService.Stop (); + i2p::client::I2PService::Stop (); + } + + void TorrentsTunnel::ReadTorrentFile (const std::filesystem::path& torrentFilePath) + { + std::shared_ptr torrent; + std::ifstream s(torrentFilePath, std::ifstream::binary); + if (s) + { + s.seekg (0,std::ios::end); + size_t len = s.tellg (); + if (len > 0) + { + s.seekg(0, std::ios::beg); + char * buf = new char[len]; + s.read(buf, len); + torrent = std::make_shared(std::string_view{buf, len}); + delete[] buf; + } + else + LogPrint (eLogError, "Torrents: Empty file ", torrentFilePath); + } + else + LogPrint (eLogError, "Torrents: Can't open file ", torrentFilePath); + + if (torrent && !torrent->IsValid ()) + { + // a torrent whose parsing stopped, an unsafe name among the rest, must + // not be used even in part: it would leave stray files behind + LogPrint (eLogError, "Torrents: Invalid torrent file ", torrentFilePath, ". Skipped"); + torrent = nullptr; + } + if (torrent) + { + torrent->SetFullPath (m_TorrentsDir/std::filesystem::path (torrent->GetName ())); + InitTorrentFiles (torrent); + InsertTorrent (torrent); + } + } + + void TorrentsTunnel::InitTorrentFiles (std::shared_ptr torrent) + { + if (!torrent) return; + if (torrent->GetFiles ().empty ()) + { + if (std::filesystem::exists (torrent->GetFullPath ())) + torrent->SetComplete (); + else + { + auto partFilePath = torrent->GetFullPath (); partFilePath += ".part"; + if (!std::filesystem::exists (partFilePath)) + CreateAndReserveFile (partFilePath, torrent->GetLength ()); + } + } + else + { + bool completed = true; + for (auto& [filePath, fileLength]: torrent->GetFiles ()) + { + filePath = torrent->GetFullPath ()/filePath; + if (!std::filesystem::exists (filePath)) + { + auto partFilePath = filePath; partFilePath += ".part"; + if (!std::filesystem::exists (partFilePath)) + CreateAndReserveFile (partFilePath, fileLength); + completed = false; + } + } + if (completed) torrent->SetComplete (); + } + auto resumeFilePath = torrent->GetFullPath (); resumeFilePath += ".resume"; + if (std::filesystem::exists (resumeFilePath)) + { + if (!torrent->IsComplete ()) + { + std::ifstream rs(resumeFilePath, std::ifstream::binary); + if (rs) + { + rs.seekg (0,std::ios::end); + size_t l = rs.tellg (); + if (l > 0) + { + rs.seekg(0, std::ios::beg); + std::vector bitfield(l); + rs.read((char *)bitfield.data (), l); + if (torrent->ApplyBitfield (bitfield)) + CompleteTorrent (torrent); + } + } + } + else + std::filesystem::remove (resumeFilePath); + } + } + + bool TorrentsTunnel::CreateAndReserveFile (const std::filesystem::path& filePath, size_t reserve) + { + if (std::filesystem::exists (filePath)) return false; + auto subdirs = filePath.parent_path (); + if (!subdirs.empty ()) + { + // try to create all subdirs + try + { + std::filesystem::create_directories (subdirs); + } + catch (std::exception& ex) + { + LogPrint (eLogError, "Torrents: Can't create subdirs ", subdirs, " : ", ex.what()); + return false; + } + } + // create file + std::ofstream f(filePath, std::ios::binary); + if (!f) return false; + f.close (); + if (reserve > 0) + { + // resize + try + { + std::filesystem::resize_file (filePath, reserve); + } + catch (std::exception& ex) + { + LogPrint (eLogError, "Torrents: Can't resize file ", filePath, " to ", reserve, " : ", ex.what()); + return false; + } + } + return true; + } + + void TorrentsTunnel::CompleteTorrent (std::shared_ptr torrent) + { + boost::asio::post (GetDiskIOService (), [this, torrent]() + { + bool completed = false; + if (torrent->GetFiles ().empty ()) + { + auto partFilePath = torrent->GetFullPath (); partFilePath += ".part"; + std::error_code ec; + std::filesystem::rename (partFilePath, torrent->GetFullPath (), ec); + if (!ec) + completed = true; + else + LogPrint (eLogError, "Torrents: Can't rename ", partFilePath); + } + else + { + completed = true; + for (const auto& [filePath, fileSize]: torrent->GetFiles ()) + { + auto partFilePath = filePath; partFilePath += ".part"; + std::error_code ec; + std::filesystem::rename (partFilePath, filePath, ec); + if (ec) + { + completed = false; + LogPrint (eLogError, "Torrents: Can't rename ", partFilePath); + } + } + } + if (completed) + { + torrent->SetComplete (); + auto resumeFilePath = torrent->GetFullPath (); resumeFilePath += ".resume"; + if (!std::filesystem::remove (resumeFilePath)) + LogPrint (eLogError, "Torrents: Can't delete resume file ", resumeFilePath); + LogPrint (eLogInfo, "Torrents: Download complete ", torrent->GetFullPath ()); + + boost::asio::post (GetService (), [this, torrent]() + { + // inform tracker that we are done + RequestTorrentTrackers (torrent, "completed"); + // close connections with seeds and reset stats for remaining + auto conns = GetTorrentConnections (torrent); + for (auto it: conns) + { + if (it->GetRemoteBitfield ().all ()) // seed + it->Close (); + else + it->ResetStats (); + } + }); + } + }); + } + + std::shared_ptr TorrentsTunnel::FindTorrent (const Torrent::InfoHash& infoHash) const + { + std::lock_guard l(m_TorrentsMutex); + auto it = m_Torrents.find (infoHash); + if (it != m_Torrents.end ()) + return it->second; + return nullptr; + } + + std::shared_ptr TorrentsTunnel::FindTorrentByID (int id) const + { + std::lock_guard l(m_TorrentsMutex); + auto it = m_TorrentsByID.find (id); + if (it != m_TorrentsByID.end ()) + return it->second.lock (); + return nullptr; + } + + std::vector TorrentsTunnel::GetTorrentIDs () const + { + std::vector ids; + std::lock_guard l(m_TorrentsMutex); + for (const auto& it: m_TorrentsByID) + if (!it.second.expired ()) ids.push_back (it.first); + return ids; + } + + std::pair, int> TorrentsTunnel::AddTorrent (std::string_view torrentFileContent) + { + auto torrent = std::make_shared (torrentFileContent); + if (m_Torrents.find (torrent->GetInfoHash ()) == m_Torrents.end ()) + { + torrent->SetFullPath (m_TorrentsDir/std::filesystem::path (torrent->GetName ())); + { + auto torrentFilePath = torrent->GetFullPath (); torrentFilePath += ".torrent"; + std::ofstream f(torrentFilePath, std::ofstream::binary); + if (f) + f.write (torrentFileContent.data (), torrentFileContent.size ()); + else + return { torrent, 0 }; + } + InitTorrentFiles (torrent); + return { torrent, InsertTorrent (torrent) }; + } + return { torrent, 0 }; + } + + int TorrentsTunnel::InsertTorrent (std::shared_ptr torrent) + { + if (!torrent) return 0; + std::lock_guard l(m_TorrentsMutex); + if (m_Torrents.emplace (torrent->GetInfoHash (), torrent).second) + { + int id = 1; + if (!m_TorrentsByID.empty ()) + id = m_TorrentsByID.rbegin ()->first + 1; + m_TorrentsByID.emplace (id, torrent); + return id; + } + return 0; + } + + bool TorrentsTunnel::RemoveTorrent (int id, bool deleteFiles) + { + std::shared_ptr torrent; + { + std::lock_guard l(m_TorrentsMutex); + auto it = m_TorrentsByID.find (id); + if (it == m_TorrentsByID.end ()) return false; + torrent = it->second.lock (); + m_TorrentsByID.erase (it); + if (!torrent) return false; + m_Torrents.erase (torrent->GetInfoHash ()); + } + boost::asio::post (GetService (), [this, torrent, deleteFiles]() + { + RemoveTorrent (torrent, deleteFiles); + }); + return true; + } + + void TorrentsTunnel::RemoveTorrent (std::shared_ptr torrent, bool deleteFiles) + { + if (!torrent) return; + auto connections = GetTorrentConnections (torrent); + // close connections + for (auto it: connections) + it->Close (); + if (deleteFiles) + boost::asio::post (GetDiskIOService (), [torrent]() + { + auto fullPath = torrent->GetFullPath (); + auto torrentFilePath = fullPath; torrentFilePath += ".torrent"; + std::error_code ec; + std::filesystem::remove (torrentFilePath, ec); + if (ec) + LogPrint (eLogError, "Torrents: Can't delete ", torrentFilePath); + auto resumeFilePath = fullPath; resumeFilePath += ".resume"; + if (std::filesystem::exists (resumeFilePath)) + { + std::filesystem::remove (resumeFilePath, ec); + if (ec) + LogPrint (eLogError, "Torrents: Can't delete ", resumeFilePath); + } + if (torrent->IsComplete () || !torrent->GetFiles ().empty ()) + { + std::filesystem::remove_all (fullPath, ec); + if (ec) + LogPrint (eLogError, "Torrents: Can't delete ", fullPath); + } + else + { + auto partFilePath = fullPath; partFilePath += ".part"; + std::filesystem::remove (partFilePath, ec); + if (ec) + LogPrint (eLogError, "Torrents: Can't delete ", partFilePath); + } + }); + } + + void TorrentsTunnel::Accept () + { + auto localDestination = GetLocalDestination (); + if (localDestination) + { + if (!localDestination->IsAcceptingStreams ()) // set it as default if not set yet + localDestination->AcceptStreams ([this](std::shared_ptr stream) + { + if (stream) + { + auto conn = std::make_shared (shared_from_this (), stream); + AddHandler (conn); + conn->ReceiveHandshake (); + } + }); + } + else + LogPrint (eLogError, "Torrents: Local destination not set"); + } + + void TorrentsTunnel::RequestTorrentTrackers (std::shared_ptr torrent, std::string_view event) + { + if (!m_Trackers.empty ()) + for (size_t i = 0; i < m_Trackers.size (); i++) + RequestTracker (i, torrent, event); + else + RequestTracker (0, torrent, event); // from announce + } + + void TorrentsTunnel::RequestTracker (size_t trackerID, std::shared_ptr torrent, std::string_view event) + { + if (!torrent) return; + i2p::http::URL reqURL; + if (trackerID < m_Trackers.size()) + reqURL.parse (m_Trackers[trackerID]); + else + reqURL.parse (torrent->GetAnnounce ()); +#if __cplusplus >= 202002L // C++20 + if (!reqURL.host.ends_with (".i2p")) +#else + if (reqURL.host.find(".i2p") == reqURL.host.npos) +#endif + { + LogPrint (eLogWarning, "Torrents: Non-I2P address ", reqURL.host, " for torrent ", torrent->GetName ()); + return; + } + if (reqURL.schema == "udp") + { + ConnectToDatagramTracker (reqURL.host, reqURL.port); + return; + } + std::map params; + params.emplace ("info_hash", torrent->GetHexStringInfoHash ()); + params.emplace ("peer_id", m_PeerID); + params.emplace ("ip", GetLocalDestination ()->GetIdentity ()->ToBase64 () + ".i2p"); + params.emplace ("port", std::to_string (TORRENT_PORT)); // 6881 + params.emplace ("compact", "1"); + params.emplace ("uploaded", std::to_string (torrent->GetUploaded ())); + params.emplace ("downloaded", std::to_string (torrent->GetLength () - torrent->GetLeft ())); + params.emplace ("left", std::to_string (torrent->GetLeft ())); + params.emplace ("numwant", torrent->IsComplete () ? "0" : "25"); // max num of peers, 0 if seeding + if (!event.empty ()) + params.emplace ("event", event); + reqURL.create_query (params); + + auto req = std::make_shared >(boost::beast::http::verb::get, reqURL.to_string (true), 11); // HTTP 1.1 + req->set (boost::beast::http::field::host, reqURL.host); + req->set (boost::beast::http::field::user_agent, "I2PSocketEepGet"); + req->keep_alive (false); // Connection: close + CreateStream ([this, req, torrent, trackerID](std::shared_ptr stream) + { + if (stream) + { + auto httpStream = std::make_shared(stream); + boost::beast::http::async_write (*httpStream, *req, + std::bind (&TorrentsTunnel::TrackerRequestSent, this, std::placeholders::_1, + std::placeholders::_2, httpStream, torrent, req, trackerID)); + } + }, reqURL.host, reqURL.port); + } + + void TorrentsTunnel::TrackerRequestSent (const boost::beast::error_code& ecode, size_t bytes_transferred, + std::shared_ptr httpStream, std::shared_ptr torrent, + std::shared_ptr > req, size_t trackerID) + { + if (!ecode) + { + // receive + auto buf = std::make_shared (); + auto res = std::make_shared >(); + boost::beast::http::async_read (*httpStream, *buf, *res, + [this, httpStream, torrent, buf, res, trackerID](const boost::beast::error_code& ecode, size_t bytes_transferred) + { + httpStream->GetStream ()->AsyncClose (); + if (!ecode) + { + if (res->result () == boost::beast::http::status::ok) + { + torrent->ParseTrackerResponse (trackerID, res->body ()); + ConnectToPeers (torrent); + ScheduleReconnectCheck (); + } + else + LogPrint (eLogWarning, "Torrents: Tracker ", trackerID, " response code ", res->result_int()); + } + }); + } + } + + void TorrentsTunnel::ConnectToPeer (std::shared_ptr torrent, const i2p::data::IdentHash& peer) + { + if (!torrent) return; + LogPrint (eLogDebug, "Torrents: Connecting to peer ", peer.ToBase32 () + ".b32.i2p"); + if (peer == GetLocalDestination ()->GetIdentHash ()) + { + LogPrint (eLogInfo, "Torrents: Can't connect to self"); + return; + } + CreateStream ([this, torrent, peer](std::shared_ptr stream) + { + if (stream) + { + LogPrint (eLogDebug, "Torrents: Connected to peer ", peer.ToBase32 () + ".b32.i2p"); + auto connection = std::make_shared(shared_from_this (), stream, torrent); + AddHandler (connection); + connection->Connect (); + } + else + LogPrint (eLogInfo, "Torrents: Can't connect to peer ", peer.ToBase32 () + ".b32.i2p"); + }, std::make_shared(peer), TORRENT_PORT); + } + + size_t TorrentsTunnel::ConnectToPeers (std::shared_ptr torrent) + { + if (!torrent) return 0; + auto peersToConnect = GetNonConnectedPeers (torrent); + if (!peersToConnect.empty ()) + { + for (const auto& it: peersToConnect) + ConnectToPeer (torrent, it); + } + return peersToConnect.size (); + } + + void TorrentsTunnel::ScheduleTrackerRequestsCheck () + { + m_TrackerRequestsCheckTimer.expires_after (std::chrono::milliseconds(TRACKER_REQUESTS_CHECK_TIMEOUT)); + m_TrackerRequestsCheckTimer.async_wait (std::bind (&TorrentsTunnel::HandleTrackerRequestsCheckTimer, + this, std::placeholders::_1)); + } + + void TorrentsTunnel::HandleTrackerRequestsCheckTimer (const boost::system::error_code& ecode) + { + if (ecode != boost::asio::error::operation_aborted) + { + auto ts = i2p::util::GetMonotonicMilliseconds (); + for (auto it: m_Torrents) + for (size_t i = 0; i < m_Trackers.size (); i++) + if (ts > it.second->GetNextTrackerRequestTime (i)) + { + auto nextInterval = it.second->GetInterval (i) + GetLocalDestination ()->GetRng()() % TRACKER_REQUESTS_INTERVAL_VARIANCE; + it.second->SetNextTrackerRequestTime (i, ts + nextInterval); + RequestTracker (i, it.second); + } + ScheduleTrackerRequestsCheck (); + } + } + + void TorrentsTunnel::ScheduleKeepAliveCheck () + { + m_KeepAliveCheckTimer.expires_after (std::chrono::seconds(PEER_KEEP_ALIVE_CHECK_INTERVAL)); + m_KeepAliveCheckTimer.async_wait (std::bind (&TorrentsTunnel::HandleKeepAliveCheckTimer, + this, std::placeholders::_1)); + } + + void TorrentsTunnel::HandleKeepAliveCheckTimer (const boost::system::error_code& ecode) + { + if (ecode != boost::asio::error::operation_aborted) + { + auto ts = i2p::util::GetMonotonicSeconds (); + IterateHandlers ([ts](std::shared_ptr handler) + { + if (handler) + std::static_pointer_cast(handler)->CheckKeepAlive (ts); + }); + ScheduleKeepAliveCheck (); + } + } + + void TorrentsTunnel::ScheduleReconnectCheck () + { + m_ReconnectCheckTimer.cancel (); + m_ReconnectCheckTimer.expires_after (std::chrono::seconds(RECONNECT_CHECK_INTERVAL)); + m_ReconnectCheckTimer.async_wait (std::bind (&TorrentsTunnel::HandleReconnectCheckTimer, + this, std::placeholders::_1)); + } + + void TorrentsTunnel::HandleReconnectCheckTimer (const boost::system::error_code& ecode) + { + if (ecode != boost::asio::error::operation_aborted) + { + for (auto it: m_Torrents) + { + if (!it.second->IsComplete ()) + { + auto numPeers = ConnectToPeers (it.second); + if (numPeers) + LogPrint (eLogDebug, "Torrents: Reconnecting to ", numPeers, " peers"); + } + } + ScheduleReconnectCheck (); + } + } + + void TorrentsTunnel::ScheduleStatusUpdate () + { + m_TorrentsStatusUpdateTimer.cancel (); + m_TorrentsStatusUpdateTimer.expires_after (std::chrono::seconds(TORRENTS_STATUS_UPDATE_INTERVAL)); + m_TorrentsStatusUpdateTimer.async_wait (std::bind (&TorrentsTunnel::HandleTorrentsStatusUpdateTimer, + this, std::placeholders::_1)); + } + + void TorrentsTunnel::HandleTorrentsStatusUpdateTimer (const boost::system::error_code& ecode) + { + if (ecode != boost::asio::error::operation_aborted) + { + auto ts = i2p::util::GetMonotonicSeconds (); + for (auto it: m_Torrents) + { + if (!it.second->IsComplete ()) + { + if (it.second->UpdateStatus (ts)) + CompleteTorrent (it.second); + else + UpdatePeersPerPiece (it.second); + } + } + UpdateStats (); + ScheduleStatusUpdate (); + } + } + + std::list > TorrentsTunnel::GetTorrentConnections (std::shared_ptr torrent) + { + std::list > ret; + if (torrent) + { + IterateHandlers ([&ret, torrent](std::shared_ptr handler) + { + if (handler) + { + auto conn = std::static_pointer_cast(handler); + if (conn->GetTorrent () == torrent && conn->GetStream ()) + ret.emplace_back (conn); + } + }); + } + return ret; + } + + std::unordered_set TorrentsTunnel::GetNonConnectedPeers (std::shared_ptr torrent) + { + std::unordered_set ret; + if (torrent) + { + ret = torrent->GetPeers (); + if(!ret.empty ()) + { + IterateHandlers ([&ret, torrent](std::shared_ptr handler) + { + if (handler) + { + auto conn = std::static_pointer_cast(handler); + if (conn->GetTorrent () == torrent && conn->GetStream ()) + { + auto ident = conn->GetStream ()->GetRemoteIdentity (); + if (ident) + ret.erase (ident->GetIdentHash ()); + } + } + }); + } + } + return ret; + } + + void TorrentsTunnel::UpdatePeersPerPiece (std::shared_ptr torrent) + { + if (!torrent) return; + torrent->StartCountingPeers (); + IterateHandlers ([torrent](std::shared_ptr handler) + { + if (handler) + { + auto conn = std::static_pointer_cast(handler); + if (conn->GetTorrent () == torrent) + torrent->ApplyPeerRemoteBitfield (conn->GetRemoteBitfield ()); + } + }); + } + + void TorrentsTunnel::UpdateStats () + { + for (auto it: m_Torrents) + it.second->ResetStats (); + IterateHandlers ([](std::shared_ptr handler) mutable + { + if (handler) + { + auto conn = std::static_pointer_cast(handler); + auto torrent = conn->GetTorrent (); + if (torrent) + { + torrent->SetDownloadRate (torrent->GetDownloadRate () + conn->GetDownloadRate ()); + torrent->SetUploadRate (torrent->GetUploadRate () + conn->GetUploadRate ()); + if (conn->IsDownloading ()) + torrent->SetNumDownloadingFromPeers (torrent->GetNumDownloadingFromPeers () + 1); + if (conn->IsUploading ()) + torrent->SetNumUploadingToPeers (torrent->GetNumUploadingToPeers () + 1); + } + } + }); + } + + void TorrentsTunnel::HandleRecvFromI2PRaw (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len) + { + // response from tracker + if (len < 8) return; + uint32_t action = bufbe32toh (buf); + switch (action) + { + case eDatagramTrackerActionConnect: + LogPrint (eLogDebug, "Torrents: action connect"); + break; + case eDatagramTrackerActionAnnounce: + LogPrint (eLogDebug, "Torrents: action announce"); + break; + case eDatagramTrackerActionError: + LogPrint (eLogDebug, "Torrents: action error"); + break; + default: + LogPrint (eLogInfo, "Torrents: Unexpected action ", action, " from tracker"); + } + } + + void TorrentsTunnel::ConnectToDatagramTracker (std::string_view dest, uint16_t port) + { + LogPrint (eLogDebug, "Torrents: Connecting to datagram tracker ", dest, ":", port); + auto address = i2p::client::context.GetAddressBook ().GetAddress (dest); + if (address && address->IsIdentHash ()) + { + auto localDestination = GetLocalDestination (); + auto dgramDest = localDestination->GetDatagramDestination (); + if (dgramDest) + { + uint8_t connectRequest[16]; + htobe64buf (connectRequest, 0x41727101980); // protocol_id + htobe32buf (connectRequest + 8, eDatagramTrackerActionConnect); // action + htobe32buf (connectRequest + 12, localDestination->GetRng()()); // transactionID + uint16_t fromPort = localDestination->GetRng()() % 1000 + 6000; + auto session = dgramDest->GetSession (address->identHash); + if (session) + { + session->SetVersion (i2p::datagram::eDatagramV2); // send datagram2 + dgramDest->SendDatagram (session, connectRequest, 16, fromPort, port); + } + else + LogPrint (eLogInfo, "Torrents: Can't obtain datagram session to ", dest); + } + else + LogPrint (eLogError, "Torrents: Datagram destination is not avaliable"); + } + else + LogPrint (eLogInfo, "Torrents: Tracker not found: ", dest); + } +} +} diff --git a/libi2pd_client/TorrentsTunnel.h b/libi2pd_client/TorrentsTunnel.h new file mode 100644 index 00000000..54a3b426 --- /dev/null +++ b/libi2pd_client/TorrentsTunnel.h @@ -0,0 +1,126 @@ +/* +* 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 TORRENTS_TUNNEL_H__ +#define TORRENTS_TUNNEL_H__ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "I2PService.h" +#include "util.h" +#include "BoostStream.h" +#include "Torrents.h" + +namespace i2p +{ +namespace torrents +{ + enum DatagramTrackerAction + { + eDatagramTrackerActionConnect = 0, + eDatagramTrackerActionAnnounce = 1, + eDatagramTrackerActionError = 3 + }; + + class TorrentsTunnel final: public i2p::client::I2PService + { + private: + + class DiskIOService: private i2p::util::RunnableServiceWithWork + { + public: + + DiskIOService (): RunnableServiceWithWork ("TDiskIO") {} + auto& GetService () { return GetIOService (); } + void Start () { StartIOService (); } + void Stop () { StopWorkAndFinishTasks (); } + }; + + public: + + TorrentsTunnel (std::string_view name, std::shared_ptr localDestination, + std::string_view torrentsDir, std::string_view trackers = ""); + + void Start () override; + void Stop () override; + auto& GetDiskIOService () { return m_DiskIOService.GetService (); }; + + const std::string& GetPeerID () const { return m_PeerID; } + const std::vector& GetTrackers () const { return m_Trackers; } + std::shared_ptr FindTorrent (const Torrent::InfoHash& infoHash) const; + std::shared_ptr FindTorrentByID (int id) const; + std::vector GetTorrentIDs () const; + std::pair, int> AddTorrent (std::string_view torrentFileContent); // (tunnel, id) + bool RemoveTorrent (int id, bool deleteFiles); + std::list > GetTorrentConnections (std::shared_ptr torrent); + + const char* GetName() const override { return m_Name.c_str (); } + + private: + + + void Accept (); + void ReadTorrentFile (const std::filesystem::path& torrentFilePath); + void InitTorrentFiles (std::shared_ptr torrent); + int InsertTorrent (std::shared_ptr torrent); // returns id > 0 if success and 0 if failed + void RemoveTorrent (std::shared_ptr torrent, bool deleteFiles); + bool CreateAndReserveFile (const std::filesystem::path& filePath, size_t reserve); + void CompleteTorrent (std::shared_ptr torrent); + void RequestTracker (size_t trackerID, std::shared_ptr torrent, std::string_view event = ""); + void RequestTorrentTrackers (std::shared_ptr torrent, std::string_view event = ""); + void TrackerRequestSent (const boost::beast::error_code& ecode, size_t bytes_transferred, + std::shared_ptr httpStream, std::shared_ptr torrent, + std::shared_ptr > req, size_t trackerID); + + void ScheduleTrackerRequestsCheck (); + void HandleTrackerRequestsCheckTimer (const boost::system::error_code& ecode); + + void ScheduleKeepAliveCheck (); + void HandleKeepAliveCheckTimer (const boost::system::error_code& ecode); + + void ScheduleReconnectCheck (); + void HandleReconnectCheckTimer (const boost::system::error_code& ecode); + + void ScheduleStatusUpdate (); + void HandleTorrentsStatusUpdateTimer (const boost::system::error_code& ecode); + + std::unordered_set GetNonConnectedPeers (std::shared_ptr torrent); + void ConnectToPeer (std::shared_ptr torrent, const i2p::data::IdentHash& peer); + size_t ConnectToPeers (std::shared_ptr torrent); + void UpdatePeersPerPiece (std::shared_ptr torrent); + void UpdateStats (); + + void HandleRecvFromI2PRaw (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len); + void ConnectToDatagramTracker (std::string_view dest, uint16_t port); + + private: + + std::string m_Name, m_PeerID; // 20 characters + std::filesystem::path m_TorrentsDir; + std::vector m_Trackers; + std::map > m_Torrents; + std::map > m_TorrentsByID; + mutable std::mutex m_TorrentsMutex; + boost::asio::steady_timer m_TrackerRequestsCheckTimer, m_KeepAliveCheckTimer, + m_ReconnectCheckTimer, m_TorrentsStatusUpdateTimer; + DiskIOService m_DiskIOService; + }; + +} +} + +#endif