copilot/infinipaint
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions 1

Fixed message ordering issues

Browse source
This commit is contained in:
Yousef Khadadeh 2025-10-02 19:10:58 +03:00
parent 61e0b2ff24
commit cda773915e
8 changed files with 48 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

14
include/Helpers/Networking/NetClient.cpp
View file

@ -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<std::shared_ptr<std::stringstream>>();
messageQueue = std::queue<OutgoingMessage>();
#endif
}
}

10
include/Helpers/Networking/NetClient.hpp
View file

@ -28,10 +28,10 @@ class NetClient : public std::enable_shared_from_this<NetClient> {
std::vector<std::shared_ptr<std::stringstream>> 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<NetClient> {
void send_queued_messages();
void send_str_as_bytes(std::shared_ptr<rtc::DataChannel> channel, const std::string& str);
std::unordered_map<std::string, std::queue<std::shared_ptr<std::stringstream>>> messageQueues;
struct OutgoingMessage {
MessageOrder order;
std::shared_ptr<std::stringstream> ss;
};
std::unordered_map<std::string, std::queue<OutgoingMessage>> messageQueues;
struct ReceiveQueue {
std::mutex mut;

11
include/Helpers/Networking/NetLibrary.cpp
View file

@ -341,13 +341,20 @@ void NetLibrary::channel_created_in_client(std::shared_ptr<rtc::DataChannel> 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;
}

7
include/Helpers/Networking/NetLibrary.hpp
View file

@ -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<NetClient> client);
static void assign_client_connection_to_server(const std::string& serverLocalID, const std::string& clientLocalID, std::shared_ptr<rtc::PeerConnection> connection);
static void channel_created_in_client(std::shared_ptr<rtc::DataChannel> 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 {

14
include/Helpers/Networking/NetServer.cpp
View file

@ -95,16 +95,16 @@ void NetServer::client_connected(std::shared_ptr<rtc::PeerConnection> 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<std::shared_ptr<std::stringstream>>();
messageQueue = std::queue<OutgoingMessage>();
#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();

14
include/Helpers/Networking/NetServer.hpp
View file

@ -31,7 +31,11 @@ class NetServer : public std::enable_shared_from_this<NetServer> {
};
std::queue<ReceivedMessage> receivedMessages;
std::unordered_map<std::string, std::queue<std::shared_ptr<std::stringstream>>> messageQueues;
struct OutgoingMessage {
MessageOrder order;
std::shared_ptr<std::stringstream> ss;
};
std::unordered_map<std::string, std::queue<OutgoingMessage>> messageQueues;
uint64_t customID;
@ -62,10 +66,10 @@ class NetServer : public std::enable_shared_from_this<NetServer> {
std::vector<std::shared_ptr<std::stringstream>> 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 <typename... Args> void send_items_to_all_clients(const std::string& channel, Args&&... items) {
@ -91,10 +95,10 @@ class NetServer : public std::enable_shared_from_this<NetServer> {
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);
}
}
}

4
src/DrawComponents/DrawComponent.cpp
View file

@ -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<std::pair<ServerClientID, CoordSpaceHelper>>& transforms) {

3
src/DrawingProgram/DrawingProgram.cpp
View file

@ -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;