Remove all remaining mutexes from core replayer code

The replayer now only has one mutex left, which is specific
to our implementation of Vulkan replay. See
doc/Multithreading.md for more information on it.

Change lava-replay output file to JSON. Also add some extra
info, like frames rendered.

Author: per-mathisen-arm

CMakeLists.txt

@@ -223,6 +223,7 @@ add_executable(mprotect tests/mprotect.cpp src/util.cpp src/util.h)
 target_include_directories(mprotect ${COMMON_INCLUDE})
 target_link_libraries(mprotect ${MOST_COMMON_LIBRARIES} pthread)
 target_compile_options(mprotect PRIVATE ${COMMON_FLAGS})
+target_compile_definitions(mprotect PUBLIC COMPILE_LAYER PUBLIC NO_JSON)
 
 #add_executable(userfaultfd tests/userfaultfd.cpp src/util.cpp src/util.h)
 #target_include_directories(userfaultfd ${COMMON_INCLUDE})
@@ -370,20 +371,23 @@ add_executable(write4 tests/write4.cpp src/filewriter.cpp src/util.cpp)
 target_include_directories(write4 ${COMMON_INCLUDE})
 target_link_libraries(write4 ${MOST_COMMON_LIBRARIES} density LZ4::LZ4 pthread)
 target_compile_options(write4 PRIVATE ${COMMON_FLAGS})
+target_compile_definitions(write4 PUBLIC COMPILE_LAYER NO_JSON)
 add_dependencies(write4 sync_generated)
 add_test(NAME write_test_4 COMMAND ${CMAKE_CURRENT_BINARY_DIR}/write4)
 
 add_executable(read4 tests/read4.cpp src/filereader.cpp src/util.cpp)
 target_include_directories(read4 ${COMMON_INCLUDE})
 target_link_libraries(read4 ${MOST_COMMON_LIBRARIES} density LZ4::LZ4 pthread)
 target_compile_options(read4 PRIVATE ${COMMON_FLAGS})
+target_compile_definitions(read4 PUBLIC COMPILE_LAYER NO_JSON)
 add_dependencies(read4 sync_generated)
 add_test(NAME read_test_4 COMMAND ${CMAKE_CURRENT_BINARY_DIR}/read4)
 set_property(TEST read_test_4 APPEND PROPERTY DEPENDS write_test_4)
 
 add_executable(write5 tests/write5.cpp src/filewriter.cpp src/filereader.cpp src/util.cpp)
 target_include_directories(write5 ${COMMON_INCLUDE})
 target_link_libraries(write5 ${MOST_COMMON_LIBRARIES} density LZ4::LZ4 pthread)
+target_compile_definitions(write5 PUBLIC COMPILE_LAYER NO_JSON)
 target_compile_options(write5 PRIVATE ${COMMON_FLAGS})
 add_dependencies(write5 sync_generated)
 add_test(NAME write_test_5 COMMAND ${CMAKE_CURRENT_BINARY_DIR}/write5)

src/filereader.cpp

@@ -134,7 +134,6 @@ void file_reader::decompressor()
 		decompress_chunk();
 	}
 
-	chunk_mutex.lock();
 	clockid_t id;
 	int r = pthread_getcpuclockid(pthread_self(), &id);
 	if (r != 0)
@@ -145,7 +144,6 @@ void file_reader::decompressor()
 	{
 		ELOG("Failed to get worker thread %u CPU usage: %s", tid, strerror(errno));
 	}
-	chunk_mutex.unlock();
 }
 
 void file_reader::start_measurement()
@@ -184,7 +182,6 @@ void file_reader::start_measurement()
 
 void file_reader::stop_measurement(uint64_t& worker, uint64_t& runner)
 {
-	chunk_mutex.lock();
 	struct timespec stop_runner_cpu_usage;
 	clockid_t id;
 	int r = pthread_getcpuclockid(pthread_self(), &id);
@@ -202,7 +199,6 @@ void file_reader::stop_measurement(uint64_t& worker, uint64_t& runner)
 	if (!multithreaded_read)
 	{
 		worker = 0;
-		chunk_mutex.unlock();
 		return;
 	}
 	pthread_t t = decompressor_thread.native_handle();
@@ -218,5 +214,4 @@ void file_reader::stop_measurement(uint64_t& worker, uint64_t& runner)
 	}
 	assert(stop_worker_cpu_usage.tv_sec >= worker_cpu_usage.tv_sec);
 	worker = diff_timespec(&stop_worker_cpu_usage, &worker_cpu_usage);
