make coalesce reads and coalesce writes actually work

include/libtorrent/disk_io_job.hpp

@@ -133,10 +133,7 @@ namespace libtorrent
 
 			// this job is currently being performed, or it's hanging
 			// on a cache piece that may be flushed soon
-			in_progress = 0x20,
-
-			// turns into file::coalesce_buffers in the file operation
-			coalesce_buffers = 0x40
+			in_progress = 0x20
 		};
 
 		// for write jobs, returns true if its block

src/disk_io_thread.cpp

@@ -131,12 +131,12 @@ namespace libtorrent
 #endif
 	}
 
-	int file_flags_for_job(disk_io_job* j)
+	int file_flags_for_job(disk_io_job* j
+		, bool const coalesce_buffers)
 	{
 		int ret = 0;
-
 		if (!(j->flags & disk_io_job::sequential_access)) ret |= file::random_access;
-		if (j->flags & disk_io_job::coalesce_buffers) ret |= file::coalesce_buffers;
+		if (coalesce_buffers) ret |= file::coalesce_buffers;
 		return ret;
 	}
 
@@ -639,6 +639,9 @@ namespace libtorrent
 		DLOG("]\n");
 #endif
 
+		int const file_flags = m_settings.get_bool(settings_pack::coalesce_writes)
+			? file::coalesce_buffers : 0;
+
 		// issue the actual write operation
 		file::iovec_t const* iov_start = iov;
 		int flushing_start = 0;
@@ -652,7 +655,7 @@ namespace libtorrent
 				, i - flushing_start
 				, piece + flushing[flushing_start] / blocks_in_piece
 				, (flushing[flushing_start] % blocks_in_piece) * block_size
-				, 0, error);
+				, file_flags, error);
 			if (ret < 0 || error) failed = true;
 			iov_start = &iov[i];
 			flushing_start = i;
@@ -1188,7 +1191,8 @@ namespace libtorrent
 
 		time_point start_time = clock_type::now();
 
