Merge pull request #1122 from ssiloti/peers-storage-policy

new peers storage policy

include/libtorrent/kademlia/dht_storage.hpp

@@ -121,11 +121,11 @@ namespace dht
 		// consider the value of dht_settings::max_peers_reply.
 		// If noseed is true only peers marked as no seed should be included.
 		//
-		// returns true if an entry with the info_hash is found and
-		// the data is returned inside the (entry) out parameter peers.
+		// returns true if the maximum number of peers are stored
+		// for this info_hash.
 		//
 		virtual bool get_peers(sha1_hash const& info_hash, udp protocol
-			, bool noseed, bool scrape
+			, bool noseed, bool scrape, address const& requester
 			, entry& peers) const = 0;
 
 		// This function is named announce_peer for consistency with the

include/libtorrent/kademlia/node.hpp

@@ -219,10 +219,8 @@ private:
 
 	void send_single_refresh(udp::endpoint const& ep, int bucket
 		, node_id const& id = node_id());
-	void lookup_peers(sha1_hash const& info_hash, entry& reply
-		, bool noseed, bool scrape) const;
-	bool lookup_torrents(sha1_hash const& target, entry& reply
-		, char* tags) const;
+	bool lookup_peers(sha1_hash const& info_hash, entry& reply
+		, bool noseed, bool scrape, address const& requester) const;
 
 	libtorrent::dht_settings const& m_settings;
 

src/kademlia/dht_storage.cpp

@@ -186,12 +186,11 @@ namespace
 		}
 
 		bool get_peers(sha1_hash const& info_hash, udp const protocol
-			, bool const noseed, bool const scrape
+			, bool const noseed, bool const scrape, address const& requester
 			, entry& peers) const override
 		{
-			auto const i = m_map.lower_bound(info_hash);
-			if (i == m_map.end()) return false;
-			if (i->first != info_hash) return false;
+			auto const i = m_map.find(info_hash);
+			if (i == m_map.end()) return int(m_map.size()) >= m_settings.max_torrents;
 
 			torrent_entry const& v = i->second;
 
@@ -257,7 +256,19 @@ namespace
 					++m;
 				}
 			}
-			return true;
+
+			if (int(i->second.peers.size()) < m_settings.max_peers)
+				return false;
+
+			// we're at the max peers stored for this torrent
+			// only send a write token if the requester is already in the set
+			// only check for a match on IP because the peer may be announcing
+			// a different port than the one it is using to send DHT messages
+			peer_entry requester_entry;
+			requester_entry.addr.address(requester);
+			auto requester_iter = i->second.peers.lower_bound(requester_entry);
+			return requester_iter == i->second.peers.end()
+				|| requester_iter->addr.address() != requester;
 		}
 
 		void announce_peer(sha1_hash const& info_hash
@@ -268,26 +279,12 @@ namespace
 			torrent_entry* v;
 			if (ti == m_map.end())
 			{
-				// we don't have this torrent, add it
-				// do we need to remove another one first?
-				if (!m_map.empty() && int(m_map.size()) >= m_settings.max_torrents)
+				if (int(m_map.size()) >= m_settings.max_torrents)
 				{
-					// we need to remove some. Remove the ones with the
-					// fewest peers
-					int num_peers = int(m_map.begin()->second.peers.size());
-					auto candidate = m_map.begin();
-					for (auto i = m_map.begin()
-						, end(m_map.end()); i != end; ++i)
-					{
-						if (int(i->second.peers.size()) > num_peers) continue;
-						if (i->first == info_hash) continue;
-						num_peers = int(i->second.peers.size());
-						candidate = i;
-					}
-					m_map.erase(candidate);
-					m_counters.peers -= num_peers;
-					m_counters.torrents -= 1;
+					// we're at capacity, drop the announce
+					return;
 				}
+
 				m_counters.torrents += 1;
 				v = &m_map[info_hash];
 			}
@@ -315,13 +312,8 @@ namespace
 			}
 			else if (v->peers.size() >= m_settings.max_peers)
 			{
-				// when we're at capacity, there's a 50/50 chance of dropping the
-				// announcing peer or an existing peer
-				if (random(1)) return;
-				i = v->peers.lower_bound(peer);
-				if (i == v->peers.end()) --i;
-				v->peers.erase(i++);
-				m_counters.peers -= 1;
+				// we're at capacity, drop the announce
+				return;
 			}
 			v->peers.insert(i, peer);
 			m_counters.peers += 1;

