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

Compress large messages with zstd

Browse source
This commit is contained in:
Yousef Khadadeh 2025-07-28 17:07:39 +03:00
parent eb9db31d31
commit 1300be0227
10 changed files with 81 additions and 60 deletions
Download patch file Download diff file Expand all files Collapse all files

57
include/Helpers/Networking/ByteStream.cpp
View file

@ -2,6 +2,7 @@
#include <cereal/archives/portable_binary.hpp>
#include <cereal/types/string.hpp>
#include "NetLibrary.hpp"
#include <zstd.h>
ByteMemBuf::ByteMemBuf(char* p, size_t l) {
setg(p, p, p + l);
@ -15,16 +16,30 @@ ByteMemStream::ByteMemStream(char* p, size_t l):
rdbuf(&buffer);
}
std::vector<std::shared_ptr<std::stringstream>> fragment_message(std::string_view v, uint64_t stride) {
if(v.length() <= stride)
std::vector<std::shared_ptr<std::stringstream>> fragment_message(std::string_view uncompressedView, uint64_t stride) {
if(uncompressedView.length() <= stride)
return {};
std::vector<char> compressedData(ZSTD_compressBound(uncompressedView.size()));
size_t trueCompressedSize = ZSTD_compress(compressedData.data(), compressedData.size(), uncompressedView.data(), uncompressedView.size(), ZSTD_CLEVEL_DEFAULT);
std::string_view v(compressedData.data(), trueCompressedSize);
uint64_t nextByte = 0;
std::vector<std::shared_ptr<std::stringstream>> toRet;
toRet.emplace_back(std::make_shared<std::stringstream>());
size_t messFirstStride = std::min(stride, v.size());
{
cereal::PortableBinaryOutputArchive m(*toRet.back());
m((MessageCommandType)0, (uint64_t)v.length(), cereal::binary_data(v.data() + nextByte, stride));
m((MessageCommandType)0, (uint64_t)v.length(), cereal::binary_data(v.data() + nextByte, messFirstStride));
}
if(messFirstStride == v.size())
return toRet;
nextByte += stride;
for(;;) {
if((v.length() - nextByte) <= stride) {
@ -44,3 +59,39 @@ std::vector<std::shared_ptr<std::stringstream>> fragment_message(std::string_vie
}
return toRet;
}
void uncompress_and_run_func(PartialFragmentMessage& spfm, std::function<void(cereal::PortableBinaryInputArchive& completeArchive)> runAfterDecoded) {
std::vector<char> uncompressedData(ZSTD_getFrameContentSize(spfm.partialFragmentMessage.data(), spfm.partialFragmentMessage.size()));
size_t trueUncompressedSize = ZSTD_decompress(uncompressedData.data(), uncompressedData.size(), spfm.partialFragmentMessage.data(), spfm.partialFragmentMessage.size());
ByteMemStream completeStrm(uncompressedData.data(), trueUncompressedSize);
cereal::PortableBinaryInputArchive completeArchive(completeStrm);
runAfterDecoded(completeArchive);
spfm.partialFragmentMessage.clear();
spfm.partialFragmentMessageLoc = 0;
}
void decode_fragmented_message(cereal::PortableBinaryInputArchive& inArchive, PartialFragmentMessage& spfm, size_t fragmentMessageStride, std::function<void(cereal::PortableBinaryInputArchive& completeArchive)> runAfterDecoded) {
if(spfm.partialFragmentMessage.size() == 0) {
spfm.partialFragmentMessageLoc = 0;
uint64_t messageSize;
inArchive(messageSize);
spfm.partialFragmentMessage.resize(messageSize);
size_t messStride = std::min(fragmentMessageStride, spfm.partialFragmentMessage.size());
inArchive(cereal::binary_data(spfm.partialFragmentMessage.data() + spfm.partialFragmentMessageLoc, messStride));
spfm.partialFragmentMessageLoc += messStride;
if(messStride == spfm.partialFragmentMessage.size())
uncompress_and_run_func(spfm, runAfterDecoded);
}
else if(spfm.partialFragmentMessage.size() - spfm.partialFragmentMessageLoc <= fragmentMessageStride) {
inArchive(cereal::binary_data(spfm.partialFragmentMessage.data() + spfm.partialFragmentMessageLoc, spfm.partialFragmentMessage.size() - spfm.partialFragmentMessageLoc));
uncompress_and_run_func(spfm, runAfterDecoded);
}
else {
inArchive(cereal::binary_data(spfm.partialFragmentMessage.data() + spfm.partialFragmentMessageLoc, fragmentMessageStride));
spfm.partialFragmentMessageLoc += fragmentMessageStride;
}
}

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

@ -3,6 +3,7 @@
#include <sstream>
#include <memory>
#include <vector>
#include <cereal/archives/portable_binary.hpp>
// https://tuttlem.github.io/2014/08/18/getting-istream-to-work-off-a-byte-array.html
class ByteMemBuf : public std::basic_streambuf<char> {
@ -16,4 +17,10 @@ class ByteMemStream : public std::istream {
ByteMemBuf buffer;
};
std::vector<std::shared_ptr<std::stringstream>> fragment_message(std::string_view v, uint64_t stride);
struct PartialFragmentMessage {
std::string partialFragmentMessage;
uint64_t partialFragmentMessageLoc = 0;
};
std::vector<std::shared_ptr<std::stringstream>> fragment_message(std::string_view uncompressedView, uint64_t stride);
void decode_fragmented_message(cereal::PortableBinaryInputArchive& inArchive, PartialFragmentMessage& spfm, size_t fragmentMessageStride, std::function<void(cereal::PortableBinaryInputArchive& completeArchive)> runAfterDecoded);

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

@ -80,28 +80,11 @@ void NetClient::parse_received_messages() {
inArchive(commandID);
if(commandID == 0) {
if(spfm.partialFragmentMessage.size() == 0) {
spfm.partialFragmentMessageLoc = 0;
uint64_t messageSize;
inArchive(messageSize);
spfm.partialFragmentMessage.resize(messageSize);
inArchive(cereal::binary_data(spfm.partialFragmentMessage.data() + spfm.partialFragmentMessageLoc, NetLibrary::FRAGMENT_MESSAGE_STRIDE));
spfm.partialFragmentMessageLoc += NetLibrary::FRAGMENT_MESSAGE_STRIDE;
}
else if(spfm.partialFragmentMessage.size() - spfm.partialFragmentMessageLoc <= NetLibrary::FRAGMENT_MESSAGE_STRIDE) {
inArchive(cereal::binary_data(spfm.partialFragmentMessage.data() + spfm.partialFragmentMessageLoc, spfm.partialFragmentMessage.size() - spfm.partialFragmentMessageLoc));
ByteMemStream completeStrm(spfm.partialFragmentMessage.data(), spfm.partialFragmentMessage.size());
cereal::PortableBinaryInputArchive completeArchive(completeStrm);
decode_fragmented_message(inArchive, spfm, NetLibrary::FRAGMENT_MESSAGE_STRIDE, [&](cereal::PortableBinaryInputArchive& completeArchive) {
MessageCommandType completeCommandID;
completeArchive(completeCommandID);
recvCallbacks[completeCommandID](completeArchive);
spfm.partialFragmentMessage.clear();
spfm.partialFragmentMessageLoc = 0;
}
else {
inArchive(cereal::binary_data(spfm.partialFragmentMessage.data() + spfm.partialFragmentMessageLoc, NetLibrary::FRAGMENT_MESSAGE_STRIDE));
spfm.partialFragmentMessageLoc += NetLibrary::FRAGMENT_MESSAGE_STRIDE;
}
});
}
else
recvCallbacks[commandID](inArchive);

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

@ -59,10 +59,6 @@ class NetClient : public std::enable_shared_from_this<NetClient> {
std::unordered_map<std::string, std::shared_ptr<rtc::DataChannel>> channels;
struct PartialFragmentMessage {
std::string partialFragmentMessage;
uint64_t partialFragmentMessageLoc = 0;
};
std::unordered_map<std::string, PartialFragmentMessage> pfm;
std::unordered_map<MessageCommandType, NetClientRecvCallback> recvCallbacks;

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

@ -1,10 +1,12 @@
#include "NetServer.hpp"
#include "Helpers/Networking/ByteStream.hpp"
#include "NetLibrary.hpp"
#include <chrono>
#include <rtc/peerconnection.hpp>
#include <Helpers/Random.hpp>
#include <Helpers/Logger.hpp>
#include "NetClient.hpp"
#include "cereal/archives/portable_binary.hpp"
NetServer::NetServer(const std::string& serverLocalID) {
localID = serverLocalID;
@ -163,28 +165,11 @@ void NetServer::ClientData::parse_received_messages(NetServer& server) {
inArchive(commandID);
if(commandID == 0) {
if(spfm.partialFragmentMessage.size() == 0) {
spfm.partialFragmentMessageLoc = 0;
uint64_t messageSize;
inArchive(messageSize);
spfm.partialFragmentMessage.resize(messageSize);
inArchive(cereal::binary_data(spfm.partialFragmentMessage.data() + spfm.partialFragmentMessageLoc, NetLibrary::FRAGMENT_MESSAGE_STRIDE));
spfm.partialFragmentMessageLoc += NetLibrary::FRAGMENT_MESSAGE_STRIDE;
}
else if(spfm.partialFragmentMessage.size() - spfm.partialFragmentMessageLoc <= NetLibrary::FRAGMENT_MESSAGE_STRIDE) {
inArchive(cereal::binary_data(spfm.partialFragmentMessage.data() + spfm.partialFragmentMessageLoc, spfm.partialFragmentMessage.size() - spfm.partialFragmentMessageLoc));
ByteMemStream completeStrm(spfm.partialFragmentMessage.data(), spfm.partialFragmentMessage.size());
cereal::PortableBinaryInputArchive completeArchive(completeStrm);
decode_fragmented_message(inArchive, spfm, NetLibrary::FRAGMENT_MESSAGE_STRIDE, [&](cereal::PortableBinaryInputArchive& completeArchive) {
MessageCommandType completeCommandID;
completeArchive(completeCommandID);
server.recvCallbacks[completeCommandID](shared_from_this(), completeArchive);
spfm.partialFragmentMessage.clear();
spfm.partialFragmentMessageLoc = 0;
}
else {
inArchive(cereal::binary_data(spfm.partialFragmentMessage.data() + spfm.partialFragmentMessageLoc, NetLibrary::FRAGMENT_MESSAGE_STRIDE));
spfm.partialFragmentMessageLoc += NetLibrary::FRAGMENT_MESSAGE_STRIDE;
}
});
}
else
server.recvCallbacks[commandID](shared_from_this(), inArchive);

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

@ -17,10 +17,6 @@ class NetServer : public std::enable_shared_from_this<NetServer> {
typedef std::function<void(std::shared_ptr<ClientData>)> NetServerDisconnectCallback;
struct ClientData : public std::enable_shared_from_this<ClientData> {
std::unordered_map<std::string, std::queue<std::shared_ptr<std::stringstream>>> messages;
struct PartialFragmentMessage {
std::string partialFragmentMessage;
uint64_t partialFragmentMessageLoc = 0;
};
std::unordered_map<std::string, PartialFragmentMessage> pfm;
MessageOrder nextMessageOrderToSend = 0;