From 262f183126cb14fc744448d5008ecce4e167e12e Mon Sep 17 00:00:00 2001 From: orignal Date: Wed, 5 Aug 2026 19:01:36 -0400 Subject: [PATCH] one connection to remote I2P address per torrent. Close connection if no data or keepalive --- libi2pd_client/Torrents.cpp | 51 ++++++++++++++++++++++++------------- libi2pd_client/Torrents.h | 13 +++++----- 2 files changed, 41 insertions(+), 23 deletions(-) diff --git a/libi2pd_client/Torrents.cpp b/libi2pd_client/Torrents.cpp index 2b3ef0f5..0f1eb346 100644 --- a/libi2pd_client/Torrents.cpp +++ b/libi2pd_client/Torrents.cpp @@ -360,7 +360,7 @@ namespace torrents auto [hashes, len] = ExtractByteString (buf); while (!hashes.empty ()) { - m_Peers.emplace_back (i2p::data::IdentHash ((const uint8_t *)hashes.substr (0, i2p::data::IdentHash::len).data ())); + m_Peers.emplace (i2p::data::IdentHash ((const uint8_t *)hashes.substr (0, i2p::data::IdentHash::len).data ())); hashes = hashes.substr (i2p::data::IdentHash::len); } return len; @@ -517,12 +517,20 @@ namespace torrents void PeerConnection::CheckKeepAlive (uint64_t ts) { - if (m_IsEstablished && ts > m_LastSendTime) + if (m_IsEstablished) { - // send keep-alive - uint32_t len = 0; - WriteToStream ((const uint8_t *)&len, 4); - m_LastSendTime = ts; + if (ts > m_LastReceiveTime + PEER_KEEP_ALIVE_TIMEOUT) + { + LogPrint (eLogInfo, "Torrent: Peer timeout expired"); + Terminate (); + } + else if (ts > m_LastSendTime + PEER_KEEP_SEND_INTERVAL) + { + // send keep-alive + uint32_t len = 0; + WriteToStream ((const uint8_t *)&len, 4); + m_LastSendTime = ts; + } } } @@ -1115,29 +1123,30 @@ namespace torrents { if (!torrent) return; LogPrint (eLogDebug, "Torrents: Connecting to peer ", peer.ToBase32 () + ".b32.i2p"); - CreateStream ([this, torrent](std::shared_ptr stream) + CreateStream ([this, torrent, peer](std::shared_ptr stream) { if (stream) { - LogPrint (eLogDebug, "Torrents: Connected to peer"); + LogPrint (eLogDebug, "Torrents: Connected to peer ", peer.ToBase32 () + ".b32.i2p"); auto connection = std::make_shared(this, stream, torrent); AddHandler (connection); connection->Connect (); } else - LogPrint (eLogInfo, "Torrents: Can't connect to peer"); + LogPrint (eLogInfo, "Torrents: Can't connect to peer ", peer.ToBase32 () + ".b32.i2p"); }, std::make_shared(peer), TORRENT_PORT); } - void TorrentsTunnel::ConnectToPeers (std::shared_ptr torrent) + size_t TorrentsTunnel::ConnectToPeers (std::shared_ptr torrent) { - if (!torrent) return; + if (!torrent) return 0; auto peersToConnect = GetNonConnectedPeers (torrent); if (!peersToConnect.empty ()) { for (const auto& it: peersToConnect) ConnectToPeer (torrent, it); } + return peersToConnect.size (); } std::string TorrentsTunnel::GetTorrentFilePath (const std::string& filename) const @@ -1172,7 +1181,7 @@ namespace torrents void TorrentsTunnel::ScheduleKeepAliveCheck () { - m_KeepAliveCheckTimer.expires_after (std::chrono::seconds(PEER_KEEP_ALIVE_CHECK_TIMEOUT)); + m_KeepAliveCheckTimer.expires_after (std::chrono::seconds(PEER_KEEP_ALIVE_CHECK_INTERVAL)); m_KeepAliveCheckTimer.async_wait (std::bind (&TorrentsTunnel::HandleKeepAliveCheckTimer, this, std::placeholders::_1)); } @@ -1204,7 +1213,11 @@ namespace torrents if (ecode != boost::asio::error::operation_aborted) { for (auto it: m_Torrents) - ConnectToPeers (it.second); + { + auto numPeers = ConnectToPeers (it.second); + if (numPeers) + LogPrint (eLogDebug, "Torrents: Reconnecting to ", numPeers, " peers"); + } ScheduleReconnectCheck (); } } @@ -1223,7 +1236,11 @@ namespace torrents auto ident = conn->GetStream ()->GetRemoteIdentity (); if (ident) { - if (std::find (peers.begin (), peers.end (), ident->GetIdentHash ()) != peers.end ()) +#if __cplusplus >= 202002L // C++20 + if (peers.contains (ident->GetIdentHash ())) +#else + if (peers.count (ident->GetIdentHash ()) > 0) +#endif ret.emplace_back (conn); } } @@ -1232,9 +1249,9 @@ namespace torrents return ret; } - std::list TorrentsTunnel::GetNonConnectedPeers (std::shared_ptr torrent) + std::unordered_set TorrentsTunnel::GetNonConnectedPeers (std::shared_ptr torrent) { - std::list ret; + std::unordered_set ret; if (torrent) { ret = torrent->GetPeers (); @@ -1247,7 +1264,7 @@ namespace torrents auto conn = std::static_pointer_cast(handler); auto ident = conn->GetStream ()->GetRemoteIdentity (); if (ident) - (void)std::remove (ret.begin (), ret.end (), ident->GetIdentHash ()); + ret.erase (ident->GetIdentHash ()); } }); } diff --git a/libi2pd_client/Torrents.h b/libi2pd_client/Torrents.h index 62a9380a..18048ab5 100644 --- a/libi2pd_client/Torrents.h +++ b/libi2pd_client/Torrents.h @@ -43,8 +43,9 @@ namespace torrents 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_INTERVAL = 120; // in seconds - constexpr int PEER_KEEP_ALIVE_CHECK_TIMEOUT = 15; // 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 = 8; constexpr size_t HANDSHAKE_MSG_LENGTH = 68; @@ -127,7 +128,7 @@ namespace torrents Piece& GetPiece (int index) { return m_Pieces[index]; } std::pair, bool> CreateBitfield () const; // (bitfield, empty) void ApplyBitfield (const std::vector& bitfield); - const std::list& GetPeers () const { return m_Peers; } + const std::unordered_set& GetPeers () const { return m_Peers; } std::tuple GetNextBlockToRequest (std::shared_ptr conn); // return (index, offset, len) void ClearAllRequests (); void Complete (); @@ -151,7 +152,7 @@ namespace torrents uint64_t m_NextTrackerRequestTime; // monotonic millicesonds InfoHash m_InfoHash; // SHA1 std::vector m_Pieces; - std::list m_Peers; + std::unordered_set m_Peers; }; class TorrentsTunnel; @@ -261,9 +262,9 @@ namespace torrents void ScheduleReconnectCheck (); void HandleReconnectCheckTimer (const boost::system::error_code& ecode); - std::list GetNonConnectedPeers (std::shared_ptr torrent); + std::unordered_set GetNonConnectedPeers (std::shared_ptr torrent); void ConnectToPeer (std::shared_ptr torrent, const i2p::data::IdentHash& peer); - void ConnectToPeers (std::shared_ptr torrent); + size_t ConnectToPeers (std::shared_ptr torrent); private: