Merge pull request #91725 from RandomShaper/cmd_queue_avoid_skip

`CommandQueueMT`: Fix sync command awaiters missing the chance

1 files changed, 19 insertions, 2 deletions

diff --git a/core/templates/command_queue_mt.h b/core/templates/command_queue_mt.h
index 9f9b77d9d0..dbf938a117 100644
--- a/core/templates/command_queue_mt.h
+++ b/core/templates/command_queue_mt.h
@@ -327,7 +327,6 @@ class CommandQueueMT {
 
 	enum {
 		DEFAULT_COMMAND_MEM_SIZE_KB = 256,
-		SYNC_SEMAPHORES = 8
 	};
 
 	BinaryMutex mutex;
@@ -335,6 +334,7 @@ class CommandQueueMT {
 	ConditionVariable sync_cond_var;
 	uint32_t sync_head = 0;
 	uint32_t sync_tail = 0;
+	uint32_t sync_awaiters = 0;
 	WorkerThreadPool::TaskID pump_task_id = WorkerThreadPool::INVALID_TASK_ID;
 	uint64_t flush_read_ptr = 0;
 
@@ -349,6 +349,15 @@ class CommandQueueMT {
 		return cmd;
 	}
 
+	_FORCE_INLINE_ void _prevent_sync_wraparound() {
+		bool safe_to_reset = !sync_awaiters;
+		bool already_sync_to_latest = sync_head == sync_tail;
+		if (safe_to_reset && already_sync_to_latest) {
+			sync_head = 0;
+			sync_tail = 0;
+		}
+	}
+
 	void _flush() {
 		if (unlikely(flush_read_ptr)) {
 			// Re-entrant call.
@@ -365,7 +374,9 @@ class CommandQueueMT {
 			cmd->call();
 			if (unlikely(cmd->sync)) {
 				sync_head++;
+				unlock(); // Give an opportunity to awaiters right away.
 				sync_cond_var.notify_all();
+				lock();
 			}
 
 			// If the command involved reallocating the buffer, the address may have changed.
@@ -378,14 +389,20 @@ class CommandQueueMT {
 
 		command_mem.clear();
 		flush_read_ptr = 0;
+
+		_prevent_sync_wraparound();
+
 		unlock();
 	}
 
 	_FORCE_INLINE_ void _wait_for_sync(MutexLock<BinaryMutex> &p_lock) {
+		sync_awaiters++;
 		uint32_t sync_head_goal = sync_tail;
 		do {
 			sync_cond_var.wait(p_lock);
-		} while (sync_head != sync_head_goal); // Can't use lower-than because of wraparound.
+		} while (sync_head < sync_head_goal);
+		sync_awaiters--;
+		_prevent_sync_wraparound();
 	}
 
 	void _no_op() {}