-	chunk_mutex.unlock();
 }

src/filereader.h

@@ -9,7 +9,6 @@
 #include <stdio.h>
 #include <stdarg.h>
 
-#include "lavamutex.h"
 #include "packfile.h"
 #include "containers.h"
 #include "util.h"
@@ -163,11 +162,9 @@ class file_reader
 
 	void disable_multithreaded_read() // we can only disable on the fly, enable makes less sense
 	{
-		chunk_mutex.lock();
 		done_decompressing = true;
 		decompressor_thread.join();
 		multithreaded_read = false;
-		chunk_mutex.unlock();
 	}
 
 	void self_test() const
@@ -186,7 +183,6 @@ class file_reader
 
 	bool multithreaded_read = true;
 	unsigned tid = -1; // only used for logging
-	lava::mutex chunk_mutex;
 	/// Pointer to mapped memory of compressed file
 	char* compressed_data = nullptr; // current position in compressed buffer
 	char* fstart = nullptr; // start position

src/read.cpp

@@ -96,26 +96,25 @@ uint16_t lava_file_reader::read_apicall()
 
 lava_reader::~lava_reader()
 {
-	global_mutex.lock();
 	delete thread_call_numbers;
 	for (auto& t : thread_streams)
 	{
-		delete t.second;
+		delete t;
 	}
 	thread_streams.clear();
-	global_mutex.unlock();
 }
 
 void lava_reader::finalize(bool terminate)
 {
 	const double total_time_ms = ((gettime() - mStartTime.load()) / 1000000UL);
 	const double fps = (double)mGlobalFrames / (total_time_ms / 1000.0);
 	ILOG("==== %.2f ms, %u frames (%.2f fps) ====", total_time_ms, mGlobalFrames, fps);
-	fprintf(out_fptr, "%.2f", fps);
-	fclose(out_fptr);
+	Json::Value out;
+	out["fps"] = fps;
+	out["frames"] = mGlobalFrames;
+	out["time"] = total_time_ms;
 	uint64_t runner = 0;
 	uint64_t worker = 0;
-	global_mutex.lock();
 	for (unsigned i = 0; i < threads.size(); i++)
 	{
 		uint64_t runner_local = 0;
@@ -133,6 +132,9 @@ void lava_reader::finalize(bool terminate)
 	assert(stop_process_cpu_usage.tv_sec >= process_cpu_usage.tv_sec);
 	const uint64_t process_time = diff_timespec(&stop_process_cpu_usage, &process_cpu_usage);
 	ILOG("CPU time spent in ms - readhead workers %lu, API runners %lu, full process %lu", (long unsigned)worker, (long unsigned)runner, (long unsigned)process_time);
+	out["readahead_workers_time"] = (Json::Value::UInt64)worker;
+	out["api_runners_time"] = (Json::Value::UInt64)runner;
+	out["process_time"] = (Json::Value::UInt64)process_time;
 	if (terminate)
 	{
 		for (auto& v : *thread_call_numbers) v = 0; // stop waiting threads from progressing
@@ -141,12 +143,12 @@ void lava_reader::finalize(bool terminate)
 			if (!thread_streams[i]->terminated.load()) pthread_cancel(threads[i].native_handle());
 		}
 	}
-	global_mutex.unlock();
+	write_json(out_fptr, out);
+	fclose(out_fptr);
 }
 
 lava_file_reader& lava_reader::file_reader(uint16_t thread_id)
 {
-	lava::lock_guard keep(global_mutex);
 	return *thread_streams.at(thread_id);
 }
 
@@ -201,10 +203,14 @@ void lava_reader::init(const std::string& path, int heap_size)
 	}
 
 	Json::Value meta = packed_json("metadata.json", mPackedFile);
-	mGlobalFrames = meta["global_frames"].asInt();
 	const int num_threads = meta["threads"].asInt();
