introduce a low watermark for disk io blocking connections

include/libtorrent/disk_io_thread.hpp

@@ -403,8 +403,8 @@ namespace libtorrent
 		int cache_piece(disk_io_job const& j, cache_piece_index_t::iterator& p
 			, bool& hit, int options, mutex::scoped_lock& l);
 
-		// this mutex only protects m_jobs, m_queue_buffer_size
+		// this mutex only protects m_jobs, m_queue_buffer_size,
-		// and m_abort
+		// m_exceeded_write_queue and m_abort
 		mutable mutex m_queue_mutex;
 		event m_signal;
 		bool m_abort;
@@ -443,6 +443,11 @@ namespace libtorrent
 		// the amount of physical ram in the machine
 		boost::uint64_t m_physical_ram;
 
+		// if we exceeded the max queue disk write size
+		// this is set to true. It remains true until the
+		// queue is smaller than the low watermark
+		bool m_exceeded_write_queue;
+
 		io_service& m_ios;
 
 		boost::function<void()> m_queue_callback;

include/libtorrent/session_settings.hpp

@@ -129,6 +129,7 @@ namespace libtorrent
 			, allowed_fast_set_size(10)
 			, suggest_mode(no_piece_suggestions)
 			, max_queued_disk_bytes(256 * 1024)
+			, max_queued_disk_bytes_low_watermark(0)
 			, handshake_timeout(10)
 #ifndef TORRENT_DISABLE_DHT
 			, use_dht_as_fallback(false)
@@ -438,6 +439,14 @@ namespace libtorrent
 		// this is a per session setting.
 		int max_queued_disk_bytes;
 
+		// this is the low watermark for the disk buffer queue.
+		// whenever the number of queued bytes exceed the
+		// max_queued_disk_bytes, libtorrent will wait for
+		// it to drop below this value before issuing more
+		// reads from the sockets. If set to 0, the
+		// low watermark will be half of the max queued disk bytes
+		int max_queued_disk_bytes_low_watermark;
+
 		// the number of seconds to wait for a handshake
 		// response from a peer. If no response is received
 		// within this time, the peer is disconnected.

parse_session_stats.py

@@ -35,7 +35,9 @@ def gen_report(name, lines):
 			column = column + 1
 			continue
 		if not first: print >>out, ', ',
-		print >>out, ' "%s" using 1:%d title "%s" axes %s with steps' % (sys.argv[1], column, k, axes[column-2]),
+		axis = 'x1y1'
+		if column-2 < len(axes): axis = axes[column-2]
+		print >>out, ' "%s" using 1:%d title "%s" axes %s with steps' % (sys.argv[1], column, k, axis),
 		first = False
 		column = column + 1
 	print >>out, ''
@@ -43,7 +45,12 @@ def gen_report(name, lines):
 	os.system('gnuplot session_stats_%s.gnuplot' % name);
 
 gen_report('torrents', ['downloading torrents', 'seeding torrents', 'checking torrents', 'stopped torrents'])
