From b16a8b8a76079f06a2ef5f6efd95c47a3e22604a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pedro=20J=2E=20Est=C3=A9banez?= Date: Tue, 11 Nov 2025 13:05:42 +0100 Subject: [PATCH 1/2] CommandQueueMT: Fix race conditions --- core/templates/command_queue_mt.h | 24 +++++++++++++++++++----- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/core/templates/command_queue_mt.h b/core/templates/command_queue_mt.h index 94a03878d40..7eaf2f94af1 100644 --- a/core/templates/command_queue_mt.h +++ b/core/templates/command_queue_mt.h @@ -39,6 +39,8 @@ #include "core/typedefs.h" class CommandQueueMT { + static const size_t MAX_COMMAND_SIZE = 1024; + struct CommandBase { bool sync = false; virtual void call() = 0; @@ -154,19 +156,28 @@ class CommandQueueMT { } void _flush() { + MutexLock lock(mutex); + if (unlikely(flush_read_ptr)) { // Re-entrant call. return; } - MutexLock lock(mutex); + char cmd_backup[MAX_COMMAND_SIZE]; while (flush_read_ptr < command_mem.size()) { uint64_t size = *(uint64_t *)&command_mem[flush_read_ptr]; - flush_read_ptr += 8; + flush_read_ptr += sizeof(uint64_t); + CommandBase *cmd = reinterpret_cast(&command_mem[flush_read_ptr]); + + // Protect against race condition between this thread + // during the call to the command and other threads potentially + // invalidating the pointer due to reallocs. + memcpy(cmd_backup, (char *)cmd, size); + uint32_t allowance_id = WorkerThreadPool::thread_enter_unlock_allowance_zone(lock); - cmd->call(); + ((CommandBase *)cmd_backup)->call(); WorkerThreadPool::thread_exit_unlock_allowance_zone(allowance_id); // Handle potential realloc due to the command and unlock allowance. @@ -174,9 +185,9 @@ class CommandQueueMT { if (unlikely(cmd->sync)) { sync_head++; - lock.~MutexLock(); // Give an opportunity to awaiters right away. + lock.temp_unlock(); // Give an opportunity to awaiters right away. sync_cond_var.notify_all(); - new (&lock) MutexLock(mutex); + lock.temp_relock(); // Handle potential realloc happened during unlock. cmd = reinterpret_cast(&command_mem[flush_read_ptr]); } @@ -210,6 +221,7 @@ public: void push(T *p_instance, M p_method, Args &&...p_args) { // Standard command, no sync. using CommandType = Command; + static_assert(sizeof(CommandType) <= MAX_COMMAND_SIZE); _push_internal(p_instance, p_method, std::forward(p_args)...); } @@ -217,6 +229,7 @@ public: void push_and_sync(T *p_instance, M p_method, Args... p_args) { // Standard command, sync. using CommandType = Command; + static_assert(sizeof(CommandType) <= MAX_COMMAND_SIZE); _push_internal(p_instance, p_method, std::forward(p_args)...); } @@ -224,6 +237,7 @@ public: void push_and_ret(T *p_instance, M p_method, R *r_ret, Args... p_args) { // Command with return value, sync. using CommandType = CommandRet; + static_assert(sizeof(CommandType) <= MAX_COMMAND_SIZE); _push_internal(p_instance, p_method, r_ret, std::forward(p_args)...); } From 4ba4558cf6e0550190ee2cb7fbc3c6af3fa2e677 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pedro=20J=2E=20Est=C3=A9banez?= Date: Fri, 7 Nov 2025 12:38:29 +0100 Subject: [PATCH 2/2] CommandQueueMT: Reduce lock contention in cases of single flusher --- core/templates/command_queue_mt.h | 18 ++++++++++++++---- modules/betsy/image_compress_betsy.h | 2 +- servers/rendering/rendering_server_default.h | 2 +- 3 files changed, 16 insertions(+), 6 deletions(-) diff --git a/core/templates/command_queue_mt.h b/core/templates/command_queue_mt.h index 7eaf2f94af1..f03dd771378 100644 --- a/core/templates/command_queue_mt.h +++ b/core/templates/command_queue_mt.h @@ -107,6 +107,7 @@ class CommandQueueMT { static const uint32_t DEFAULT_COMMAND_MEM_SIZE_KB = 64; + bool unique_flusher = false; BinaryMutex mutex; LocalVector command_mem; ConditionVariable sync_cond_var; @@ -176,9 +177,17 @@ class CommandQueueMT { // invalidating the pointer due to reallocs. memcpy(cmd_backup, (char *)cmd, size); - uint32_t allowance_id = WorkerThreadPool::thread_enter_unlock_allowance_zone(lock); - ((CommandBase *)cmd_backup)->call(); - WorkerThreadPool::thread_exit_unlock_allowance_zone(allowance_id); + if (unique_flusher) { + // A single thread will pump; the lock is only needed for the command queue itself. + lock.temp_unlock(); + ((CommandBase *)cmd_backup)->call(); + lock.temp_relock(); + } else { + // At least we can unlock during WTP operations. + uint32_t allowance_id = WorkerThreadPool::thread_enter_unlock_allowance_zone(lock); + ((CommandBase *)cmd_backup)->call(); + WorkerThreadPool::thread_exit_unlock_allowance_zone(allowance_id); + } // Handle potential realloc due to the command and unlock allowance. cmd = reinterpret_cast(&command_mem[flush_read_ptr]); @@ -266,7 +275,8 @@ public: pump_task_id = p_task_id; } - CommandQueueMT() { + CommandQueueMT(bool p_unique_flusher = false) : + unique_flusher(p_unique_flusher) { command_mem.reserve(DEFAULT_COMMAND_MEM_SIZE_KB * 1024); } }; diff --git a/modules/betsy/image_compress_betsy.h b/modules/betsy/image_compress_betsy.h index 15b58894970..78352a12947 100644 --- a/modules/betsy/image_compress_betsy.h +++ b/modules/betsy/image_compress_betsy.h @@ -103,7 +103,7 @@ Error _betsy_compress_s3tc(Image *r_img, Image::UsedChannels p_channels); class BetsyCompressor : public Object { GDSOFTCLASS(BetsyCompressor, Object); - mutable CommandQueueMT command_queue; + mutable CommandQueueMT command_queue = CommandQueueMT(true); bool exit = false; WorkerThreadPool::TaskID task_id = WorkerThreadPool::INVALID_TASK_ID; diff --git a/servers/rendering/rendering_server_default.h b/servers/rendering/rendering_server_default.h index 34faef662c4..e9ac6098ec6 100644 --- a/servers/rendering/rendering_server_default.h +++ b/servers/rendering/rendering_server_default.h @@ -74,7 +74,7 @@ class RenderingServerDefault : public RenderingServer { uint64_t print_frame_profile_ticks_from = 0; uint32_t print_frame_profile_frame_count = 0; - mutable CommandQueueMT command_queue; + mutable CommandQueueMT command_queue = CommandQueueMT(true); Thread::ID server_thread = Thread::MAIN_ID; WorkerThreadPool::TaskID server_task_id = WorkerThreadPool::INVALID_TASK_ID;