core/object/message_queue.cpp: Calls to statistics() function needs to be removed

[EvanNibbe]

Godot version

Godot 4.0 alpha 10, alpha 12, dev (v4.0.alpha.custom_build [3fe89e7])

System information

Ubuntu 20.04.3 LTS Memory: 31.3 GB

Issue description

Based on a plain reading of the statistics function in core/object/message_queue.cpp, the function should be entirely unnecessary, since it does not directly operate on any non-local memory, and returns nothing. However, this function gets called so continually in my project, that the project simply froze until I used kill -9 on Godot. I proceeded to make a new version of Godot where the sole change was to remove all calls to the statistics(); function in message_queue.cpp, and return FAILED; instead of freezing my project due to this function, the project "worked", but there was a regression in the form of the cursor being registered in the Editor as 1 inch away from where my cursor actually was, causing many problems thus (but detached windows were unaffected).

Steps to reproduce

/* message_queue.cpp use this to replace the message_queue.cpp source file,
then compile with
scons -j23 platform=linuxbsd tools=yes use_lto=yes CXXFLAGS=-O3

Then open Godot with a project with sufficiently large object files in the 3D editor.
My own project used many instances of the files contained here: https://utdallas.box.com/s/oy709oo4pes6sqzjuhyazvsywseyw76t

/*************************************************************************/
/*                       This file is part of:                           */
/*                           GODOT ENGINE                                */
/*                      https://godotengine.org                          */
/*************************************************************************/
/* Copyright (c) 2007-2022 Juan Linietsky, Ariel Manzur.                 */
/* Copyright (c) 2014-2022 Godot Engine contributors (cf. AUTHORS.md).   */
/*                                                                       */
/* Permission is hereby granted, free of charge, to any person obtaining */
/* a copy of this software and associated documentation files (the       */
/* "Software"), to deal in the Software without restriction, including   */
/* without limitation the rights to use, copy, modify, merge, publish,   */
/* distribute, sublicense, and/or sell copies of the Software, and to    */
/* permit persons to whom the Software is furnished to do so, subject to */
/* the following conditions:                                             */
/*                                                                       */
/* The above copyright notice and this permission notice shall be        */
/* included in all copies or substantial portions of the Software.       */
/*                                                                       */
/* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,       */
/* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF    */
/* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.*/
/* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY  */
/* CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,  */
/* TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE     */
/* SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.                */
/*************************************************************************/

#include "message_queue.h"

#include "core/config/project_settings.h"
#include "core/core_string_names.h"
#include "core/object/class_db.h"
#include "core/object/script_language.h"

MessageQueue *MessageQueue::singleton = nullptr;

MessageQueue *MessageQueue::get_singleton() {
	return singleton;
}

Error MessageQueue::push_callp(ObjectID p_id, const StringName &p_method, const Variant **p_args, int p_argcount, bool p_show_error) {
	return push_callablep(Callable(p_id, p_method), p_args, p_argcount, p_show_error);
}

Error MessageQueue::push_set(ObjectID p_id, const StringName &p_prop, const Variant &p_value) {
	_THREAD_SAFE_METHOD_

	uint8_t room_needed = sizeof(Message) + sizeof(Variant);

	if ((buffer_end + room_needed) >= buffer_size) {
		String type;
		if (ObjectDB::get_instance(p_id)) {
			type = ObjectDB::get_instance(p_id)->get_class();
		}
		print_line("Failed set: " + type + ":" + p_prop + " target ID: " + itos(p_id));
		//statistics();
		return FAILED;
		ERR_FAIL_V_MSG(ERR_OUT_OF_MEMORY, "Message queue out of memory. Try increasing 'memory/limits/message_queue/max_size_kb' in project settings.");
	}

	Message *msg = memnew_placement(&buffer[buffer_end], Message);
	msg->args = 1;
	msg->callable = Callable(p_id, p_prop);
	msg->type = TYPE_SET;

	buffer_end += sizeof(Message);

	Variant *v = memnew_placement(&buffer[buffer_end], Variant);
	buffer_end += sizeof(Variant);
	*v = p_value;

	return OK;
}

Error MessageQueue::push_notification(ObjectID p_id, int p_notification) {
	_THREAD_SAFE_METHOD_

	ERR_FAIL_COND_V(p_notification < 0, ERR_INVALID_PARAMETER);

	uint8_t room_needed = sizeof(Message);

	if ((buffer_end + room_needed) >= buffer_size) {
		print_line("Failed notification: " + itos(p_notification) + " target ID: " + itos(p_id));
		//statistics();
		return FAILED;
		ERR_FAIL_V_MSG(ERR_OUT_OF_MEMORY, "Message queue out of memory. Try increasing 'memory/limits/message_queue/max_size_kb' in project settings.");
	}

	Message *msg = memnew_placement(&buffer[buffer_end], Message);

	msg->type = TYPE_NOTIFICATION;
	msg->callable = Callable(p_id, CoreStringNames::get_singleton()->notification); //name is meaningless but callable needs it
	//msg->target;
	msg->notification = p_notification;

	buffer_end += sizeof(Message);

	return OK;
}