-gen_report('peers', ['peers', 'connecting peers', 'unchoked peers', 'peers disk-up', 'peers disk-down', 'peers bw-up', 'peers bw-down')
+gen_report('peers', ['peers', 'connecting peers', 'unchoked peers', 'peers disk-up', 'peers disk-down', 'peers bw-up', 'peers bw-down', 'error peers'])
-gen_report('peers_list', ['num list peers')
+gen_report('peers_list', ['num list peers'])
-gen_report('rates', ['upload rate', 'download rate', 'disk block buffers'])
+gen_report('overall_rates', ['upload rate', 'download rate', 'smooth upload rate', 'smooth download rate'])
+gen_report('peer_dl_rates', ['peers down 0', 'peers down 0-10', 'peers down 10-50', 'peers down 50-100', 'peers down 500-1000', 'peers down 1000-'])
+gen_report('peer_ul_rates', ['peers up 0', 'peers up 0-10', 'peers up 10-50', 'peers up 50-100', 'peers up 500-1000', 'peers up 1000-'])
+gen_report('disk', ['disk write queued bytes', 'disk block buffers'])
+gen_report('peers_upload', ['peers up interested', 'peers up unchoked', 'peers up requests'])
+gen_report('peers_download', ['peers down interested', 'peers down unchoked', 'peers down requests'])
 

src/disk_io_thread.cpp

@@ -1445,7 +1445,12 @@ namespace libtorrent
 		m_jobs.back().start_time = time_now_hires();
 
 		if (j.action == disk_io_job::write)
+		{
 			m_queue_buffer_size += j.buffer_size;
+			if (m_queue_buffer_size >= m_settings.max_queued_disk_bytes
+				&& m_settings.max_queued_disk_bytes > 0)
+				m_exceeded_write_queue;
+		}
 		m_signal.signal(l);
 	}
 
@@ -1546,6 +1551,10 @@ namespace libtorrent
 
 		for (;;)
 		{
+			// used to indicate whether or not we should post the
+			// 'restart download' event or not.
+			bool post = false;
+
 #ifdef TORRENT_DISK_STATS
 			m_log << log_time() << " idle" << std::endl;
 #endif
@@ -1607,6 +1616,24 @@ namespace libtorrent
 				{
 					TORRENT_ASSERT(m_queue_buffer_size >= j.buffer_size);
 					m_queue_buffer_size -= j.buffer_size;
+					
+					if (m_exceeded_write_queue)
+					{
+						int low_watermark = m_settings.max_queued_disk_bytes_low_watermark == 0
+							? m_settings.max_queued_disk_bytes / 2
+							: m_settings.max_queued_disk_bytes_low_watermark;
+						if (low_watermark >= m_settings.max_queued_disk_bytes)
+							low_watermark = m_settings.max_queued_disk_bytes / 2;
+
+						if (m_queue_buffer_size < low_watermark
+							|| m_settings.max_queued_disk_bytes > 0)
+						{
+							// we just dropped below the high watermark of number of bytes
+							// queued for writing to the disk. Notify the session so that it
+							// can trigger all the connections waiting for this event
+							post = true;
+						}
+					}
 				}
 
 				jl.unlock();
@@ -1703,19 +1730,8 @@ namespace libtorrent
 			disk_buffer_holder holder(*this
 				, operation_has_buffer(j) ? j.buffer : 0);
 
-			bool post = false;
+			if (post && m_queue_callback)
-			if (m_queue_buffer_size + j.buffer_size >= m_settings.max_queued_disk_bytes
+				m_ios.post(m_queue_callback);
-				&& m_queue_buffer_size < m_settings.max_queued_disk_bytes
-				&& m_queue_callback
-				&& m_settings.max_queued_disk_bytes > 0)
-			{
-				// we just dropped below the high watermark of number of bytes
-				// queued for writing to the disk. Notify the session so that it
-				// can trigger all the connections waiting for this event
-				post = true;
-			}
-
-			if (post) m_ios.post(m_queue_callback);
 
 			flush_expired_pieces();
 

src/session_impl.cpp

@@ -237,6 +237,7 @@ namespace aux {
 		TORRENT_SETTING(integer, allowed_fast_set_size)
 		TORRENT_SETTING(integer, suggest_mode)
 		TORRENT_SETTING(integer, max_queued_disk_bytes)
+		TORRENT_SETTING(integer, max_queued_disk_bytes_low_watermark)
 		TORRENT_SETTING(integer, handshake_timeout)
 #ifndef TORRENT_DISABLE_DHT
 		TORRENT_SETTING(boolean, use_dht_as_fallback)
@@ -786,7 +787,6 @@ namespace aux {
 #endif
 
 #ifdef TORRENT_STATS
-//#error log wasted bytes, failed bytes, peers disconnected with an error, disk buffer usage
 		m_stats_logger.open("session_stats.log", std::ios::trunc);
 		m_stats_logger <<
 			"second:upload rate:download rate:downloading torrents:seeding torrents"
@@ -1476,6 +1476,7 @@ namespace aux {
 			|| m_settings.coalesce_writes != s.coalesce_writes
 			|| m_settings.coalesce_reads != s.coalesce_reads
 			|| m_settings.max_queued_disk_bytes != s.max_queued_disk_bytes
+			|| m_settings.max_queued_disk_bytes_low_watermark != s.max_queued_disk_bytes_low_watermark
 			|| m_settings.disable_hash_checks != s.disable_hash_checks
 			|| m_settings.explicit_read_cache != s.explicit_read_cache
 #ifndef TORRENT_DISABLE_MLOCK