optimize allow-fast logic (#800)

optimize allow-fast logic

ChangeLog

@@ -1,5 +1,6 @@
 1.1.1 release
 
+	* optimize allow-fast logic
 	* fix issue where FAST extension messages were not used during handshake
 	* fixed crash on invalid input in http_parser
 	* upgraded to libtommath 1.0

include/libtorrent/bt_peer_connection.hpp

@@ -356,6 +356,10 @@ private:
 		// and can send bittorrent messages
 		bool m_sent_handshake:1;
 
+		// set to true once we send the allowed-fast messages. This is
+		// only done once per connection
+		bool m_sent_allowed_fast:1;
+
 #if !defined(TORRENT_DISABLE_ENCRYPTION) && !defined(TORRENT_DISABLE_EXTENSIONS)
 		// this is set to true after the encryption method has been
 		// successfully negotiated (either plaintext or rc4), to signal

simulation/Jamfile

@@ -41,5 +41,6 @@ alias libtorrent-sims :
 	[ run test_trackers_extension.cpp ]
 	[ run test_tracker.cpp ]
 	[ run test_ip_filter.cpp ]
+	[ run test_fast_extensions.cpp ]
 	;
 

simulation/create_torrent.cpp

@@ -46,7 +46,8 @@ std::string save_path(int idx)
 	return path;
 }
 
-lt::add_torrent_params create_torrent(int idx, bool seed)
+lt::add_torrent_params create_torrent(int const idx, bool const seed
+	, int const num_pieces)
 {
 	// TODO: if we want non-seeding torrents, that could be a bit cheaper to
 	// create
@@ -60,7 +61,7 @@ lt::add_torrent_params create_torrent(int idx, bool seed)
 	if (ec) fprintf(stderr, "failed to create directory: \"%s\": %s\n"
 		, path.c_str(), ec.message().c_str());
 	std::ofstream file(lt::combine_path(path, name).c_str());
-	params.ti = ::create_torrent(&file, name, 0x4000, 9 + idx, false);
+	params.ti = ::create_torrent(&file, name, 0x4000, num_pieces + idx, false);
 	file.close();
 
 	// by setting the save path to a dummy path, it won't be seeding

simulation/create_torrent.hpp

@@ -37,7 +37,8 @@ POSSIBILITY OF SUCH DAMAGE.
 #include "libtorrent/add_torrent_params.hpp"
 
 std::string save_path(int idx);
-libtorrent::add_torrent_params create_torrent(int idx, bool seed = true);
+libtorrent::add_torrent_params create_torrent(int idx, bool seed = true
+	, int num_pieces = 9);
 
 #endif
 

simulation/fake_peer.hpp

@@ -34,6 +34,7 @@ POSSIBILITY OF SUCH DAMAGE.
 #define SIMULATION_FAKE_PEER_HPP
 
 #include <array>
+#include <functional>
 #include "test.hpp"
 #include "simulator/simulator.hpp"
 #include "libtorrent/session.hpp"
@@ -41,6 +42,7 @@ POSSIBILITY OF SUCH DAMAGE.
 #include "libtorrent/torrent_handle.hpp"
 #include "libtorrent/sha1_hash.hpp"
 #include "libtorrent/torrent_info.hpp"
+#include "libtorrent/io.hpp"
 
 using namespace sim;
 
@@ -89,19 +91,56 @@ struct fake_peer
 
 	bool tripped() const { return m_tripped; }
 
+	void send_interested()
+	{
+		m_send_buffer.resize(m_send_buffer.size() + 5);
+		char* ptr = m_send_buffer.data() + m_send_buffer.size() - 5;
+
+		lt::detail::write_uint32(1, ptr);
+		lt::detail::write_uint8(2, ptr);
+	}
+
+	void send_bitfield(std::vector<bool> const& pieces)
+	{
+		int const bytes = (pieces.size() + 7) / 8;
+		m_send_buffer.resize(m_send_buffer.size() + 5 + bytes);
+		char* ptr = m_send_buffer.data() + m_send_buffer.size() - 5 - bytes;
+
+		lt::detail::write_uint32(1 + bytes, ptr);
+		lt::detail::write_uint8(5, ptr);
+
+		boost::uint8_t b = 0;
+		int cnt = 7;
+		for (std::vector<bool>::const_iterator i = pieces.begin()
+			, end(pieces.end()); i != end; ++i)
+		{
+			if (*i) b |= 1 << cnt;
+			--cnt;
+			if (cnt < 0)
+			{
+				lt::detail::write_uint8(b, ptr);
+				b = 0;
+				cnt = 7;
+			}
+		}
+		lt::detail::write_uint8(b, ptr);
+	}
+
 private:
 
-	void write_handshake(boost::system::error_code const& ec, lt::sha1_hash ih)
+	void write_handshake(boost::system::error_code const& ec
+		, lt::sha1_hash ih)
 	{
-		asio::ip::tcp::endpoint ep = m_out_socket.remote_endpoint();
+		using namespace std::placeholders;
+
+		asio::ip::tcp::endpoint const ep = m_out_socket.remote_endpoint();
 		printf("fake_peer::connect (%s) -> (%d) %s\n"
 			, lt::print_endpoint(ep).c_str(), ec.value()
 			, ec.message().c_str());
 		static char const handshake[]
 		= "\x13" "BitTorrent protocol\0\0\0\0\0\0\0\x04"
 			"                    " // space for info-hash
-			"aaaaaaaaaaaaaaaaaaaa" // peer-id
-			"\0\0\0\x01\x02"; // interested
+			"aaaaaaaaaaaaaaaaaaaa"; // peer-id
 		int const len = sizeof(handshake) - 1;
 		memcpy(m_out_buffer, handshake, len);
 		memcpy(&m_out_buffer[28], ih.data(), 20);
@@ -112,10 +151,28 @@ private:
 			printf("fake_peer::write_handshake(%s) -> (%d) %s\n"
 				, lt::print_endpoint(ep).c_str(), ec.value()
 				, ec.message().c_str());
+			if (m_send_buffer.empty())
+			{
 				this->m_out_socket.close();
+			}
+			else
+			{
+				asio::async_write(m_out_socket, asio::const_buffers_1(
+					m_send_buffer.data(), m_send_buffer.size())
+					, std::bind(&fake_peer::write_send_buffer, this, _1, _2));
+			}
 		});
 	}
 
