limit number of torrents tracked by DHT and support DHT name lookups

ChangeLog

@@ -1,3 +1,4 @@
+	* support DHT name lookup
 	* optimized memory usage of torrent_info and file_storage, forcing some API changes
 	  around file_storage and file_entry
 	* support trackerid tracker extension
@@ -57,6 +58,7 @@
 	  incoming connection
 	* added more detailed instrumentation of the disk I/O thread
 
+	* limit number of torrents tracked by DHT
 	* fixed bug when allow_multiple_connections_per_ip was enabled
 	* potential WOW64 fix for unbuffered I/O (windows)
 	* expose set_alert_queue_size_limit to python binding

docs/manual.rst

@@ -3908,6 +3908,7 @@ session_settings
 		bool low_prio_disk;
 		int local_service_announce_interval;
 		int dht_announce_interval;
+		int dht_max_torrents;
 
 		int udp_tracker_token_expiry;
 		bool volatile_read_cache;
@@ -4512,6 +4513,9 @@ This interval is specified in seconds.
 torrents to the distributed hash table (DHT). This is specified to
 be 15 minutes which is its default.
 
+``dht_max_torrents`` is the max number of torrents we will track
+in the DHT.
+
 ``udp_tracker_token_expiry`` is the number of seconds libtorrent
 will keep UDP tracker connection tokens around for. This is specified
 to be 60 seconds, and defaults to that. The higher this value is, the

include/libtorrent/kademlia/node.hpp

@@ -95,6 +95,7 @@ struct peer_entry
 // this is a group. It contains a set of group members
 struct torrent_entry
 {
+	std::string name;
 	std::set<peer_entry> peers;
 };
 

include/libtorrent/session_settings.hpp

@@ -206,6 +206,7 @@ namespace libtorrent
 			, low_prio_disk(true)
 			, local_service_announce_interval(5 * 60)
 			, dht_announce_interval(15 * 60)
+			, dht_max_torrents(3000)
 			, udp_tracker_token_expiry(60)
 			, volatile_read_cache(false)
 			, guided_read_cache(true)
@@ -780,6 +781,9 @@ namespace libtorrent
 		// torrents. Defaults to 15 minutes
 		int dht_announce_interval;
 
+		// this is the max number of torrents the DHT will track
+		int dht_max_torrents;
+
 		// the number of seconds a connection ID received
 		// from a UDP tracker is valid for. This is specified
 		// as 60 seconds

src/kademlia/dht_tracker.cpp

@@ -511,7 +511,7 @@ namespace libtorrent { namespace dht
 		lazy_entry e;
 		int pos;
 		error_code ec;
-		int ret = lazy_bdecode(buf, buf + bytes_transferred, e, ec, &pos);
+		int ret = lazy_bdecode(buf, buf + bytes_transferred, e, ec, &pos, 10, 500);
 		if (ret != 0)
 		{
 #ifdef TORRENT_DHT_VERBOSE_LOGGING

src/kademlia/node.cpp

@@ -536,6 +536,8 @@ bool node_impl::lookup_peers(sha1_hash const& info_hash, entry& reply) const
 	torrent_entry const& v = i->second;
 	if (v.peers.empty()) return false;
 
+	if (!v.name.empty()) reply["n"] = v.name;
+
 	int num = (std::min)((int)v.peers.size(), m_settings.max_peers_reply);
 	int t = 0;
 	int m = 0;
@@ -753,10 +755,11 @@ void node_impl::incoming_request(msg const& m, entry& e)
 			{"info_hash", lazy_entry::string_t, 20, 0},
 			{"port", lazy_entry::int_t, 0, 0},
 			{"token", lazy_entry::string_t, 0, 0},
+			{"n", lazy_entry::string_t, 0, key_desc_t::optional},
 		};
 
-		lazy_entry const* msg_keys[3];
+		lazy_entry const* msg_keys[4];
-		if (!verify_message(arg_ent, msg_desc, msg_keys, 3, error_string, sizeof(error_string)))
+		if (!verify_message(arg_ent, msg_desc, msg_keys, 4, error_string, sizeof(error_string)))
 		{
 #ifdef TORRENT_DHT_VERBOSE_LOGGING
 			++g_failed_announces;
@@ -795,7 +798,33 @@ void node_impl::incoming_request(msg const& m, entry& e)
 		// the table get a chance to add it.
 		m_table.node_seen(id, m.addr);
 
+		if (!m_map.empty() && m_map.size() >= m_ses.settings().dht_max_torrents)
+		{
+			// we need to remove some. Remove the ones with the
+			// fewest peers
+			int num_peers = m_map.begin()->second.peers.size();
+			table_t::iterator candidate = m_map.begin();
+			for (table_t::iterator i = m_map.begin()
+				, end(m_map.end()); i != end; ++i)
+			{
+				if (i->second.peers.size() > num_peers) continue;
+				if (i->first == info_hash) continue;
+				num_peers = i->second.peers.size();
+				candidate = i;
+			}
+			m_map.erase(candidate);
+		}
 		torrent_entry& v = m_map[info_hash];
+
+		// the peer announces a torrent name, and we don't have a name
+		// for this torrent. Store it.
+		if (msg_keys[3] && v.name.empty())
+		{
+			std::string name = msg_keys[3]->string_value();
+			if (name.size() > 50) name.resize(50);
+			v.name = name;
+		}
+
 		peer_entry e;
 		e.addr = tcp::endpoint(m.addr.address(), port);
 		e.added = time_now();

test/test_dht.cpp

@@ -43,7 +43,9 @@ using namespace libtorrent;
 
 int dht_port = 48199;
 
-void send_dht_msg(datagram_socket& sock, char const* msg, lazy_entry* reply, char const* t = "10")
+void send_dht_msg(datagram_socket& sock, char const* msg, lazy_entry* reply
+	, char const* t = "10", char const* info_hash = 0, char const* name = 0
+	, char const* token = 0, int port = 0)
 {
 	entry e;
 	e["q"] = msg;
@@ -51,6 +53,10 @@ void send_dht_msg(datagram_socket& sock, char const* msg, lazy_entry* reply, cha
 	e["y"] = "q";
 	entry::dictionary_type& a = e["a"].dict();
 	a["id"] = "00000000000000000000";
+	if (info_hash) a["info_hash"] = info_hash;
+	if (name) a["n"] = name;
+	if (token) a["token"] = token;
+	if (port) a["port"] = port;
 	char msg_buf[1500];
 	int size = bencode(msg_buf, e);
 
@@ -98,6 +104,7 @@ int test_main()
 		{"t", lazy_entry::string_t, 2, 0},
 	};
 
+	fprintf(stderr, "msg: %s\n", print_entry(response).c_str());
 	ret = dht::verify_message(&response, pong_desc, parsed, 2, error_string, sizeof(error_string));
 	TEST_CHECK(ret);
 	if (ret)
@@ -105,6 +112,10 @@ int test_main()
 		TEST_CHECK(parsed[0]->string_value() == "r");
 		TEST_CHECK(parsed[1]->string_value() == "10");
 	}
+	else
+	{
+		fprintf(stderr, "invalid ping response: %s\n", error_string);
+	}
 
 	// ====== invalid message ======
 
@@ -115,6 +126,7 @@ int test_main()
 		{"e", lazy_entry::list_t, 0, 0},
 	};
 
+	fprintf(stderr, "msg: %s\n", print_entry(response).c_str());
 	ret = dht::verify_message(&response, err_desc, parsed, 2, error_string, sizeof(error_string));
 	TEST_CHECK(ret);
 	if (ret)
@@ -132,6 +144,75 @@ int test_main()
 			TEST_ERROR("invalid error response");
 		}
 	}
