maxmustermann/godot
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Page allocation in MessageQueue

Browse source 
Fixes #35653
This commit is contained in:
Stanislav 2021-08-28 20:47:54 +03:00
parent dcf2d09231
commit 91bfe0e135
2 changed files with 144 additions and 119 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

256
core/object/message_queue.cpp
View file

@ -64,7 +64,7 @@ Error MessageQueue::push_set(ObjectID p_id, const StringName &p_prop, const Vari
uint8_t room_needed = sizeof(Message) + sizeof(Variant);
if ((buffer_end + room_needed) >= buffer_size) {
if (room_needed > buffer_size) {
String type;
if (ObjectDB::get_instance(p_id)) {
type = ObjectDB::get_instance(p_id)->get_class();
@ -72,17 +72,20 @@ Error MessageQueue::push_set(ObjectID p_id, const StringName &p_prop, const Vari
print_line("Failed set: " + type + ":" + p_prop + " target ID: " + itos(p_id));
statistics();
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.");
} else if ((buffers[buffers.size() - 1].buffer_end + room_needed) >= buffer_size) {
buffers.push_back({ memnew_arr(uint8_t, buffer_size), 0 });
}
Buffer &buf = buffers.write[buffers.size() - 1];
Message *msg = memnew_placement(&buffer[buffer_end], Message);
Message *msg = memnew_placement(&buf.buffer[buf.buffer_end], Message);
msg->args = 1;
msg->callable = Callable(p_id, p_prop);
msg->type = TYPE_SET;
buffer_end += sizeof(Message);
buf.buffer_end += sizeof(Message);
Variant *v = memnew_placement(&buffer[buffer_end], Variant);
buffer_end += sizeof(Variant);
Variant *v = memnew_placement(&buf.buffer[buf.buffer_end], Variant);
buf.buffer_end += sizeof(Variant);
*v = p_value;
return OK;
@ -95,20 +98,23 @@ Error MessageQueue::push_notification(ObjectID p_id, int p_notification) {
uint8_t room_needed = sizeof(Message);
if ((buffer_end + room_needed) >= buffer_size) {
if (room_needed > buffer_size) {
print_line("Failed notification: " + itos(p_notification) + " target ID: " + itos(p_id));
statistics();
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.");
} else if ((buffers[buffers.size() - 1].buffer_end + room_needed) >= buffer_size) {
buffers.push_back({ memnew_arr(uint8_t, buffer_size), 0 });
}
Buffer &buf = buffers.write[buffers.size() - 1];
Message *msg = memnew_placement(&buffer[buffer_end], Message);
Message *msg = memnew_placement(&buf.buffer[buf.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);
buf.buffer_end += sizeof(Message);
return OK;
}
@ -128,15 +134,18 @@ Error MessageQueue::push_set(Object *p_object, const StringName &p_prop, const V
Error MessageQueue::push_callable(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;
uint32_t room_needed = sizeof(Message) + sizeof(Variant) * p_argcount;
if ((buffer_end + room_needed) >= buffer_size) {
if (room_needed > buffer_size) {
print_line("Failed method: " + p_callable);
statistics();
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.");
} else if ((buffers[buffers.size() - 1].buffer_end + room_needed) >= buffer_size) {
buffers.push_back({ memnew_arr(uint8_t, buffer_size), 0 });
}
Buffer &buf = buffers.write[buffers.size() - 1];
Message *msg = memnew_placement(&buffer[buffer_end], Message);
Message *msg = memnew_placement(&buf.buffer[buf.buffer_end], Message);
msg->args = p_argcount;
msg->callable = p_callable;
msg->type = TYPE_CALL;
@ -144,11 +153,11 @@ Error MessageQueue::push_callable(const Callable &p_callable, const Variant **p_
msg->type |= FLAG_SHOW_ERROR;
}
buffer_end += sizeof(Message);
buf.buffer_end += sizeof(Message);
for (int i = 0; i < p_argcount; i++) {
Variant *v = memnew_placement(&buffer[buffer_end], Variant);
buffer_end += sizeof(Variant);
Variant *v = memnew_placement(&buf.buffer[buf.buffer_end], Variant);
buf.buffer_end += sizeof(Variant);
*v = *p_args[i];
}
@ -176,55 +185,60 @@ void MessageQueue::statistics() {
Map<Callable, int> call_count;
int null_count = 0;
uint32_t read_pos = 0;
while (read_pos < buffer_end) {
Message *message = (Message *)&buffer[read_pos];
uint32_t total_bytes = 0;
for (int i = 0; i < buffers.size(); ++i) {
const Buffer &buf = buffers[i];
total_bytes += buf.buffer_end;
uint32_t read_pos = 0;
while (read_pos < buf.buffer_end) {
Message *message = (Message *)&buf.buffer[read_pos];
Object *target = message->callable.get_object();
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;
}
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]++;
call_count[message->callable]++;
} break;
case TYPE_NOTIFICATION: {
if (!notify_count.has(message->notification)) {
notify_count[message->notification] = 0;
}
} break;
case TYPE_NOTIFICATION: {
if (!notify_count.has(message->notification)) {
notify_count[message->notification] = 0;
}
notify_count[message->notification]++;
notify_count[message->notification]++;
} break;
case TYPE_SET: {
StringName t = message->callable.get_method();
if (!set_count.has(t)) {
set_count[t] = 0;
}
} break;
case TYPE_SET: {
StringName t = message->callable.get_method();
if (!set_count.has(t)) {
set_count[t] = 0;
}
set_count[t]++;
set_count[t]++;
} break;
} break;
}
} else {
//object was deleted
print_line("Object was deleted while awaiting a callback");
null_count++;
}
} else {
//object was deleted
print_line("Object was deleted while awaiting a callback");
null_count++;
}
read_pos += sizeof(Message);
if ((message->type & FLAG_MASK) != TYPE_NOTIFICATION) {
read_pos += sizeof(Variant) * message->args;
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("TOTAL BYTES: " + itos(total_bytes));
print_line("NULL count: " + itos(null_count));
for (Map<StringName, int>::Element *E = set_count.front(); E; E = E->next()) {
@ -262,12 +276,11 @@ void MessageQueue::_call_function(const Callable &p_callable, const Variant *p_a
}
void MessageQueue::flush() {
if (buffer_end > buffer_max_used) {
buffer_max_used = buffer_end;
uint32_t used = (buffers.size() - 1) * buffer_size + buffers[buffers.size() - 1].buffer_end;
if (used > buffer_max_used) {
buffer_max_used = used;
}
uint32_t read_pos = 0;
//using reverse locking strategy
_THREAD_SAFE_LOCK_
@ -277,60 +290,68 @@ void MessageQueue::flush() {
}
flushing = true;
while (read_pos < buffer_end) {
//lock on each iteration, so a call can re-add itself to the message queue
for (int j = 0; j < buffers.size(); ++j) {
const Buffer &buf = buffers[j];
uint32_t read_pos = 0;
while (read_pos < buf.buffer_end) {
//lock on each iteration, so a call can re-add itself to the message queue
Message *message = (Message *)&buffer[read_pos];
Message *message = (Message *)&buf.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;
uint32_t advance = sizeof(Message);
if ((message->type & FLAG_MASK) != TYPE_NOTIFICATION) {
advance += sizeof(Variant) * message->args;
}
}
if ((message->type & FLAG_MASK) != TYPE_NOTIFICATION) {
Variant *args = (Variant *)(message + 1);
for (int i = 0; i < message->args; i++) {
args[i].~Variant();
//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_
}
if (j) {
memdelete_arr(buf.buffer);
}
message->~Message();
_THREAD_SAFE_LOCK_
}
buffer_end = 0; // reset buffer
buffers.resize(1);
buffers.write[0].buffer_end = 0; // reset buffer
flushing = false;
_THREAD_SAFE_UNLOCK_
}
@ -346,29 +367,30 @@ MessageQueue::MessageQueue() {
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);
buffers.push_back({ memnew_arr(uint8_t, buffer_size), 0 });
}
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();
for (int j = 0; j < buffers.size(); ++j) {
const Buffer &buf = buffers[j];
uint32_t read_pos = 0;
while (read_pos < buf.buffer_end) {
Message *message = (Message *)&buf.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;
}
}
message->~Message();
read_pos += sizeof(Message);
if ((message->type & FLAG_MASK) != TYPE_NOTIFICATION) {
read_pos += sizeof(Variant) * message->args;
}
memdelete_arr(buf.buffer);
}
singleton = nullptr;
memdelete_arr(buffer);
}

7
core/object/message_queue.h
View file

@ -59,8 +59,11 @@ class MessageQueue {
};
};
uint8_t *buffer;
uint32_t buffer_end = 0;
struct Buffer {
uint8_t *buffer;
uint32_t buffer_end = 0;
};
Vector<Buffer> buffers;
uint32_t buffer_max_used = 0;
uint32_t buffer_size;