From 2aab3e9e60f07d146dfe490eceb7dba31029c4e3 Mon Sep 17 00:00:00 2001 From: Yousef Khadadeh Date: Fri, 10 Oct 2025 10:34:12 +0300 Subject: [PATCH] Prevent head of line blocking and track server resource download progress --- include/Helpers/Networking/NetClient.cpp | 9 ++-- include/Helpers/Networking/NetClient.hpp | 24 ++++++---- include/Helpers/Networking/NetLibrary.hpp | 9 +++- include/Helpers/Networking/NetServer.cpp | 16 ++++++- include/Helpers/Networking/NetServer.hpp | 56 +++++++++++++++-------- src/ConnectionManager.cpp | 19 ++++++-- src/ConnectionManager.hpp | 5 +- src/DrawingProgram/DrawingProgram.cpp | 4 +- src/ResourceManager.cpp | 50 ++++++++++++++------ src/ResourceManager.hpp | 7 +-- src/Server/ClientData.hpp | 3 +- src/Server/CommandList.hpp | 4 +- src/Server/MainServer.cpp | 55 +++++++++++++--------- src/Server/MainServer.hpp | 1 + 14 files changed, 179 insertions(+), 83 deletions(-) diff --git a/include/Helpers/Networking/NetClient.cpp b/include/Helpers/Networking/NetClient.cpp index 5ad9365..a6f9abe 100644 --- a/include/Helpers/Networking/NetClient.cpp +++ b/include/Helpers/Networking/NetClient.cpp @@ -93,8 +93,11 @@ void NetClient::parse_received_messages() { } } -std::pair NetClient::get_progress_into_fragmented_message(const std::string& channel) { - auto& m = pfm[channel]; +NetLibrary::DownloadProgress NetClient::get_progress_into_fragmented_message(const std::string& channel) const { + auto it = pfm.find(channel); + if(it == pfm.end()) + return {0, 0}; + auto& m = it->second; if(m.partialFragmentMessage.empty()) return {0, 0}; return {m.partialFragmentMessageLoc, m.partialFragmentMessage.size()}; @@ -142,7 +145,7 @@ void NetClient::send_queued_messages() { } } -bool NetClient::is_disconnected() { +bool NetClient::is_disconnected() const { return isDisconnected; } diff --git a/include/Helpers/Networking/NetClient.hpp b/include/Helpers/Networking/NetClient.hpp index 776a827..112dc51 100644 --- a/include/Helpers/Networking/NetClient.hpp +++ b/include/Helpers/Networking/NetClient.hpp @@ -25,19 +25,27 @@ class NetClient : public std::enable_shared_from_this { cereal::PortableBinaryOutputArchive m(*ss); (m(items), ...); } - std::vector> fragmentedMessage = fragment_message(ss->view(), NetLibrary::FRAGMENT_MESSAGE_STRIDE); + auto& messageQueue = messageQueues[channel]; - if(fragmentedMessage.empty()) - 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(NetLibrary::calc_order_for_queued_message(channel, nextMessageOrderToSend), ss2); + + if(channel == UNRELIABLE_COMMAND_CHANNEL) { + if(ss->view().length() <= NetLibrary::MAX_UNRELIABLE_MESSAGE_SIZE) // Drop unreliable messages that are too big + messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, nextMessageOrderToSend), ss); + } + else { + std::vector> fragmentedMessage = fragment_message(ss->view(), NetLibrary::FRAGMENT_MESSAGE_STRIDE); + if(fragmentedMessage.empty()) + messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, nextMessageOrderToSend), ss); + else { + for(auto& ss2 : fragmentedMessage) + messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, nextMessageOrderToSend), ss2); + } } } void update(); void add_recv_callback(MessageCommandType commandID, const NetClientRecvCallback& callback); - bool is_disconnected(); - std::pair get_progress_into_fragmented_message(const std::string& channel); + bool is_disconnected() const; + NetLibrary::DownloadProgress get_progress_into_fragmented_message(const std::string& channel) const; private: std::string localID; diff --git a/include/Helpers/Networking/NetLibrary.hpp b/include/Helpers/Networking/NetLibrary.hpp index a44b635..127248e 100644 --- a/include/Helpers/Networking/NetLibrary.hpp +++ b/include/Helpers/Networking/NetLibrary.hpp @@ -31,8 +31,15 @@ class NetLibrary { static constexpr size_t LOCALID_LEN = 5; static constexpr size_t GLOBALID_LEN = 20; + + struct DownloadProgress { + uint64_t downloadedBytes; + uint64_t totalBytes; + }; + private: - static constexpr size_t FRAGMENT_MESSAGE_STRIDE = 4096; + static constexpr size_t FRAGMENT_MESSAGE_STRIDE = 512; + static constexpr size_t MAX_UNRELIABLE_MESSAGE_SIZE = 4096; static constexpr size_t MAX_BUFFERED_DATA_PER_CHANNEL = 64000; static constexpr std::chrono::seconds TIMEOUT_DURATION = std::chrono::seconds(30); diff --git a/include/Helpers/Networking/NetServer.cpp b/include/Helpers/Networking/NetServer.cpp index 189b0d4..864f48b 100644 --- a/include/Helpers/Networking/NetServer.cpp +++ b/include/Helpers/Networking/NetServer.cpp @@ -41,7 +41,7 @@ void NetServer::update() { clientsToAdd.clear(); } -bool NetServer::is_disconnected() { +bool NetServer::is_disconnected() const { return isDisconnected; } @@ -178,6 +178,16 @@ void NetServer::ClientData::parse_received_messages(NetServer& server) { } } +NetLibrary::DownloadProgress NetServer::ClientData::get_progress_into_fragmented_message(const std::string& channel) const { + auto it = pfm.find(channel); + if(it == pfm.end()) + return {0, 0}; + auto& m = it->second; + if(m.partialFragmentMessage.empty()) + return {0, 0}; + return {m.partialFragmentMessageLoc, m.partialFragmentMessage.size()}; +} + void NetServer::add_recv_callback(MessageCommandType commandID, const NetServerRecvCallback& callback) { recvCallbacks[commandID] = callback; } @@ -186,6 +196,10 @@ void NetServer::add_disconnect_callback(const NetServerDisconnectCallback& callb disconnectCallback = callback; } +const std::vector> NetServer::get_client_list() const { + return clients; +} + NetServer::~NetServer() { if(directConnectClient) directConnectClient->isDisconnected = true; diff --git a/include/Helpers/Networking/NetServer.hpp b/include/Helpers/Networking/NetServer.hpp index 9b2afe2..5b3c5c4 100644 --- a/include/Helpers/Networking/NetServer.hpp +++ b/include/Helpers/Networking/NetServer.hpp @@ -42,6 +42,7 @@ class NetServer : public std::enable_shared_from_this { void send_queued_messages(NetServer& server); void parse_received_messages(NetServer& server); void send_str_as_bytes(std::shared_ptr channel, const std::string& str); + NetLibrary::DownloadProgress get_progress_into_fragmented_message(const std::string& channel) const; std::atomic setToDisconnect = false; std::chrono::steady_clock::time_point lastMessageTime; @@ -63,13 +64,19 @@ class NetServer : public std::enable_shared_from_this { (m(items), ...); } - std::vector> fragmentedMessage = fragment_message(ss->view(), NetLibrary::FRAGMENT_MESSAGE_STRIDE); auto& messageQueue = client->messageQueues[channel]; - if(fragmentedMessage.empty()) - 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(NetLibrary::calc_order_for_queued_message(channel, client->nextMessageOrderToSend), ss2); + if(channel == UNRELIABLE_COMMAND_CHANNEL) { + if(ss->view().length() <= NetLibrary::MAX_UNRELIABLE_MESSAGE_SIZE) // Drop unreliable messages that are too big + messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, client->nextMessageOrderToSend), ss); + } + else { + std::vector> fragmentedMessage = fragment_message(ss->view(), NetLibrary::FRAGMENT_MESSAGE_STRIDE); + if(fragmentedMessage.empty()) + messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, client->nextMessageOrderToSend), ss); + else { + for(auto& ss2 : fragmentedMessage) + 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) { @@ -89,25 +96,40 @@ class NetServer : public std::enable_shared_from_this { (m(items), ...); } - std::vector> fragmentedMessage = fragment_message(ss->view(), NetLibrary::FRAGMENT_MESSAGE_STRIDE); + if(channel == UNRELIABLE_COMMAND_CHANNEL) { + if(ss->view().length() <= NetLibrary::MAX_UNRELIABLE_MESSAGE_SIZE) { // Drop unreliable messages that are too big + for(auto& client : clients) { + if(client && clientChecker(client)) { + auto& messageQueue = client->messageQueues[channel]; + messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, client->nextMessageOrderToSend), ss); + } + } + } + } + else { + std::vector> fragmentedMessage = fragment_message(ss->view(), NetLibrary::FRAGMENT_MESSAGE_STRIDE); - for(auto& client : clients) { - if(client && clientChecker(client)) { - auto& messageQueue = client->messageQueues[channel]; - if(fragmentedMessage.empty()) - 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(NetLibrary::calc_order_for_queued_message(channel, client->nextMessageOrderToSend), ss2); + for(auto& client : clients) { + if(client && clientChecker(client)) { + auto& messageQueue = client->messageQueues[channel]; + if(fragmentedMessage.empty()) + messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, client->nextMessageOrderToSend), ss); + else { + for(auto& ss2 : fragmentedMessage) + messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, client->nextMessageOrderToSend), ss2); + } } } } } - bool is_disconnected(); + bool is_disconnected() const; void add_recv_callback(MessageCommandType commandID, const NetServerRecvCallback& callback); void add_disconnect_callback(const NetServerDisconnectCallback& callback); + + const std::vector> get_client_list() const; + private: void parse_received_messages(); void send_queued_messages(); @@ -117,8 +139,6 @@ class NetServer : public std::enable_shared_from_this { std::string localID; - std::mutex clientList; - std::vector> clients; // Adding an add list, because i dont want to block the entire clients vector with a mutex diff --git a/src/ConnectionManager.cpp b/src/ConnectionManager.cpp index 7095d45..185ce00 100644 --- a/src/ConnectionManager.cpp +++ b/src/ConnectionManager.cpp @@ -40,9 +40,18 @@ bool ConnectionManager::is_host_disconnected() { return localServer && localServer->netServer->is_disconnected(); } -std::pair ConnectionManager::client_get_resource_retrieval_progress() { - if(client) - return client->get_progress_into_fragmented_message(RESOURCE_COMMAND_CHANNEL); - else - return {0, 0}; +bool ConnectionManager::host_exists() { + return localServer != nullptr; +} + +bool ConnectionManager::client_exists() { + return client != nullptr; +} + +std::unordered_map ConnectionManager::server_get_resource_retrieval_progress() { + return localServer->resource_download_progress(); +} + +NetLibrary::DownloadProgress ConnectionManager::client_get_resource_retrieval_progress() { + return client->get_progress_into_fragmented_message(RESOURCE_COMMAND_CHANNEL); } diff --git a/src/ConnectionManager.hpp b/src/ConnectionManager.hpp index e94a1c7..816c81a 100644 --- a/src/ConnectionManager.hpp +++ b/src/ConnectionManager.hpp @@ -14,7 +14,10 @@ class ConnectionManager { } bool is_client_disconnected(); bool is_host_disconnected(); - std::pair client_get_resource_retrieval_progress(); + bool host_exists(); + bool client_exists(); + std::unordered_map server_get_resource_retrieval_progress(); + NetLibrary::DownloadProgress client_get_resource_retrieval_progress(); private: std::shared_ptr client; std::unique_ptr localServer; diff --git a/src/DrawingProgram/DrawingProgram.cpp b/src/DrawingProgram/DrawingProgram.cpp index 5c7dda5..5ec6879 100644 --- a/src/DrawingProgram/DrawingProgram.cpp +++ b/src/DrawingProgram/DrawingProgram.cpp @@ -725,8 +725,8 @@ void DrawingProgram::draw(SkCanvas* canvas, const DrawData& drawData) { for(auto& c : updateableComponents) { if(c->get_type() == DRAWCOMPONENT_IMAGE) { auto img = std::static_pointer_cast(c); - if(img->d.imageID == drawData.main->world->rMan.get_resource_being_retrieved() && img->d.imageID != ServerClientID{0, 0}) - img->draw_download_progress_bar(canvas, drawData, drawData.main->world->rMan.get_resource_retrieval_progress()); + float progress = drawData.main->world->rMan.get_resource_retrieval_progress(img->d.imageID); + img->draw_download_progress_bar(canvas, drawData, progress); } } } diff --git a/src/ResourceManager.cpp b/src/ResourceManager.cpp index 84d8ee5..a0dea7b 100644 --- a/src/ResourceManager.cpp +++ b/src/ResourceManager.cpp @@ -19,28 +19,48 @@ ResourceManager::ResourceManager(World& initWorld): void ResourceManager::init_client_callbacks() { world.con.client_add_recv_callback(CLIENT_NEW_RESOURCE_ID, [&](cereal::PortableBinaryInputArchive& message) { - message(resourceBeingRetrieved); + message(resourceBeingRetrievedClient); }); world.con.client_add_recv_callback(CLIENT_NEW_RESOURCE_DATA, [&](cereal::PortableBinaryInputArchive& message) { - message(resources[resourceBeingRetrieved]); - Logger::get().log("INFO", "Received new resource with id: " + std::to_string(resourceBeingRetrieved.first) + " " + std::to_string(resourceBeingRetrieved.second) + " of size " + std::to_string(resources[resourceBeingRetrieved].data->size())); - resourceBeingRetrieved = {0, 0}; + message(resources[resourceBeingRetrievedClient]); + Logger::get().log("INFO", "Received new resource with id: " + std::to_string(resourceBeingRetrievedClient.first) + " " + std::to_string(resourceBeingRetrievedClient.second) + " of size " + std::to_string(resources[resourceBeingRetrievedClient].data->size())); + resourceBeingRetrievedClient = {0, 0}; }); } -ServerClientID ResourceManager::get_resource_being_retrieved() { - return resourceBeingRetrieved; -} - -float ResourceManager::get_resource_retrieval_progress() { - std::pair progressBytes = world.con.client_get_resource_retrieval_progress(); - if(progressBytes.second == 0) +float ResourceManager::get_resource_retrieval_progress(const ServerClientID& id) { + if(id == ServerClientID{0, 0}) return 0.0f; - else - return static_cast(progressBytes.first) / static_cast(progressBytes.second); + if(world.con.host_exists() && !world.con.is_host_disconnected()) { + auto it = serverDownloadProgress.find(id); + if(it == serverDownloadProgress.end()) + return 0.0f; + else { + auto& progressBytes = it->second; + if(progressBytes.totalBytes == 0) + return 0.0f; + return static_cast(progressBytes.downloadedBytes) / static_cast(progressBytes.totalBytes); + } + } + else if(world.con.client_exists() && !world.con.is_client_disconnected()) { + if(id != resourceBeingRetrievedClient) + return 0.0f; + else { + NetLibrary::DownloadProgress progressBytes = world.con.client_get_resource_retrieval_progress(); + if(progressBytes.totalBytes == 0) + return 0.0f; + return static_cast(progressBytes.downloadedBytes) / static_cast(progressBytes.totalBytes); + } + } + return 0.0f; } void ResourceManager::update() { + if(!world.con.host_exists() || world.con.is_host_disconnected()) + serverDownloadProgress.clear(); + else + serverDownloadProgress = world.con.server_get_resource_retrieval_progress(); + for(auto& [k, v] : displays) v->update(world); } @@ -92,8 +112,8 @@ ServerClientID ResourceManager::add_resource(const ResourceData& resource) { } ServerClientID newID = world.get_new_id(); resources[newID] = resource; - world.con.client_send_items_to_server(RELIABLE_COMMAND_CHANNEL, SERVER_RESOURCE_INIT, newID); - world.con.client_send_items_to_server(RESOURCE_COMMAND_CHANNEL, SERVER_RESOURCE_DATA, resource); + world.con.client_send_items_to_server(RESOURCE_COMMAND_CHANNEL, SERVER_NEW_RESOURCE_ID, newID); + world.con.client_send_items_to_server(RESOURCE_COMMAND_CHANNEL, SERVER_NEW_RESOURCE_DATA, resource); return newID; } diff --git a/src/ResourceManager.hpp b/src/ResourceManager.hpp index 6f00531..fb7444c 100644 --- a/src/ResourceManager.hpp +++ b/src/ResourceManager.hpp @@ -7,6 +7,7 @@ #include #include #include +#include class World; @@ -44,12 +45,12 @@ class ResourceManager { std::optional get_resource(const ServerClientID& id) const; ClientPortionID get_max_id(ServerPortionID serverID) const; const std::unordered_map& resource_list(); - ServerClientID get_resource_being_retrieved(); - float get_resource_retrieval_progress(); + float get_resource_retrieval_progress(const ServerClientID& id); World& world; private: - ServerClientID resourceBeingRetrieved = {0, 0}; + std::unordered_map serverDownloadProgress; + ServerClientID resourceBeingRetrievedClient = {0, 0}; std::unordered_map resources; std::unordered_map> displays; diff --git a/src/Server/ClientData.hpp b/src/Server/ClientData.hpp index 7285d34..7c5d38f 100644 --- a/src/Server/ClientData.hpp +++ b/src/Server/ClientData.hpp @@ -20,8 +20,7 @@ class ClientData { uint64_t canvasScale = 0; - std::queue pendingResourceData; - std::queue pendingResourceID; + ServerClientID pendingResourceID = {0, 0}; template void serialize(Archive& a) { a(camCoords, windowSize, cursorPos, cursorColor, displayName); diff --git a/src/Server/CommandList.hpp b/src/Server/CommandList.hpp index 069aafb..ac42ed5 100644 --- a/src/Server/CommandList.hpp +++ b/src/Server/CommandList.hpp @@ -13,8 +13,8 @@ enum ServerCommands : MessageCommandType { SERVER_ERASE_SINGLE_COMPONENT, SERVER_ERASE_MANY_COMPONENTS, SERVER_UPDATE_COMPONENT, - SERVER_RESOURCE_INIT, - SERVER_RESOURCE_DATA, + SERVER_NEW_RESOURCE_ID, + SERVER_NEW_RESOURCE_DATA, SERVER_CHAT_MESSAGE, SERVER_NEW_BOOKMARK, SERVER_REMOVE_BOOKMARK, diff --git a/src/Server/MainServer.cpp b/src/Server/MainServer.cpp index 59f3b94..051141e 100644 --- a/src/Server/MainServer.cpp +++ b/src/Server/MainServer.cpp @@ -185,33 +185,22 @@ MainServer::MainServer(World& initWorld, const std::string& serverLocalID): comp->server_send_update(*this, isFinalUpdate); } }); - netServer->add_recv_callback(SERVER_RESOURCE_INIT, [&](std::shared_ptr client, cereal::PortableBinaryInputArchive& message) { + netServer->add_recv_callback(SERVER_NEW_RESOURCE_ID, [&](std::shared_ptr client, cereal::PortableBinaryInputArchive& message) { auto& c = clients[client->customID]; - ServerClientID newID; - message(newID); - if(c.pendingResourceData.empty()) - c.pendingResourceID.emplace(newID); - else { - netServer->send_items_to_all_clients_except(client, RESOURCE_COMMAND_CHANNEL, CLIENT_NEW_RESOURCE_ID, newID); - netServer->send_items_to_all_clients_except(client, RESOURCE_COMMAND_CHANNEL, CLIENT_NEW_RESOURCE_DATA, c.pendingResourceData.front()); - data.resources.emplace(newID, c.pendingResourceData.front()); - c.pendingResourceData.pop(); - } + message(c.pendingResourceID); + if(c.pendingResourceID == ServerClientID{0, 0}) + throw std::runtime_error("[MainServer::MainServer] Received a null resource ID"); }); - netServer->add_recv_callback(SERVER_RESOURCE_DATA, [&](std::shared_ptr client, cereal::PortableBinaryInputArchive& message) { + netServer->add_recv_callback(SERVER_NEW_RESOURCE_DATA, [&](std::shared_ptr client, cereal::PortableBinaryInputArchive& message) { ResourceData resourceData; message(resourceData); auto& c = clients[client->customID]; - if(c.pendingResourceID.empty()) { - c.pendingResourceData.emplace(); - c.pendingResourceData.back() = std::move(resourceData); - } - else { - netServer->send_items_to_all_clients_except(client, RESOURCE_COMMAND_CHANNEL, CLIENT_NEW_RESOURCE_ID, c.pendingResourceID.front()); - netServer->send_items_to_all_clients_except(client, RESOURCE_COMMAND_CHANNEL, CLIENT_NEW_RESOURCE_DATA, resourceData); - data.resources.emplace(c.pendingResourceID.front(), resourceData); - c.pendingResourceID.pop(); - } + if(c.pendingResourceID == ServerClientID{0, 0}) + throw std::runtime_error("[MainServer::MainServer] Received data for a null resource ID"); + netServer->send_items_to_all_clients_except(client, RESOURCE_COMMAND_CHANNEL, CLIENT_NEW_RESOURCE_ID, c.pendingResourceID); + netServer->send_items_to_all_clients_except(client, RESOURCE_COMMAND_CHANNEL, CLIENT_NEW_RESOURCE_DATA, resourceData); + data.resources.emplace(c.pendingResourceID, resourceData); + c.pendingResourceID = ServerClientID{0, 0}; }); netServer->add_recv_callback(SERVER_CHAT_MESSAGE, [&](std::shared_ptr client, cereal::PortableBinaryInputArchive& message) { std::string chatMessage; @@ -326,6 +315,28 @@ void MainServer::update() { } } +std::unordered_map MainServer::resource_download_progress() { + std::unordered_map toRet; + + if(!netServer) + return toRet; + + for(auto& [serverID, c] : clients) { + if(c.pendingResourceID != ServerClientID{0, 0}) { + const std::vector>& clientListInNetServer = netServer->get_client_list(); + auto netClientIt = std::find_if(clientListInNetServer.begin(), clientListInNetServer.end(), [&serverID](auto& nC) { + return nC->customID == serverID; + }); + if(netClientIt != clientListInNetServer.end()) { + auto& netClient = *netClientIt; + toRet.emplace(c.pendingResourceID, netClient->get_progress_into_fragmented_message(RESOURCE_COMMAND_CHANNEL)); + } + } + } + + return toRet; +} + MainServer::~MainServer() { //netServer->close_server(); } diff --git a/src/Server/MainServer.hpp b/src/Server/MainServer.hpp index fc9a80e..3b2d704 100644 --- a/src/Server/MainServer.hpp +++ b/src/Server/MainServer.hpp @@ -12,6 +12,7 @@ class MainServer { ~MainServer(); std::shared_ptr netServer; std::unordered_map clients; + std::unordered_map resource_download_progress(); private: World& world; void ensure_display_name_unique(std::string& displayName);