/**************************************************************************/
/* test_command_queue.h */
/**************************************************************************/
/* This file is part of: */
/* GODOT ENGINE */
/* https://godotengine.org */
/**************************************************************************/
/* Copyright (c) 2014-present Godot Engine contributors (see AUTHORS.md). */
/* Copyright (c) 2007-2014 Juan Linietsky, Ariel Manzur. */
/* */
/* 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 "core/config/project_settings.h"
#include "core/math/random_number_generator.h"
#include "core/object/worker_thread_pool.h"
#include "core/os/os.h"
#include "core/os/thread.h"
#include "core/templates/command_queue_mt.h"
#include "tests/test_macros.h"
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;
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;
WorkerThreadPool::TaskID reader_task_id = WorkerThreadPool::INVALID_TASK_ID;
Thread writer_thread;
int func1_count = 0;
void func1(Transform3D t) {
func1_count++;
}
void func2(Transform3D t, float f) {
func1_count++;
}
void func3(Transform3D t1, Transform3D t2, Transform3D t3, Transform3D t4, Transform3D t5, Transform3D t6) {
func1_count++;
}
Transform3D func1r(Transform3D t) {
func1_count++;
return t;
}
Transform3D func2r(Transform3D 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 (reader_task_id == WorkerThreadPool::INVALID_TASK_ID) {
command_queue.flush_all();
} else {
if (message_count_to_read < 0) {
command_queue.flush_all();
}
for (int i = 0; i < message_count_to_read; i++) {
WorkerThreadPool::get_singleton()->yield();
command_queue.wait_and_flush();
}
}
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) {
Transform3D tr;
Transform3D 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(bool p_use_thread_pool_sync = false) {
if (p_use_thread_pool_sync) {
reader_task_id = WorkerThreadPool::get_singleton()->add_native_task(&SharedThreadState::static_reader_thread_loop, this, true);
command_queue.set_pump_task_id(reader_task_id);
} else {
reader_thread.start(&SharedThreadState::static_reader_thread_loop, this);
}
writer_thread.start(&SharedThreadState::static_writer_thread_loop, this);
}
void destroy_threads() {
exit_threads = true;
reader_threadwork.main_start_work();
writer_threadwork.main_start_work();
if (reader_task_id != WorkerThreadPool::INVALID_TASK_ID) {
WorkerThreadPool::get_singleton()->wait_for_task_completion(reader_task_id);
} else {
reader_thread.wait_to_finish();
}
writer_thread.wait_to_finish();
}
};
static void test_command_queue_basic(bool p_use_thread_pool_sync) {
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(p_use_thread_pool_sync);
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 Queue Basics") {
test_command_queue_basic(false);
}
TEST_CASE("[CommandQueue] Test Queue Basics with WorkerThreadPool sync.") {
test_command_queue_basic(true);
}
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 // TEST_COMMAND_QUEUE_H
Codebase Browser
godot