+	void write_send_buffer(boost::system::error_code const& ec
+		, size_t bytes_transferred)
+	{
+		printf("fake_peer::write_send_buffer() -> (%d) %s\n"
+			, ec.value(), ec.message().c_str());
+
+		m_out_socket.close();
+	}
+
 	char m_out_buffer[300];
 
 	asio::io_service m_ios;
@@ -123,6 +180,8 @@ private:
 	asio::ip::tcp::socket m_in_socket;
 	asio::ip::tcp::socket m_out_socket;
 	bool m_tripped;
+
+	std::vector<char> m_send_buffer;
 };
 
 inline void add_fake_peers(lt::torrent_handle h)

simulation/test_fast_extensions.cpp

@@ -0,0 +1,206 @@
+/*
+
+Copyright (c) 2016, Arvid Norberg
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions
+are met:
+
+    * Redistributions of source code must retain the above copyright
+      notice, this list of conditions and the following disclaimer.
+    * Redistributions in binary form must reproduce the above copyright
+      notice, this list of conditions and the following disclaimer in
+      the documentation and/or other materials provided with the distribution.
+    * Neither the name of the author nor the names of its
+      contributors may be used to endorse or promote products derived
+      from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+POSSIBILITY OF SUCH DAMAGE.
+
+*/
+
+#include "test.hpp"
+#include "utils.hpp"
+#include "libtorrent/alert.hpp"
+#include "libtorrent/alert_types.hpp"
+#include "libtorrent/session.hpp"
+#include "libtorrent/add_torrent_params.hpp"
+#include "create_torrent.hpp"
+#include "settings.hpp"
+#include "fake_peer.hpp"
+#include "setup_transfer.hpp" // for ep()
+#include "simulator/utils.hpp"
+
+namespace lt = libtorrent;
+
+template <typename Sett, typename Alert>
+void run_fake_peer_test(
+	lt::add_torrent_params params
+	, Sett const& sett
+	, Alert const& alert)
+{
+	sim::default_config cfg;
+	sim::simulation sim{cfg};
+
+	sim::asio::io_service ios(sim, lt::address_v4::from_string("50.0.0.1"));
+	lt::session_proxy zombie;
+
+	// setup settings pack to use for the session (customization point)
+	lt::settings_pack pack = settings();
+	pack.set_str(lt::settings_pack::listen_interfaces, "0.0.0.0:6881");
+	sett(pack);
+	// create session
+	std::shared_ptr<lt::session> ses = std::make_shared<lt::session>(pack, ios);
+
+	fake_peer p1(sim, "60.0.0.0");
+
+	params.flags &= ~lt::add_torrent_params::flag_auto_managed;
+	params.flags &= ~lt::add_torrent_params::flag_paused;
+	ses->async_add_torrent(params);
+
+	// the alert notification function is called from within libtorrent's
+	// context. It's not OK to talk to libtorrent in there, post it back out and
+	// then ask for alerts.
+	print_alerts(*ses, [&](lt::session& ses, lt::alert const* a) {
+		alert(ses, a, p1);
+	});
+
+	sim::timer t(sim, lt::seconds(1)
+		, [&](boost::system::error_code const& ec)
+	{
+		ses->set_alert_notify([]{});
+		// shut down
+		zombie = ses->abort();
+
+		p1.close();
+
+		ses.reset();
+	});
+
+	sim.run();
+}
+
+// make sure we consistently send the same allow-fast pieces, regardless
+// of which pieces the peer has.
+TORRENT_TEST(allow_fast)
+{
+	std::set<int> allowed_fast;
+
+	int const num_pieces = 50;
+	lt::add_torrent_params params = create_torrent(0, false, num_pieces);
+	std::vector<bool> bitfield(num_pieces, false);
+
+	for (int i = 0; i < num_pieces + 1; ++i)
+	{
+		// just for this one session, to check for duplicates
+		std::set<int> local_allowed_fast;
+
+		run_fake_peer_test(params, [] (lt::settings_pack& pack) {
+			pack.set_int(lt::settings_pack::allowed_fast_set_size, 13);
+		}
+		, [&] (lt::session& ses, lt::alert const* a, fake_peer& p1)
+		{
+			if (auto at = lt::alert_cast<lt::add_torrent_alert>(a))
+			{
+				lt::torrent_handle h = at->handle;
+				p1.connect_to(ep("50.0.0.1", 6881)
+					, h.torrent_file()->info_hash());
+				p1.send_bitfield(bitfield);
+				p1.send_interested();
+			}
+			else if (auto l = lt::alert_cast<lt::peer_log_alert>(a))
+			{
+				if (strcmp(l->event_type, "ALLOWED_FAST") != 0) return;
+
+				int const piece = atoi(l->msg());
+				// make sure we don't get the same allowed piece more than once
+				TEST_EQUAL(local_allowed_fast.count(piece), 0);
+
+				// build the union of all allow-fast pieces we've received, across
+				// simulations.
+				allowed_fast.insert(piece);
+				local_allowed_fast.insert(piece);
+
+				// make sure this is a valid piece
+				TEST_CHECK(piece < num_pieces);
+				TEST_CHECK(piece >= 0);
+				// and make sure it's not one of the pieces we have
+				// because that would be redundant
+				TEST_EQUAL(bitfield[piece], false);
+			}
+		});
+
+		// i goes from [0, mum_pieces + 1) to cover the have-none and have-all
+		// cases. After the last iteration, we can't add another piece.
+		if (i < int(bitfield.size()))
+			bitfield[i] = true;
+	}
+
+	// we should never have sent any other pieces than the 13 designated for this
+	// peer's IP.
+	TEST_EQUAL(int(allowed_fast.size()), 13);
+}
+
+// This tests a worst case scenario of allow-fast configuration where we must
+// verify that libtorrent correctly aborts before satisfying the settings
+// (because doing so would be too expensive)
+//
+// we have a torrent with a lot of pieces, and we want to send that many minus
+// one allow-fast pieces. The way allow-fast pieces are computed is by hashing
+// the peer's IP modulus the number of pieces. To actually compute which pieces
+// to send (or which one piece _not_ to send) we would have to work hard through
+// a lot of duplicates. This test makes sure we don't, and abort well before
+// then
+TORRENT_TEST(allow_fast_stress)
+{
+	std::set<int> allowed_fast;
+
+	int const num_pieces = 50000;
+	lt::add_torrent_params params = create_torrent(0, false, num_pieces);
+
+	run_fake_peer_test(params, [&] (lt::settings_pack& pack) {
+		pack.set_int(lt::settings_pack::allowed_fast_set_size, num_pieces - 1);
+	}
+	, [&] (lt::session& ses, lt::alert const* a, fake_peer& p1)
+	{
+		if (auto at = lt::alert_cast<lt::add_torrent_alert>(a))
+		{
+			lt::torrent_handle h = at->handle;
+			p1.connect_to(ep("50.0.0.1", 6881)
+				, h.torrent_file()->info_hash());
+			p1.send_interested();
+		}
+		else if (auto l = lt::alert_cast<lt::peer_log_alert>(a))
+		{
+			if (strcmp(l->event_type, "ALLOWED_FAST") != 0) return;
+
+			int const piece = atoi(l->msg());
+
+			// make sure we don't get the same allowed piece more than once
+			TEST_EQUAL(allowed_fast.count(piece), 0);
+
+			// build the union of all allow-fast pieces we've received, across
+			// simulations.
+			allowed_fast.insert(piece);
+
+			// make sure this is a valid piece
+			TEST_CHECK(piece < num_pieces);
+			TEST_CHECK(piece >= 0);
+		}
+	});
+
+	std::printf("received %d allowed fast, out of %d configured ones\n"
+		, int(allowed_fast.size()), num_pieces - 1);
+	TEST_CHECK(int(allowed_fast.size()) < num_pieces / 80);
+}