-		int file_flags = file_flags_for_job(j);
+		int const file_flags = file_flags_for_job(j
+			, m_settings.get_bool(settings_pack::coalesce_reads));
 		file::iovec_t b = { j->buffer.disk_block, size_t(j->d.io.buffer_size) };
 
 		int ret = j->storage->get_storage_impl()->readv(&b, 1
@@ -1262,7 +1266,8 @@ namespace libtorrent
 		// can remove them. We can now release the cache mutex and dive into the
 		// disk operations.
 
-		int file_flags = file_flags_for_job(j);
+		int const file_flags = file_flags_for_job(j
+			, m_settings.get_bool(settings_pack::coalesce_reads));
 		time_point start_time = clock_type::now();
 
 		ret = j->storage->get_storage_impl()->readv(iov, iov_len
@@ -1270,7 +1275,7 @@ namespace libtorrent
 
 		if (!j->error.ec)
 		{
-			boost::uint32_t read_time = total_microseconds(clock_type::now() - start_time);
+			boost::uint32_t const read_time = total_microseconds(clock_type::now() - start_time);
 			m_read_time.add_sample(read_time / iov_len);
 
 			m_stats_counters.inc_stats_counter(counters::num_blocks_read, iov_len);
@@ -1419,8 +1424,9 @@ namespace libtorrent
 	{
 		time_point start_time = clock_type::now();
 
-		file::iovec_t b = { j->buffer.disk_block, size_t(j->d.io.buffer_size) };
-		int file_flags = file_flags_for_job(j);
+		file::iovec_t const b = { j->buffer.disk_block, size_t(j->d.io.buffer_size) };
+		int const file_flags = file_flags_for_job(j
+			, m_settings.get_bool(settings_pack::coalesce_writes));
 
 		m_stats_counters.inc_stats_counter(counters::num_writing_threads, 1);
 
@@ -2232,7 +2238,8 @@ namespace libtorrent
 		int const piece_size = j->storage->files()->piece_size(j->piece);
 		int const block_size = m_disk_cache.block_size();
 		int const blocks_in_piece = (piece_size + block_size - 1) / block_size;
-		int const file_flags = file_flags_for_job(j);
+		int const file_flags = file_flags_for_job(j
+			, m_settings.get_bool(settings_pack::coalesce_reads));
 
 		file::iovec_t iov;
 		iov.iov_base = m_disk_cache.allocate_buffer("hashing");
@@ -2278,7 +2285,8 @@ namespace libtorrent
 		INVARIANT_CHECK;
 
 		int const piece_size = j->storage->files()->piece_size(j->piece);
-		int const file_flags = file_flags_for_job(j);
+		int const file_flags = file_flags_for_job(j
+			, m_settings.get_bool(settings_pack::coalesce_reads));
 
 		mutex::scoped_lock l(m_cache_mutex);
 
@@ -2628,7 +2636,8 @@ namespace libtorrent
 			|| m_settings.get_bool(settings_pack::use_read_cache) == false)
 			return 0;
 
-		int file_flags = file_flags_for_job(j);
+		int const file_flags = file_flags_for_job(j
+			, m_settings.get_bool(settings_pack::coalesce_reads));
 
 		mutex::scoped_lock l(m_cache_mutex);
 

src/file.cpp

@@ -1827,8 +1827,16 @@ typedef struct _FILE_ALLOCATED_RANGE_BUFFER {
 		int ret = iov(&::preadv, native_handle(), file_offset, bufs, num_bufs, ec);
 #else
 
+		// there's no point in coalescing single buffer writes
+		if (num_bufs == 1)
+		{
+			flags &= ~file::coalesce_buffers;
+		}
+
 		file::iovec_t tmp;
-		if (flags & file::coalesce_buffers)
+		file::iovec_t const* const orig_bufs = bufs;
+		int const orig_num_bufs = num_bufs;
+		if ((flags & file::coalesce_buffers))
 		{
 			if (!coalesce_read_buffers(bufs, num_bufs, &tmp))
 				// ok, that failed, don't coalesce this read
@@ -1841,8 +1849,8 @@ typedef struct _FILE_ALLOCATED_RANGE_BUFFER {
 		int ret = iov(&::read, native_handle(), file_offset, bufs, num_bufs, ec);
 #endif
 
-		if (flags & file::coalesce_buffers)
-			coalesce_read_buffers_end(bufs, num_bufs
+		if ((flags & file::coalesce_buffers))
+			coalesce_read_buffers_end(orig_bufs, orig_num_bufs
 				, static_cast<char*>(tmp.iov_base), !ec);
 
 #endif
@@ -1877,6 +1885,12 @@ typedef struct _FILE_ALLOCATED_RANGE_BUFFER {
 		int ret = iov(&::pwritev, native_handle(), file_offset, bufs, num_bufs, ec);
 #else
 
+		// there's no point in coalescing single buffer writes
+		if (num_bufs == 1)
+		{
+			flags &= ~file::coalesce_buffers;
+		}
+
 		file::iovec_t tmp;
 		if (flags & file::coalesce_buffers)
 		{

src/storage.cpp

@@ -230,7 +230,9 @@ namespace libtorrent
 			, m_flags(flags)
 		{}
 
-		int file_op(int file_index, boost::int64_t file_offset, int size
+		int file_op(int const file_index
+			, boost::int64_t const file_offset
+			, int const size
 			, file::iovec_t const* bufs, storage_error& ec)
 			TORRENT_OVERRIDE TORRENT_FINAL
 		{
@@ -314,12 +316,14 @@ namespace libtorrent
 
 	struct read_fileop : fileop
 	{
-		read_fileop(default_storage& st, int flags)
+		read_fileop(default_storage& st, int const flags)
 			: m_storage(st)
 			, m_flags(flags)
 		{}
 
-		int file_op(int file_index, boost::int64_t file_offset, int size
+		int file_op(int const file_index
+			, boost::int64_t const file_offset
+			, int const size
 			, file::iovec_t const* bufs, storage_error& ec)
 			TORRENT_OVERRIDE TORRENT_FINAL
 		{
@@ -396,7 +400,7 @@ namespace libtorrent
 
 	private:
 		default_storage& m_storage;
-		int m_flags;
+		int const m_flags;
 	};
 
 	default_storage::default_storage(storage_params const& params)

test/test_file.cpp

@@ -360,3 +360,35 @@ TORRENT_TEST(hard_link)
 		fprintf(stderr, "remove failed: [%s] %s\n", ec.category().name(), ec.message().c_str());
 }
 
+TORRENT_TEST(coalesce_buffer)
+{
+	error_code ec;
+	file f;
+	TEST_CHECK(f.open("test_file", file::read_write, ec));
+	if (ec)
+		fprintf(stderr, "open failed: [%s] %s\n", ec.category().name(), ec.message().c_str());
+	TEST_EQUAL(ec, error_code());
+	if (ec) fprintf(stderr, "%s\n", ec.message().c_str());
+	file::iovec_t b[2] = {{(void*)"test", 4}, {(void*)"foobar", 6}};
+	TEST_EQUAL(f.writev(0, b, 2, ec, file::coalesce_buffers), 4 + 6);
+	if (ec)
+		fprintf(stderr, "writev failed: [%s] %s\n", ec.category().name(), ec.message().c_str());
+	TEST_CHECK(!ec);
+	char test_buf1[5] = {0};
+	char test_buf2[7] = {0};
+	b[0].iov_base = test_buf1;
+	b[0].iov_len = 4;
+	b[1].iov_base = test_buf2;
+	b[1].iov_len = 6;
+	TEST_EQUAL(f.readv(0, b, 2, ec), 4 + 6);
+	if (ec)
+	{
+		fprintf(stderr, "readv failed: [%s] %s\n"
+			, ec.category().name(), ec.message().c_str());
+	}
+	TEST_EQUAL(ec, error_code());
+	TEST_CHECK(strcmp(test_buf1, "test") == 0);
+	TEST_CHECK(strcmp(test_buf2, "foobar") == 0);
+	f.close();
+}
+

test/test_transfer.cpp

@@ -228,7 +228,7 @@ void test_transfer(int proxy_type, settings_pack const& sett
 
 	create_directory("tmp1_transfer", ec);
 	std::ofstream file("tmp1_transfer/temporary");
-	boost::shared_ptr<torrent_info> t = ::create_torrent(&file, "temporary", 16 * 1024, 13, false);
+	boost::shared_ptr<torrent_info> t = ::create_torrent(&file, "temporary", 32 * 1024, 13, false);
 	file.close();
 
 	TEST_CHECK(exists(combine_path("tmp1_transfer", "temporary")));
@@ -421,6 +421,30 @@ TORRENT_TEST(allow_fast)
 	cleanup();
 }
 
+TORRENT_TEST(coalesce_reads)
+{
+	using namespace libtorrent;
+	// test allowed fast
+	settings_pack p;
+	p.set_int(settings_pack::read_cache_line_size, 16);
+	p.set_bool(settings_pack::coalesce_reads, true);
+	test_transfer(0, p, false);
+
+	cleanup();
+}
+
+TORRENT_TEST(coalesce_writes)
+{
+	using namespace libtorrent;
+	// test allowed fast
+	settings_pack p;
+	p.set_bool(settings_pack::coalesce_writes, true);
+	test_transfer(0, p, false);
+
+	cleanup();
+}
+
+
 TORRENT_TEST(allocate)
 {
 	using namespace libtorrent;