Merge pull request #90268 from RandomShaper/wtp_servers

Use WorkerThreadPool for Server threads (enhanced)

core/config/engine.cpp

@@ -82,6 +82,17 @@ int Engine::get_audio_output_latency() const {
 	return _audio_output_latency;
 }
 
+void Engine::increment_frames_drawn() {
+	if (frame_server_synced) {
+		server_syncs++;
+	} else {
+		server_syncs = 0;
+	}
+	frame_server_synced = false;
+
+	frames_drawn++;
+}
+
 uint64_t Engine::get_frames_drawn() {
 	return frames_drawn;
 }
@@ -364,6 +375,11 @@ Engine *Engine::get_singleton() {
 	return singleton;
 }
 
+bool Engine::notify_frame_server_synced() {
+	frame_server_synced = true;
+	return server_syncs > SERVER_SYNC_FRAME_COUNT_WARNING;
+}
+
 Engine::Engine() {
 	singleton = this;
 }

core/config/engine.h

@@ -91,6 +91,10 @@ private:
 	String write_movie_path;
 	String shader_cache_path;
 
+	static constexpr int SERVER_SYNC_FRAME_COUNT_WARNING = 5;
+	int server_syncs = 0;
+	bool frame_server_synced = false;
+
 public:
 	static Engine *get_singleton();
 
@@ -179,6 +183,9 @@ public:
 	bool is_generate_spirv_debug_info_enabled() const;
 	int32_t get_gpu_index() const;
 
+	void increment_frames_drawn();
+	bool notify_frame_server_synced();
+
 	Engine();
 	virtual ~Engine() {}
 };

core/object/worker_thread_pool.cpp

@@ -35,6 +35,8 @@
 #include "core/os/thread_safe.h"
 #include "core/templates/command_queue_mt.h"
 