simulation/test_swarm.cpp

@@ -418,7 +418,6 @@ TORRENT_TEST(delete_partfile)
 // outgoing connections
 // TODO: add test that makes sure a torrent in graceful pause mode won't accept
 // incoming connections
-// TODO: test allow-fast
 // TODO: test the different storage allocation modes
 // TODO: test contiguous buffers
 

src/bt_peer_connection.cpp

@@ -110,6 +110,7 @@ namespace libtorrent
 		, m_supports_fast(false)
 		, m_sent_bitfield(false)
 		, m_sent_handshake(false)
+		, m_sent_allowed_fast(false)
 #if !defined(TORRENT_DISABLE_ENCRYPTION) && !defined(TORRENT_DISABLE_EXTENSIONS)
 		, m_encrypted(false)
 		, m_rc4_encrypted(false)
@@ -366,6 +367,10 @@ namespace libtorrent
 
 		if (!m_supports_fast) return;
 
+#ifndef TORRENT_DISABLE_LOGGING
+		peer_log(peer_log_alert::outgoing_message, "ALLOWED_FAST", "%d", piece);
+#endif
+
 		TORRENT_ASSERT(m_sent_handshake);
 		TORRENT_ASSERT(m_sent_bitfield);
 		TORRENT_ASSERT(associated_torrent().lock()->valid_metadata());