src/kademlia/node.cpp

@@ -750,13 +750,13 @@ void node::status(session_status& s)
 }
 #endif
 
-void node::lookup_peers(sha1_hash const& info_hash, entry& reply
-	, bool noseed, bool scrape) const
+bool node::lookup_peers(sha1_hash const& info_hash, entry& reply
+	, bool noseed, bool scrape, address const& requester) const
 {
 	if (m_observer)
 		m_observer->get_peers(info_hash);
 
-	m_storage.get_peers(info_hash, protocol(), noseed, scrape, reply);
+	return m_storage.get_peers(info_hash, protocol(), noseed, scrape, requester, reply);
 }
 
 entry write_nodes_entry(std::vector<node_entry> const& nodes)
@@ -848,7 +848,6 @@ void node::incoming_request(msg const& m, entry& e)
 		}
 
 		sha1_hash const info_hash(msg_keys[0].string_ptr());
-		reply["token"] = generate_token(m.addr, info_hash);
 
 		m_counters.inc_stats_counter(counters::dht_get_peers_in);
 
@@ -859,7 +858,12 @@ void node::incoming_request(msg const& m, entry& e)
 		bool scrape = false;
 		if (msg_keys[1] && msg_keys[1].int_value() != 0) noseed = true;
 		if (msg_keys[2] && msg_keys[2].int_value() != 0) scrape = true;
-		lookup_peers(info_hash, reply, noseed, scrape);
+		// If our storage is full we want to withhold the write token so that
+		// announces will spill over to our neighbors. This widens the
+		// perimeter of nodes which store peers for this torrent
+		bool full = lookup_peers(info_hash, reply, noseed, scrape, m.addr.address());
+		if (!full) reply["token"] = generate_token(m.addr, info_hash);
+
 #ifndef TORRENT_DISABLE_LOGGING
 		if (reply.find_key("values") && m_observer)
 		{

test/test_dht_storage.cpp

@@ -100,7 +100,7 @@ TORRENT_TEST(announce_peer)
 	std::unique_ptr<dht_storage_interface> s(create_default_dht_storage(sett));
 
 	entry peers;
-	s->get_peers(n1, udp::v4(), false, false, peers);
+	s->get_peers(n1, udp::v4(), false, false, address(), peers);
 
 	TEST_CHECK(peers["n"].string().empty())
 	TEST_CHECK(peers["values"].list().empty());
@@ -111,15 +111,17 @@ TORRENT_TEST(announce_peer)
 	tcp::endpoint const p4 = ep("124.31.75.24", 1);
 
 	s->announce_peer(n1, p1, "torrent_name", false);
-	s->get_peers(n1, udp::v4(), false, false, peers);
+	peers = entry();
+	s->get_peers(n1, udp::v4(), false, false, address(), peers);
 	TEST_EQUAL(peers["n"].string(), "torrent_name")
 	TEST_EQUAL(peers["values"].list().size(), 1)
 
 	s->announce_peer(n2, p2, "torrent_name1", false);
 	s->announce_peer(n2, p3, "torrent_name1", false);
 	s->announce_peer(n3, p4, "torrent_name2", false);
-	bool r = s->get_peers(n1, udp::v4(), false, false, peers);
-	TEST_CHECK(!r);
+	peers = entry();
+	s->get_peers(n3, udp::v4(), false, false, address(), peers);
+	TEST_CHECK(!peers.find_key("values"));
 }
 
 TORRENT_TEST(dual_stack)
@@ -140,11 +142,11 @@ TORRENT_TEST(dual_stack)
 	s->announce_peer(n1, p5, "torrent_name", false);
 
 	entry peers4;
-	s->get_peers(n1, udp::v4(), false, false, peers4);
+	s->get_peers(n1, udp::v4(), false, false, address(), peers4);
 	TEST_EQUAL(peers4["values"].list().size(), 3);
 
 	entry peers6;
-	s->get_peers(n1, udp::v6(), false, false, peers6);
+	s->get_peers(n1, udp::v6(), false, false, address(), peers6);
 	TEST_EQUAL(peers6["values"].list().size(), 2);
 }