Skip to content

add compression and decompression of slow controls and alert payloads #74

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
brichards64 merged 3 commits into from
Aug 27, 2026
Merged

add compression and decompression of slow controls and alert payloads #74

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
dac67ad
add compression and decompression of slow controls and alert payloads
Aug 10, 2026
2192f24
add decompression to RemoteControl
Aug 10, 2026
82c640a
Merge branch 'main' into scc_compress
brichards64 Aug 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 56 additions & 13 deletions src/RemoteControl/RemoteControl.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

#include "ServiceDiscovery.h"
#include "zmq.hpp"
#include <zstd.h>

#include <boost/uuid/uuid.hpp> // uuid class
#include <boost/uuid/uuid_generators.hpp> // generators
Expand All @@ -18,9 +19,13 @@
#define GROUP_COMMAND_REPLY_WAIT 2000
#define FILE_SEND_WAIT 120000
#define FILE_SEND_PORT 24001
#define MAX_DECOMPRESSED_SIZE 655355
const unsigned char ZSTD_MAGIC_BYTES[4] = {0x28,0xB5,0x2F,0xFD}; // ZSTD_MAGICNUMBER from zstd.h BUT REVERSED!

using namespace ToolFramework;

bool ZstdDecompress(ZSTD_DCtx* zstd_dctx, char* msg, uint64_t msgsize, std::string& decompress_buffer);

