Add the possibility to compile filewriter in single-thread mode

Now we can also measure the perf impact of multi-threading when
turning off the compression thread, the write-out threa, or
both.

Also adjust sleep times of the write helper threads as our own
unit tests ran much faster without threads until they were
lowered.

Author: per-mathisen-arm

src/filereader.cpp

@@ -23,7 +23,7 @@ file_reader::file_reader(const std::string& filename, unsigned mytid) : tid(myti
 	done_reading = false;
 	readahead_chunks.exchange(2);
 	decompressor_thread = std::thread(&file_reader::decompressor, this);
-	DLOG("%s opened for reading (size %lu) and decompressor thread launched!", filename.c_str(), (unsigned long)total_left);
+	DLOG("%s opened for reading (size %lu) and decompressor thread %u launched!", filename.c_str(), (unsigned long)total_left, tid);
 }
 
 file_reader::file_reader(packed pf, unsigned mytid) : tid(mytid), mFilename(pf.inside)
@@ -51,7 +51,7 @@ file_reader::~file_reader()
 	fclose(fp);
 	if (times_caught_decompressor) // this would be bad
 	{
-		ELOG("We caught up with the decompressor thread %u times!", (unsigned)times_caught_decompressor);
+		ELOG("We caught up with the decompressor thread %u times on thread %u!", (unsigned)times_caught_decompressor, tid);
 	}
 }
 

src/filewriter.cpp

@@ -98,8 +98,19 @@ void file_writer::finalize()
 	}
 	// whatever is left in our current buffer, move to work list
 	chunk_mutex.lock();
+	printf("Filewriter finalizing thread %u: %lu total bytes, %lu in last chunk, %d uncompressed chunks, and %d compressed chunks to be written out\n",
+	       mTid, (unsigned long)uncompressed_bytes, (unsigned long)uidx, (int)uncompressed_chunks.size(), (int)compressed_chunks.size());
 	chunk.shrink(uidx);
+#ifndef MULTITHREADED_COMPRESS
+	chunk = compress_chunk(chunk);
+#ifdef MULTITHREADED_WRITE
+	compressed_chunks.push_front(chunk);
+#else
+	write_chunk(chunk);
+#endif
+#else
 	uncompressed_chunks.push_front(chunk);
+#endif
 	chunk = buffer(uncompressed_chunk_size); // ready to go again
 	chunk_mutex.unlock();
 	// wrap up work in work lists
@@ -127,10 +138,31 @@ file_writer::~file_writer()
 	chunk.release();
 }
 
