core/command_queue_mt: Customizable size and tests

Adds unit tests for command_queue_mt.h/cpp
In this revision, some unit tests will fail due to issue #42107.

4 files changed, 493 insertions, 5 deletions

diff --git a/core/command_queue_mt.cpp b/core/command_queue_mt.cpp
index ace210ca2c..95fcd2c70e 100644
--- a/core/command_queue_mt.cpp
+++ b/core/command_queue_mt.cpp
@@ -31,6 +31,7 @@
 #include "command_queue_mt.h"
 
 #include "core/os/os.h"
+#include "core/project_settings.h"
 
 void CommandQueueMT::lock() {
 	mutex.lock();
@@ -94,6 +95,10 @@ tryagain:
 }
 
 CommandQueueMT::CommandQueueMT(bool p_sync) {
+	command_mem_size = GLOBAL_DEF_RST("memory/limits/command_queue/multithreading_queue_size_kb", DEFAULT_COMMAND_MEM_SIZE_KB);
+	ProjectSettings::get_singleton()->set_custom_property_info("memory/limits/command_queue/multithreading_queue_size_kb", PropertyInfo(Variant::INT, "memory/limits/command_queue/multithreading_queue_size_kb", PROPERTY_HINT_RANGE, "1,4096,1,or_greater"));
+	command_mem_size *= 1024;
+	command_mem = (uint8_t *)memalloc(command_mem_size);
 	if (p_sync) {
 		sync = memnew(Semaphore);
 	}
diff --git a/core/command_queue_mt.h b/core/command_queue_mt.h
index d7a6a5bc43..7369c655ba 100644
--- a/core/command_queue_mt.h
+++ b/core/command_queue_mt.h
@@ -330,15 +330,15 @@ class CommandQueueMT {
 	/***** BASE *******/
 
 	enum {
-		COMMAND_MEM_SIZE_KB = 256,
-		COMMAND_MEM_SIZE = COMMAND_MEM_SIZE_KB * 1024,
+		DEFAULT_COMMAND_MEM_SIZE_KB = 256,
 		SYNC_SEMAPHORES = 8
 	};
 
-	uint8_t *command_mem = (uint8_t *)memalloc(COMMAND_MEM_SIZE);
+	uint8_t *command_mem = nullptr;
 	uint32_t read_ptr = 0;
 	uint32_t write_ptr = 0;
 	uint32_t dealloc_ptr = 0;
+	uint32_t command_mem_size = 0;
 	SyncSemaphore sync_sems[SYNC_SEMAPHORES];
 	Mutex mutex;
 	Semaphore *sync = nullptr;
@@ -362,7 +362,7 @@ class CommandQueueMT {
 		} else {
 			// ahead of dealloc_ptr, check that there is room
 
-			if ((COMMAND_MEM_SIZE - write_ptr) < alloc_size + sizeof(uint32_t)) {
+			if ((command_mem_size - write_ptr) < alloc_size + sizeof(uint32_t)) {
 				// no room at the end, wrap down;
 
 				if (dealloc_ptr == 0) { // don't want write_ptr to become dealloc_ptr
@@ -375,7 +375,7 @@ class CommandQueueMT {
 				}
 
 				// if this happens, it's a bug
-				ERR_FAIL_COND_V((COMMAND_MEM_SIZE - write_ptr) < 8, nullptr);
+				ERR_FAIL_COND_V((command_mem_size - write_ptr) < 8, nullptr);
 				// zero means, wrap to beginning
 
 				uint32_t *p = (uint32_t *)&command_mem[write_ptr];
diff --git a/tests/test_command_queue.h b/tests/test_command_queue.h
new file mode 100644
index 0000000000..a66fd3c86e
--- /dev/null
+++ b/tests/test_command_queue.h
@@ -0,0 +1,482 @@
+/*************************************************************************/
+/*  test_command_queue.h                                                 */
+/*************************************************************************/
+/*                       This file is part of:                           */
+/*                           GODOT ENGINE                                */
+/*                      https://godotengine.org                          */
+/*************************************************************************/
+/* Copyright (c) 2007-2020 Juan Linietsky, Ariel Manzur.                 */
+/* Copyright (c) 2014-2020 Godot Engine contributors (cf. AUTHORS.md).   */
+/*                                                                       */
+/* Permission is hereby granted, free of charge, to any person obtaining */
+/* a copy of this software and associated documentation files (the       */
+/* "Software"), to deal in the Software without restriction, including   */
+/* without limitation the rights to use, copy, modify, merge, publish,   */
+/* distribute, sublicense, and/or sell copies of the Software, and to    */
+/* permit persons to whom the Software is furnished to do so, subject to */
+/* the following conditions:                                             */
+/*                                                                       */
+/* The above copyright notice and this permission notice shall be        */
+/* included in all copies or substantial portions of the Software.       */
+/*                                                                       */
+/* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,       */
+/* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF    */
+/* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.*/
+/* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY  */
+/* CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,  */
+/* TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE     */
+/* SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.                */
+/*************************************************************************/
+
+#ifndef TEST_COMMAND_QUEUE_H
+#define TEST_COMMAND_QUEUE_H
+
+#include "test_command_queue.h"
+
+#include "core/command_queue_mt.h"
+#include "core/os/mutex.h"
+#include "core/os/os.h"
+#include "core/os/semaphore.h"
+#include "core/os/thread.h"
+#include "core/project_settings.h"
+
+#if !defined(NO_THREADS)
+
+namespace TestCommandQueue {
+
+class ThreadWork {
+	Semaphore thread_sem;
+	Semaphore main_sem;
+	Mutex mut;
+	int threading_errors = 0;
+	enum State {
+		MAIN_START,
+		MAIN_DONE,
+		THREAD_START,
+		THREAD_DONE,
+	} state;
+
+public:
+	ThreadWork() {
+		mut.lock();
+		state = MAIN_START;
+	}
+	~ThreadWork() {
+		CHECK_MESSAGE(threading_errors == 0, "threads did not lock/unlock correctly");
+	}
+	void thread_wait_for_work() {
+		thread_sem.wait();
+		mut.lock();
+		if (state != MAIN_DONE) {
+			threading_errors++;
+		}
+		state = THREAD_START;
+	}
+	void thread_done_work() {
+		if (state != THREAD_START) {
+			threading_errors++;
+		}
+		state = THREAD_DONE;
+		mut.unlock();
+		main_sem.post();
+	}
+
+	void main_wait_for_done() {
+		main_sem.wait();
+		mut.lock();
+		if (state != THREAD_DONE) {
+			threading_errors++;
+		}
+		state = MAIN_START;
+	}
+	void main_start_work() {
+		if (state != MAIN_START) {
+			threading_errors++;
+		}
+		state = MAIN_DONE;
+		mut.unlock();
+		thread_sem.post();
+	}
+};
+
+class SharedThreadState {
+public:
+	ThreadWork reader_threadwork;
+	ThreadWork writer_threadwork;
+
+	CommandQueueMT command_queue = CommandQueueMT(true);
+
+	enum TestMsgType {
+		TEST_MSG_FUNC1_TRANSFORM,
+		TEST_MSG_FUNC2_TRANSFORM_FLOAT,
+		TEST_MSG_FUNC3_TRANSFORMx6,
+		TEST_MSGSYNC_FUNC1_TRANSFORM,
+		TEST_MSGSYNC_FUNC2_TRANSFORM_FLOAT,
+		TEST_MSGRET_FUNC1_TRANSFORM,
+		TEST_MSGRET_FUNC2_TRANSFORM_FLOAT,
+		TEST_MSG_MAX
+	};
+
+	Vector<TestMsgType> message_types_to_write;
+	bool during_writing = false;
+	int message_count_to_read = 0;
+	bool exit_threads = false;
+
+	Thread *reader_thread = nullptr;
+	Thread *writer_thread = nullptr;
+
+	int func1_count = 0;
+
+	void func1(Transform t) {
+		func1_count++;
+	}
+	void func2(Transform t, float f) {
+		func1_count++;
+	}
+	void func3(Transform t1, Transform t2, Transform t3, Transform t4, Transform t5, Transform t6) {
+		func1_count++;
+	}
+	Transform func1r(Transform t) {
+		func1_count++;
+		return t;
+	}
+	Transform func2r(Transform t, float f) {
+		func1_count++;
+		return t;
+	}
+
+	void add_msg_to_write(TestMsgType type) {
+		message_types_to_write.push_back(type);
+	}
+
+	void reader_thread_loop() {
+		reader_threadwork.thread_wait_for_work();
+		while (!exit_threads) {
+			if (message_count_to_read < 0) {
+				command_queue.flush_all();
+			}
+			for (int i = 0; i < message_count_to_read; i++) {
+				command_queue.wait_and_flush_one();
+			}
+			message_count_to_read = 0;
+
+			reader_threadwork.thread_done_work();
+			reader_threadwork.thread_wait_for_work();
+		}
+		command_queue.flush_all();
+		reader_threadwork.thread_done_work();
+	}
+	static void static_reader_thread_loop(void *stsvoid) {
+		SharedThreadState *sts = static_cast<SharedThreadState *>(stsvoid);
+		sts->reader_thread_loop();
+	}
+
+	void writer_thread_loop() {
+		during_writing = false;
+		writer_threadwork.thread_wait_for_work();
+		while (!exit_threads) {
+			Transform tr;
+			Transform otr;
+			float f = 1;
+			during_writing = true;
+			for (int i = 0; i < message_types_to_write.size(); i++) {
+				TestMsgType msg_type = message_types_to_write[i];
+				switch (msg_type) {
+					case TEST_MSG_FUNC1_TRANSFORM:
+						command_queue.push(this, &SharedThreadState::func1, tr);
+						break;
+					case TEST_MSG_FUNC2_TRANSFORM_FLOAT:
+						command_queue.push(this, &SharedThreadState::func2, tr, f);
+						break;
+					case TEST_MSG_FUNC3_TRANSFORMx6:
+						command_queue.push(this, &SharedThreadState::func3, tr, tr, tr, tr, tr, tr);
+						break;
+					case TEST_MSGSYNC_FUNC1_TRANSFORM:
+						command_queue.push_and_sync(this, &SharedThreadState::func1, tr);
+						break;
+					case TEST_MSGSYNC_FUNC2_TRANSFORM_FLOAT:
+						command_queue.push_and_sync(this, &SharedThreadState::func2, tr, f);
+						break;
+					case TEST_MSGRET_FUNC1_TRANSFORM:
+						command_queue.push_and_ret(this, &SharedThreadState::func1r, tr, &otr);
+						break;
+					case TEST_MSGRET_FUNC2_TRANSFORM_FLOAT:
+						command_queue.push_and_ret(this, &SharedThreadState::func2r, tr, f, &otr);
+						break;
+					default:
+						break;
+				}
+			}
+			message_types_to_write.clear();
+			during_writing = false;
+
+			writer_threadwork.thread_done_work();
+			writer_threadwork.thread_wait_for_work();
+		}
+		writer_threadwork.thread_done_work();
+	}
+	static void static_writer_thread_loop(void *stsvoid) {
+		SharedThreadState *sts = static_cast<SharedThreadState *>(stsvoid);
+		sts->writer_thread_loop();
+	}
+
+	void init_threads() {
+		reader_thread = Thread::create(&SharedThreadState::static_reader_thread_loop, this);
+		writer_thread = Thread::create(&SharedThreadState::static_writer_thread_loop, this);
+	}
+	void destroy_threads() {
+		exit_threads = true;
+		reader_threadwork.main_start_work();
+		writer_threadwork.main_start_work();
+
+		Thread::wait_to_finish(reader_thread);
+		memdelete(reader_thread);
+		reader_thread = nullptr;
+		Thread::wait_to_finish(writer_thread);
+		memdelete(writer_thread);
+		writer_thread = nullptr;
+	}
+};
+
+TEST_CASE("[CommandQueue] Test Queue Basics") {
+	const char *COMMAND_QUEUE_SETTING = "memory/limits/command_queue/multithreading_queue_size_kb";
+	ProjectSettings::get_singleton()->set_setting(COMMAND_QUEUE_SETTING, 1);
+	SharedThreadState sts;
+	sts.init_threads();
+
+	sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC1_TRANSFORM);
+	sts.writer_threadwork.main_start_work();
+	sts.writer_threadwork.main_wait_for_done();
+	CHECK_MESSAGE(sts.func1_count == 0,
+			"Control: no messages read before reader has run.");
+
+	sts.message_count_to_read = 1;
+	sts.reader_threadwork.main_start_work();
+	sts.reader_threadwork.main_wait_for_done();
+	CHECK_MESSAGE(sts.func1_count == 1,
+			"Reader should have read one message");
+
+	sts.message_count_to_read = -1;
+	sts.reader_threadwork.main_start_work();
+	sts.reader_threadwork.main_wait_for_done();
+	CHECK_MESSAGE(sts.func1_count == 1,
+			"Reader should have read no additional messages from flush_all");
+
+	sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC1_TRANSFORM);
+	sts.writer_threadwork.main_start_work();
+	sts.writer_threadwork.main_wait_for_done();
+
+	sts.message_count_to_read = -1;
+	sts.reader_threadwork.main_start_work();
+	sts.reader_threadwork.main_wait_for_done();
+	CHECK_MESSAGE(sts.func1_count == 2,
+			"Reader should have read one additional message from flush_all");
+
+	sts.destroy_threads();
+
+	CHECK_MESSAGE(sts.func1_count == 2,
+			"Reader should have read no additional messages after join");
+	ProjectSettings::get_singleton()->set_setting(COMMAND_QUEUE_SETTING,
+			ProjectSettings::get_singleton()->property_get_revert(COMMAND_QUEUE_SETTING));
+}
+
+TEST_CASE("[CommandQueue] Test Waiting at Queue Full") {
+	const char *COMMAND_QUEUE_SETTING = "memory/limits/command_queue/multithreading_queue_size_kb";
+	ProjectSettings::get_singleton()->set_setting(COMMAND_QUEUE_SETTING, 1);
+	SharedThreadState sts;
+	sts.init_threads();
+
+	int msgs_to_add = 24; // a queue of size 1kB fundamentally cannot fit 24 matrices.
+	for (int i = 0; i < msgs_to_add; i++) {
+		sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC1_TRANSFORM);
+	}
+	sts.writer_threadwork.main_start_work();
+	// If we call main_wait_for_done, we will deadlock. So instead...
+	sts.message_count_to_read = 1;
+	sts.reader_threadwork.main_start_work();
+	sts.reader_threadwork.main_wait_for_done();
+	CHECK_MESSAGE(sts.func1_count == 1,
+			"Reader should have read one message");
+	CHECK_MESSAGE(sts.during_writing,
+			"Writer thread should still be blocked on writing.");
+	sts.message_count_to_read = msgs_to_add - 3;
+	sts.reader_threadwork.main_start_work();
+	sts.reader_threadwork.main_wait_for_done();
+	CHECK_MESSAGE(sts.func1_count >= msgs_to_add - 3,
+			"Reader should have read most messages");
+	sts.writer_threadwork.main_wait_for_done();
+	CHECK_MESSAGE(sts.during_writing == false,
+			"Writer thread should no longer be blocked on writing.");
+	sts.message_count_to_read = 2;
+	sts.reader_threadwork.main_start_work();
+	sts.reader_threadwork.main_wait_for_done();
+	sts.message_count_to_read = -1;
+	sts.reader_threadwork.main_start_work();
+	sts.reader_threadwork.main_wait_for_done();
+	CHECK_MESSAGE(sts.func1_count == msgs_to_add,
+			"Reader should have read all messages");
+
+	sts.destroy_threads();
+
+	CHECK_MESSAGE(sts.func1_count == msgs_to_add,
+			"Reader should have read no additional messages after join");
+	ProjectSettings::get_singleton()->set_setting(COMMAND_QUEUE_SETTING,
+			ProjectSettings::get_singleton()->property_get_revert(COMMAND_QUEUE_SETTING));
+}
+
+TEST_CASE("[CommandQueue] Test Queue Wrapping to same spot.") {
+	const char *COMMAND_QUEUE_SETTING = "memory/limits/command_queue/multithreading_queue_size_kb";
+	ProjectSettings::get_singleton()->set_setting(COMMAND_QUEUE_SETTING, 1);
+	SharedThreadState sts;
+	sts.init_threads();
+
+	sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC3_TRANSFORMx6);
+	sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC3_TRANSFORMx6);
+	sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC1_TRANSFORM);
+	sts.writer_threadwork.main_start_work();
+	sts.writer_threadwork.main_wait_for_done();
+
+	sts.message_count_to_read = -1;
+	sts.reader_threadwork.main_start_work();
+	sts.reader_threadwork.main_wait_for_done();
+	CHECK_MESSAGE(sts.func1_count == 3,
+			"Reader should have read at least three messages");
+
+	sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC3_TRANSFORMx6);
+	sts.writer_threadwork.main_start_work();
+	sts.writer_threadwork.main_wait_for_done();
+	sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC1_TRANSFORM);
+	sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC3_TRANSFORMx6);
+	sts.writer_threadwork.main_start_work();
+	OS::get_singleton()->delay_usec(1000);
+
+	sts.message_count_to_read = -1;
+	sts.reader_threadwork.main_start_work();
+	OS::get_singleton()->delay_usec(1000);
+
+	sts.writer_threadwork.main_wait_for_done();
+	sts.reader_threadwork.main_wait_for_done();
+	CHECK_MESSAGE(sts.func1_count >= 3,
+			"Reader should have read at least three messages");
+
+	sts.message_count_to_read = 6 - sts.func1_count;
+	sts.reader_threadwork.main_start_work();
+
+	// The following will fail immediately.
+	// The reason it hangs indefinitely in engine, is all subsequent calls to
+	// CommandQueue.wait_and_flush_one will also fail.
+	sts.reader_threadwork.main_wait_for_done();
+
+	// Because looping around uses an extra message, easiest to consume all.
+	sts.message_count_to_read = -1;
+	sts.reader_threadwork.main_start_work();
+	sts.reader_threadwork.main_wait_for_done();
+	CHECK_MESSAGE(sts.func1_count == 6,
+			"Reader should have read both message sets");
+
+	sts.destroy_threads();
+
+	CHECK_MESSAGE(sts.func1_count == 6,
+			"Reader should have read no additional messages after join");
+	ProjectSettings::get_singleton()->set_setting(COMMAND_QUEUE_SETTING,
+			ProjectSettings::get_singleton()->property_get_revert(COMMAND_QUEUE_SETTING));
+}
+
+TEST_CASE("[CommandQueue] Test Queue Lapping") {
+	const char *COMMAND_QUEUE_SETTING = "memory/limits/command_queue/multithreading_queue_size_kb";
+	ProjectSettings::get_singleton()->set_setting(COMMAND_QUEUE_SETTING, 1);
+	SharedThreadState sts;
+	sts.init_threads();
+
+	sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC1_TRANSFORM);
+	sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC3_TRANSFORMx6);
+	sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC3_TRANSFORMx6);
+	sts.writer_threadwork.main_start_work();
+	sts.writer_threadwork.main_wait_for_done();
+
+	// We need to read an extra message so that it triggers the dealloc logic once.
+	// Otherwise, the queue will be considered full.
+	sts.message_count_to_read = 3;
+	sts.reader_threadwork.main_start_work();
+	sts.reader_threadwork.main_wait_for_done();
+	CHECK_MESSAGE(sts.func1_count == 3,
+			"Reader should have read first set of messages");
+
+	sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC3_TRANSFORMx6);
+	sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC3_TRANSFORMx6);
+	sts.writer_threadwork.main_start_work();
+	// Don't wait for these, because the queue isn't big enough.
+	sts.writer_threadwork.main_wait_for_done();
+
+	sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC2_TRANSFORM_FLOAT);
+	sts.writer_threadwork.main_start_work();
+	OS::get_singleton()->delay_usec(1000);
+
+	sts.message_count_to_read = 3;
+	sts.reader_threadwork.main_start_work();
+	sts.reader_threadwork.main_wait_for_done();
+
+	sts.writer_threadwork.main_wait_for_done();
+
+	sts.message_count_to_read = -1;
+	sts.reader_threadwork.main_start_work();
+	sts.reader_threadwork.main_wait_for_done();
+
+	CHECK_MESSAGE(sts.func1_count == 6,
+			"Reader should have read rest of the messages after lapping writers.");
+
+	sts.destroy_threads();
+
+	CHECK_MESSAGE(sts.func1_count == 6,
+			"Reader should have read no additional messages after join");
+	ProjectSettings::get_singleton()->set_setting(COMMAND_QUEUE_SETTING,
+			ProjectSettings::get_singleton()->property_get_revert(COMMAND_QUEUE_SETTING));
+}
+
+TEST_CASE("[Stress][CommandQueue] Stress test command queue") {
+	const char *COMMAND_QUEUE_SETTING = "memory/limits/command_queue/multithreading_queue_size_kb";
+	ProjectSettings::get_singleton()->set_setting(COMMAND_QUEUE_SETTING, 1);
+	SharedThreadState sts;
+	sts.init_threads();
+
+	RandomNumberGenerator rng;
+
+	rng.set_seed(1837267);
+
+	int msgs_to_add = 2048;
+
+	for (int i = 0; i < msgs_to_add; i++) {
+		// randi_range is inclusive, so allow any enum value except MAX.
+		sts.add_msg_to_write((SharedThreadState::TestMsgType)rng.randi_range(0, SharedThreadState::TEST_MSG_MAX - 1));
+	}
+	sts.writer_threadwork.main_start_work();
+
+	int max_loop_iters = msgs_to_add * 2;
+	int loop_iters = 0;
+	while (sts.func1_count < msgs_to_add && loop_iters < max_loop_iters) {
+		int remaining = (msgs_to_add - sts.func1_count);
+		sts.message_count_to_read = rng.randi_range(1, remaining < 128 ? remaining : 128);
+		if (loop_iters % 3 == 0) {
+			sts.message_count_to_read = -1;
+		}
+		sts.reader_threadwork.main_start_work();
+		sts.reader_threadwork.main_wait_for_done();
+		loop_iters++;
+	}
+	CHECK_MESSAGE(loop_iters < max_loop_iters,
+			"Reader needed too many iterations to read messages!");
+	sts.writer_threadwork.main_wait_for_done();
+
+	sts.destroy_threads();
+
+	CHECK_MESSAGE(sts.func1_count == msgs_to_add,
+			"Reader should have read no additional messages after join");
+	ProjectSettings::get_singleton()->set_setting(COMMAND_QUEUE_SETTING,
+			ProjectSettings::get_singleton()->property_get_revert(COMMAND_QUEUE_SETTING));
+}
+
+} // namespace TestCommandQueue
+
+#endif // !defined(NO_THREADS)
+
+#endif // TEST_COMMAND_QUEUE_H
diff --git a/tests/test_main.cpp b/tests/test_main.cpp
index f0dec22738..60674c79c0 100644
--- a/tests/test_main.cpp
+++ b/tests/test_main.cpp
@@ -36,6 +36,7 @@
 #include "test_basis.h"
 #include "test_class_db.h"
 #include "test_color.h"
+#include "test_command_queue.h"
 #include "test_expression.h"
 #include "test_gradient.h"
 #include "test_gui.h"