+WorkerThreadPool::Task *const WorkerThreadPool::ThreadData::YIELDING = (Task *)1;
+
 void WorkerThreadPool::Task::free_template_userdata() {
 	ERR_FAIL_NULL(template_userdata);
 	ERR_FAIL_NULL(native_func_userdata);
@@ -60,11 +62,13 @@ void WorkerThreadPool::_process_task(Task *p_task) {
 		// its pre-created threads can't have ScriptServer::thread_enter() called on them early.
 		// Therefore, we do it late at the first opportunity, so in case the task
 		// about to be run uses scripting, guarantees are held.
+		task_mutex.lock();
 		if (!curr_thread.ready_for_scripting && ScriptServer::are_languages_initialized()) {
+			task_mutex.unlock();
 			ScriptServer::thread_enter();
+			task_mutex.lock();
 			curr_thread.ready_for_scripting = true;
 		}
-		task_mutex.lock();
 		p_task->pool_thread_index = pool_thread_index;
 		prev_task = curr_thread.current_task;
 		curr_thread.current_task = p_task;
@@ -389,83 +393,117 @@ Error WorkerThreadPool::wait_for_task_completion(TaskID p_task_id) {
 	task_mutex.unlock();
 
 	if (caller_pool_thread) {
-		while (true) {
-			Task *task_to_process = nullptr;
-			{
-				MutexLock lock(task_mutex);
-				bool was_signaled = caller_pool_thread->signaled;
-				caller_pool_thread->signaled = false;
-
-				if (task->completed) {
-					// This thread was awaken also for some reason, but it's about to exit.
-					// Let's find out what may be pending and forward the requests.
-					if (!exit_threads && was_signaled) {
-						uint32_t to_process = task_queue.first() ? 1 : 0;
-						uint32_t to_promote = caller_pool_thread->current_task->low_priority && low_priority_task_queue.first() ? 1 : 0;
-						if (to_process || to_promote) {
-							// This thread must be left alone since it won't loop again.
-							caller_pool_thread->signaled = true;
-							_notify_threads(caller_pool_thread, to_process, to_promote);
-						}
-					}
+		_wait_collaboratively(caller_pool_thread, task);
+		task->waiting_pool--;
+		if (task->waiting_pool == 0 && task->waiting_user == 0) {
+			tasks.erase(p_task_id);
+			task_allocator.free(task);
+		}
+	} else {
+		task->done_semaphore.wait();
+		task_mutex.lock();
+		task->waiting_user--;
+		if (task->waiting_pool == 0 && task->waiting_user == 0) {
+			tasks.erase(p_task_id);
+			task_allocator.free(task);
+		}
+		task_mutex.unlock();
+	}
 
-					task->waiting_pool--;
-					if (task->waiting_pool == 0 && task->waiting_user == 0) {
-						tasks.erase(p_task_id);
-						task_allocator.free(task);
-					}
+	return OK;
+}
 
-					break;
-				}
+void WorkerThreadPool::_wait_collaboratively(ThreadData *p_caller_pool_thread, Task *p_task) {
+	// Keep processing tasks until the condition to stop waiting is met.
 
-				if (!exit_threads) {
-					// This is a thread from the pool. It shouldn't just idle.
-					// Let's try to process other tasks while we wait.
+#define IS_WAIT_OVER (unlikely(p_task == ThreadData::YIELDING) ? p_caller_pool_thread->yield_is_over : p_task->completed)
 
-					if (caller_pool_thread->current_task->low_priority && low_priority_task_queue.first()) {
-						if (_try_promote_low_priority_task()) {
-							_notify_threads(caller_pool_thread, 1, 0);
-						}
+	while (true) {
+		Task *task_to_process = nullptr;
+		{
+			MutexLock lock(task_mutex);
+			bool was_signaled = p_caller_pool_thread->signaled;
+			p_caller_pool_thread->signaled = false;
+
+			if (IS_WAIT_OVER) {
+				p_caller_pool_thread->yield_is_over = false;
+				if (!exit_threads && was_signaled) {
+					// This thread was awaken for some additional reason, but it's about to exit.
+					// Let's find out what may be pending and forward the requests.
+					uint32_t to_process = task_queue.first() ? 1 : 0;
+					uint32_t to_promote = p_caller_pool_thread->current_task->low_priority && low_priority_task_queue.first() ? 1 : 0;
+					if (to_process || to_promote) {
+						// This thread must be left alone since it won't loop again.
+						p_caller_pool_thread->signaled = true;
+						_notify_threads(p_caller_pool_thread, to_process, to_promote);
 					}
+				}
+
+				break;
+			}
 
-					if (singleton->task_queue.first()) {
-						task_to_process = task_queue.first()->self();
-						task_queue.remove(task_queue.first());
+			if (!exit_threads) {
+				if (p_caller_pool_thread->current_task->low_priority && low_priority_task_queue.first()) {
+					if (_try_promote_low_priority_task()) {
+						_notify_threads(p_caller_pool_thread, 1, 0);
 					}
+				}
 
-					if (!task_to_process) {
-						caller_pool_thread->awaited_task = task;
+				if (singleton->task_queue.first()) {
+					task_to_process = task_queue.first()->self();
+					task_queue.remove(task_queue.first());
+				}
 
-						if (flushing_cmd_queue) {
-							flushing_cmd_queue->unlock();
-						}
-						caller_pool_thread->cond_var.wait(lock);
-						if (flushing_cmd_queue) {
-							flushing_cmd_queue->lock();
-						}
+				if (!task_to_process) {
+					p_caller_pool_thread->awaited_task = p_task;
 
-						DEV_ASSERT(exit_threads || caller_pool_thread->signaled || task->completed);
-						caller_pool_thread->awaited_task = nullptr;
+					if (flushing_cmd_queue) {
+						flushing_cmd_queue->unlock();
+					}
+					p_caller_pool_thread->cond_var.wait(lock);
+					if (flushing_cmd_queue) {
+						flushing_cmd_queue->lock();
 					}
-				}
-			}
 
-			if (task_to_process) {
-				_process_task(task_to_process);
+					DEV_ASSERT(exit_threads || p_caller_pool_thread->signaled || IS_WAIT_OVER);
+					p_caller_pool_thread->awaited_task = nullptr;
+				}
 			}
 		}
-	} else {
-		task->done_semaphore.wait();
-		task_mutex.lock();
-		task->waiting_user--;
-		if (task->waiting_pool == 0 && task->waiting_user == 0) {
-			tasks.erase(p_task_id);
-			task_allocator.free(task);
+
+		if (task_to_process) {
+			_process_task(task_to_process);
 		}
+	}
+}
+
+void WorkerThreadPool::yield() {
+	int th_index = get_thread_index();
+	ERR_FAIL_COND_MSG(th_index == -1, "This function can only be called from a worker thread.");
+	_wait_collaboratively(&threads[th_index], ThreadData::YIELDING);
+}
+
+void WorkerThreadPool::notify_yield_over(TaskID p_task_id) {
+	task_mutex.lock();
+	Task **taskp = tasks.getptr(p_task_id);
+	if (!taskp) {
 		task_mutex.unlock();
+		ERR_FAIL_MSG("Invalid Task ID.");
 	}
+	Task *task = *taskp;
 
-	return OK;
+#ifdef DEBUG_ENABLED
+	if (task->pool_thread_index == get_thread_index()) {
+		WARN_PRINT("A worker thread is attempting to notify itself. That makes no sense.");
+	}
+#endif
+
+	ThreadData &td = threads[task->pool_thread_index];
+	td.yield_is_over = true;
+	td.signaled = true;
+	td.cond_var.notify_one();
+
+	task_mutex.unlock();
 }
 
 WorkerThreadPool::GroupID WorkerThreadPool::_add_group_task(const Callable &p_callable, void (*p_func)(void *, uint32_t), void *p_userdata, BaseTemplateUserdata *p_template_userdata, int p_elements, int p_tasks, bool p_high_priority, const String &p_description) {

core/object/worker_thread_pool.h

@@ -107,13 +107,21 @@ private:
 	BinaryMutex task_mutex;
 
 	struct ThreadData {
+		static Task *const YIELDING; // Too bad constexpr doesn't work here.
+
 		uint32_t index = 0;
 		Thread thread;
-		bool ready_for_scripting = false;
-		bool signaled = false;
+		bool ready_for_scripting : 1;
+		bool signaled : 1;
+		bool yield_is_over : 1;
 		Task *current_task = nullptr;
-		Task *awaited_task = nullptr; // Null if not awaiting the condition variable. Special value for idle-waiting.
+		Task *awaited_task = nullptr; // Null if not awaiting the condition variable, or special value (YIELDING).
 		ConditionVariable cond_var;
+
+		ThreadData() :
+				ready_for_scripting(false),
+				signaled(false),
+				yield_is_over(false) {}
 	};
 
 	TightLocalVector<ThreadData> threads;
@@ -177,6 +185,8 @@ private:
 		}
 	};
 
+	void _wait_collaboratively(ThreadData *p_caller_pool_thread, Task *p_task);
+
 protected:
 	static void _bind_methods();
 
@@ -196,6 +206,9 @@ public:
 	bool is_task_completed(TaskID p_task_id) const;
 	Error wait_for_task_completion(TaskID p_task_id);
 
+	void yield();
+	void notify_yield_over(TaskID p_task_id);
+
 	template <typename C, typename M, typename U>
 	GroupID add_template_group_task(C *p_instance, M p_method, U p_userdata, int p_elements, int p_tasks = -1, bool p_high_priority = false, const String &p_description = String()) {
 		typedef GroupUserData<C, M, U> GroupUD;

core/templates/command_queue_mt.cpp

@@ -70,14 +70,8 @@ CommandQueueMT::SyncSemaphore *CommandQueueMT::_alloc_sync_sem() {
 	return &sync_sems[idx];
 }
 
-CommandQueueMT::CommandQueueMT(bool p_sync) {
-	if (p_sync) {
-		sync = memnew(Semaphore);
-	}
+CommandQueueMT::CommandQueueMT() {
 }
 
 CommandQueueMT::~CommandQueueMT() {
-	if (sync) {
-		memdelete(sync);
-	}
 }

core/templates/command_queue_mt.h

@@ -248,16 +248,17 @@
 #define CMD_TYPE(N) Command##N<T, M COMMA(N) COMMA_SEP_LIST(TYPE_ARG, N)>
 #define CMD_ASSIGN_PARAM(N) cmd->p##N = p##N
 
-#define DECL_PUSH(N)                                                         \
-	template <typename T, typename M COMMA(N) COMMA_SEP_LIST(TYPE_PARAM, N)> \
-	void push(T *p_instance, M p_method COMMA(N) COMMA_SEP_LIST(PARAM, N)) { \
-		CMD_TYPE(N) *cmd = allocate_and_lock<CMD_TYPE(N)>();                 \
-		cmd->instance = p_instance;                                          \
-		cmd->method = p_method;                                              \
-		SEMIC_SEP_LIST(CMD_ASSIGN_PARAM, N);                                 \
-		unlock();                                                            \
-		if (sync)                                                            \
-			sync->post();                                                    \
+#define DECL_PUSH(N)                                                            \
+	template <typename T, typename M COMMA(N) COMMA_SEP_LIST(TYPE_PARAM, N)>    \
+	void push(T *p_instance, M p_method COMMA(N) COMMA_SEP_LIST(PARAM, N)) {    \
+		CMD_TYPE(N) *cmd = allocate_and_lock<CMD_TYPE(N)>();                    \
+		cmd->instance = p_instance;                                             \
+		cmd->method = p_method;                                                 \
+		SEMIC_SEP_LIST(CMD_ASSIGN_PARAM, N);                                    \
+		if (pump_task_id != WorkerThreadPool::INVALID_TASK_ID) {                \
+			WorkerThreadPool::get_singleton()->notify_yield_over(pump_task_id); \
+		}                                                                       \
+		unlock();                                                               \
 	}
 
 #define CMD_RET_TYPE(N) CommandRet##N<T, M, COMMA_SEP_LIST(TYPE_ARG, N) COMMA(N) R>
@@ -272,9 +273,10 @@
 		SEMIC_SEP_LIST(CMD_ASSIGN_PARAM, N);                                                   \
 		cmd->ret = r_ret;                                                                      \
 		cmd->sync_sem = ss;                                                                    \
+		if (pump_task_id != WorkerThreadPool::INVALID_TASK_ID) {                               \
+			WorkerThreadPool::get_singleton()->notify_yield_over(pump_task_id);                \
+		}                                                                                      \
 		unlock();                                                                              \
-		if (sync)                                                                              \
-			sync->post();                                                                      \
 		ss->sem.wait();                                                                        \
 		ss->in_use = false;                                                                    \
 	}
@@ -290,9 +292,10 @@
 		cmd->method = p_method;                                                       \
 		SEMIC_SEP_LIST(CMD_ASSIGN_PARAM, N);                                          \
 		cmd->sync_sem = ss;                                                           \
+		if (pump_task_id != WorkerThreadPool::INVALID_TASK_ID) {                      \
+			WorkerThreadPool::get_singleton()->notify_yield_over(pump_task_id);       \
+		}                                                                             \
 		unlock();                                                                     \
-		if (sync)                                                                     \
-			sync->post();                                                             \
 		ss->sem.wait();                                                               \
 		ss->in_use = false;                                                           \
 	}
@@ -340,7 +343,7 @@ class CommandQueueMT {
 	LocalVector<uint8_t> command_mem;
 	SyncSemaphore sync_sems[SYNC_SEMAPHORES];
 	Mutex mutex;
-	Semaphore *sync = nullptr;
+	WorkerThreadPool::TaskID pump_task_id = WorkerThreadPool::INVALID_TASK_ID;
 	uint64_t flush_read_ptr = 0;
 
 	template <typename T>
@@ -420,12 +423,16 @@ public:
 	}
 
 	void wait_and_flush() {
-		ERR_FAIL_NULL(sync);
-		sync->wait();
+		ERR_FAIL_COND(pump_task_id == WorkerThreadPool::INVALID_TASK_ID);
+		WorkerThreadPool::get_singleton()->wait_for_task_completion(pump_task_id);
 		_flush();
 	}
 
-	CommandQueueMT(bool p_sync);
+	void set_pump_task_id(WorkerThreadPool::TaskID p_task_id) {
+		pump_task_id = p_task_id;
+	}
+
+	CommandQueueMT();
 	~CommandQueueMT();
 };
 

drivers/egl/egl_manager.cpp

@@ -260,21 +260,6 @@ void EGLManager::release_current() {
 	eglMakeCurrent(current_display.egl_display, EGL_NO_SURFACE, EGL_NO_SURFACE, EGL_NO_CONTEXT);
 }
 
-void EGLManager::make_current() {
-	if (!current_window) {
-		return;
-	}
-
-	if (!current_window->initialized) {
-		WARN_PRINT("Current OpenGL window is uninitialized!");
-		return;
-	}
-
-	GLDisplay &current_display = displays[current_window->gldisplay_id];
-
-	eglMakeCurrent(current_display.egl_display, current_window->egl_surface, current_window->egl_surface, current_display.egl_context);
-}
-
 void EGLManager::swap_buffers() {
 	if (!current_window) {
 		return;

drivers/egl/egl_manager.h

@@ -98,7 +98,6 @@ public:
 	void window_destroy(DisplayServer::WindowID p_window_id);
 
 	void release_current();
-	void make_current();
 	void swap_buffers();
 
 	void window_make_current(DisplayServer::WindowID p_window_id);

drivers/gles3/storage/particles_storage.cpp

@@ -135,7 +135,6 @@ bool ParticlesStorage::particles_get_emitting(RID p_particles) {
 		return false;
 	}
 
-	ERR_FAIL_COND_V_MSG(RSG::threaded, false, "This function should never be used with threaded rendering, as it stalls the renderer.");
 	Particles *particles = particles_owner.get_or_null(p_particles);
 	ERR_FAIL_NULL_V(particles, false);
 
@@ -380,10 +379,6 @@ void ParticlesStorage::particles_request_process(RID p_particles) {
 }
 
 AABB ParticlesStorage::particles_get_current_aabb(RID p_particles) {
-	if (RSG::threaded) {
-		WARN_PRINT_ONCE("Calling this function with threaded rendering enabled stalls the renderer, use with care.");
-	}
-
 	const Particles *particles = particles_owner.get_or_null(p_particles);
 	ERR_FAIL_NULL_V(particles, AABB());
 
@@ -1207,7 +1202,6 @@ Dependency *ParticlesStorage::particles_get_dependency(RID p_particles) const {
 }
 
 bool ParticlesStorage::particles_is_inactive(RID p_particles) const {
-	ERR_FAIL_COND_V_MSG(RSG::threaded, false, "This function should never be used with threaded rendering, as it stalls the renderer.");
 	const Particles *particles = particles_owner.get_or_null(p_particles);
 	ERR_FAIL_NULL_V(particles, false);
 	return !particles->emitting && particles->inactive;

main/main.cpp

@@ -2276,6 +2276,9 @@ Error Main::setup(const char *execpath, int argc, char *argv[], bool p_second_ph
 			// Editor and project manager cannot run with rendering in a separate thread (they will crash on startup).
 			rtm = OS::RENDER_THREAD_SAFE;
 		}
+#if !defined(THREADS_ENABLED)
+		rtm = OS::RENDER_THREAD_SAFE;
+#endif
 		OS::get_singleton()->_render_thread_mode = OS::RenderThreadMode(rtm);
 	}
 
@@ -2719,7 +2722,9 @@ Error Main::setup2() {
 	}
 
 	if (OS::get_singleton()->_render_thread_mode == OS::RENDER_SEPARATE_THREAD) {
-		WARN_PRINT("The Multi-Threaded rendering thread model is experimental, and has known issues which can lead to project crashes. Use the Single-Safe option in the project settings instead.");
+		WARN_PRINT("The Multi-Threaded rendering thread model is experimental. Feel free to try it since it will eventually become a stable feature.\n"
+				   "However, bear in mind that at the moment it can lead to project crashes or instability.\n"
+				   "So, unless you want to test the engine, use the Single-Safe option in the project settings instead.");
 	}
 
 	/* Initialize Pen Tablet Driver */
@@ -4027,11 +4032,11 @@ bool Main::iteration() {
 		if ((!force_redraw_requested) && OS::get_singleton()->is_in_low_processor_usage_mode()) {
 			if (RenderingServer::get_singleton()->has_changed()) {
 				RenderingServer::get_singleton()->draw(true, scaled_step); // flush visual commands
-				Engine::get_singleton()->frames_drawn++;
+				Engine::get_singleton()->increment_frames_drawn();
 			}
 		} else {
 			RenderingServer::get_singleton()->draw(true, scaled_step); // flush visual commands
-			Engine::get_singleton()->frames_drawn++;
+			Engine::get_singleton()->increment_frames_drawn();
 			force_redraw_requested = false;
 		}
 	}

platform/linuxbsd/wayland/display_server_wayland.cpp

@@ -1174,14 +1174,6 @@ void DisplayServerWayland::release_rendering_thread() {
 #endif
 }
 
-void DisplayServerWayland::make_rendering_thread() {
-#ifdef GLES3_ENABLED
-	if (egl_manager) {
-		egl_manager->make_current();
-	}
-#endif
-}
-
 void DisplayServerWayland::swap_buffers() {
 #ifdef GLES3_ENABLED
 	if (egl_manager) {

platform/linuxbsd/wayland/display_server_wayland.h

@@ -276,7 +276,6 @@ public:
 	virtual void process_events() override;
 
 	virtual void release_rendering_thread() override;
-	virtual void make_rendering_thread() override;
 	virtual void swap_buffers() override;
 
 	virtual void set_context(Context p_context) override;

platform/linuxbsd/x11/display_server_x11.cpp

@@ -4268,7 +4268,7 @@ bool DisplayServerX11::_window_focus_check() {
 }
 
 void DisplayServerX11::process_events() {
-	_THREAD_SAFE_METHOD_
+	_THREAD_SAFE_LOCK_
 
 #ifdef DISPLAY_SERVER_X11_DEBUG_LOGS_ENABLED
 	static int frame = 0;
@@ -5097,6 +5097,8 @@ void DisplayServerX11::process_events() {
 		*/
 	}
 
+	_THREAD_SAFE_UNLOCK_
+
 	Input::get_singleton()->flush_buffered_events();
 }
 
@@ -5111,17 +5113,6 @@ void DisplayServerX11::release_rendering_thread() {
 #endif
 }
 
-void DisplayServerX11::make_rendering_thread() {
-#if defined(GLES3_ENABLED)
-	if (gl_manager) {
-		gl_manager->make_current();
-	}
-	if (gl_manager_egl) {
-		gl_manager_egl->make_current();
-	}
-#endif
-}
-
 void DisplayServerX11::swap_buffers() {
 #if defined(GLES3_ENABLED)
 	if (gl_manager) {

platform/linuxbsd/x11/display_server_x11.h

@@ -526,7 +526,6 @@ public:
 	virtual void process_events() override;
 
 	virtual void release_rendering_thread() override;
-	virtual void make_rendering_thread() override;
 	virtual void swap_buffers() override;
 
 	virtual void set_context(Context p_context) override;

platform/linuxbsd/x11/gl_manager_x11.cpp

@@ -311,20 +311,6 @@ void GLManager_X11::window_make_current(DisplayServer::WindowID p_window_id) {
 	_internal_set_current_window(&win);
 }
 
-void GLManager_X11::make_current() {
-	if (!_current_window) {
-		return;
-	}
-	if (!_current_window->in_use) {
-		WARN_PRINT("current window not in use!");
-		return;
-	}
-	const GLDisplay &disp = get_current_display();
-	if (!glXMakeCurrent(_x_windisp.x11_display, _x_windisp.x11_window, disp.context->glx_context)) {
-		ERR_PRINT("glXMakeCurrent failed");
-	}
-}
-
 void GLManager_X11::swap_buffers() {
 	if (!_current_window) {
 		return;

platform/linuxbsd/x11/gl_manager_x11.h

@@ -117,7 +117,6 @@ public:
 	void window_resize(DisplayServer::WindowID p_window_id, int p_width, int p_height);
 
 	void release_current();
-	void make_current();
 	void swap_buffers();
 
 	void window_make_current(DisplayServer::WindowID p_window_id);

platform/macos/display_server_macos.h

@@ -426,7 +426,6 @@ public:
 	virtual void force_process_and_drop_events() override;
 
 	virtual void release_rendering_thread() override;
-	virtual void make_rendering_thread() override;
 	virtual void swap_buffers() override;
 
 	virtual void set_native_icon(const String &p_filename) override;

platform/macos/display_server_macos.mm

@@ -358,7 +358,6 @@ void DisplayServerMacOS::_dispatch_input_events(const Ref<InputEvent> &p_event)
 }
 
 void DisplayServerMacOS::_dispatch_input_event(const Ref<InputEvent> &p_event) {
-	_THREAD_SAFE_METHOD_
 	if (!in_dispatch_input_event) {
 		in_dispatch_input_event = true;
 
@@ -2986,7 +2985,7 @@ Key DisplayServerMacOS::keyboard_get_label_from_physical(Key p_keycode) const {
 }
 
 void DisplayServerMacOS::process_events() {
-	_THREAD_SAFE_METHOD_
+	_THREAD_SAFE_LOCK_
 
 	while (true) {
 		NSEvent *event = [NSApp
@@ -3019,7 +3018,9 @@ void DisplayServerMacOS::process_events() {
 
 	if (!drop_events) {
 		_process_key_events();
+		_THREAD_SAFE_UNLOCK_
 		Input::get_singleton()->flush_buffered_events();
+		_THREAD_SAFE_LOCK_
 	}
 
 	for (KeyValue<WindowID, WindowData> &E : windows) {
@@ -3045,6 +3046,8 @@ void DisplayServerMacOS::process_events() {
 			}
 		}
 	}
+
+	_THREAD_SAFE_UNLOCK_
 }
 
 void DisplayServerMacOS::force_process_and_drop_events() {
@@ -3056,9 +3059,14 @@ void DisplayServerMacOS::force_process_and_drop_events() {
 }
 
 void DisplayServerMacOS::release_rendering_thread() {
-}
-
-void DisplayServerMacOS::make_rendering_thread() {
+#if defined(GLES3_ENABLED)
+	if (gl_manager_angle) {
+		gl_manager_angle->release_current();
+	}
+	if (gl_manager_legacy) {
+		gl_manager_legacy->release_current();
+	}
+#endif
 }
 
 void DisplayServerMacOS::swap_buffers() {

platform/macos/gl_manager_macos_legacy.h

@@ -73,7 +73,6 @@ public:
 	void window_resize(DisplayServer::WindowID p_window_id, int p_width, int p_height);
 
 	void release_current();
-	void make_current();
 	void swap_buffers();
 
 	void window_make_current(DisplayServer::WindowID p_window_id);

platform/macos/gl_manager_macos_legacy.mm

@@ -117,6 +117,7 @@ void GLManagerLegacy_MacOS::release_current() {
 	}
 
 	[NSOpenGLContext clearCurrentContext];
+	current_window = DisplayServer::INVALID_WINDOW_ID;
 }
 
 void GLManagerLegacy_MacOS::window_make_current(DisplayServer::WindowID p_window_id) {
@@ -133,18 +134,6 @@ void GLManagerLegacy_MacOS::window_make_current(DisplayServer::WindowID p_window
 	current_window = p_window_id;
 }
 
-void GLManagerLegacy_MacOS::make_current() {
-	if (current_window == DisplayServer::INVALID_WINDOW_ID) {
-		return;
-	}
-	if (!windows.has(current_window)) {
-		return;
-	}
-
-	GLWindow &win = windows[current_window];
-	[win.context makeCurrentContext];
-}
-
 void GLManagerLegacy_MacOS::swap_buffers() {
 	GLWindow &win = windows[current_window];
 	[win.context flushBuffer];

platform/windows/display_server_windows.cpp

@@ -2962,7 +2962,7 @@ String DisplayServerWindows::keyboard_get_layout_name(int p_index) const {
 }
 
 void DisplayServerWindows::process_events() {
-	_THREAD_SAFE_METHOD_
+	_THREAD_SAFE_LOCK_
 
 	MSG msg;
 
@@ -2977,7 +2977,10 @@ void DisplayServerWindows::process_events() {
 
 	if (!drop_events) {
 		_process_key_events();
+		_THREAD_SAFE_UNLOCK_
 		Input::get_singleton()->flush_buffered_events();
+	} else {
+		_THREAD_SAFE_UNLOCK_
 	}
 }
 
@@ -2990,9 +2993,14 @@ void DisplayServerWindows::force_process_and_drop_events() {
 }
 
 void DisplayServerWindows::release_rendering_thread() {
-}
-
-void DisplayServerWindows::make_rendering_thread() {
+#if defined(GLES3_ENABLED)
+	if (gl_manager_angle) {
+		gl_manager_angle->release_current();
+	}
+	if (gl_manager_native) {
+		gl_manager_native->release_current();
+	}
+#endif
 }
 
 void DisplayServerWindows::swap_buffers() {
@@ -3433,7 +3441,6 @@ void DisplayServerWindows::_dispatch_input_events(const Ref<InputEvent> &p_event
 }
 
 void DisplayServerWindows::_dispatch_input_event(const Ref<InputEvent> &p_event) {
-	_THREAD_SAFE_METHOD_
 	if (in_dispatch_input_event) {
 		return;
 	}

platform/windows/display_server_windows.h

@@ -679,7 +679,6 @@ public:
 	virtual void force_process_and_drop_events() override;
 
 	virtual void release_rendering_thread() override;
-	virtual void make_rendering_thread() override;
 	virtual void swap_buffers() override;
 
 	virtual void set_native_icon(const String &p_filename) override;

platform/windows/gl_manager_windows_native.cpp

@@ -427,10 +427,6 @@ Error GLManagerNative_Windows::window_create(DisplayServer::WindowID p_window_id
 	return OK;
 }
 
-void GLManagerNative_Windows::_internal_set_current_window(GLWindow *p_win) {
-	_current_window = p_win;
-}
-
 void GLManagerNative_Windows::window_destroy(DisplayServer::WindowID p_window_id) {
 	GLWindow &win = get_window(p_window_id);
 	if (_current_window == &win) {
@@ -447,6 +443,8 @@ void GLManagerNative_Windows::release_current() {
 	if (!gd_wglMakeCurrent(_current_window->hDC, nullptr)) {
 		ERR_PRINT("Could not detach OpenGL context from window marked current: " + format_error_message(GetLastError()));
 	}
+
+	_current_window = nullptr;
 }
 
 void GLManagerNative_Windows::window_make_current(DisplayServer::WindowID p_window_id) {
@@ -467,17 +465,7 @@ void GLManagerNative_Windows::window_make_current(DisplayServer::WindowID p_wind
 		ERR_PRINT("Could not switch OpenGL context to other window: " + format_error_message(GetLastError()));
 	}
 
-	_internal_set_current_window(&win);
-}
-
-void GLManagerNative_Windows::make_current() {
-	if (!_current_window) {
-		return;
-	}
-	const GLDisplay &disp = get_current_display();
-	if (!gd_wglMakeCurrent(_current_window->hDC, disp.hRC)) {
-		ERR_PRINT("Could not switch OpenGL context to window marked current: " + format_error_message(GetLastError()));
-	}
+	_current_window = &win;
 }
 
 void GLManagerNative_Windows::swap_buffers() {
@@ -491,7 +479,6 @@ Error GLManagerNative_Windows::initialize() {
 
 void GLManagerNative_Windows::set_use_vsync(DisplayServer::WindowID p_window_id, bool p_use) {
 	GLWindow &win = get_window(p_window_id);
-	GLWindow *current = _current_window;
 
 	if (&win != _current_window) {
 		window_make_current(p_window_id);
@@ -506,11 +493,6 @@ void GLManagerNative_Windows::set_use_vsync(DisplayServer::WindowID p_window_id,
 	} else {
 		WARN_PRINT("Could not set V-Sync mode. V-Sync is not supported.");
 	}
-
-	if (current != _current_window) {
-		_current_window = current;
-		make_current();
-	}
 }
 
 bool GLManagerNative_Windows::is_using_vsync(DisplayServer::WindowID p_window_id) const {

platform/windows/gl_manager_windows_native.h

@@ -68,9 +68,6 @@ private:
 
 	PFNWGLSWAPINTERVALEXTPROC wglSwapIntervalEXT = nullptr;
 
-	// funcs
-	void _internal_set_current_window(GLWindow *p_win);
-
 	GLWindow &get_window(unsigned int id) { return _windows[id]; }
 	const GLWindow &get_window(unsigned int id) const { return _windows[id]; }
 
@@ -91,7 +88,6 @@ public:
 	void window_resize(DisplayServer::WindowID p_window_id, int p_width, int p_height) {}
 
 	void release_current();
-	void make_current();
 	void swap_buffers();
 
 	void window_make_current(DisplayServer::WindowID p_window_id);

scene/main/window.cpp

@@ -1579,6 +1579,7 @@ bool Window::_can_consume_input_events() const {
 }
 
 void Window::_window_input(const Ref<InputEvent> &p_ev) {
+	ERR_MAIN_THREAD_GUARD;
 	if (EngineDebugger::is_active()) {
 		// Quit from game window using the stop shortcut (F8 by default).
 		// The custom shortcut is provided via environment variable when running from the editor.

servers/display_server.cpp

@@ -697,10 +697,6 @@ void DisplayServer::release_rendering_thread() {
 	WARN_PRINT("Rendering thread not supported by this display server.");
 }
 
-void DisplayServer::make_rendering_thread() {
-	WARN_PRINT("Rendering thread not supported by this display server.");
-}
-
 void DisplayServer::swap_buffers() {
 	WARN_PRINT("Swap buffers not supported by this display server.");
 }

servers/display_server.h

@@ -559,7 +559,6 @@ public:
 	virtual void force_process_and_drop_events();
 
 	virtual void release_rendering_thread();
-	virtual void make_rendering_thread();
 	virtual void swap_buffers();
 
 	virtual void set_native_icon(const String &p_filename);

servers/physics_server_2d_wrap_mt.cpp

@@ -33,7 +33,7 @@
 #include "core/os/os.h"
 
 void PhysicsServer2DWrapMT::thread_exit() {
-	exit.set();
+	exit = true;
 }
 
 void PhysicsServer2DWrapMT::thread_step(real_t p_delta) {
@@ -41,25 +41,18 @@ void PhysicsServer2DWrapMT::thread_step(real_t p_delta) {
 	step_sem.post();
 }
 
-void PhysicsServer2DWrapMT::_thread_callback(void *_instance) {
-	PhysicsServer2DWrapMT *vsmt = reinterpret_cast<PhysicsServer2DWrapMT *>(_instance);
-
-	vsmt->thread_loop();
-}
-
 void PhysicsServer2DWrapMT::thread_loop() {
 	server_thread = Thread::get_caller_id();
 
 	physics_server_2d->init();
 
-	exit.clear();
-	step_thread_up.set();
-	while (!exit.is_set()) {
-		// flush commands one by one, until exit is requested
-		command_queue.wait_and_flush();
+	command_queue.set_pump_task_id(server_task_id);
+	while (!exit) {
+		WorkerThreadPool::get_singleton()->yield();
+		command_queue.flush_all();
 	}
 
-	command_queue.flush_all(); // flush all
+	command_queue.flush_all();
 
 	physics_server_2d->finish();
 }
@@ -70,18 +63,14 @@ void PhysicsServer2DWrapMT::step(real_t p_step) {
 	if (create_thread) {
 		command_queue.push(this, &PhysicsServer2DWrapMT::thread_step, p_step);
 	} else {
-		command_queue.flush_all(); //flush all pending from other threads
+		command_queue.flush_all(); // Flush all pending from other threads.
 		physics_server_2d->step(p_step);
 	}
 }
 
 void PhysicsServer2DWrapMT::sync() {
 	if (create_thread) {
-		if (first_frame) {
-			first_frame = false;
-		} else {
-			step_sem.wait(); //must not wait if a step was not issued
-		}
+		step_sem.wait();
 	}
 	physics_server_2d->sync();
 }
@@ -96,40 +85,34 @@ void PhysicsServer2DWrapMT::end_sync() {
 
 void PhysicsServer2DWrapMT::init() {
 	if (create_thread) {
-		//OS::get_singleton()->release_rendering_thread();
-		thread.start(_thread_callback, this);
-		while (!step_thread_up.is_set()) {
-			OS::get_singleton()->delay_usec(1000);
-		}
+		exit = false;
+		server_task_id = WorkerThreadPool::get_singleton()->add_task(callable_mp(this, &PhysicsServer2DWrapMT::thread_loop), true);
+		step_sem.post();
 	} else {
 		physics_server_2d->init();
 	}
 }
 
 void PhysicsServer2DWrapMT::finish() {
-	if (thread.is_started()) {
+	if (create_thread) {
 		command_queue.push(this, &PhysicsServer2DWrapMT::thread_exit);
-		thread.wait_to_finish();
+		if (server_task_id != WorkerThreadPool::INVALID_TASK_ID) {
+			WorkerThreadPool::get_singleton()->wait_for_task_completion(server_task_id);
+			server_task_id = WorkerThreadPool::INVALID_TASK_ID;
+		}
 	} else {
 		physics_server_2d->finish();
 	}
 }
 
-PhysicsServer2DWrapMT::PhysicsServer2DWrapMT(PhysicsServer2D *p_contained, bool p_create_thread) :
-		command_queue(p_create_thread) {
+PhysicsServer2DWrapMT::PhysicsServer2DWrapMT(PhysicsServer2D *p_contained, bool p_create_thread) {
 	physics_server_2d = p_contained;
 	create_thread = p_create_thread;
-
-	if (!p_create_thread) {
-		server_thread = Thread::get_caller_id();
-	} else {
-		server_thread = 0;
+	if (!create_thread) {
+		server_thread = Thread::MAIN_ID;
 	}
-
-	main_thread = Thread::get_caller_id();
 }
 
 PhysicsServer2DWrapMT::~PhysicsServer2DWrapMT() {
 	memdelete(physics_server_2d);
-	//finish();
 }

servers/physics_server_2d_wrap_mt.h

@@ -32,6 +32,7 @@
 #define PHYSICS_SERVER_2D_WRAP_MT_H
 
 #include "core/config/project_settings.h"
+#include "core/object/worker_thread_pool.h"
 #include "core/os/thread.h"
 #include "core/templates/command_queue_mt.h"
 #include "core/templates/safe_refcount.h"
@@ -43,30 +44,27 @@
 #define SYNC_DEBUG
 #endif
 
+#ifdef DEBUG_ENABLED
+#define MAIN_THREAD_SYNC_WARN WARN_PRINT("Call to " + String(__FUNCTION__) + " causing PhysicsServer2D synchronizations on every frame. This significantly affects performance.");
+#endif
+
 class PhysicsServer2DWrapMT : public PhysicsServer2D {
-	mutable PhysicsServer2D *physics_server_2d;
+	mutable PhysicsServer2D *physics_server_2d = nullptr;
 
 	mutable CommandQueueMT command_queue;
 
-	static void _thread_callback(void *_instance);
 	void thread_loop();
 
-	Thread::ID server_thread;
-	Thread::ID main_thread;
-	SafeFlag exit;
-	Thread thread;
-	SafeFlag step_thread_up;
+	Thread::ID server_thread = Thread::UNASSIGNED_ID;
+	WorkerThreadPool::TaskID server_task_id = WorkerThreadPool::INVALID_TASK_ID;
+	bool exit = false;
+	Semaphore step_sem;
 	bool create_thread = false;
 
-	Semaphore step_sem;
 	void thread_step(real_t p_delta);
 
 	void thread_exit();
 
-	bool first_frame = true;
-
-	Mutex alloc_mutex;
-
 public:
 #define ServerName PhysicsServer2D
 #define ServerNameWrapMT PhysicsServer2DWrapMT
@@ -94,7 +92,7 @@ public:
 
 	//these work well, but should be used from the main thread only
 	bool shape_collide(RID p_shape_A, const Transform2D &p_xform_A, const Vector2 &p_motion_A, RID p_shape_B, const Transform2D &p_xform_B, const Vector2 &p_motion_B, Vector2 *r_results, int p_result_max, int &r_result_count) override {
-		ERR_FAIL_COND_V(main_thread != Thread::get_caller_id(), false);
+		ERR_FAIL_COND_V(!Thread::is_main_thread(), false);
 		return physics_server_2d->shape_collide(p_shape_A, p_xform_A, p_motion_A, p_shape_B, p_xform_B, p_motion_B, r_results, p_result_max, r_result_count);
 	}
 
@@ -109,18 +107,18 @@ public:
 
 	// this function only works on physics process, errors and returns null otherwise
 	PhysicsDirectSpaceState2D *space_get_direct_state(RID p_space) override {
-		ERR_FAIL_COND_V(main_thread != Thread::get_caller_id(), nullptr);
+		ERR_FAIL_COND_V(!Thread::is_main_thread(), nullptr);
 		return physics_server_2d->space_get_direct_state(p_space);
 	}
 
 	FUNC2(space_set_debug_contacts, RID, int);
 	virtual Vector<Vector2> space_get_contacts(RID p_space) const override {
-		ERR_FAIL_COND_V(main_thread != Thread::get_caller_id(), Vector<Vector2>());
+		ERR_FAIL_COND_V(!Thread::is_main_thread(), Vector<Vector2>());
 		return physics_server_2d->space_get_contacts(p_space);
 	}
 
 	virtual int space_get_contact_count(RID p_space) const override {
-		ERR_FAIL_COND_V(main_thread != Thread::get_caller_id(), 0);
+		ERR_FAIL_COND_V(!Thread::is_main_thread(), 0);
 		return physics_server_2d->space_get_contact_count(p_space);
 	}
 
@@ -261,13 +259,13 @@ public:
 	FUNC2(body_set_pickable, RID, bool);
 
 	bool body_test_motion(RID p_body, const MotionParameters &p_parameters, MotionResult *r_result = nullptr) override {
-		ERR_FAIL_COND_V(main_thread != Thread::get_caller_id(), false);
+		ERR_FAIL_COND_V(!Thread::is_main_thread(), false);
 		return physics_server_2d->body_test_motion(p_body, p_parameters, r_result);
 	}
 
 	// this function only works on physics process, errors and returns null otherwise
 	PhysicsDirectBodyState2D *body_get_direct_state(RID p_body) override {
-		ERR_FAIL_COND_V(main_thread != Thread::get_caller_id(), nullptr);
+		ERR_FAIL_COND_V(!Thread::is_main_thread(), nullptr);
 		return physics_server_2d->body_get_direct_state(p_body);
 	}
 
@@ -338,4 +336,8 @@ public:
 #endif
 #undef SYNC_DEBUG
 
+#ifdef DEBUG_ENABLED
+#undef MAIN_THREAD_SYNC_WARN
+#endif
+
 #endif // PHYSICS_SERVER_2D_WRAP_MT_H

servers/physics_server_3d_wrap_mt.cpp

@@ -41,22 +41,15 @@ void PhysicsServer3DWrapMT::thread_step(real_t p_delta) {
 	step_sem.post();
 }
 
-void PhysicsServer3DWrapMT::_thread_callback(void *_instance) {
-	PhysicsServer3DWrapMT *vsmt = reinterpret_cast<PhysicsServer3DWrapMT *>(_instance);
-
-	vsmt->thread_loop();
-}
-
 void PhysicsServer3DWrapMT::thread_loop() {
 	server_thread = Thread::get_caller_id();
 
 	physics_server_3d->init();
 
-	exit = false;
-	step_thread_up = true;
+	command_queue.set_pump_task_id(server_task_id);
 	while (!exit) {
-		// flush commands one by one, until exit is requested
-		command_queue.wait_and_flush();
+		WorkerThreadPool::get_singleton()->yield();
+		command_queue.flush_all();
 	}
 
 	command_queue.flush_all(); // flush all
@@ -70,18 +63,14 @@ void PhysicsServer3DWrapMT::step(real_t p_step) {
 	if (create_thread) {
 		command_queue.push(this, &PhysicsServer3DWrapMT::thread_step, p_step);
 	} else {
-		command_queue.flush_all(); //flush all pending from other threads
+		command_queue.flush_all(); // Flush all pending from other threads.
 		physics_server_3d->step(p_step);
 	}
 }
 
 void PhysicsServer3DWrapMT::sync() {
 	if (create_thread) {
-		if (first_frame) {
-			first_frame = false;
-		} else {
-			step_sem.wait(); //must not wait if a step was not issued
-		}
+		step_sem.wait();
 	}
 	physics_server_3d->sync();
 }
@@ -96,40 +85,34 @@ void PhysicsServer3DWrapMT::end_sync() {
 
 void PhysicsServer3DWrapMT::init() {
 	if (create_thread) {
-		//OS::get_singleton()->release_rendering_thread();
-		thread.start(_thread_callback, this);
-		while (!step_thread_up) {
-			OS::get_singleton()->delay_usec(1000);
-		}
+		exit = false;
+		server_task_id = WorkerThreadPool::get_singleton()->add_task(callable_mp(this, &PhysicsServer3DWrapMT::thread_loop), true);
+		step_sem.post();
 	} else {
 		physics_server_3d->init();
 	}
 }
 
 void PhysicsServer3DWrapMT::finish() {
-	if (thread.is_started()) {
+	if (create_thread) {
 		command_queue.push(this, &PhysicsServer3DWrapMT::thread_exit);
-		thread.wait_to_finish();
+		if (server_task_id != WorkerThreadPool::INVALID_TASK_ID) {
+			WorkerThreadPool::get_singleton()->wait_for_task_completion(server_task_id);
+			server_task_id = WorkerThreadPool::INVALID_TASK_ID;
+		}
 	} else {
 		physics_server_3d->finish();
 	}
 }
 
-PhysicsServer3DWrapMT::PhysicsServer3DWrapMT(PhysicsServer3D *p_contained, bool p_create_thread) :
-		command_queue(p_create_thread) {
+PhysicsServer3DWrapMT::PhysicsServer3DWrapMT(PhysicsServer3D *p_contained, bool p_create_thread) {
 	physics_server_3d = p_contained;
 	create_thread = p_create_thread;
-
-	if (!p_create_thread) {
-		server_thread = Thread::get_caller_id();
-	} else {
-		server_thread = 0;
+	if (!create_thread) {
+		server_thread = Thread::MAIN_ID;
 	}
-
-	main_thread = Thread::get_caller_id();
 }
 
 PhysicsServer3DWrapMT::~PhysicsServer3DWrapMT() {
 	memdelete(physics_server_3d);
-	//finish();
 }

servers/physics_server_3d_wrap_mt.h

@@ -32,6 +32,7 @@
 #define PHYSICS_SERVER_3D_WRAP_MT_H
 
 #include "core/config/project_settings.h"
+#include "core/object/worker_thread_pool.h"
 #include "core/os/thread.h"
 #include "core/templates/command_queue_mt.h"
 #include "servers/physics_server_3d.h"
@@ -42,30 +43,27 @@
 #define SYNC_DEBUG
 #endif
 
+#ifdef DEBUG_ENABLED
+#define MAIN_THREAD_SYNC_WARN WARN_PRINT("Call to " + String(__FUNCTION__) + " causing PhysicsServer3D synchronizations on every frame. This significantly affects performance.");
+#endif
+
 class PhysicsServer3DWrapMT : public PhysicsServer3D {
-	mutable PhysicsServer3D *physics_server_3d;
+	mutable PhysicsServer3D *physics_server_3d = nullptr;
 
 	mutable CommandQueueMT command_queue;
 
-	static void _thread_callback(void *_instance);
 	void thread_loop();
 
-	Thread::ID server_thread;
-	Thread::ID main_thread;
-	volatile bool exit = false;
-	Thread thread;
-	volatile bool step_thread_up = false;
+	Thread::ID server_thread = Thread::UNASSIGNED_ID;
+	WorkerThreadPool::TaskID server_task_id = WorkerThreadPool::INVALID_TASK_ID;
+	bool exit = false;
+	Semaphore step_sem;
 	bool create_thread = false;
 
-	Semaphore step_sem;
 	void thread_step(real_t p_delta);
 
 	void thread_exit();
 
-	bool first_frame = true;
-
-	Mutex alloc_mutex;
-
 public:
 #define ServerName PhysicsServer3D
 #define ServerNameWrapMT PhysicsServer3DWrapMT
@@ -98,7 +96,7 @@ public:
 #if 0
 	//these work well, but should be used from the main thread only
 	bool shape_collide(RID p_shape_A, const Transform &p_xform_A, const Vector3 &p_motion_A, RID p_shape_B, const Transform &p_xform_B, const Vector3 &p_motion_B, Vector3 *r_results, int p_result_max, int &r_result_count) {
-		ERR_FAIL_COND_V(main_thread != Thread::get_caller_id(), false);
+		ERR_FAIL_COND_V(!Thread::is_main_thread(), false);
 		return physics_server_3d->shape_collide(p_shape_A, p_xform_A, p_motion_A, p_shape_B, p_xform_B, p_motion_B, r_results, p_result_max, r_result_count);
 	}
 #endif
@@ -113,18 +111,18 @@ public:
 
 	// this function only works on physics process, errors and returns null otherwise
 	PhysicsDirectSpaceState3D *space_get_direct_state(RID p_space) override {
-		ERR_FAIL_COND_V(main_thread != Thread::get_caller_id(), nullptr);
+		ERR_FAIL_COND_V(!Thread::is_main_thread(), nullptr);
 		return physics_server_3d->space_get_direct_state(p_space);
 	}
 
 	FUNC2(space_set_debug_contacts, RID, int);
 	virtual Vector<Vector3> space_get_contacts(RID p_space) const override {
-		ERR_FAIL_COND_V(main_thread != Thread::get_caller_id(), Vector<Vector3>());
+		ERR_FAIL_COND_V(!Thread::is_main_thread(), Vector<Vector3>());
 		return physics_server_3d->space_get_contacts(p_space);
 	}
 
 	virtual int space_get_contact_count(RID p_space) const override {
-		ERR_FAIL_COND_V(main_thread != Thread::get_caller_id(), 0);
+		ERR_FAIL_COND_V(!Thread::is_main_thread(), 0);
 		return physics_server_3d->space_get_contact_count(p_space);
 	}
 
@@ -260,13 +258,13 @@ public:
 	FUNC2(body_set_ray_pickable, RID, bool);
 
 	bool body_test_motion(RID p_body, const MotionParameters &p_parameters, MotionResult *r_result = nullptr) override {
-		ERR_FAIL_COND_V(main_thread != Thread::get_caller_id(), false);
+		ERR_FAIL_COND_V(!Thread::is_main_thread(), false);
 		return physics_server_3d->body_test_motion(p_body, p_parameters, r_result);
 	}
 
 	// this function only works on physics process, errors and returns null otherwise
 	PhysicsDirectBodyState3D *body_get_direct_state(RID p_body) override {
-		ERR_FAIL_COND_V(main_thread != Thread::get_caller_id(), nullptr);
+		ERR_FAIL_COND_V(!Thread::is_main_thread(), nullptr);
 		return physics_server_3d->body_get_direct_state(p_body);
 	}
 
@@ -411,4 +409,8 @@ public:
 #endif
 #undef SYNC_DEBUG
 
+#ifdef DEBUG_ENABLED
+#undef MAIN_THREAD_SYNC_WARN
+#endif
+
 #endif // PHYSICS_SERVER_3D_WRAP_MT_H

servers/rendering/renderer_rd/storage_rd/particles_storage.cpp

@@ -257,7 +257,6 @@ void ParticlesStorage::particles_set_emitting(RID p_particles, bool p_emitting)
 }
 
 bool ParticlesStorage::particles_get_emitting(RID p_particles) {
-	ERR_FAIL_COND_V_MSG(RSG::threaded, false, "This function should never be used with threaded rendering, as it stalls the renderer.");
 	Particles *particles = particles_owner.get_or_null(p_particles);
 	ERR_FAIL_NULL_V(particles, false);
 
@@ -608,10 +607,6 @@ void ParticlesStorage::particles_request_process(RID p_particles) {
 }
 
 AABB ParticlesStorage::particles_get_current_aabb(RID p_particles) {
-	if (RSG::threaded) {
-		WARN_PRINT_ONCE("Calling this function with threaded rendering enabled stalls the renderer, use with care.");
-	}
-
 	const Particles *particles = particles_owner.get_or_null(p_particles);
 	ERR_FAIL_NULL_V(particles, AABB());
 
@@ -1642,7 +1637,6 @@ Dependency *ParticlesStorage::particles_get_dependency(RID p_particles) const {
 }
 
 bool ParticlesStorage::particles_is_inactive(RID p_particles) const {
-	ERR_FAIL_COND_V_MSG(RSG::threaded, false, "This function should never be used with threaded rendering, as it stalls the renderer.");
 	const Particles *particles = particles_owner.get_or_null(p_particles);
 	ERR_FAIL_NULL_V(particles, false);
 	return !particles->emitting && particles->inactive;

servers/rendering/rendering_server_default.cpp

@@ -69,9 +69,6 @@ void RenderingServerDefault::request_frame_drawn_callback(const Callable &p_call
 }
 
 void RenderingServerDefault::_draw(bool p_swap_buffers, double frame_step) {
-	//needs to be done before changes is reset to 0, to not force the editor to redraw
-	RS::get_singleton()->emit_signal(SNAME("frame_pre_draw"));
-
 	changes = 0;
 
 	RSG::rasterizer->begin_frame(frame_step);
@@ -220,16 +217,9 @@ void RenderingServerDefault::_finish() {
 
 void RenderingServerDefault::init() {
 	if (create_thread) {
-		print_verbose("RenderingServerWrapMT: Creating render thread");
+		print_verbose("RenderingServerWrapMT: Starting render thread");
 		DisplayServer::get_singleton()->release_rendering_thread();
-		if (create_thread) {
-			thread.start(_thread_callback, this);
-			print_verbose("RenderingServerWrapMT: Starting render thread");
-		}
-		while (!draw_thread_up.is_set()) {
-			OS::get_singleton()->delay_usec(1000);
-		}
-		print_verbose("RenderingServerWrapMT: Finished render thread");
+		server_task_id = WorkerThreadPool::get_singleton()->add_task(callable_mp(this, &RenderingServerDefault::_thread_loop), true);
 	} else {
 		_init();
 	}
@@ -238,8 +228,9 @@ void RenderingServerDefault::init() {
 void RenderingServerDefault::finish() {
 	if (create_thread) {
 		command_queue.push(this, &RenderingServerDefault::_thread_exit);
-		if (thread.is_started()) {
-			thread.wait_to_finish();
+		if (server_task_id != WorkerThreadPool::INVALID_TASK_ID) {
+			WorkerThreadPool::get_singleton()->wait_for_task_completion(server_task_id);
+			server_task_id = WorkerThreadPool::INVALID_TASK_ID;
 		}
 	} else {
 		_finish();
@@ -337,38 +328,29 @@ Size2i RenderingServerDefault::get_maximum_viewport_size() const {
 }
 
 void RenderingServerDefault::_thread_exit() {
-	exit.set();
+	exit = true;
 }
 
 void RenderingServerDefault::_thread_draw(bool p_swap_buffers, double frame_step) {
 	_draw(p_swap_buffers, frame_step);
 }
 
-void RenderingServerDefault::_thread_flush() {
-}
-
-void RenderingServerDefault::_thread_callback(void *_instance) {
-	RenderingServerDefault *vsmt = reinterpret_cast<RenderingServerDefault *>(_instance);
-
-	vsmt->_thread_loop();
-}
-
 void RenderingServerDefault::_thread_loop() {
 	server_thread = Thread::get_caller_id();
 
-	DisplayServer::get_singleton()->make_rendering_thread();
-
+	DisplayServer::get_singleton()->gl_window_make_current(DisplayServer::MAIN_WINDOW_ID); // Move GL to this thread.
 	_init();
 
-	draw_thread_up.set();
-	while (!exit.is_set()) {
-		// flush commands one by one, until exit is requested
-		command_queue.wait_and_flush();
+	command_queue.set_pump_task_id(server_task_id);
+	while (!exit) {
+		WorkerThreadPool::get_singleton()->yield();
+		command_queue.flush_all();
 	}
 
-	command_queue.flush_all(); // flush all
+	command_queue.flush_all();
 
 	_finish();
+	DisplayServer::get_singleton()->release_rendering_thread();
 }
 
 /* INTERPOLATION */
@@ -384,15 +366,15 @@ void RenderingServerDefault::set_physics_interpolation_enabled(bool p_enabled) {
 /* EVENT QUEUING */
 
 void RenderingServerDefault::sync() {
-	if (create_thread) {
-		command_queue.push_and_sync(this, &RenderingServerDefault::_thread_flush);
-	} else {
-		command_queue.flush_all(); //flush all pending from other threads
+	if (!create_thread) {
+		command_queue.flush_all(); // Flush all pending from other threads.
 	}
 }
 
 void RenderingServerDefault::draw(bool p_swap_buffers, double frame_step) {
 	ERR_FAIL_COND_MSG(!Thread::is_main_thread(), "Manually triggering the draw function from the RenderingServer can only be done on the main thread. Call this function from the main thread or use call_deferred().");
+	// Needs to be done before changes is reset to 0, to not force the editor to redraw.
+	RS::get_singleton()->emit_signal(SNAME("frame_pre_draw"));
 	if (create_thread) {
 		command_queue.push(this, &RenderingServerDefault::_thread_draw, p_swap_buffers, frame_step);
 	} else {
@@ -404,21 +386,14 @@ void RenderingServerDefault::_call_on_render_thread(const Callable &p_callable)
 	p_callable.call();
 }
 
-RenderingServerDefault::RenderingServerDefault(bool p_create_thread) :
-		command_queue(p_create_thread) {
+RenderingServerDefault::RenderingServerDefault(bool p_create_thread) {
 	RenderingServer::init();
 
-#ifdef THREADS_ENABLED
 	create_thread = p_create_thread;
 	if (!create_thread) {
-		server_thread = Thread::get_caller_id();
-	} else {
-		server_thread = 0;
+		server_thread = Thread::MAIN_ID;
 	}
-#else
-	create_thread = false;
-	server_thread = Thread::get_main_id();
-#endif
+
 	RSG::threaded = create_thread;
 
 	RSG::canvas = memnew(RendererCanvasCull);

servers/rendering/rendering_server_default.h

@@ -31,6 +31,7 @@
 #ifndef RENDERING_SERVER_DEFAULT_H
 #define RENDERING_SERVER_DEFAULT_H
 
+#include "core/object/worker_thread_pool.h"
 #include "core/os/thread.h"
 #include "core/templates/command_queue_mt.h"
 #include "core/templates/hash_map.h"
@@ -75,22 +76,17 @@ class RenderingServerDefault : public RenderingServer {
 
 	mutable CommandQueueMT command_queue;
 
-	static void _thread_callback(void *_instance);
 	void _thread_loop();
 
-	Thread::ID server_thread = 0;
-	SafeFlag exit;
-	Thread thread;
-	SafeFlag draw_thread_up;
-	bool create_thread;
+	Thread::ID server_thread = Thread::UNASSIGNED_ID;
+	WorkerThreadPool::TaskID server_task_id = WorkerThreadPool::INVALID_TASK_ID;
+	bool exit = false;
+	bool create_thread = false;
 
 	void _thread_draw(bool p_swap_buffers, double frame_step);
-	void _thread_flush();
 
 	void _thread_exit();
 
-	Mutex alloc_mutex;
-
 	void _draw(bool p_swap_buffers, double frame_step);
 	void _init();
 	void _finish();
@@ -127,6 +123,10 @@ public:
 #define SYNC_DEBUG
 #endif
 
+#ifdef DEBUG_ENABLED
+#define MAIN_THREAD_SYNC_WARN WARN_PRINT("Call to " + String(__FUNCTION__) + " causing RenderingServer synchronizations on every frame. This significantly affects performance.");
+#endif
+
 #include "servers/server_wrap_mt_common.h"
 
 	/* TEXTURE API */
@@ -1013,6 +1013,9 @@ public:
 #undef ServerName
 #undef WRITE_ACTION
 #undef SYNC_DEBUG
+#ifdef DEBUG_ENABLED
+#undef MAIN_THREAD_SYNC_WARN
+#endif
 
 	virtual uint64_t get_rendering_info(RenderingInfo p_info) override;
 	virtual RenderingDevice::DeviceType get_video_adapter_type() const override;

servers/rendering_server.cpp

@@ -83,25 +83,16 @@ static PackedInt64Array to_int_array(const Vector<ObjectID> &ids) {
 }
 
 PackedInt64Array RenderingServer::_instances_cull_aabb_bind(const AABB &p_aabb, RID p_scenario) const {
-	if (RSG::threaded) {
-		WARN_PRINT_ONCE("Using this function with a threaded renderer hurts performance, as it causes a server stall.");
-	}
 	Vector<ObjectID> ids = instances_cull_aabb(p_aabb, p_scenario);
 	return to_int_array(ids);
 }
 
 PackedInt64Array RenderingServer::_instances_cull_ray_bind(const Vector3 &p_from, const Vector3 &p_to, RID p_scenario) const {
-	if (RSG::threaded) {
-		WARN_PRINT_ONCE("Using this function with a threaded renderer hurts performance, as it causes a server stall.");
-	}
 	Vector<ObjectID> ids = instances_cull_ray(p_from, p_to, p_scenario);
 	return to_int_array(ids);
 }
 
 PackedInt64Array RenderingServer::_instances_cull_convex_bind(const TypedArray<Plane> &p_convex, RID p_scenario) const {
-	if (RSG::threaded) {
-		WARN_PRINT_ONCE("Using this function with a threaded renderer hurts performance, as it causes a server stall.");
-	}
 	Vector<Plane> planes;
 	for (int i = 0; i < p_convex.size(); ++i) {
 		const Variant &v = p_convex[i];

servers/server_wrap_mt_common.h

@@ -31,12 +31,22 @@
 #ifndef SERVER_WRAP_MT_COMMON_H
 #define SERVER_WRAP_MT_COMMON_H
 
+#ifdef DEBIG_ENABLED
+#define MAIN_THREAD_SYNC_CHECK                                                                         \
+	if (unlikely(Thread::is_main_thread() && Engine::get_singleton()->notify_frame_server_synced())) { \
+		MAIN_THREAD_SYNC_WARN                                                                          \
+	}
+#else
+#define MAIN_THREAD_SYNC_CHECK
+#endif
+
 #define FUNC0R(m_r, m_type)                                                     \
 	virtual m_r m_type() override {                                             \
 		if (Thread::get_caller_id() != server_thread) {                         \
 			m_r ret;                                                            \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, &ret); \
 			SYNC_DEBUG                                                          \
+			MAIN_THREAD_SYNC_CHECK                                              \
 			return ret;                                                         \
 		} else {                                                                \
 			command_queue.flush_if_pending();                                   \
@@ -68,6 +78,7 @@
 			m_r ret;                                                            \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, &ret); \
 			SYNC_DEBUG                                                          \
+			MAIN_THREAD_SYNC_CHECK                                              \
 			return ret;                                                         \
 		} else {                                                                \
 			command_queue.flush_if_pending();                                   \
@@ -102,6 +113,7 @@
 		if (Thread::get_caller_id() != server_thread) {                    \
 			command_queue.push_and_sync(server_name, &ServerName::m_type); \
 			SYNC_DEBUG                                                     \
+			MAIN_THREAD_SYNC_CHECK                                         \
 		} else {                                                           \
 			command_queue.flush_if_pending();                              \
 			server_name->m_type();                                         \
@@ -113,6 +125,7 @@
 		if (Thread::get_caller_id() != server_thread) {                    \
 			command_queue.push_and_sync(server_name, &ServerName::m_type); \
 			SYNC_DEBUG                                                     \
+			MAIN_THREAD_SYNC_CHECK                                         \
 		} else {                                                           \
 			command_queue.flush_if_pending();                              \
 			server_name->m_type();                                         \
@@ -128,6 +141,7 @@
 			m_r ret;                                                                \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, &ret); \
 			SYNC_DEBUG                                                              \
+			MAIN_THREAD_SYNC_CHECK                                                  \
 			return ret;                                                             \
 		} else {                                                                    \
 			command_queue.flush_if_pending();                                       \
@@ -141,6 +155,7 @@
 			m_r ret;                                                                \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, &ret); \
 			SYNC_DEBUG                                                              \
+			MAIN_THREAD_SYNC_CHECK                                                  \
 			return ret;                                                             \
 		} else {                                                                    \
 			command_queue.flush_if_pending();                                       \
@@ -154,6 +169,7 @@
 		if (Thread::get_caller_id() != server_thread) {                        \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1); \
 			SYNC_DEBUG                                                         \
+			MAIN_THREAD_SYNC_CHECK                                             \
 		} else {                                                               \
 			command_queue.flush_if_pending();                                  \
 			server_name->m_type(p1);                                           \
@@ -165,6 +181,7 @@
 		if (Thread::get_caller_id() != server_thread) {                        \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1); \
 			SYNC_DEBUG                                                         \
+			MAIN_THREAD_SYNC_CHECK                                             \
 		} else {                                                               \
 			command_queue.flush_if_pending();                                  \
 			server_name->m_type(p1);                                           \
@@ -199,6 +216,7 @@
 			m_r ret;                                                                    \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, p2, &ret); \
 			SYNC_DEBUG                                                                  \
+			MAIN_THREAD_SYNC_CHECK                                                      \
 			return ret;                                                                 \
 		} else {                                                                        \
 			command_queue.flush_if_pending();                                           \
@@ -212,6 +230,7 @@
 			m_r ret;                                                                    \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, p2, &ret); \
 			SYNC_DEBUG                                                                  \
+			MAIN_THREAD_SYNC_CHECK                                                      \
 			return ret;                                                                 \
 		} else {                                                                        \
 			command_queue.flush_if_pending();                                           \
@@ -225,6 +244,7 @@
 		if (Thread::get_caller_id() != server_thread) {                            \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1, p2); \
 			SYNC_DEBUG                                                             \
+			MAIN_THREAD_SYNC_CHECK                                                 \
 		} else {                                                                   \
 			command_queue.flush_if_pending();                                      \
 			server_name->m_type(p1, p2);                                           \
@@ -236,6 +256,7 @@
 		if (Thread::get_caller_id() != server_thread) {                            \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1, p2); \
 			SYNC_DEBUG                                                             \
+			MAIN_THREAD_SYNC_CHECK                                                 \
 		} else {                                                                   \
 			command_queue.flush_if_pending();                                      \
 			server_name->m_type(p1, p2);                                           \
@@ -270,6 +291,7 @@
 			m_r ret;                                                                        \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, p2, p3, &ret); \
 			SYNC_DEBUG                                                                      \
+			MAIN_THREAD_SYNC_CHECK                                                          \
 			return ret;                                                                     \
 		} else {                                                                            \
 			command_queue.flush_if_pending();                                               \
@@ -283,6 +305,7 @@
 			m_r ret;                                                                        \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, p2, p3, &ret); \
 			SYNC_DEBUG                                                                      \
+			MAIN_THREAD_SYNC_CHECK                                                          \
 			return ret;                                                                     \
 		} else {                                                                            \
 			command_queue.flush_if_pending();                                               \
@@ -296,6 +319,7 @@
 		if (Thread::get_caller_id() != server_thread) {                                \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1, p2, p3); \
 			SYNC_DEBUG                                                                 \
+			MAIN_THREAD_SYNC_CHECK                                                     \
 		} else {                                                                       \
 			command_queue.flush_if_pending();                                          \
 			server_name->m_type(p1, p2, p3);                                           \
@@ -307,6 +331,7 @@
 		if (Thread::get_caller_id() != server_thread) {                                \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1, p2, p3); \
 			SYNC_DEBUG                                                                 \
+			MAIN_THREAD_SYNC_CHECK                                                     \
 		} else {                                                                       \
 			command_queue.flush_if_pending();                                          \
 			server_name->m_type(p1, p2, p3);                                           \
@@ -341,6 +366,7 @@
 			m_r ret;                                                                            \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, p2, p3, p4, &ret); \
 			SYNC_DEBUG                                                                          \
+			MAIN_THREAD_SYNC_CHECK                                                              \
 			return ret;                                                                         \
 		} else {                                                                                \
 			command_queue.flush_if_pending();                                                   \
@@ -354,6 +380,7 @@
 			m_r ret;                                                                            \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, p2, p3, p4, &ret); \
 			SYNC_DEBUG                                                                          \
+			MAIN_THREAD_SYNC_CHECK                                                              \
 			return ret;                                                                         \
 		} else {                                                                                \
 			command_queue.flush_if_pending();                                                   \
@@ -367,6 +394,7 @@
 		if (Thread::get_caller_id() != server_thread) {                                    \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1, p2, p3, p4); \
 			SYNC_DEBUG                                                                     \
+			MAIN_THREAD_SYNC_CHECK                                                         \
 		} else {                                                                           \
 			command_queue.flush_if_pending();                                              \
 			server_name->m_type(p1, p2, p3, p4);                                           \
@@ -378,6 +406,7 @@
 		if (Thread::get_caller_id() != server_thread) {                                    \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1, p2, p3, p4); \
 			SYNC_DEBUG                                                                     \
+			MAIN_THREAD_SYNC_CHECK                                                         \
 		} else {                                                                           \
 			command_queue.flush_if_pending();                                              \
 			server_name->m_type(p1, p2, p3, p4);                                           \
@@ -412,6 +441,7 @@
 			m_r ret;                                                                                \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, p2, p3, p4, p5, &ret); \
 			SYNC_DEBUG                                                                              \
+			MAIN_THREAD_SYNC_CHECK                                                                  \
 			return ret;                                                                             \
 		} else {                                                                                    \
 			command_queue.flush_if_pending();                                                       \
@@ -425,6 +455,7 @@
 			m_r ret;                                                                                \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, p2, p3, p4, p5, &ret); \
 			SYNC_DEBUG                                                                              \
+			MAIN_THREAD_SYNC_CHECK                                                                  \
 			return ret;                                                                             \
 		} else {                                                                                    \
 			command_queue.flush_if_pending();                                                       \
@@ -438,6 +469,7 @@
 		if (Thread::get_caller_id() != server_thread) {                                        \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1, p2, p3, p4, p5); \
 			SYNC_DEBUG                                                                         \
+			MAIN_THREAD_SYNC_CHECK                                                             \
 		} else {                                                                               \
 			command_queue.flush_if_pending();                                                  \
 			server_name->m_type(p1, p2, p3, p4, p5);                                           \
@@ -449,6 +481,7 @@
 		if (Thread::get_caller_id() != server_thread) {                                         \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1, p2, p3, p4, p5);  \
 			SYNC_DEBUG                                                                          \
+			MAIN_THREAD_SYNC_CHECK                                                              \
 		} else {                                                                                \
 			command_queue.flush_if_pending();                                                   \
 			server_name->m_type(p1, p2, p3, p4, p5);                                            \
@@ -483,6 +516,7 @@
 			m_r ret;                                                                                    \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, p2, p3, p4, p5, p6, &ret); \
 			SYNC_DEBUG                                                                                  \
+			MAIN_THREAD_SYNC_CHECK                                                                      \
 			return ret;                                                                                 \
 		} else {                                                                                        \
 			command_queue.flush_if_pending();                                                           \
@@ -496,6 +530,7 @@
 			m_r ret;                                                                                      \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, p2, p3, p4, p5, p6, &ret);   \
 			SYNC_DEBUG                                                                                    \
+			MAIN_THREAD_SYNC_CHECK                                                                        \
 			return ret;                                                                                   \
 		} else {                                                                                          \
 			command_queue.flush_if_pending();                                                             \
@@ -509,6 +544,7 @@
 		if (Thread::get_caller_id() != server_thread) {                                              \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1, p2, p3, p4, p5, p6);   \
 			SYNC_DEBUG                                                                               \
+			MAIN_THREAD_SYNC_CHECK                                                                   \
 		} else {                                                                                     \
 			command_queue.flush_if_pending();                                                        \
 			server_name->m_type(p1, p2, p3, p4, p5, p6);                                             \
@@ -520,6 +556,7 @@
 		if (Thread::get_caller_id() != server_thread) {                                                    \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1, p2, p3, p4, p5, p6);         \
 			SYNC_DEBUG                                                                                     \
+			MAIN_THREAD_SYNC_CHECK                                                                         \
 		} else {                                                                                           \
 			command_queue.flush_if_pending();                                                              \
 			server_name->m_type(p1, p2, p3, p4, p5, p6);                                                   \
@@ -554,6 +591,7 @@
 			m_r ret;                                                                                           \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, p2, p3, p4, p5, p6, p7, &ret);    \
 			SYNC_DEBUG                                                                                         \
+			MAIN_THREAD_SYNC_CHECK                                                                             \
 			return ret;                                                                                        \
 		} else {                                                                                               \
 			command_queue.flush_if_pending();                                                                  \
@@ -567,6 +605,7 @@
 			m_r ret;                                                                                                 \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, p2, p3, p4, p5, p6, p7, &ret);          \
 			SYNC_DEBUG                                                                                               \
+			MAIN_THREAD_SYNC_CHECK                                                                                   \
 			return ret;                                                                                              \
 		} else {                                                                                                     \
 			command_queue.flush_if_pending();                                                                        \
@@ -580,6 +619,7 @@
 		if (Thread::get_caller_id() != server_thread) {                                                         \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1, p2, p3, p4, p5, p6, p7);          \
 			SYNC_DEBUG                                                                                          \
+			MAIN_THREAD_SYNC_CHECK                                                                              \
 		} else {                                                                                                \
 			command_queue.flush_if_pending();                                                                   \
 			server_name->m_type(p1, p2, p3, p4, p5, p6, p7);                                                    \
@@ -591,6 +631,7 @@
 		if (Thread::get_caller_id() != server_thread) {                                                               \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1, p2, p3, p4, p5, p6, p7);                \
 			SYNC_DEBUG                                                                                                \
+			MAIN_THREAD_SYNC_CHECK                                                                                    \
 		} else {                                                                                                      \
 			command_queue.flush_if_pending();                                                                         \
 			server_name->m_type(p1, p2, p3, p4, p5, p6, p7);                                                          \
@@ -625,6 +666,7 @@
 			m_r ret;                                                                                                      \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, p2, p3, p4, p5, p6, p7, p8, &ret);           \
 			SYNC_DEBUG                                                                                                    \
+			MAIN_THREAD_SYNC_CHECK                                                                                        \
 			return ret;                                                                                                   \
 		} else {                                                                                                          \
 			command_queue.flush_if_pending();                                                                             \
@@ -638,6 +680,7 @@
 			m_r ret;                                                                                                            \
 			command_queue.push_and_ret(server_name, &ServerName::m_type, p1, p2, p3, p4, p5, p6, p7, p8, &ret);                 \
 			SYNC_DEBUG                                                                                                          \
+			MAIN_THREAD_SYNC_CHECK                                                                                              \
 			return ret;                                                                                                         \
 		} else {                                                                                                                \
 			command_queue.flush_if_pending();                                                                                   \
@@ -651,6 +694,7 @@
 		if (Thread::get_caller_id() != server_thread) {                                                                    \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1, p2, p3, p4, p5, p6, p7, p8);                 \
 			SYNC_DEBUG                                                                                                     \
+			MAIN_THREAD_SYNC_CHECK                                                                                         \
 		} else {                                                                                                           \
 			command_queue.flush_if_pending();                                                                              \
 			server_name->m_type(p1, p2, p3, p4, p5, p6, p7, p8);                                                           \
@@ -662,6 +706,7 @@
 		if (Thread::get_caller_id() != server_thread) {                                                                          \
 			command_queue.push_and_sync(server_name, &ServerName::m_type, p1, p2, p3, p4, p5, p6, p7, p8);                       \
 			SYNC_DEBUG                                                                                                           \
+			MAIN_THREAD_SYNC_CHECK                                                                                               \
 		} else {                                                                                                                 \
 			command_queue.flush_if_pending();                                                                                    \
 			server_name->m_type(p1, p2, p3, p4, p5, p6, p7, p8);                                                                 \

tests/core/templates/test_command_queue.h

@@ -33,6 +33,7 @@
 
 #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"
@@ -100,7 +101,7 @@ public:
 	ThreadWork reader_threadwork;
 	ThreadWork writer_threadwork;
 
-	CommandQueueMT command_queue = CommandQueueMT(true);
+	CommandQueueMT command_queue;
 
 	enum TestMsgType {
 		TEST_MSG_FUNC1_TRANSFORM,
@@ -119,6 +120,7 @@ public:
 	bool exit_threads = false;
 
 	Thread reader_thread;
+	WorkerThreadPool::TaskID reader_task_id = WorkerThreadPool::INVALID_TASK_ID;
 	Thread writer_thread;
 
 	int func1_count = 0;
@@ -148,11 +150,16 @@ public:
 	void reader_thread_loop() {
 		reader_threadwork.thread_wait_for_work();
 		while (!exit_threads) {
-			if (message_count_to_read < 0) {
+			if (reader_task_id == WorkerThreadPool::INVALID_TASK_ID) {
 				command_queue.flush_all();
-			}
-			for (int i = 0; i < message_count_to_read; i++) {
-				command_queue.wait_and_flush();
+			} 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;
 
@@ -216,8 +223,13 @@ public:
 		sts->writer_thread_loop();
 	}
 
-	void init_threads() {
-		reader_thread.start(&SharedThreadState::static_reader_thread_loop, this);
+	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() {
@@ -225,16 +237,20 @@ public:
 		reader_threadwork.main_start_work();
 		writer_threadwork.main_start_work();
 
-		reader_thread.wait_to_finish();
+		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();
 	}
 };
 
-TEST_CASE("[CommandQueue] Test Queue Basics") {
+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();
+	sts.init_threads(p_use_thread_pool_sync);
 
 	sts.add_msg_to_write(SharedThreadState::TEST_MSG_FUNC1_TRANSFORM);
 	sts.writer_threadwork.main_start_work();
@@ -272,6 +288,14 @@ TEST_CASE("[CommandQueue] Test Queue Basics") {
 			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);

tests/core/threads/test_worker_thread_pool.h

@@ -38,6 +38,7 @@
 namespace TestWorkerThreadPool {
 
 static LocalVector<SafeNumeric<int>> counter;
+static SafeFlag exit;
 
 static void static_test(void *p_arg) {
 	counter[(uint64_t)p_arg].increment();
@@ -106,6 +107,72 @@ TEST_CASE("[WorkerThreadPool] Process elements using group tasks") {
 	}
 }
 
+static void static_test_daemon(void *p_arg) {
+	while (!exit.is_set()) {
+		counter[0].add(1);
+		WorkerThreadPool::get_singleton()->yield();
+	}
+}
+
+static void static_busy_task(void *p_arg) {
+	while (!exit.is_set()) {
+		OS::get_singleton()->delay_usec(1);
+	}
+}
+
+static void static_legit_task(void *p_arg) {
+	*((bool *)p_arg) = counter[0].get() > 0;
+	counter[1].add(1);
+}
+
+TEST_CASE("[WorkerThreadPool] Run a yielding daemon as the only hope for other tasks to run") {
+	exit.clear();
+	counter.clear();
+	counter.resize(2);
+
+	WorkerThreadPool::TaskID daemon_task_id = WorkerThreadPool::get_singleton()->add_native_task(static_test_daemon, nullptr, true);
+
+	int num_threads = WorkerThreadPool::get_singleton()->get_thread_count();
+
+	// Keep all the other threads busy.
+	LocalVector<WorkerThreadPool::TaskID> task_ids;
+	for (int i = 0; i < num_threads - 1; i++) {
+		task_ids.push_back(WorkerThreadPool::get_singleton()->add_native_task(static_busy_task, nullptr, true));
+	}
+
+	LocalVector<WorkerThreadPool::TaskID> legit_task_ids;
+	LocalVector<bool> legit_task_needed_yield;
+	int legit_tasks_count = num_threads * 4;
+	legit_task_needed_yield.resize(legit_tasks_count);
+	for (int i = 0; i < legit_tasks_count; i++) {
+		legit_task_needed_yield[i] = false;
+		task_ids.push_back(WorkerThreadPool::get_singleton()->add_native_task(static_legit_task, &legit_task_needed_yield[i], i >= legit_tasks_count / 2));
+	}
+
+	while (counter[1].get() != legit_tasks_count) {
+		OS::get_singleton()->delay_usec(1);
+	}
+
+	exit.set();
+	for (uint32_t i = 0; i < task_ids.size(); i++) {
+		WorkerThreadPool::get_singleton()->wait_for_task_completion(task_ids[i]);
+	}
+	WorkerThreadPool::get_singleton()->notify_yield_over(daemon_task_id);
+	WorkerThreadPool::get_singleton()->wait_for_task_completion(daemon_task_id);
+
+	CHECK_MESSAGE(counter[0].get() > 0, "Daemon task should have looped at least once.");
+	CHECK_MESSAGE(counter[1].get() == legit_tasks_count, "All legit tasks should have been able to run.");
+
+	bool all_needed_yield = true;
+	for (int i = 0; i < legit_tasks_count; i++) {
+		if (!legit_task_needed_yield[i]) {
+			all_needed_yield = false;
+			break;
+		}
+	}
+	CHECK_MESSAGE(all_needed_yield, "All legit tasks should have needed the daemon yielding to run.");
+}
+
 } // namespace TestWorkerThreadPool
 
 #endif // TEST_WORKER_THREAD_POOL_H