diff --git a/libi2pd_client/Torrents.cpp b/libi2pd_client/Torrents.cpp index 32627599..e859af35 100644 --- a/libi2pd_client/Torrents.cpp +++ b/libi2pd_client/Torrents.cpp @@ -530,6 +530,24 @@ namespace torrents return len; } + void Torrent::HandleDatagramTrackerResponse (size_t trackerID, uint32_t interval, const uint8_t * hashes, size_t hashesLen) + { + if (trackerID >= m_TrackersInfo.size ()) + m_TrackersInfo.resize (trackerID + 1, TrackerInfo{{}, MIN_TRACKER_REQUESTS_INTERVAL, 0}); + auto& [peers, trackerRequestInterval, nextRequestTime] = m_TrackersInfo[trackerID]; + trackerRequestInterval = interval*1000; // milliseconds + nextRequestTime = i2p::util::GetMonotonicMilliseconds () + trackerRequestInterval; + peers.clear (); + size_t offset = 0; + while (offset + i2p::data::IdentHash::len <= hashesLen) + { + i2p::data::IdentHash ident (hashes + offset); + if (ident.IsZero ()) break; + peers.emplace (std::move (ident)); + offset += i2p::data::IdentHash::len; + } + } + std::pair, bool> Torrent::CreateBitfield () const { size_t numPieces = m_Pieces.size (); diff --git a/libi2pd_client/Torrents.h b/libi2pd_client/Torrents.h index a151c3ed..ee02d37c 100644 --- a/libi2pd_client/Torrents.h +++ b/libi2pd_client/Torrents.h @@ -33,20 +33,14 @@ namespace torrents { constexpr size_t REQUEST_BLOCK_SIZE = 16384; constexpr uint16_t TORRENT_PORT = 6881; // not used by required by protocol - constexpr int TRACKER_RESPONSE_TIMEOUT = 8; // in seconds - constexpr int TRACKER_REQUESTS_CHECK_TIMEOUT = 1900; // in milliseconds - constexpr int RECONNECT_CHECK_INTERVAL = 70; // in seconds constexpr int MIN_TRACKER_REQUESTS_INTERVAL = 15000; // in milliseconds - constexpr int TRACKER_REQUESTS_INTERVAL_VARIANCE = 3000; // in milliseconds constexpr size_t PEER_CONNECTION_RECEIVE_BUFFER_SIZE = 65535; constexpr int PEER_CONNECTION_MAX_IDLE = 3600; // in seconds constexpr int PEER_KEEP_ALIVE_TIMEOUT = 120; // in seconds constexpr int PEER_KEEP_SEND_INTERVAL = 95; // in seconds - constexpr int PEER_KEEP_ALIVE_CHECK_INTERVAL = 15; // in seconds constexpr size_t MAX_NUM_REQUESTS = 12; constexpr size_t MAX_NUM_PIECES = 6; constexpr int PIECE_INACTIVITY_TIMEOUT = 60; // in seconds - constexpr int TORRENTS_STATUS_UPDATE_INTERVAL = 25; // in seconds constexpr int HANDSHAKE_RECEIVE_TIMEOUT = 20; // in seconds constexpr int BANDWIDTH_RATE_SAMPLING_INTERVAL = 20; // in milliseconds @@ -159,6 +153,7 @@ namespace torrents Torrent (std::string_view buf); void ParseTrackerResponse (size_t trackerID, std::string_view buf); + void HandleDatagramTrackerResponse (size_t trackerID, uint32_t interval, const uint8_t * hashes, size_t hashesLen); bool IsComplete () const { return m_IsComplete; } void SetComplete (); diff --git a/libi2pd_client/TorrentsTunnel.cpp b/libi2pd_client/TorrentsTunnel.cpp index 90e6109d..7d5e8284 100644 --- a/libi2pd_client/TorrentsTunnel.cpp +++ b/libi2pd_client/TorrentsTunnel.cpp @@ -442,7 +442,7 @@ namespace torrents } if (reqURL.schema == "udp") { - ConnectToDatagramTracker (torrent, reqURL.host, reqURL.port); + ConnectToDatagramTracker (torrent, trackerID, reqURL.host, reqURL.port); return; } std::map params; @@ -550,6 +550,18 @@ namespace torrents if (ecode != boost::asio::error::operation_aborted) { auto ts = i2p::util::GetMonotonicMilliseconds (); + if (!m_DatragramTrackerTransactions.empty ()) + { + // cleanup expired transactions + auto it = m_DatragramTrackerTransactions.begin (); + while (it != m_DatragramTrackerTransactions.end ()) + { + if (ts > (std::get<5>(it->second) + DATAGRAM_TRACKER_TRANSACTION_TIMEOUT)*1000LL) + it = m_DatragramTrackerTransactions.erase (it); + else + it++; + } + } for (auto it: m_Torrents) for (size_t i = 0; i < m_Trackers.size (); i++) if (ts > it.second->GetNextTrackerRequestTime (i)) @@ -729,7 +741,7 @@ namespace torrents HandleConnectResponse (buf + 4, len - 4); break; case eDatagramTrackerActionAnnounce: - LogPrint (eLogDebug, "TorrentsTunnel: action announce"); + HandleAnnounceResponse (buf + 4, len - 4); break; case eDatagramTrackerActionError: HandleErrorResponse (buf + 4, len - 4); @@ -741,13 +753,14 @@ namespace torrents void TorrentsTunnel::HandleErrorResponse (const uint8_t * buf, size_t len) { - LogPrint (eLogDebug, "TorrentsTunnel: action error response"); uint32_t transactionID = bufbe32toh (buf); + LogPrint (eLogDebug, "TorrentsTunnel: Datagram tracker action error response ", transactionID); m_DatragramTrackerTransactions.erase (transactionID); LogPrint (eLogInfo, "TorrentsTunnel: Datagram tracker error response: ", std::string_view ((const char *)(buf + 4), len - 4)); } - void TorrentsTunnel::ConnectToDatagramTracker (std::shared_ptr torrent, std::string_view dest, uint16_t port) + void TorrentsTunnel::ConnectToDatagramTracker (std::shared_ptr torrent, + size_t trackerID, std::string_view dest, uint16_t port) { LogPrint (eLogDebug, "TorrentsTunnel: Connecting to datagram tracker ", dest, ":", port); auto address = i2p::client::context.GetAddressBook ().GetAddress (dest); @@ -766,7 +779,8 @@ namespace torrents auto session = dgramDest->GetSession (address->identHash); if (session) { - m_DatragramTrackerTransactions.emplace (transactionID, std::make_tuple (torrent, address->identHash, port, fromPort, 0)); + m_DatragramTrackerTransactions.emplace (transactionID, std::make_tuple (torrent, + trackerID, address->identHash, port, fromPort, i2p::util::GetMonotonicSeconds ())); session->SetVersion (i2p::datagram::eDatagramV2); // send datagram2 dgramDest->SendDatagram (session, connectRequest, 16, fromPort, port); } @@ -782,13 +796,13 @@ namespace torrents void TorrentsTunnel::HandleConnectResponse (const uint8_t * buf, size_t len) { - LogPrint (eLogDebug, "TorrentsTunnel: action connect response"); if (len < 12) { LogPrint (eLogInfo, "TorrentsTunnel: Unexpected connect response length ", len + 4); return; } uint32_t transactionID = bufbe32toh (buf); + LogPrint (eLogDebug, "TorrentsTunnel: Datagram tracker action connect response ", transactionID); auto it = m_DatragramTrackerTransactions.find (transactionID); if (it == m_DatragramTrackerTransactions.end ()) { @@ -796,7 +810,7 @@ namespace torrents return; } uint64_t connectionID = bufbe64toh (buf + 4); - auto [torrent, ident, port, fromPort, ts] = it->second; + auto [torrent, trackerID, ident, port, fromPort, ts] = it->second; if (!torrent.expired ()) SendAnnounceToDatagramTracker (transactionID, connectionID, torrent.lock (), ident, port, fromPort); else @@ -835,5 +849,29 @@ namespace torrents LogPrint (eLogInfo, "TorrentsTunnel: Can't obtain datagram session"); } } + + void TorrentsTunnel::HandleAnnounceResponse (const uint8_t * buf, size_t len) + { + if (len < 16) + { + LogPrint (eLogInfo, "TorrentsTunnel: Unexpected announce response length ", len + 4); + return; + } + uint32_t transactionID = bufbe32toh (buf); + LogPrint (eLogDebug, "TorrentsTunnel: Datagram tracker action announce response ", transactionID); + auto it = m_DatragramTrackerTransactions.find (transactionID); + if (it == m_DatragramTrackerTransactions.end ()) + { + LogPrint (eLogInfo, "TorrentsTunnel: Datagram tracker transaction ", transactionID, " not found"); + return; + } + auto [torrent, trackerID, ident, port, fromPort, ts] = it->second; + if (!torrent.expired ()) + { + uint32_t interval = bufbe32toh (buf + 4); + torrent.lock ()->HandleDatagramTrackerResponse (trackerID, interval, buf + 16, len - 16); + } + m_DatragramTrackerTransactions.erase (it); + } } } diff --git a/libi2pd_client/TorrentsTunnel.h b/libi2pd_client/TorrentsTunnel.h index 6ec744aa..6957809c 100644 --- a/libi2pd_client/TorrentsTunnel.h +++ b/libi2pd_client/TorrentsTunnel.h @@ -30,6 +30,14 @@ namespace i2p { namespace torrents { + constexpr int TRACKER_RESPONSE_TIMEOUT = 8; // in seconds + constexpr int DATAGRAM_TRACKER_TRANSACTION_TIMEOUT = 60; // in seconds + constexpr int TRACKER_REQUESTS_CHECK_TIMEOUT = 1900; // in milliseconds + constexpr int RECONNECT_CHECK_INTERVAL = 70; // in seconds + constexpr int TRACKER_REQUESTS_INTERVAL_VARIANCE = 3000; // in milliseconds + constexpr int PEER_KEEP_ALIVE_CHECK_INTERVAL = 15; // in seconds + constexpr int TORRENTS_STATUS_UPDATE_INTERVAL = 25; // in seconds + enum DatagramTrackerAction { eDatagramTrackerActionConnect = 0, @@ -106,9 +114,10 @@ namespace torrents void UpdateStats (); void HandleRecvFromI2PRaw (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len); - void ConnectToDatagramTracker (std::shared_ptr torrent, std::string_view dest, uint16_t port); + void ConnectToDatagramTracker (std::shared_ptr torrent, size_t trackerID, std::string_view dest, uint16_t port); void HandleConnectResponse (const uint8_t * buf, size_t len); void HandleErrorResponse (const uint8_t * buf, size_t len); + void HandleAnnounceResponse (const uint8_t * buf, size_t len); void SendAnnounceToDatagramTracker (uint32_t transactionID, uint64_t connectionID, std::shared_ptr torrent, const i2p::data::IdentHash& ident, uint16_t port, uint16_t fromPort); @@ -123,8 +132,8 @@ namespace torrents boost::asio::steady_timer m_TrackerRequestsCheckTimer, m_KeepAliveCheckTimer, m_ReconnectCheckTimer, m_TorrentsStatusUpdateTimer; DiskIOService m_DiskIOService; - std::unordered_map, i2p::data::IdentHash, uint16_t, uint16_t, uint64_t> > - m_DatragramTrackerTransactions; // transactionID->(torrent, ident, port, fromPort, ts in monotonic seconds) + std::unordered_map, size_t, i2p::data::IdentHash, uint16_t, uint16_t, uint64_t> > + m_DatragramTrackerTransactions; // transactionID->(torrent, trackerID, ident, port, fromPort, ts in monotonic seconds) }; }