-	global_mutex.lock();
+	mGlobalFrames = meta["global_frames"].asInt();
+
+	// initialize threads -- note that this happens before threading begins, so thread safe
+	threads.resize(num_threads);
+	thread_streams.resize(num_threads);
 	thread_call_numbers = new std::vector<std::atomic_uint_fast32_t>(num_threads);
+
 	for (int thread_id = 0; thread_id < num_threads; thread_id++)
 	{
 		Json::Value frameinfo = packed_json("frames_" + _to_string(thread_id) + ".json", path);
@@ -217,12 +223,10 @@ void lava_reader::init(const std::string& path, int heap_size)
 				if (v["global_frame"] == mEnd + 1) { uncompressed_target = v["position"].asUInt(); break; }
 			}
 		}
-		lava_file_reader* f = new lava_file_reader(this, mPackedFile, thread_id, mGlobalFrames, frameinfo, uncompressed_size, uncompressed_target, mStart, mEnd);
-		thread_streams.emplace(thread_id, std::move(f));
+		thread_streams[thread_id] = new lava_file_reader(this, mPackedFile, thread_id, mGlobalFrames, frameinfo, uncompressed_size, uncompressed_target, mStart, mEnd);
 	}
-	global_mutex.unlock();
 
-	out_fptr = fopen("lavaresults.txt", "w");
+	out_fptr = fopen("lavaresults.json", "w");
 	if (!out_fptr) ABORT("Failed to open results file: %s", strerror(errno));
 	mStartTime.store(gettime());
 }

src/read.h

@@ -100,9 +100,8 @@ class lava_reader
 	std::atomic_uint64_t mStartTime{ 0 };
 	/// Start CPU usage for whole process
 	struct timespec process_cpu_usage;
-	lava::mutex global_mutex;
 	std::string mPackedFile;
-	std::unordered_map<int, lava_file_reader*> thread_streams GUARDED_BY(global_mutex);
+	std::vector<lava_file_reader*> thread_streams;
 	int mStart = 0;
 	int mEnd = -1;
 	int mGlobalFrames = 0;
@@ -182,12 +181,10 @@ class lava_file_reader : public file_reader
 					ELOG("Failed to get process CPU usage: %s", strerror(errno));
 				}
 				// Set start time in all threads
-				parent->global_mutex.lock();
 				for (unsigned i = 0; i < parent->threads.size(); i++)
 				{
 					parent->thread_streams[i]->start_measurement();
 				}
-				parent->global_mutex.unlock();
 			}
 		}
 		current.frame++;

src/replay.cpp

@@ -115,17 +115,17 @@ static void replay_thread(int thread_id)
 	}
 }
 
-static void run_multithreaded(int n)
+static void run_multithreaded()
 {
 	if (sandbox)
 	{
 		const char* err = sandbox_replay_start();
 		if (err) WLOG("Warning: Failed to change sandbox to replay mode: %s", err);
 	}
 
-	for (int i = 0; i < n; i++)
+	for (unsigned i = 0; i < replayer.threads.size(); i++)
 	{
-		replayer.threads.emplace_back(replay_thread, i);
+		replayer.threads[i] = std::thread(replay_thread, i);
 	}
 
 	for (unsigned i = 0; i < replayer.threads.size(); i++)
@@ -318,10 +318,7 @@ int main(int argc, char **argv)
 		exit(EXIT_SUCCESS);
 	}
 
-	// Read all thread files
-	std::vector<std::string> threadfiles = packed_files(filename, "thread_");
-	if (threadfiles.size() == 0) DIE("Failed to find any threads in %s!", filename.c_str());
-	run_multithreaded(threadfiles.size());
+	run_multithreaded();
 	if (p__custom_allocator) allocators_print(stdout);
 	vkuDestroyWrapper(library);
 	if (p__debug_destination) fclose(p__debug_destination);

src/tool.cpp

@@ -271,24 +271,20 @@ int main(int argc, char **argv)
 		replayer.init(filename_input, heap_size);
 		replayer.remap = validate_remap;
 