@@ -991,6 +996,15 @@ namespace libtorrent
 		}
 		if (!m_recv_buffer.packet_finished()) return;
 
+		// we defer sending the allowed set until the peer says it's interested in
+		// us. This saves some bandwidth and allows us to omit messages for pieces
+		// that the peer already has
+		if (!m_sent_allowed_fast && m_supports_fast)
+		{
+			m_sent_allowed_fast = true;
+			send_allowed_set();
+		}
+
 		incoming_interested();
 	}
 
@@ -2116,13 +2130,11 @@ namespace libtorrent
 		else if (m_supports_fast && t->is_seed() && !m_settings.get_bool(settings_pack::lazy_bitfields))
 		{
 			write_have_all();
-			send_allowed_set();
 			return;
 		}
 		else if (m_supports_fast && t->num_have() == 0)
 		{
 			write_have_none();
-			send_allowed_set();
 			return;
 		}
 		else if (t->num_have() == 0)
@@ -2237,9 +2249,6 @@ namespace libtorrent
 			}
 			// TODO: if we're finished, send upload_only message
 		}
-
-		if (m_supports_fast)
-			send_allowed_set();
 	}
 
 #ifndef TORRENT_DISABLE_EXTENSIONS

src/peer_connection.cpp

@@ -596,9 +596,6 @@ namespace libtorrent
 				// that the peer already has
 				if (has_piece(i)) continue;
 
-#ifndef TORRENT_DISABLE_LOGGING
-				peer_log(peer_log_alert::outgoing_message, "ALLOWED_FAST", "%d", i);
-#endif
 				write_allow_fast(i);
 				TORRENT_ASSERT(std::find(m_accept_fast.begin()
 					, m_accept_fast.end(), i)
@@ -631,18 +628,26 @@ namespace libtorrent
 		x.append(t->torrent_file().info_hash().data(), 20);
 
 		sha1_hash hash = hasher(x.c_str(), x.size()).final();
+		int attempts = 0;
+		int loops = 0;
 		for (;;)
 		{
-			char* p = hash.data();
-			for (int i = 0; i < 5; ++i)
+			char const* p = hash.data();
+			for (int i = 0; i < hash.size / sizeof(boost::uint32_t); ++i)
 			{
-				int piece = detail::read_uint32(p) % num_pieces;
+				++loops;
+				int const piece = detail::read_uint32(p) % num_pieces;
 				if (std::find(m_accept_fast.begin(), m_accept_fast.end(), piece)
-					== m_accept_fast.end())
+					!= m_accept_fast.end())
+				{
+					// this is our safety-net to make sure this loop terminates, even
+					// under the worst conditions
+					if (++loops > 500) return;
+					continue;
+				}
+
+				if (!has_piece(piece))
 				{
-#ifndef TORRENT_DISABLE_LOGGING
-					peer_log(peer_log_alert::outgoing_message, "ALLOWED_FAST", "%d", piece);
-#endif
 					write_allow_fast(piece);
 					if (m_accept_fast.empty())
 					{
@@ -651,9 +656,8 @@ namespace libtorrent
 					}
 					m_accept_fast.push_back(piece);
 					m_accept_fast_piece_cnt.push_back(0);
-					if (int(m_accept_fast.size()) >= num_allowed_pieces
-						|| int(m_accept_fast.size()) == num_pieces) return;
 				}
+				if (++attempts >= num_allowed_pieces) return;
 			}
 			hash = hasher(hash.data(), 20).final();
 		}