int main(int argc, char** argv){

// if (argc!=3) return 1;
Expand All @@ -30,6 +35,7 @@ int main(int argc, char** argv){


zmq::context_t context(3);
ZSTD_DCtx* zstd_dctx = ZSTD_createDCtx();

//std::string address(argv[1]);
// std::stringstream tmp (argv[2]);
Expand Down Expand Up @@ -236,14 +242,17 @@ int main(int argc, char** argv){

zmq::message_t receive;
if(ServiceSend.recv(&receive)){
std::istringstream iss(static_cast<char*>(receive.data()));

std::string answer;
answer=iss.str();
if(!ZstdDecompress(zstd_dctx, (char*)receive.data(), receive.size(), answer)){
std::cerr<<"failed to decompress reply!"<<std::endl;

} else {
Store rr;
rr.JsonParser(answer);
if(rr.Get<std::string>("msg_type")=="Command Reply") std::cout<<std::endl<<rr.Get<std::string>("msg_value")<<std::endl<<std::endl;

}

Store rr;
rr.JsonParser(answer);
if(rr.Get<std::string>("msg_type")=="Command Reply") std::cout<<std::endl<<(rr.Get<std::string>("msg_value"))<<std::endl<<std::endl;
}
else std::cout<<std::endl<<"message timed out"<<std::endl;

Expand Down Expand Up @@ -364,15 +373,18 @@ int main(int argc, char** argv){

zmq::message_t receive;
if(ServiceSend.recv(&receive)){
std::istringstream iss(static_cast<char*>(receive.data()));

std::string answer;
answer=iss.str();

Store rr;
rr.JsonParser(answer);
if(rr.Get<std::string>("msg_type")=="Command Reply") std::cout<<std::endl<<rr.Get<std::string>("msg_value")<<std::endl<<std::endl;
if(!ZstdDecompress(zstd_dctx, (char*)receive.data(), receive.size(), answer)){
std::cerr<<"failed to decompress reply!"<<std::endl;

} else {
Store rr;
rr.JsonParser(answer);
if(rr.Get<std::string>("msg_type")=="Command Reply") std::cout<<std::endl<<rr.Get<std::string>("msg_value")<<std::endl<<std::endl;

}
}

}
}

Expand All @@ -392,3 +404,34 @@ int main(int argc, char** argv){
return 0;

}

bool ZstdDecompress(ZSTD_DCtx* zstd_dctx, char* msg, uint64_t msgsize, std::string& decompress_buffer){
std::string errmsg;
if(msgsize>4 && std::memcmp(msg,ZSTD_MAGIC_BYTES,4)==0){
uint64_t decompressed_bytes = ZSTD_getFrameContentSize(msg, msgsize);
if(decompressed_bytes==ZSTD_CONTENTSIZE_UNKNOWN || decompressed_bytes==ZSTD_CONTENTSIZE_ERROR){
// bad response
errmsg = std::string{"Received corrupt zstd message "}+ZSTD_getErrorName(decompressed_bytes);
goto decompress_error;
}
if(decompressed_bytes > MAX_DECOMPRESSED_SIZE){
errmsg = "Compressed message with oversized payload: "+std::to_string(decompressed_bytes)+" bytes";
goto decompress_error;
}
decompress_buffer.resize(decompressed_bytes);
decompressed_bytes = ZSTD_decompressDCtx(zstd_dctx,(void*)decompress_buffer.data(),decompressed_bytes, msg, msgsize);
if(ZSTD_isError(decompressed_bytes)){
errmsg = std::string{"zstd error decompressing response: "}+ZSTD_getErrorName(decompressed_bytes);
goto decompress_error;
}
} else {
// message not compressed
decompress_buffer.assign(msg, msgsize);
}
return true;

decompress_error:
std::cerr << errmsg << std::endl;
decompress_buffer.clear();
return false;
}
96 changes: 82 additions & 14 deletions src/ServiceDiscovery/SlowControlCollection.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,10 @@

using namespace ToolFramework;

namespace {
const unsigned char ZSTD_MAGIC_BYTES[4] = {0x28,0xB5,0x2F,0xFD}; // ZSTD_MAGICNUMBER from zstd.h BUT REVERSED!
}

SlowControlCollectionThread_args::SlowControlCollectionThread_args(){

sock=0;
Expand All @@ -12,9 +16,6 @@ SlowControlCollectionThread_args::SlowControlCollectionThread_args(){
alert_functions_mutex=0;
SC_vars=0;

m_pub = 0;
pub_monitor_socket = 0;
pub_connected_mtx = 0;

}

Expand Down Expand Up @@ -55,6 +56,9 @@ SlowControlCollection::SlowControlCollection(){
Add("NewConfig",SlowControlElementType(INFO),0,0,false,false);
SC_vars["NewConfig"]->SetValue(0);

zstd_cctx = ZSTD_createCCtx();
zstd_dctx = ZSTD_createDCtx();

}

SlowControlCollection::~SlowControlCollection(){
Expand Down Expand Up @@ -289,10 +293,14 @@ void SlowControlCollection::Thread(Thread_args* arg){
return;
}

std::istringstream iss(static_cast<char*>(message.data()));
std::string payload;
if(!args->SCC->ZstdDecompress(args->SCC, (char*)message.data(), message.size(), payload)){
std::cerr<<"failed to decompress slow control message!"<<std::endl;
return;
}

Store tmp;
//printf("iss=%s\n",iss.str().c_str());
tmp.JsonParser(iss.str());
tmp.JsonParser(payload);
//tmp.Print();
if(!tmp.Has("msg_value")){
std::cerr<<"error: Poorly formatted slowcontrol input [no msg_value]"<<std::endl;
Expand Down Expand Up @@ -371,9 +379,8 @@ void SlowControlCollection::Thread(Thread_args* arg){
std::string tmp2="";
rr>>tmp2;
//printf("reply message is= %s \n",tmp2.c_str());
zmq::message_t send(tmp2.length()+1);
snprintf ((char *) send.data(), tmp2.length()+1 , "%s" ,tmp2.c_str()) ;

zmq::message_t send = args->SCC->ZstdCompress(args->SCC, tmp2);

bool tmp_ok = args->sock->send(identity, ZMQ_SNDMORE);
if(tmp_ok) tmp_ok = tmp_ok && args->sock->send(blank, ZMQ_SNDMORE);
Expand Down Expand Up @@ -405,17 +412,20 @@ void SlowControlCollection::Thread(Thread_args* arg){
ok = args->sub->recv(&message);
if(ok==0){
// FIXME this case should be handled! what do we do?
std::cerr<<"failed to receive alert payload!"<<std::endl;
std::cerr<<"failed to receive "<<iss.str() << " alert payload!"<<std::endl;
return;
}
if(!args->SCC->ZstdDecompress(args->SCC, (char*)message.data(), message.size(), payload)){
std::cerr<<"failed to decompress "<<iss.str() << " alert payload!"<<std::endl;
return;
}
payload.resize(message.size(),'\0');
memcpy((void*)payload.data(),message.data(),message.size());
has_data=true;
}

//int a=0;
while(message.more()){

args->sub->recv(&message);
args->sub->recv(&message); // FIXME do we want any warnings or handling here?
//memcpy((void*)payload.data(),message.data(),message.size());
//a++;
}
Expand Down Expand Up @@ -592,8 +602,8 @@ bool SlowControlCollection::AlertSend(std::string alert, std::string payload){
// if we didn't return, we have a payload as well
bool ok = m_pub->send(message, ZMQ_SNDMORE);
if(!ok) return false; // err: "zmq send "+zmq_strerror(errno)
zmq::message_t message2(payload.length()+1);
snprintf((char*) message2.data(), payload.length()+1, "%s", payload.c_str());

zmq::message_t message2 = args->SCC->ZstdCompress(args->SCC, payload);
return m_pub->send(message2); // err: "zmq send "+zmq_strerror(errno)

}
Expand Down Expand Up @@ -868,3 +878,61 @@ void SlowControlCollection::ClearState(){
SC_vars["State"]->SetValue(m_state);
return;
}

zmq::message_t SlowControlCollection::ZstdCompress(SlowControlCollection* SCC, std::string& msg){
if(msg.length()<SCC->COMPRESS_THRESHOLD){
zmq::message_t zmsg(msg.size());
memcpy(zmsg.data(), msg.data(), msg.size());
return zmsg;
}

std::unique_lock<std::mutex> locker(*SCC->zstd_cctx_mtx);
std::string compressed_msg_buf;
compressed_msg_buf.resize(ZSTD_compressBound(msg.size()));
uint64_t bytes_to_send = ZSTD_compressCCtx(SCC->zstd_cctx, (void*)compressed_msg_buf.data(), compressed_msg_buf.size(), msg.data(), msg.size(), SCC->zstd_compression_level);
if(ZSTD_isError(bytes_to_send)){
locker.unlock();
std::string errmsg = std::string{"Warning: error compressing multicast message "}+ZSTD_getErrorName(bytes_to_send);
std::clog << errmsg << std::endl;
// send it uncompressed
zmq::message_t zmsg(msg.size());
memcpy(zmsg.data(), msg.data(), msg.size());
return zmsg;
}
zmq::message_t zmsg(msg.size());
memcpy(zmsg.data(), compressed_msg_buf.data(), bytes_to_send);
return zmsg;
}

bool SlowControlCollection::ZstdDecompress(SlowControlCollection* SCC, char* msg, uint64_t msgsize, std::string& decompress_buffer){
std::string errmsg;
std::unique_lock<std::mutex> locker(*SCC->zstd_dctx_mtx);
if(msgsize>4 && std::memcmp(msg,ZSTD_MAGIC_BYTES,4)==0){
uint64_t decompressed_bytes = ZSTD_getFrameContentSize(msg, msgsize);
if(decompressed_bytes==ZSTD_CONTENTSIZE_UNKNOWN || decompressed_bytes==ZSTD_CONTENTSIZE_ERROR){
// bad response
errmsg = std::string{"Received corrupt zstd message "}+ZSTD_getErrorName(decompressed_bytes);
goto decompress_error;
}
if(decompressed_bytes > SCC->MAX_DECOMPRESSED_SIZE){
errmsg = "Compressed message with oversized payload: "+std::to_string(decompressed_bytes)+" bytes";
goto decompress_error;
}
decompress_buffer.resize(decompressed_bytes);
decompressed_bytes = ZSTD_decompressDCtx(SCC->zstd_dctx,(void*)decompress_buffer.data(),decompressed_bytes, msg, msgsize);
if(ZSTD_isError(decompressed_bytes)){
errmsg = std::string{"zstd error decompressing response: "}+ZSTD_getErrorName(decompressed_bytes);
goto decompress_error;
}
} else {
// message not compressed
decompress_buffer.assign(msg, msgsize);
}
return true;

decompress_error:
locker.unlock();
std::clog << errmsg << std::endl;
decompress_buffer.clear();
return false;
}
12 changes: 12 additions & 0 deletions src/ServiceDiscovery/SlowControlCollection.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
#include <zmq.hpp>
#include "DAQUtilities.h"
#include <functional>
#include <mutex>
#include <zstd.h>

namespace ToolFramework{

Expand Down Expand Up @@ -68,6 +70,8 @@ namespace ToolFramework{
void SetError(bool error);
void SetWarning(bool warn);
void ClearState();
zmq::message_t ZstdCompress(SlowControlCollection* SCC, std::string& msg);
bool ZstdDecompress(SlowControlCollection* SCC, char* msg, uint64_t msgsize, std::string& decompress_buffer);

template<typename T> T GetValue(std::string name){
if(!SC_vars.count(name)) return T{};
Expand All @@ -81,6 +85,14 @@ namespace ToolFramework{
std::map<std::string, AlertFunction> m_alert_functions;
std::mutex m_alert_functions_mutex;

ZSTD_CCtx* zstd_cctx;
std::mutex* zstd_cctx_mtx;
ZSTD_DCtx* zstd_dctx;
std::mutex* zstd_dctx_mtx;
int zstd_compression_level=1;
uint32_t COMPRESS_THRESHOLD=0; //1024; // compress any send messages > this many bytes
uint32_t MAX_DECOMPRESSED_SIZE=655355; // refuse to decompress messages that will exceed this size once decompressed

DAQUtilities* m_util;
zmq::context_t* m_context;
zmq::socket_t* m_pub;
Expand Down
Loading