seedlingattempt/godot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Redo of Message Queue

Browse Source 
* Functionality moved to a base class CallQueue, which will be used for inter-thread communication within the scene.
* MessageQueue now uses growing pages, starts from a single 4k page.
* Limit still exists, but because its not allocated by default, it can be much higher.
This commit is contained in:
Juan Linietsky
2023-04-11 17:55:06 +02:00
parent 6d7413be74
commit 6055e4424a
3 changed files with 354 additions and 198 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
core/object/message_queue.h
View File

@ -33,26 +33,45 @@
#include "core/object/object_id.h"
#include "core/os/thread_safe.h"
#include "core/templates/local_vector.h"
#include "core/templates/paged_allocator.h"
#include "core/variant/variant.h"
class Object;
class MessageQueue {
_THREAD_SAFE_CLASS_
class CallQueue {
public:
enum {
DEFAULT_QUEUE_SIZE_KB = 4096
PAGE_SIZE_BYTES = 4096
};
private:
enum {
TYPE_CALL,
TYPE_NOTIFICATION,
TYPE_SET,
TYPE_END, // End marker.
FLAG_SHOW_ERROR = 1 << 14,
FLAG_MASK = FLAG_SHOW_ERROR - 1
FLAG_MASK = FLAG_SHOW_ERROR - 1,
};
struct Page {
uint8_t data[PAGE_SIZE_BYTES];
};
Mutex mutex;
typedef PagedAllocator<Page, true> Allocator;
Allocator *allocator = nullptr;
bool allocator_is_custom = false;
LocalVector<Page *> pages;
LocalVector<uint32_t> page_messages;
uint32_t max_pages = 0;
uint32_t pages_used = 0;
uint32_t page_offset = 0;
bool flushing = false;
struct Message {
Callable callable;
int16_t type;
@ -62,20 +81,21 @@ class MessageQueue {
};
};
uint8_t *buffer = nullptr;
uint32_t buffer_end = 0;
uint32_t buffer_max_used = 0;
uint32_t buffer_size = 0;
_FORCE_INLINE_ void _ensure_first_page() {
if (unlikely(pages.is_empty())) {
pages.push_back(allocator->alloc());
page_messages.push_back(0);
pages_used = 1;
}
}
void _add_page();
void _call_function(const Callable &p_callable, const Variant *p_args, int p_argcount, bool p_show_error);
static MessageQueue *singleton;
bool flushing = false;
String error_text;
public:
static MessageQueue *get_singleton();
Error push_callp(ObjectID p_id, const StringName &p_method, const Variant **p_args, int p_argcount, bool p_show_error = false);
template <typename... VarArgs>
Error push_call(ObjectID p_id, const StringName &p_method, VarArgs... p_args) {
@ -87,9 +107,9 @@ public:
return push_callp(p_id, p_method, sizeof...(p_args) == 0 ? nullptr : (const Variant **)argptrs, sizeof...(p_args));
}
Error push_notification(ObjectID p_id, int p_notification);
Error push_set(ObjectID p_id, const StringName &p_prop, const Variant &p_value);
Error push_callablep(const Callable &p_callable, const Variant **p_args, int p_argcount, bool p_show_error = false);
Error push_set(ObjectID p_id, const StringName &p_prop, const Variant &p_value);
Error push_notification(ObjectID p_id, int p_notification);
template <typename... VarArgs>
Error push_callable(const Callable &p_callable, VarArgs... p_args) {
@ -115,13 +135,22 @@ public:
Error push_notification(Object *p_object, int p_notification);
Error push_set(Object *p_object, const StringName &p_prop, const Variant &p_value);
Error flush();
void clear();
void statistics();
void flush();
bool is_flushing() const;
int get_max_buffer_usage() const;
CallQueue(Allocator *p_custom_allocator = 0, uint32_t p_max_pages = 8192, const String &p_error_text = String());
virtual ~CallQueue();
};
class MessageQueue : public CallQueue {
static MessageQueue *singleton;
public:
_FORCE_INLINE_ static MessageQueue *get_singleton() { return singleton; }
MessageQueue();
~MessageQueue();
};