Error MessageQueue::push_callp(Object *p_object, const StringName &p_method, const Variant **p_args, int p_argcount, bool p_show_error) {
	return push_callp(p_object->get_instance_id(), p_method, p_args, p_argcount, p_show_error);
}

Error MessageQueue::push_notification(Object *p_object, int p_notification) {
	return push_notification(p_object->get_instance_id(), p_notification);
}

Error MessageQueue::push_set(Object *p_object, const StringName &p_prop, const Variant &p_value) {
	return push_set(p_object->get_instance_id(), p_prop, p_value);
}

Error MessageQueue::push_callablep(const Callable &p_callable, const Variant **p_args, int p_argcount, bool p_show_error) {
	_THREAD_SAFE_METHOD_

	int room_needed = sizeof(Message) + sizeof(Variant) * p_argcount;

	if ((buffer_end + room_needed) >= buffer_size) {
		print_line("Failed method: " + p_callable);
		//statistics(); //it was Evan's doing to remove this line
		return FAILED;
		ERR_FAIL_V_MSG(ERR_OUT_OF_MEMORY, "Message queue out of memory. Try increasing 'memory/limits/message_queue/max_size_kb' in project settings.");
	}

	Message *msg = memnew_placement(&buffer[buffer_end], Message);
	msg->args = p_argcount;
	msg->callable = p_callable;
	msg->type = TYPE_CALL;
	if (p_show_error) {
		msg->type |= FLAG_SHOW_ERROR;
	}

	buffer_end += sizeof(Message);

	for (int i = 0; i < p_argcount; i++) {
		Variant *v = memnew_placement(&buffer[buffer_end], Variant);
		buffer_end += sizeof(Variant);
		*v = *p_args[i];
	}

	return OK;
}

void MessageQueue::statistics() {
	//Note: set_count, notify_count, call_count, null_count, message, target, are local variables
	//it would appear that nothing is done by this function, so it should be able to be removed
	HashMap<StringName, int> set_count;
	HashMap<int, int> notify_count;
	HashMap<Callable, int> call_count;
	int null_count = 0;

	uint32_t read_pos = 0;
	while (read_pos < buffer_end) {
		Message *message = (Message *)&buffer[read_pos];

		Object *target = message->callable.get_object();

		if (target != nullptr) {
			switch (message->type & FLAG_MASK) {
				case TYPE_CALL: {
					if (!call_count.has(message->callable)) {
						call_count[message->callable] = 0;
					}

					call_count[message->callable]++;

				} break;
				case TYPE_NOTIFICATION: {
					if (!notify_count.has(message->notification)) {
						notify_count[message->notification] = 0;
					}

					notify_count[message->notification]++;

				} break;
				case TYPE_SET: {
					StringName t = message->callable.get_method();
					if (!set_count.has(t)) {
						set_count[t] = 0;
					}

					set_count[t]++;

				} break;
			}

		} else {
			//object was deleted
			//print_line("Object was deleted while awaiting a callback");
			//the above line seemed to have caused the most problems in slowing down the system.

			null_count++;
		}

		read_pos += sizeof(Message);
		if ((message->type & FLAG_MASK) != TYPE_NOTIFICATION) {
			read_pos += sizeof(Variant) * message->args;
		}
	}

	print_line("TOTAL BYTES: " + itos(buffer_end));
	print_line("NULL count: " + itos(null_count));

	for (const KeyValue<StringName, int> &E : set_count) {
		print_line("SET " + E.key + ": " + itos(E.value));
	}

	for (const KeyValue<Callable, int> &E : call_count) {
		print_line("CALL " + E.key + ": " + itos(E.value));
	}

	for (const KeyValue<int, int> &E : notify_count) {
		print_line("NOTIFY " + itos(E.key) + ": " + itos(E.value));
	}
}

int MessageQueue::get_max_buffer_usage() const {
	return buffer_max_used;
}

void MessageQueue::_call_function(const Callable &p_callable, const Variant *p_args, int p_argcount, bool p_show_error) {
	const Variant **argptrs = nullptr;
	if (p_argcount) {
		argptrs = (const Variant **)alloca(sizeof(Variant *) * p_argcount);
		for (int i = 0; i < p_argcount; i++) {
			argptrs[i] = &p_args[i];
		}
	}

	Callable::CallError ce;
	Variant ret;
	p_callable.call(argptrs, p_argcount, ret, ce);
	if (p_show_error && ce.error != Callable::CallError::CALL_OK) {
		ERR_PRINT("Error calling deferred method: " + Variant::get_callable_error_text(p_callable, argptrs, p_argcount, ce) + ".");
	}
}

