lt sync

libtorrent/include/libtorrent/bandwidth_manager.hpp

@@ -123,6 +123,7 @@ struct bandwidth_manager
 #ifndef NDEBUG
 	bool is_in_history(PeerConnection const* peer) const
 	{
+		mutex_t::scoped_lock l(m_mutex);
 		for (typename history_t::const_iterator i
 			= m_history.begin(), end(m_history.end()); i != end; ++i)
 		{
@@ -142,6 +143,7 @@ struct bandwidth_manager
 		INVARIANT_CHECK;
 		TORRENT_ASSERT(blk > 0);
 
+		mutex_t::scoped_lock l(m_mutex);
 		TORRENT_ASSERT(!peer->ignore_bandwidth_limits());
 
 		// make sure this peer isn't already in line
@@ -187,7 +189,7 @@ struct bandwidth_manager
 #ifdef TORRENT_VERBOSE_BANDWIDTH_LIMIT
 		std::cerr << " req_bandwidht. m_queue.size() = " << m_queue.size() << std::endl;
 #endif
-		if (!m_queue.empty()) hand_out_bandwidth();
+		if (!m_queue.empty()) hand_out_bandwidth(l);
 	}
 
 #ifndef NDEBUG
@@ -239,6 +241,7 @@ private:
 
 		TORRENT_ASSERT(!m_history.empty());
 
+		mutex_t::scoped_lock l(m_mutex);
 		ptime now(time_now());
 		while (!m_history.empty() && m_history.back().expires_at <= now)
 		{
@@ -248,8 +251,10 @@ private:
 			TORRENT_ASSERT(m_current_quota >= 0);
 			intrusive_ptr<PeerConnection> c = e.peer;
 			shared_ptr<Torrent> t = e.tor.lock();
+			l.unlock();
 			if (!c->is_disconnecting()) c->expire_bandwidth(m_channel, e.amount);
 			if (t) t->expire_bandwidth(m_channel, e.amount);
+			l.lock();
 		}
 		
 		// now, wait for the next chunk to expire
@@ -262,7 +267,7 @@ private:
 		// since some bandwidth just expired, it
 		// means we can hand out more (in case there
 		// are still consumers in line)
-		if (!m_queue.empty()) hand_out_bandwidth();
+		if (!m_queue.empty()) hand_out_bandwidth(l);
 #ifndef NDEBUG
 		}
 		catch (std::exception&)
@@ -272,7 +277,7 @@ private:
 #endif
 	}
 
-	void hand_out_bandwidth() throw()
+	void hand_out_bandwidth(boost::mutex::scoped_lock& l) throw()
 	{
 		// if we're already handing out bandwidth, just return back
 		// to the loop further down on the callstack
@@ -285,9 +290,7 @@ private:
 
 		ptime now(time_now());
 
-		mutex_t::scoped_lock l(m_mutex);
 		int limit = m_limit;
-		l.unlock();
 
 		// available bandwidth to hand out
 		int amount = limit - m_current_quota;
@@ -319,7 +322,9 @@ private:
 			if (!t) continue;
 			if (qe.peer->is_disconnecting())
 			{
+				l.unlock();
 				t->expire_bandwidth(m_channel, qe.max_block_size);
+				l.lock();
 				TORRENT_ASSERT(amount == limit - m_current_quota);
 				continue;
 			}
@@ -389,8 +394,10 @@ private:
 			TORRENT_ASSERT(amount == limit - m_current_quota);
 			amount -= hand_out_amount;
 			TORRENT_ASSERT(hand_out_amount <= qe.max_block_size);
+			l.unlock();
 			t->assign_bandwidth(m_channel, hand_out_amount, qe.max_block_size);
 			qe.peer->assign_bandwidth(m_channel, hand_out_amount);
+			l.lock();
 			add_history_entry(history_entry<PeerConnection, Torrent>(
 				qe.peer, t, hand_out_amount, now + bw_window_size));
 			TORRENT_ASSERT(amount == limit - m_current_quota);

libtorrent/include/libtorrent/web_peer_connection.hpp

@@ -120,7 +120,8 @@ namespace libtorrent
 		void write_interested() {}
 		void write_not_interested() {}
 		void write_request(peer_request const& r);
-		void write_cancel(peer_request const& r) {}
+		void write_cancel(peer_request const& r)
+		{ incoming_reject_request(r); }
 		void write_have(int index) {}
 		void write_piece(peer_request const& r, char* buffer) { TORRENT_ASSERT(false); }
 		void write_keepalive() {}

libtorrent/src/broadcast_socket.cpp

@@ -170,12 +170,17 @@ namespace libtorrent
 		for (std::list<socket_entry>::iterator i = m_sockets.begin()
 			, end(m_sockets.end()); i != end; ++i)
 		{
+			if (!i->socket) continue;
 			asio::error_code e;
 			i->socket->send_to(asio::buffer(buffer, size), m_multicast_endpoint, 0, e);
 #ifndef NDEBUG
 //			std::cerr << " sending on " << i->socket->local_endpoint().address().to_string() << std::endl;
 #endif
-			if (e) ec = e;
+			if (e)
+			{
+				i->socket->close(e);
+				i->socket.reset();
+			}
 		}
 	}
 
