From cda773915e7404a873a02a61d46fd7aa2e10b8ef Mon Sep 17 00:00:00 2001 From: Yousef Khadadeh Date: Thu, 2 Oct 2025 19:10:58 +0300 Subject: [PATCH] Fixed message ordering issues --- include/Helpers/Networking/NetClient.cpp | 14 +++++++------- include/Helpers/Networking/NetClient.hpp | 10 +++++++--- include/Helpers/Networking/NetLibrary.cpp | 11 +++++++++-- include/Helpers/Networking/NetLibrary.hpp | 7 +++++-- include/Helpers/Networking/NetServer.cpp | 14 +++++++------- include/Helpers/Networking/NetServer.hpp | 14 +++++++++----- src/DrawComponents/DrawComponent.cpp | 4 ++-- src/DrawingProgram/DrawingProgram.cpp | 3 ++- 8 files changed, 48 insertions(+), 29 deletions(-) diff --git a/include/Helpers/Networking/NetClient.cpp b/include/Helpers/Networking/NetClient.cpp index 3ca51b6..9d88553 100644 --- a/include/Helpers/Networking/NetClient.cpp +++ b/include/Helpers/Networking/NetClient.cpp @@ -61,7 +61,7 @@ void NetClient::parse_received_messages() { lastMessageTime = std::chrono::steady_clock::now(); size_t messageOrderDisplacement = 0; - if(receivedMessage.channel == UNRELIABLE_COMMAND_CHANNEL || receivedMessage.channel == RELIABLE_COMMAND_CHANNEL) { + if(NetLibrary::is_ordered_channel(receivedMessage.channel)) { MessageOrder messageOrder = NetLibrary::get_message_order(receivedMessage.data); if(nextMessageOrderToExpect > messageOrder && receivedMessage.channel == UNRELIABLE_COMMAND_CHANNEL) { receiveQueue->messages.pop(); @@ -95,16 +95,16 @@ void NetClient::parse_received_messages() { void NetClient::send_queued_messages() { for(auto& [channelName, messageQueue] : messageQueues) { - bool addMessageOrder = channelName == UNRELIABLE_COMMAND_CHANNEL || channelName == RELIABLE_COMMAND_CHANNEL; + bool addMessageOrder = NetLibrary::is_ordered_channel(channelName); if(directConnectServer) { auto& c = directConnectServer->directConnectClientData; std::scoped_lock rMessLock(c->receivedMessagesMutex); while(!messageQueue.empty()) { std::string toInsert; if(addMessageOrder) - toInsert = NetLibrary::attach_message_order_and_update(nextMessageOrderToSend, messageQueue.front()->str()); + toInsert = NetLibrary::attach_message_order(messageQueue.front().order, messageQueue.front().ss->str()); else - toInsert = messageQueue.front()->str(); + toInsert = messageQueue.front().ss->str(); rtc::binary b((std::byte*)toInsert.c_str(), (std::byte*)(toInsert.c_str() + toInsert.length())); c->receivedMessages.emplace(channelName, b); messageQueue.pop(); @@ -120,14 +120,14 @@ void NetClient::send_queued_messages() { break; #endif if(addMessageOrder) - send_str_as_bytes(channel, NetLibrary::attach_message_order_and_update(nextMessageOrderToSend, messageQueue.front()->str())); + send_str_as_bytes(channel, NetLibrary::attach_message_order(messageQueue.front().order, messageQueue.front().ss->str())); else - send_str_as_bytes(channel, messageQueue.front()->str()); + send_str_as_bytes(channel, messageQueue.front().ss->str()); messageQueue.pop(); } #ifdef __EMSCRIPTEN__ if(channel->bufferedAmount() >= NetLibrary::MAX_BUFFERED_DATA_PER_CHANNEL && channelName == UNRELIABLE_COMMAND_CHANNEL) - messageQueue = std::queue>(); + messageQueue = std::queue(); #endif } } diff --git a/include/Helpers/Networking/NetClient.hpp b/include/Helpers/Networking/NetClient.hpp index 374d131..974f6a6 100644 --- a/include/Helpers/Networking/NetClient.hpp +++ b/include/Helpers/Networking/NetClient.hpp @@ -28,10 +28,10 @@ class NetClient : public std::enable_shared_from_this { std::vector> fragmentedMessage = fragment_message(ss->view(), NetLibrary::FRAGMENT_MESSAGE_STRIDE); auto& messageQueue = messageQueues[channel]; if(fragmentedMessage.empty()) - messageQueue.emplace(ss); + messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, nextMessageOrderToSend), ss); else if(channel != UNRELIABLE_COMMAND_CHANNEL) { // Just drop unreliable messages that are fragmented for(auto& ss2 : fragmentedMessage) - messageQueue.emplace(ss2); + messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, nextMessageOrderToSend), ss2); } } void update(); @@ -45,7 +45,11 @@ class NetClient : public std::enable_shared_from_this { void send_queued_messages(); void send_str_as_bytes(std::shared_ptr channel, const std::string& str); - std::unordered_map>> messageQueues; + struct OutgoingMessage { + MessageOrder order; + std::shared_ptr ss; + }; + std::unordered_map> messageQueues; struct ReceiveQueue { std::mutex mut; diff --git a/include/Helpers/Networking/NetLibrary.cpp b/include/Helpers/Networking/NetLibrary.cpp index e2a185c..6279659 100644 --- a/include/Helpers/Networking/NetLibrary.cpp +++ b/include/Helpers/Networking/NetLibrary.cpp @@ -341,13 +341,20 @@ void NetLibrary::channel_created_in_client(std::shared_ptr cha } } -std::string NetLibrary::attach_message_order_and_update(MessageOrder& order, const std::string& message) { +bool NetLibrary::is_ordered_channel(std::string_view channelName) { + return channelName == RELIABLE_COMMAND_CHANNEL || channelName == UNRELIABLE_COMMAND_CHANNEL; +} + +MessageOrder NetLibrary::calc_order_for_queued_message(std::string_view channelName, MessageOrder& mOrder) { + return is_ordered_channel(channelName) ? mOrder++ : 0; +} + +std::string NetLibrary::attach_message_order(MessageOrder order, const std::string& message) { union { MessageOrder o; char b[sizeof(MessageOrder)]; } u; u.o = (std::endian::native == std::endian::little) ? std::byteswap(order) : order; - order++; return std::string(u.b, sizeof(MessageOrder)) + message; } diff --git a/include/Helpers/Networking/NetLibrary.hpp b/include/Helpers/Networking/NetLibrary.hpp index 6569f6d..a44b635 100644 --- a/include/Helpers/Networking/NetLibrary.hpp +++ b/include/Helpers/Networking/NetLibrary.hpp @@ -33,15 +33,18 @@ class NetLibrary { static constexpr size_t GLOBALID_LEN = 20; private: static constexpr size_t FRAGMENT_MESSAGE_STRIDE = 4096; - static constexpr size_t MAX_BUFFERED_DATA_PER_CHANNEL = 65500; + static constexpr size_t MAX_BUFFERED_DATA_PER_CHANNEL = 64000; static constexpr std::chrono::seconds TIMEOUT_DURATION = std::chrono::seconds(30); + static bool is_ordered_channel(std::string_view channelName); + static MessageOrder calc_order_for_queued_message(std::string_view channelName, MessageOrder& mOrder); + static void finish_client_registration(std::shared_ptr client); static void assign_client_connection_to_server(const std::string& serverLocalID, const std::string& clientLocalID, std::shared_ptr connection); static void channel_created_in_client(std::shared_ptr channel); - static std::string attach_message_order_and_update(MessageOrder& order, const std::string& message); + static std::string attach_message_order(MessageOrder order, const std::string& message); static MessageOrder get_message_order(rtc::binary& message); struct PeerData { diff --git a/include/Helpers/Networking/NetServer.cpp b/include/Helpers/Networking/NetServer.cpp index ffddb0a..189b0d4 100644 --- a/include/Helpers/Networking/NetServer.cpp +++ b/include/Helpers/Networking/NetServer.cpp @@ -95,16 +95,16 @@ void NetServer::client_connected(std::shared_ptr connection void NetServer::ClientData::send_queued_messages(NetServer& server) { for(auto& [channelName, messageQueue] : messageQueues) { - bool addMessageOrder = channelName == UNRELIABLE_COMMAND_CHANNEL || channelName == RELIABLE_COMMAND_CHANNEL; + bool addMessageOrder = NetLibrary::is_ordered_channel(channelName); if(this == server.directConnectClientData.get()) { NetClient* c = server.directConnectClient; std::scoped_lock recvQueueLock(c->receiveQueue->mut); while(!messageQueue.empty()) { std::string toInsert; if(addMessageOrder) - toInsert = NetLibrary::attach_message_order_and_update(nextMessageOrderToSend, messageQueue.front()->str()); + toInsert = NetLibrary::attach_message_order(messageQueue.front().order, messageQueue.front().ss->str()); else - toInsert = messageQueue.front()->str(); + toInsert = messageQueue.front().ss->str(); rtc::binary b((std::byte*)toInsert.c_str(), (std::byte*)(toInsert.c_str() + toInsert.length())); c->receiveQueue->messages.emplace(channelName, b); messageQueue.pop(); @@ -120,14 +120,14 @@ void NetServer::ClientData::send_queued_messages(NetServer& server) { break; #endif if(addMessageOrder) - send_str_as_bytes(channel, NetLibrary::attach_message_order_and_update(nextMessageOrderToSend, messageQueue.front()->str())); + send_str_as_bytes(channel, NetLibrary::attach_message_order(messageQueue.front().order, messageQueue.front().ss->str())); else - send_str_as_bytes(channel, messageQueue.front()->str()); + send_str_as_bytes(channel, messageQueue.front().ss->str()); messageQueue.pop(); } #ifdef __EMSCRIPTEN__ if(channel->bufferedAmount() >= NetLibrary::MAX_BUFFERED_DATA_PER_CHANNEL && channelName == UNRELIABLE_COMMAND_CHANNEL) - messageQueue = std::queue>(); + messageQueue = std::queue(); #endif } } @@ -146,7 +146,7 @@ void NetServer::ClientData::parse_received_messages(NetServer& server) { lastMessageTime = std::chrono::steady_clock::now(); size_t messageOrderDisplacement = 0; - if(receivedMessage.channel == UNRELIABLE_COMMAND_CHANNEL || receivedMessage.channel == RELIABLE_COMMAND_CHANNEL) { + if(NetLibrary::is_ordered_channel(receivedMessage.channel)) { MessageOrder messageOrder = NetLibrary::get_message_order(receivedMessage.data); if(nextMessageOrderToExpect > messageOrder && receivedMessage.channel == UNRELIABLE_COMMAND_CHANNEL) { receivedMessages.pop(); diff --git a/include/Helpers/Networking/NetServer.hpp b/include/Helpers/Networking/NetServer.hpp index cf61488..9b2afe2 100644 --- a/include/Helpers/Networking/NetServer.hpp +++ b/include/Helpers/Networking/NetServer.hpp @@ -31,7 +31,11 @@ class NetServer : public std::enable_shared_from_this { }; std::queue receivedMessages; - std::unordered_map>> messageQueues; + struct OutgoingMessage { + MessageOrder order; + std::shared_ptr ss; + }; + std::unordered_map> messageQueues; uint64_t customID; @@ -62,10 +66,10 @@ class NetServer : public std::enable_shared_from_this { std::vector> fragmentedMessage = fragment_message(ss->view(), NetLibrary::FRAGMENT_MESSAGE_STRIDE); auto& messageQueue = client->messageQueues[channel]; if(fragmentedMessage.empty()) - messageQueue.emplace(ss); + messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, client->nextMessageOrderToSend), ss); else if(channel != UNRELIABLE_COMMAND_CHANNEL) { // Just drop unreliable messages that are fragmented for(auto& ss2 : fragmentedMessage) - messageQueue.emplace(ss2); + messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, client->nextMessageOrderToSend), ss2); } } template void send_items_to_all_clients(const std::string& channel, Args&&... items) { @@ -91,10 +95,10 @@ class NetServer : public std::enable_shared_from_this { if(client && clientChecker(client)) { auto& messageQueue = client->messageQueues[channel]; if(fragmentedMessage.empty()) - messageQueue.emplace(ss); + messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, client->nextMessageOrderToSend), ss); else if(channel != UNRELIABLE_COMMAND_CHANNEL) { // Just drop unreliable messages that are fragmented for(auto& ss2 : fragmentedMessage) - messageQueue.emplace(ss2); + messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, client->nextMessageOrderToSend), ss2); } } } diff --git a/src/DrawComponents/DrawComponent.cpp b/src/DrawComponents/DrawComponent.cpp index d7fd42d..e0b4e57 100644 --- a/src/DrawComponents/DrawComponent.cpp +++ b/src/DrawComponents/DrawComponent.cpp @@ -74,9 +74,9 @@ void DrawComponent::server_send_erase_set(MainServer& server, const std::unorder void DrawComponent::server_send_update(MainServer& server, bool final) { if(final) - server.netServer->send_items_to_all_clients(RELIABLE_COMMAND_CHANNEL, CLIENT_UPDATE_COMPONENT, id, *this); + server.netServer->send_items_to_all_clients(RELIABLE_COMMAND_CHANNEL, CLIENT_UPDATE_COMPONENT, final, id, *this); else - server.netServer->send_items_to_all_clients(UNRELIABLE_COMMAND_CHANNEL, CLIENT_UPDATE_COMPONENT, id, *this); + server.netServer->send_items_to_all_clients(UNRELIABLE_COMMAND_CHANNEL, CLIENT_UPDATE_COMPONENT, final, id, *this); } void DrawComponent::server_send_transform_many(MainServer& server, const std::vector>& transforms) { diff --git a/src/DrawingProgram/DrawingProgram.cpp b/src/DrawingProgram/DrawingProgram.cpp index 1b7e744..176ebcf 100644 --- a/src/DrawingProgram/DrawingProgram.cpp +++ b/src/DrawingProgram/DrawingProgram.cpp @@ -62,7 +62,8 @@ DrawingProgram::DrawingProgram(World& initWorld): void DrawingProgram::init_client_callbacks() { world.con.client_add_recv_callback(CLIENT_UPDATE_COMPONENT, [&](cereal::PortableBinaryInputArchive& message) { ServerClientID id; - message(id); + bool isFinal; // Not actually used, but sent for debugging purposes + message(isFinal, id); auto objPtr = components.get_item_by_id(id); if(!objPtr) return;