void MessageQueue::flush() {
	if (buffer_end > buffer_max_used) {
		buffer_max_used = buffer_end;
	}

	uint32_t read_pos = 0;

	//using reverse locking strategy
	_THREAD_SAFE_LOCK_

	if (flushing) {
		_THREAD_SAFE_UNLOCK_
		ERR_FAIL_COND(flushing); //already flushing, you did something odd
	}
	flushing = true;

	while (read_pos < buffer_end) {
		//lock on each iteration, so a call can re-add itself to the message queue

		Message *message = (Message *)&buffer[read_pos];

		uint32_t advance = sizeof(Message);
		if ((message->type & FLAG_MASK) != TYPE_NOTIFICATION) {
			advance += sizeof(Variant) * message->args;
		}

		//pre-advance so this function is reentrant
		read_pos += advance;

		_THREAD_SAFE_UNLOCK_

		Object *target = message->callable.get_object();

		if (target != nullptr) {
			switch (message->type & FLAG_MASK) {
				case TYPE_CALL: {
					Variant *args = (Variant *)(message + 1);

					// messages don't expect a return value

					_call_function(message->callable, args, message->args, message->type & FLAG_SHOW_ERROR);

				} break;
				case TYPE_NOTIFICATION: {
					// messages don't expect a return value
					target->notification(message->notification);

				} break;
				case TYPE_SET: {
					Variant *arg = (Variant *)(message + 1);
					// messages don't expect a return value
					target->set(message->callable.get_method(), *arg);

				} break;
			}
		}

		if ((message->type & FLAG_MASK) != TYPE_NOTIFICATION) {
			Variant *args = (Variant *)(message + 1);
			for (int i = 0; i < message->args; i++) {
				args[i].~Variant();
			}
		}

		message->~Message();

		_THREAD_SAFE_LOCK_
	}

	buffer_end = 0; // reset buffer
	flushing = false;
	_THREAD_SAFE_UNLOCK_
}

bool MessageQueue::is_flushing() const {
	return flushing;
}

MessageQueue::MessageQueue() {
	ERR_FAIL_COND_MSG(singleton != nullptr, "A MessageQueue singleton already exists.");
	singleton = this;

	buffer_size = GLOBAL_DEF_RST("memory/limits/message_queue/max_size_kb", DEFAULT_QUEUE_SIZE_KB);
	ProjectSettings::get_singleton()->set_custom_property_info("memory/limits/message_queue/max_size_kb", PropertyInfo(Variant::INT, "memory/limits/message_queue/max_size_kb", PROPERTY_HINT_RANGE, "1024,4096,1,or_greater"));
	buffer_size *= 1024;
	buffer = memnew_arr(uint8_t, buffer_size);
}

MessageQueue::~MessageQueue() {
	uint32_t read_pos = 0;

	while (read_pos < buffer_end) {
		Message *message = (Message *)&buffer[read_pos];
		Variant *args = (Variant *)(message + 1);
		int argc = message->args;
		if ((message->type & FLAG_MASK) != TYPE_NOTIFICATION) {
			for (int i = 0; i < argc; i++) {
				args[i].~Variant();
			}
		}
		message->~Message();

		read_pos += sizeof(Message);
		if ((message->type & FLAG_MASK) != TYPE_NOTIFICATION) {
			read_pos += sizeof(Variant) * message->args;
		}
	}

	singleton = nullptr;
	memdelete_arr(buffer);
}

Minimal reproduction project

The large object files I was using are here: https://utdallas.box.com/s/oy709oo4pes6sqzjuhyazvsywseyw76t
The file size exceeds what Godot allows, so I am using my student box drive with a public link.

[Calinou]

PS: Code blocks should use triple backticks like this (with an optional language name for syntax highlighting):

```cpp
code here
```

I edited your post accordingly, but remember to do this in the future 🙂

[lawnjelly]

This is probably due to the message queue running out of memory, and it cunningly prints out a lot of debug statistics every frame from that point lol. 😁
This should probably be changed to a CRASH_NOW, as it is pretty much a fatal error, providing it gets to print out the error message before crashing.

I can fix this soon if #65740 is merged. I did think about combining, but it would be better to bikeshed separately, as the there are a couple of solutions - CRASH_NOW or hide the output after the first occurrence.

This error will also be less likely to occur anyway with growable message queue. 👍

[akien-mga]

This may be solved already in 4.1 as the MessageQueue is growable, though I'll let you assess further @lawnjelly.

I don't remember ever seeing the statistics() method produce anything else than a crash, so it would be good to check if it's now useful, or if it can be fixed or should be removed.

[rune-scape]

may be fixed by #89490
haven't tested tho

[AThousandShips]

Can someone confirm this still occurs? Should have been fixed

[AThousandShips]

Closing as resolved, if someone can still reproduce this please comment