@@ -184,6 +189,7 @@ namespace libtorrent
 	{
 		if (ec || bytes_transferred == 0 || !m_on_receive) return;
 		m_on_receive(s->remote, s->buffer, bytes_transferred);
+		if (!s->socket) return;
 		s->socket->async_receive_from(asio::buffer(s->buffer, sizeof(s->buffer))
 			, s->remote, bind(&broadcast_socket::on_receive, this, s, _1, _2));
 	}
@@ -195,6 +201,7 @@ namespace libtorrent
 		for (std::list<socket_entry>::iterator i = m_sockets.begin()
 			, end(m_sockets.end()); i != end; ++i)
 		{
+			if (!socket) continue;
 			i->socket->close();
 		}
 	}

libtorrent/src/bt_peer_connection.cpp

@@ -295,6 +295,8 @@ namespace libtorrent
 	{
 		INVARIANT_CHECK;
 
+		if (!m_supports_fast) return;
+
 		TORRENT_ASSERT(m_sent_handshake && m_sent_bitfield);
 		TORRENT_ASSERT(associated_torrent().lock()->valid_metadata());
 
@@ -312,6 +314,7 @@ namespace libtorrent
 
 		TORRENT_ASSERT(m_sent_handshake && m_sent_bitfield);
 		TORRENT_ASSERT(associated_torrent().lock()->valid_metadata());
+		TORRENT_ASSERT(m_supports_fast);
 
 		char msg[] = {0,0,0,5, msg_allowed_fast, 0, 0, 0, 0};
 		char* ptr = msg + 5;
@@ -1297,6 +1300,9 @@ namespace libtorrent
 		detail::write_int32(r.start, ptr); // begin
 		detail::write_int32(r.length, ptr); // length
 		send_buffer(msg, sizeof(msg));
+
+		if (!m_supports_fast)
+			incoming_reject_request(r);
 	}
 
 	void bt_peer_connection::write_request(peer_request const& r)

libtorrent/src/peer_connection.cpp

@@ -759,7 +759,7 @@ namespace libtorrent
 	}
 	
 	// -----------------------------
-	// -------- REJECT PIECE -------
+	// ------- SUGGEST PIECE -------
 	// -----------------------------
 
 	void peer_connection::incoming_suggest(int index)
@@ -1127,6 +1127,18 @@ namespace libtorrent
 				"i: " << m_peer_interested << " | "
 				"t: " << (int)t->torrent_file().piece_size(r.piece) << " | "
 				"n: " << t->torrent_file().num_pieces() << " ]\n";
+
+			(*m_logger) << time_now_string()
+				<< " ==> REJECT_PIECE [ "
+				"piece: " << r.piece << " | "
+				"s: " << r.start << " | "
+				"l: " << r.length << " ]\n";
+
+			(*m_logger) << time_now_string()
+				<< " ==> REJECT_PIECE [ "
+				"piece: " << r.piece << " | "
+				"s: " << r.start << " | "
+				"l: " << r.length << " ]\n";
 #endif
 			write_reject_request(r);
 			return;
@@ -1155,8 +1167,13 @@ namespace libtorrent
 			{
 				write_reject_request(r);
 #ifdef TORRENT_VERBOSE_LOGGING
-			(*m_logger) << time_now_string()
+				(*m_logger) << time_now_string()
-				<< " *** REJECTING REQUEST [ peer choked and piece not in allowed fast set ]\n";
+					<< " *** REJECTING REQUEST [ peer choked and piece not in allowed fast set ]\n";
+				(*m_logger) << time_now_string()
+					<< " ==> REJECT_PIECE [ "
+					"piece: " << r.piece << " | "
+					"s: " << r.start << " | "
+					"l: " << r.length << " ]\n";
 #endif
 			}
 			else
@@ -1179,6 +1196,12 @@ namespace libtorrent
 				"n: " << t->torrent_file().num_pieces() << " | "
 				"h: " << t->have_piece(r.piece) << " | "
 				"block_limit: " << t->block_size() << " ]\n";
+
+			(*m_logger) << time_now_string()
+				<< " ==> REJECT_PIECE [ "
+				"piece: " << r.piece << " | "
+				"s: " << r.start << " | "
+				"l: " << r.length << " ]\n";
 #endif
 
 			write_reject_request(r);
@@ -1365,6 +1388,9 @@ namespace libtorrent
 		m_outstanding_writing_bytes += p.length;
 		TORRENT_ASSERT(!m_reading);
 		picker.mark_as_writing(block_finished, peer_info_struct());