-		// Read all thread files
-		std::vector<std::string> threadfiles = packed_files(filename_input, "thread_");
-		if (threadfiles.size() == 0) DIE("Failed to find any threads in %s!", filename_input.c_str());
-
 		// Add callbacks
 		vkCreateShaderModule_callbacks.push_back(callback_vkCreateShaderModule);
 		vkDestroyDevice_callbacks.push_back(callback_vkDestroyDevice);
 
-		for (int i = 0; i < (int)threadfiles.size(); i++)
+		for (unsigned i = 0; i < replayer.threads.size(); i++)
 		{
 			if (verbose)
 			{
 				printf("Threads:\n");
 				Json::Value frameinfo = packed_json("frames_" + _to_string(i) + ".json", filename_input);
-				printf("\t%d : [%s] with %u local frames, %d highest global frame, %u uncompressed size\n", i, frameinfo.get("thread_name", "unknown").asString().c_str(),
+				printf("\t%u : [%s] with %u local frames, %d highest global frame, %u uncompressed size\n", i, frameinfo.get("thread_name", "unknown").asString().c_str(),
 					(unsigned)frameinfo["frames"].size(), frameinfo["highest_global_frame"].asInt(), frameinfo["uncompressed_size"].asUInt());
 			}
-			replayer.threads.emplace_back(&replay_thread, &replayer, i);
+			replayer.threads[i] = std::thread(&replay_thread, &replayer, i);
 		}
 		for (unsigned i = 0; i < replayer.threads.size(); i++)
 		{
@@ -313,13 +309,9 @@ int main(int argc, char **argv)
 		// Add in the rewrite queue from the previous run
 		replayer.rewrite_queue = rewrite_queue_copy;
 
-		// Read all thread files
-		std::vector<std::string> threadfiles = packed_files(filename_input, "thread_");
-		if (threadfiles.size() == 0) DIE("Failed to find any threads in %s!", filename_input.c_str());
-
-		for (int i = 0; i < (int)threadfiles.size(); i++)
+		for (unsigned i = 0; i < replayer.threads.size(); i++)
 		{
-			replayer.threads.emplace_back(&replay_thread, &replayer, i);
+			replayer.threads[i] = std::thread(&replay_thread, &replayer, i);
 		}
 		for (unsigned i = 0; i < replayer.threads.size(); i++)
 		{

src/util.cpp

@@ -1,4 +1,5 @@
 #include "util.h"
+#include "jsoncpp/json/writer.h"
 
 #include <string.h>
 #include <spirv/unified1/spirv.h>
@@ -386,3 +387,33 @@ const char* pretty_print_VkObjectType(VkObjectType val)
 	}
 	return "Error";
 }
+
+#ifndef NO_JSON
+void write_json(FILE* fp, const Json::Value& v)
+{
+	Json::StyledWriter writer;
+	std::string data = writer.write(v);
+	size_t written;
+	int err = 0;
+	do {
+		written = fwrite(data.c_str(), data.size(), 1, fp);
+		err = ferror(fp);
+	} while (!err && !written);
+	if (err)
+	{
+		ELOG("Failed to write dictionary: %s", strerror(err));
+	}
+}
+
+void write_json(const std::string& path, const Json::Value& v)
+{
+	FILE* fp = fopen(path.c_str(), "w");
+	if (!fp)
+	{
+		ELOG("Failed to open \"%s\": %s", path.c_str(), strerror(errno));
+		return;
+	}
+	write_json(fp, v);
+	fclose(fp);
+}
+#endif

src/util.h

@@ -15,6 +15,9 @@
 #include <assert.h>
 #include <vector>
 #include <string>
+#ifndef NO_JSON
+#include "jsoncpp/json/value.h"
+#endif
 
 #define VK_NO_PROTOTYPES
 #include "vulkan_utility.h"
@@ -251,3 +254,8 @@ typedef uint64_t (VKAPI_PTR *PFN_vkGetDeviceTracingObjectPropertyTRACETOOLTEST)(
 int get_env_int(const char* name, int v);
 int get_env_bool(const char* name, int v);
 FILE* get_env_file(const char* name, FILE* fallback);
+
+#ifndef NO_JSON
+void write_json(const std::string& path, const Json::Value& v);
+void write_json(FILE* fp, const Json::Value& v);
+#endif