Refactor Node Processing (WIP)

* Node processing works on the concept of process groups.
* A node group can be inherited, run on main thread, or a sub-thread.
* Groups can be ordered.
* Process priority is now present for physics.

This is the first steps towards implementing godotengine/godot-proposals#6424.
No threading or thread guards exist yet.

core/object/message_queue.cpp

@@ -432,6 +432,17 @@ bool CallQueue::is_flushing() const {
 	return flushing;
 }
 
+bool CallQueue::has_messages() const {
+	if (pages_used == 0) {
+		return false;
+	}
+	if (pages_used == 1 && page_messages[0] == 0) {
+		return false;
+	}
+
+	return true;
+}
+
 int CallQueue::get_max_buffer_usage() const {
 	return pages.size() * PAGE_SIZE_BYTES;
 }

core/object/message_queue.h

@@ -45,6 +45,13 @@ class CallQueue {
 		PAGE_SIZE_BYTES = 4096
 	};
 
+	struct Page {
+		uint8_t data[PAGE_SIZE_BYTES];
+	};
+
+	// Needs to be public to be able to define it outside the class.
+	typedef PagedAllocator<Page, true> Allocator;
+
 private:
 	enum {
 		TYPE_CALL,
@@ -56,12 +63,7 @@ class CallQueue {
 		FLAG_MASK = FLAG_NULL_IS_OK - 1,
 	};
 
-	struct Page {
-		uint8_t data[PAGE_SIZE_BYTES];
-	};
-
 	Mutex mutex;
-	typedef PagedAllocator<Page, true> Allocator;
 
 	Allocator *allocator = nullptr;
 	bool allocator_is_custom = false;
@@ -140,6 +142,8 @@ class CallQueue {
 	void clear();
 	void statistics();
 
+	bool has_messages() const;
+
 	bool is_flushing() const;
 	int get_max_buffer_usage() const;
 

core/object/object.cpp

@@ -200,6 +200,10 @@ bool Object::_predelete() {
 	return _predelete_ok;
 }
 
+void Object::cancel_free() {
+	_predelete_ok = false;
+}
+
 void Object::_postinitialize() {
 	_class_name_ptr = _get_class_namev(); // Set the direct pointer, which is much faster to obtain, but can only happen after postinitialize.
 	_initialize_classv();
@@ -1551,6 +1555,7 @@ void Object::_bind_methods() {
 	ClassDB::bind_method(D_METHOD("tr_n", "message", "plural_message", "n", "context"), &Object::tr_n, DEFVAL(""));
 
 	ClassDB::bind_method(D_METHOD("is_queued_for_deletion"), &Object::is_queued_for_deletion);
+	ClassDB::bind_method(D_METHOD("cancel_free"), &Object::cancel_free);
 
 	ClassDB::add_virtual_method("Object", MethodInfo("free"), false);
 

core/object/object.h

@@ -924,6 +924,8 @@ class Object {
 
 	_ALWAYS_INLINE_ bool is_ref_counted() const { return type_is_reference; }
 
+	void cancel_free();
+
 	Object();
 	virtual ~Object();
 };

core/os/thread.cpp

@@ -51,7 +51,7 @@ void Thread::_set_platform_functions(const PlatformFunctions &p_functions) {
 }
 
 void Thread::callback(ID p_caller_id, const Settings &p_settings, Callback p_callback, void *p_userdata) {
-	Thread::caller_id = p_caller_id;
+	Thread::caller_id = _thread_id_hash(std::this_thread::get_id());
 	if (platform_functions.set_priority) {
 		platform_functions.set_priority(p_settings.priority);
 	}

core/os/thread.h

@@ -95,6 +95,8 @@ class Thread {
 	// get the ID of the main thread
 	_FORCE_INLINE_ static ID get_main_id() { return main_thread_id; }
 
+	_FORCE_INLINE_ static bool is_main_thread() { return main_thread_id == caller_id; }
+
 	static Error set_name(const String &p_name);
 
 	void start(Thread::Callback p_callback, void *p_user, const Settings &p_settings = Settings());

core/templates/hash_map.h

@@ -97,7 +97,7 @@ class HashMap {
 	}
 
 	bool _lookup_pos(const TKey &p_key, uint32_t &r_pos) const {
-		if (elements == nullptr) {
+		if (elements == nullptr || num_elements == 0) {
 			return false; // Failed lookups, no elements
 		}
 
@@ -252,7 +252,7 @@ class HashMap {
 	}
 
 	void clear() {
-		if (elements == nullptr) {
+		if (elements == nullptr || num_elements == 0) {
 			return;
 		}
 		uint32_t capacity = hash_table_size_primes[capacity_index];

core/templates/hash_set.h

@@ -80,7 +80,7 @@ class HashSet {
 	}
 
 	bool _lookup_pos(const TKey &p_key, uint32_t &r_pos) const {
-		if (keys == nullptr) {
+		if (keys == nullptr || num_elements == 0) {
 			return false; // Failed lookups, no elements
 		}
 
@@ -237,7 +237,7 @@ class HashSet {
 	}
 
 	void clear() {
-		if (keys == nullptr) {
+		if (keys == nullptr || num_elements == 0) {
 			return;
 		}
 		uint32_t capacity = hash_table_size_primes[capacity_index];

core/templates/local_vector.h

@@ -99,11 +99,13 @@ class LocalVector {
 		}
 	}
 
-	void erase(const T &p_val) {
+	_FORCE_INLINE_ bool erase(const T &p_val) {
 		int64_t idx = find(p_val);
 		if (idx >= 0) {
 			remove_at(idx);
+			return true;
 		}
+		return false;
 	}
 
 	void invert() {

core/templates/vector.h

@@ -71,12 +71,15 @@ class Vector {
 	void fill(T p_elem);
 
 	void remove_at(int p_index) { _cowdata.remove_at(p_index); }
-	void erase(const T &p_val) {
+	_FORCE_INLINE_ bool erase(const T &p_val) {
 		int idx = find(p_val);
 		if (idx >= 0) {
 			remove_at(idx);
+			return true;
 		}
+		return false;
 	}
+
 	void reverse();
 
 	_FORCE_INLINE_ T *ptrw() { return _cowdata.ptrw(); }

doc/classes/Node.xml

@@ -182,6 +182,18 @@
 				[b]Note:[/b] For performance reasons, the order of node groups is [i]not[/i] guaranteed. The order of node groups should not be relied upon as it can vary across project runs.
 			</description>
 		</method>
+		<method name="call_deferred_thread_group" qualifiers="vararg">
+			<return type="Variant" />
+			<param index="0" name="method" type="StringName" />
+			<description>
+			</description>
+		</method>
+		<method name="call_thread_safe" qualifiers="vararg">
+			<return type="Variant" />
+			<param index="0" name="method" type="StringName" />
+			<description>
+			</description>
+		</method>
 		<method name="can_process" qualifiers="const">
 			<return type="bool" />
 			<description>
@@ -557,6 +569,18 @@
 				[b]Note:[/b] Internal children can only be moved within their expected "internal range" (see [code]internal[/code] parameter in [method add_child]).
 			</description>
 		</method>
+		<method name="notify_deferred_thread_group">
+			<return type="void" />
+			<param index="0" name="what" type="int" />
+			<description>
+			</description>
+		</method>
+		<method name="notify_thread_safe">
+			<return type="void" />
+			<param index="0" name="what" type="int" />
+			<description>
+			</description>
+		</method>
 		<method name="print_orphan_nodes" qualifiers="static">
 			<return type="void" />
 			<description>
@@ -690,6 +714,13 @@
 				Sends a [method rpc] to a specific peer identified by [param peer_id] (see [method MultiplayerPeer.set_target_peer]). Returns [code]null[/code].
 			</description>
 		</method>
+		<method name="set_deferred_thread_group">
+			<return type="void" />
+			<param index="0" name="property" type="StringName" />
+			<param index="1" name="value" type="Variant" />
+			<description>
+			</description>
+		</method>
 		<method name="set_display_folded">
 			<return type="void" />
 			<param index="0" name="fold" type="bool" />
@@ -778,6 +809,13 @@
 				Sets whether this is an instance load placeholder. See [InstancePlaceholder].
 			</description>
 		</method>
+		<method name="set_thread_safe">
+			<return type="void" />
+			<param index="0" name="property" type="StringName" />
+			<param index="1" name="value" type="Variant" />
+			<description>
+			</description>
+		</method>
 		<method name="update_configuration_warnings">
 			<return type="void" />
 			<description>
@@ -804,9 +842,17 @@
 		<member name="process_mode" type="int" setter="set_process_mode" getter="get_process_mode" enum="Node.ProcessMode" default="0">
 			Can be used to pause or unpause the node, or make the node paused based on the [SceneTree], or make it inherit the process mode from its parent (default).
 		</member>
+		<member name="process_physics_priority" type="int" setter="set_physics_process_priority" getter="get_physics_process_priority" default="0">
+		</member>
 		<member name="process_priority" type="int" setter="set_process_priority" getter="get_process_priority" default="0">
 			The node's priority in the execution order of the enabled processing callbacks (i.e. [constant NOTIFICATION_PROCESS], [constant NOTIFICATION_PHYSICS_PROCESS] and their internal counterparts). Nodes whose process priority value is [i]lower[/i] will have their processing callbacks executed first.
 		</member>
+		<member name="process_thread_group" type="int" setter="set_process_thread_group" getter="get_process_thread_group" enum="Node.ProcessThreadGroup" default="0">
+		</member>
+		<member name="process_thread_group_order" type="int" setter="set_process_thread_group_order" getter="get_process_thread_group_order">
+		</member>
+		<member name="process_thread_messages" type="int" setter="set_process_thread_messages" getter="get_process_thread_messages" enum="Node.ProcessThreadMessages">
+		</member>
 		<member name="scene_file_path" type="String" setter="set_scene_file_path" getter="get_scene_file_path">
 			If a scene is instantiated from a file, its topmost node contains the absolute file path from which it was loaded in [member scene_file_path] (e.g. [code]res://levels/1.tscn[/code]). Otherwise, [member scene_file_path] is set to an empty string.
 		</member>
@@ -1026,6 +1072,18 @@
 		<constant name="PROCESS_MODE_DISABLED" value="4" enum="ProcessMode">
 			Never process. Completely disables processing, ignoring the [SceneTree]'s paused property. This is the inverse of [constant PROCESS_MODE_ALWAYS].
 		</constant>
+		<constant name="PROCESS_THREAD_GROUP_INHERIT" value="0" enum="ProcessThreadGroup">
+		</constant>
+		<constant name="PROCESS_THREAD_GROUP_MAIN_THREAD" value="1" enum="ProcessThreadGroup">
+		</constant>
+		<constant name="PROCESS_THREAD_GROUP_SUB_THREAD" value="2" enum="ProcessThreadGroup">
+		</constant>
+		<constant name="FLAG_PROCESS_THREAD_MESSAGES" value="1" enum="ProcessThreadMessages">
+		</constant>
+		<constant name="FLAG_PROCESS_THREAD_MESSAGES_PHYSICS" value="2" enum="ProcessThreadMessages">
+		</constant>
+		<constant name="FLAG_PROCESS_THREAD_MESSAGES_ALL" value="3" enum="ProcessThreadMessages">
+		</constant>
 		<constant name="DUPLICATE_SIGNALS" value="1" enum="DuplicateFlags">
 			Duplicate the node's signals.
 		</constant>

doc/classes/Object.xml

@@ -354,6 +354,11 @@
 				Returns [code]true[/code] if the object is allowed to translate messages with [method tr] and [method tr_n]. See also [method set_message_translation].
 			</description>
 		</method>
+		<method name="cancel_free">
+			<return type="void" />
+			<description>
+			</description>
+		</method>
 		<method name="connect">
 			<return type="int" enum="Error" />
 			<param index="0" name="signal" type="StringName" />

editor/editor_node.cpp

@@ -633,6 +633,8 @@ void EditorNode::_notification(int p_what) {
 		} break;
 
 		case NOTIFICATION_ENTER_TREE: {
+			get_tree()->set_disable_node_threading(true); // No node threading while running editor.
+
 			Engine::get_singleton()->set_editor_hint(true);
 
 			Window *window = get_window();

main/main.cpp

@@ -180,6 +180,7 @@ static int converter_max_line_length = 100000;
 
 HashMap<Main::CLIScope, Vector<String>> forwardable_cli_arguments;
 #endif
+static bool single_threaded_scene = false;
 bool use_startup_benchmark = false;
 String startup_benchmark_file;
 
@@ -423,6 +424,7 @@ void Main::print_help(const char *p_binary) {
 	OS::get_singleton()->print("  --gpu-abort                       Abort on graphics API usage errors (usually validation layer errors). May help see the problem if your system freezes.\n");
 #endif
 	OS::get_singleton()->print("  --remote-debug <uri>              Remote debug (<protocol>://<host/IP>[:<port>], e.g. tcp://127.0.0.1:6007).\n");
+	OS::get_singleton()->print("  --single-threaded-scene           Scene tree runs in single-threaded mode. Sub-thread groups are disabled and run on the main thread.\n");
 #if defined(DEBUG_ENABLED)
 	OS::get_singleton()->print("  --debug-collisions                Show collision shapes when running the scene.\n");
 	OS::get_singleton()->print("  --debug-paths                     Show path lines when running the scene.\n");
@@ -1104,6 +1106,8 @@ Error Main::setup(const char *execpath, int argc, char *argv[], bool p_second_ph
 				OS::get_singleton()->print("Missing remote debug server uri, aborting.\n");
 				goto error;
 			}
+		} else if (I->get() == "--single-threaded-scene") {
+			single_threaded_scene = true;
 		} else if (I->get() == "--build-solutions") { // Build the scripting solution such C#
 
 			auto_build_solutions = true;
@@ -2775,6 +2779,10 @@ bool Main::start() {
 		}
 #endif
 
+		if (single_threaded_scene) {
+			sml->set_disable_node_threading(true);
+		}
+
 		bool embed_subwindows = GLOBAL_GET("display/window/subwindows/embed_subwindows");
 
 		if (single_window || (!project_manager && !editor && embed_subwindows) || !DisplayServer::get_singleton()->has_feature(DisplayServer::Feature::FEATURE_SUBWINDOWS)) {

scene/3d/skeleton_3d.cpp

@@ -226,6 +226,11 @@ void Skeleton3D::_update_process_order() {
 
 void Skeleton3D::_notification(int p_what) {
 	switch (p_what) {
+		case NOTIFICATION_ENTER_TREE: {
+			if (dirty) {
+				notification(NOTIFICATION_UPDATE_SKELETON);
+			}
+		} break;
 		case NOTIFICATION_UPDATE_SKELETON: {
 			RenderingServer *rs = RenderingServer::get_singleton();
 			Bone *bonesptr = bones.ptrw();
@@ -629,7 +634,9 @@ void Skeleton3D::_make_dirty() {
 		return;
 	}
 
-	MessageQueue::get_singleton()->push_notification(this, NOTIFICATION_UPDATE_SKELETON);
+	if (is_inside_tree()) {
+		notify_deferred_thread_group(NOTIFICATION_UPDATE_SKELETON);
+	}
 	dirty = true;
 }