+	else
+	{
+		fprintf(stderr, "invalid error response: %s\n", error_string);
+	}
+
+	// ====== get_peers ======
+
+	send_dht_msg(sock, "get_peers", &response, "10", "01010101010101010101");
+
+	dht::key_desc_t peer1_desc[] = {
+		{"y", lazy_entry::string_t, 1, 0},
+		{"r", lazy_entry::dict_t, 0, 0},
+	};
+
+	std::string token;
+	fprintf(stderr, "msg: %s\n", print_entry(response).c_str());
+	ret = dht::verify_message(&response, peer1_desc, parsed, 2, error_string, sizeof(error_string));
+	TEST_CHECK(ret);
+	if (ret)
+	{
+		TEST_CHECK(parsed[0]->string_value() == "r");
+		token = parsed[1]->dict_find_string_value("token");
+	}
+	else
+	{
+		fprintf(stderr, "invalid get_peers response: %s\n", error_string);
+	}
+
+	// ====== announce ======
+
+	send_dht_msg(sock, "announce_peer", &response, "10", "01010101010101010101", "test", token.c_str(), 8080);
+
+	dht::key_desc_t ann_desc[] = {
+		{"y", lazy_entry::string_t, 1, 0},
+	};
+
+	fprintf(stderr, "msg: %s\n", print_entry(response).c_str());
+	ret = dht::verify_message(&response, ann_desc, parsed, 1, error_string, sizeof(error_string));
+	TEST_CHECK(ret);
+	if (ret)
+	{
+		TEST_CHECK(parsed[0]->string_value() == "r");
+	}
+	else
+	{
+		fprintf(stderr, "invalid announce response: %s\n", error_string);
+	}
+
+	// ====== get_peers ======
+
+	send_dht_msg(sock, "get_peers", &response, "10", "01010101010101010101");
+
+	dht::key_desc_t peer2_desc[] = {
+		{"y", lazy_entry::string_t, 1, 0},
+		{"r", lazy_entry::dict_t, 0, 0},
+	};
+
+	fprintf(stderr, "msg: %s\n", print_entry(response).c_str());
+	ret = dht::verify_message(&response, peer2_desc, parsed, 2, error_string, sizeof(error_string));
+	TEST_CHECK(ret);
+	if (ret)
+	{
+		TEST_CHECK(parsed[0]->string_value() == "r");
+		TEST_EQUAL(parsed[1]->dict_find_string_value("n"), "test");
+	}
+	else
+	{
+		fprintf(stderr, "invalid get_peers response: %s\n", error_string);
+	}
 
 	return 0;
 }