+void file_writer::write_chunk(buffer& active)
+{
+	size_t written = 0;
+	size_t size = active.size();
+	char* ptr = active.data();
+	int err = 0;
+	do {
+		written = fwrite(ptr, 1, size, fp);
+		ptr += written;
+		size -= written;
+		err = ferror(fp);
+	} while (size > 0 && (err == EAGAIN || err == EWOULDBLOCK || err == EINTR));
+	if (size > 0)
+	{
+		ELOG("Failed to write out file (%u bytes left): %s", (unsigned)size, strerror(ferror(fp)));
+	}
+	DLOG3("Filewriter thread %d wrote out compressed buffer of %lu size\n", mTid, (unsigned long)active.size());
+	active.release();
+}
+
 void file_writer::serializer()
 {
 	// lock, steal compressed buffer, unlock, store to disk, sleep, repeat
 	set_thread_name("serializer");
+#ifdef MULTITHREADED_WRITE
 	while (1)
 	{
 		buffer active;
@@ -151,36 +183,42 @@ void file_writer::serializer()
 		// save compressed buffer
 		if (active.size() > 0)
 		{
-			size_t written = 0;
-			size_t size = active.size();
-			char* ptr = active.data();
-			int err = 0;
-			DLOG3("\thunk=%u", (unsigned)size);
-			do {
-				written = fwrite(ptr, 1, size, fp);
-				DLOG3("\t\twritten=%u / %u", (unsigned)written, (unsigned)size);
-				ptr += written;
-				size -= written;
-				err = ferror(fp);
-			} while (size > 0 && (err == EAGAIN || err == EWOULDBLOCK || err == EINTR));
-			if (size > 0)
-			{
-				ELOG("Failed to write out file (%u bytes left): %s", (unsigned)size, strerror(ferror(fp)));
-			}
-			active.release();
+			write_chunk(active);
 		}
 		// if not done and no work done, wait a bit
 		else if (!done_compressing)
 		{
-			usleep(10000);
+			usleep(2000);
 		}
 	}
+#endif
+}
+
+buffer file_writer::compress_chunk(buffer& uncompressed)
+{
+	const uint64_t header_size = sizeof(uint64_t) * 2;
+	uint64_t compressed_size = density_compress_safe_size(uncompressed.size()) + header_size;
+	buffer compressed(compressed_size);
+	density_processing_result result = density_compress((const uint8_t *)uncompressed.data(), uncompressed.size(),
+	                                                    (uint8_t *)compressed.data() + header_size, compressed.size(),
+	                                                    DENSITY_ALGORITHM_CHEETAH);
+	uncompressed.release();
+	if (result.state != DENSITY_STATE_OK)
+	{
+		ABORT("Failed to compress buffer - aborting from compression thread");
+	}
+	uint64_t header[2] = { result.bytesWritten, result.bytesRead }; // store compressed and uncompressed sizes
+	memcpy(compressed.data(), header, header_size); // use memcpy to avoid aliasing issues
+	compressed.shrink(result.bytesWritten + header_size);
+	DLOG3("Filewriter thread %d handing over compressed buffer of %lu bytes, was %lu bytes uncompressed", mTid, (unsigned long)(result.bytesWritten + header_size), (unsigned long)result.bytesRead);
+	return compressed;
 }
 
 void file_writer::compressor()
 {
 	// lock, grab pointer to uncompressed, make new compressed, unlock, compress, sleep, repeat
 	set_thread_name("compressor");
+#ifdef MULTITHREADED_COMPRESS
 	while (1)
 	{
 		buffer uncompressed;
@@ -202,29 +240,20 @@ void file_writer::compressor()
 		// compress it
 		if (uncompressed.size() > 0)
 		{
-			const uint64_t header_size = sizeof(uint64_t) * 2;
-			uint64_t compressed_size = density_compress_safe_size(uncompressed.size()) + header_size;
-			buffer compressed(compressed_size);
-			density_processing_result result = density_compress((const uint8_t *)uncompressed.data(), uncompressed.size(),
-			                                                    (uint8_t *)compressed.data() + header_size, compressed.size(),
-			                                                    DENSITY_ALGORITHM_CHEETAH);
-			uncompressed.release();
-			if (result.state != DENSITY_STATE_OK)
-			{
-				ELOG("Failed to compress buffer - aborting compression thread");
-				break;
-			}
-			uint64_t header[2] = { result.bytesWritten, result.bytesRead }; // store compressed and uncompressed sizes
-			memcpy(compressed.data(), header, header_size); // use memcpy to avoid aliasing issues
-			compressed.shrink(result.bytesWritten + header_size);
+			buffer compressed = compress_chunk(uncompressed);
+#ifdef MULTITHREADED_WRITE
 			chunk_mutex.lock();
 			compressed_chunks.push_front(compressed);
 			chunk_mutex.unlock();
+#else
+			write_chunk(compressed);
+#endif
 		}
 		// if not done and no work done, wait a bit
 		else if (!done_feeding)
 		{
-			usleep(100000);
+			usleep(2000);
 		}
 	}
+#endif
 }

src/filewriter.h

@@ -11,6 +11,9 @@
 #include "lavamutex.h"
 #include "util.h"
 
+#define MULTITHREADED_COMPRESS
+#define MULTITHREADED_WRITE
+
 class file_writer
 {
 	file_writer(const file_writer&) = delete;
@@ -21,9 +24,18 @@ class file_writer
 		// shrink existing chunk to actually used size
 		chunk.shrink(uidx);
 		// move chunk into list of chunks to compress
+#ifdef MULTITHREADED_COMPRESS
 		chunk_mutex.lock();
 		uncompressed_chunks.push_front(chunk);
 		chunk_mutex.unlock();
+#else
+		buffer compressed = compress_chunk(chunk);
+#ifdef MULTITHREADED_WRITE
+		compressed_chunks.push_front(compressed);
+#else
+		write_chunk(compressed);
+#endif
+#endif
 		// create a new chunk for writing into (we could employ a free list here as a possible optimization)
 		if (size > uncompressed_chunk_size) // make sure our new chunk is big enough
 		{
@@ -141,6 +153,8 @@ class file_writer
 private:
 	void compressor(); // runs in separate thread, moves chunks from uncompressed to compressed
 	void serializer(); // runs in separate thread, moves chunks from compressed to disk
+	buffer compress_chunk(buffer& uncompressed); // returns compressed buffer
+	void write_chunk(buffer& active);
 
 	lava::mutex chunk_mutex;
 	FILE* fp = nullptr;