+#ifndef NDEBUG
+		t->check_invariant();
+#endif
 	}
 
 	void peer_connection::on_disk_write_complete(int ret, disk_io_job const& j
@@ -1399,8 +1425,10 @@ namespace libtorrent
 		
 			if (t->alerts().should_post(alert::fatal))
 			{
-				std::string err = "torrent paused: disk write error, " + j.str;
+				if (j.str != "write failed: No space left on device"){
-				t->alerts().post_alert(file_error_alert(t->get_handle(), err));
+					std::string err = "torrent paused: disk write error, " + j.str;
+					t->alerts().post_alert(file_error_alert(t->get_handle(), err));
+				}
 			}
 			t->pause();
 			return;
@@ -1479,6 +1507,14 @@ namespace libtorrent
 		if (i != m_requests.end())
 		{
 			m_requests.erase(i);
+#ifdef TORRENT_VERBOSE_LOGGING
+			(*m_logger) << time_now_string()
+				<< " ==> REJECT_PIECE [ "
+				"piece: " << r.piece << " | "
+				"s: " << r.start << " | "
+				"l: " << r.length << " ]\n";
+#endif
+			write_reject_request(r);
 		}
 		else
 		{
@@ -1735,11 +1771,6 @@ namespace libtorrent
 			// sent yet, so we don't have to send a cancel.
 			return;
 		}
-		else
-		{	
-			m_download_queue.erase(it);
-			t->picker().abort_download(block);
-		}
 
 		int block_offset = block.block_index * t->block_size();
 		int block_size
@@ -1753,13 +1784,12 @@ namespace libtorrent
 		r.start = block_offset;
 		r.length = block_size;
 
-		write_cancel(r);
-
 #ifdef TORRENT_VERBOSE_LOGGING
 		(*m_logger) << time_now_string()
 				<< " ==> CANCEL  [ piece: " << block.piece_index << " | s: "
 				<< block_offset << " | l: " << block_size << " | " << block.block_index << " ]\n";
 #endif
+		write_cancel(r);
 	}
 
 	void peer_connection::send_choke()
@@ -1783,6 +1813,19 @@ namespace libtorrent
 		// reject the requests we have in the queue
 		std::for_each(m_requests.begin(), m_requests.end()
 			, bind(&peer_connection::write_reject_request, this, _1));
+
+#ifdef TORRENT_VERBOSE_LOGGING
+		for (std::deque<peer_request>::iterator i = m_requests.begin()
+			, end(m_requests.end()); i != end; ++i)
+		{
+			peer_request const& r = *i;
+			(*m_logger) << time_now_string()
+				<< " ==> REJECT_PIECE [ "
+				"piece: " << r.piece << " | "
+				"s: " << r.start << " | "
+				"l: " << r.length << " ]\n";
+		}
+#endif
 		m_requests.clear();
 	}
 
@@ -2206,18 +2249,24 @@ namespace libtorrent
 			else
 			{
 				piece_picker& picker = t->picker();
-				while (!m_download_queue.empty())
+
+				std::deque<piece_block> dl(m_download_queue);
+				for (std::deque<piece_block>::iterator i = dl.begin()
+					, end(dl.end()); i != end; ++i)
 				{
 					piece_block const& r = m_download_queue.back();
-					picker.abort_download(r);
+#ifdef TORRENT_VERBOSE_LOGGING
+					(*m_logger) << time_now_string()
+						<< " ==> CANCEL  [ piece: " << r.piece_index
+						<< " | block: " << r.block_index
+						<< " ]\n";
+#endif
 					write_cancel(t->to_req(r));
-					m_download_queue.pop_back();
 				}
 				while (!m_request_queue.empty())
 				{
 					piece_block const& r = m_request_queue.back();
 					picker.abort_download(r);
-					write_cancel(t->to_req(r));
 					m_request_queue.pop_back();
 				}
 
@@ -2877,6 +2926,10 @@ namespace libtorrent
 				== m_ses.m_bandwidth_manager[i]->is_in_history(this)
 				|| m_bandwidth_limit[i].throttle() == bandwidth_limit::inf);
 		}
+		std::set<piece_block> unique;
+		std::copy(m_download_queue.begin(), m_download_queue.end(), std::inserter(unique, unique.begin()));
+		std::copy(m_request_queue.begin(), m_request_queue.end(), std::inserter(unique, unique.begin()));
+		TORRENT_ASSERT(unique.size() == m_download_queue.size() + m_request_queue.size());
 		if (m_peer_info)
 		{
 			TORRENT_ASSERT(m_peer_info->connection == this

libtorrent/src/torrent.cpp

@@ -2628,7 +2628,8 @@ namespace libtorrent
 			for (std::map<piece_block, int>::iterator i = num_requests.begin()
 				, end(num_requests.end()); i != end; ++i)
 			{
-				TORRENT_ASSERT(m_picker->num_peers(i->first) == i->second);
+				if (!m_picker->is_downloaded(i->first))
+					TORRENT_ASSERT(m_picker->num_peers(i->first) == i->second);
 			}
 		}