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

Prevent head of line blocking and track server resource download progress

Browse source
This commit is contained in:
Yousef Khadadeh 2025-10-10 10:34:12 +03:00
parent 344ef4093a
commit 2aab3e9e60
14 changed files with 179 additions and 83 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -93,8 +93,11 @@ void NetClient::parse_received_messages() {
}
}
std::pair<uint64_t, uint64_t> 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;
}

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

@ -25,19 +25,27 @@ class NetClient : public std::enable_shared_from_this<NetClient> {
cereal::PortableBinaryOutputArchive m(*ss);
(m(items), ...);
}
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(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<std::shared_ptr<std::stringstream>> 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<uint64_t, uint64_t> 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;

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

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

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

@ -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<std::shared_ptr<NetServer::ClientData>> NetServer::get_client_list() const {
return clients;
}
NetServer::~NetServer() {
if(directConnectClient)
directConnectClient->isDisconnected = true;

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

@ -42,6 +42,7 @@ class NetServer : public std::enable_shared_from_this<NetServer> {
void send_queued_messages(NetServer& server);
void parse_received_messages(NetServer& server);
void send_str_as_bytes(std::shared_ptr<rtc::DataChannel> channel, const std::string& str);
NetLibrary::DownloadProgress get_progress_into_fragmented_message(const std::string& channel) const;
std::atomic<bool> setToDisconnect = false;
std::chrono::steady_clock::time_point lastMessageTime;
@ -63,13 +64,19 @@ class NetServer : public std::enable_shared_from_this<NetServer> {
(m(items), ...);
}
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(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<std::shared_ptr<std::stringstream>> 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 <typename... Args> void send_items_to_all_clients(const std::string& channel, Args&&... items) {
@ -89,25 +96,40 @@ class NetServer : public std::enable_shared_from_this<NetServer> {
(m(items), ...);
}
std::vector<std::shared_ptr<std::stringstream>> 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<std::shared_ptr<std::stringstream>> 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<std::shared_ptr<ClientData>> 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<NetServer> {
std::string localID;
std::mutex clientList;
std::vector<std::shared_ptr<ClientData>> clients;
// Adding an add list, because i dont want to block the entire clients vector with a mutex

19
src/ConnectionManager.cpp
View file

@ -40,9 +40,18 @@ bool ConnectionManager::is_host_disconnected() {
return localServer && localServer->netServer->is_disconnected();
}
std::pair<uint64_t, uint64_t> 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<ServerClientID, NetLibrary::DownloadProgress> 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);
}

5
src/ConnectionManager.hpp
View file

@ -14,7 +14,10 @@ class ConnectionManager {
}
bool is_client_disconnected();
bool is_host_disconnected();
std::pair<uint64_t, uint64_t> client_get_resource_retrieval_progress();
bool host_exists();
bool client_exists();
std::unordered_map<ServerClientID, NetLibrary::DownloadProgress> server_get_resource_retrieval_progress();
NetLibrary::DownloadProgress client_get_resource_retrieval_progress();
private:
std::shared_ptr<NetClient> client;
std::unique_ptr<MainServer> localServer;

4
src/DrawingProgram/DrawingProgram.cpp
View file

@ -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<DrawImage>(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);
}
}
}

50
src/ResourceManager.cpp
View file

@ -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<uint64_t, uint64_t> 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<float>(progressBytes.first) / static_cast<float>(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<float>(progressBytes.downloadedBytes) / static_cast<float>(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<float>(progressBytes.downloadedBytes) / static_cast<float>(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;
}

7
src/ResourceManager.hpp
View file

@ -7,6 +7,7 @@
#include <cereal/types/string.hpp>
#include <cereal/types/unordered_map.hpp>
#include <queue>
#include <Helpers/Networking/NetLibrary.hpp>
class World;
@ -44,12 +45,12 @@ class ResourceManager {
std::optional<ResourceData> get_resource(const ServerClientID& id) const;
ClientPortionID get_max_id(ServerPortionID serverID) const;
const std::unordered_map<ServerClientID, ResourceData>& 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<ServerClientID, NetLibrary::DownloadProgress> serverDownloadProgress;
ServerClientID resourceBeingRetrievedClient = {0, 0};
std::unordered_map<ServerClientID, ResourceData> resources;
std::unordered_map<ServerClientID, std::shared_ptr<ResourceDisplay>> displays;

3
src/Server/ClientData.hpp
View file

@ -20,8 +20,7 @@ class ClientData {
uint64_t canvasScale = 0;
std::queue<ResourceData> pendingResourceData;
std::queue<ServerClientID> pendingResourceID;
ServerClientID pendingResourceID = {0, 0};
template <typename Archive> void serialize(Archive& a) {
a(camCoords, windowSize, cursorPos, cursorColor, displayName);

4
src/Server/CommandList.hpp
View file

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

55
src/Server/MainServer.cpp
View file

@ -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<NetServer::ClientData> client, cereal::PortableBinaryInputArchive& message) {
netServer->add_recv_callback(SERVER_NEW_RESOURCE_ID, [&](std::shared_ptr<NetServer::ClientData> 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<NetServer::ClientData> client, cereal::PortableBinaryInputArchive& message) {
netServer->add_recv_callback(SERVER_NEW_RESOURCE_DATA, [&](std::shared_ptr<NetServer::ClientData> 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<NetServer::ClientData> client, cereal::PortableBinaryInputArchive& message) {
std::string chatMessage;
@ -326,6 +315,28 @@ void MainServer::update() {
}
}
std::unordered_map<ServerClientID, NetLibrary::DownloadProgress> MainServer::resource_download_progress() {
std::unordered_map<ServerClientID, NetLibrary::DownloadProgress> toRet;
if(!netServer)
return toRet;
for(auto& [serverID, c] : clients) {
if(c.pendingResourceID != ServerClientID{0, 0}) {
const std::vector<std::shared_ptr<NetServer::ClientData>>& 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();
}

1
src/Server/MainServer.hpp
View file

@ -12,6 +12,7 @@ class MainServer {
~MainServer();
std::shared_ptr<NetServer> netServer;
std::unordered_map<ServerPortionID, ClientData> clients;
std::unordered_map<ServerClientID, NetLibrary::DownloadProgress> resource_download_progress();
private:
World& world;
void ensure_display_name_unique(std::string& displayName);