diff --git a/src/RemoteControl/RemoteControl.cpp b/src/RemoteControl/RemoteControl.cpp index 20a7ddcf..a8a6ab7a 100644 --- a/src/RemoteControl/RemoteControl.cpp +++ b/src/RemoteControl/RemoteControl.cpp @@ -5,6 +5,7 @@ #include "ServiceDiscovery.h" #include "zmq.hpp" +#include #include // uuid class #include // generators @@ -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; @@ -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]); @@ -236,14 +242,17 @@ int main(int argc, char** argv){ zmq::message_t receive; if(ServiceSend.recv(&receive)){ - std::istringstream iss(static_cast(receive.data())); - std::string answer; - answer=iss.str(); + if(!ZstdDecompress(zstd_dctx, (char*)receive.data(), receive.size(), answer)){ + std::cerr<<"failed to decompress reply!"<("msg_type")=="Command Reply") std::cout<("msg_value")<("msg_type")=="Command Reply") std::cout<("msg_value"))<(receive.data())); - std::string answer; - answer=iss.str(); - - Store rr; - rr.JsonParser(answer); - if(rr.Get("msg_type")=="Command Reply") std::cout<("msg_value")<("msg_type")=="Command Reply") std::cout<("msg_value")<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; +} diff --git a/src/ServiceDiscovery/SlowControlCollection.cpp b/src/ServiceDiscovery/SlowControlCollection.cpp index 206e1de3..3ccdbe65 100644 --- a/src/ServiceDiscovery/SlowControlCollection.cpp +++ b/src/ServiceDiscovery/SlowControlCollection.cpp @@ -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; @@ -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; } @@ -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(){ @@ -289,10 +293,14 @@ void SlowControlCollection::Thread(Thread_args* arg){ return; } - std::istringstream iss(static_cast(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!"<>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); @@ -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!"<SCC->ZstdDecompress(args->SCC, (char*)message.data(), message.size(), payload)){ + std::cerr<<"failed to decompress "<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++; } @@ -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) } @@ -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()COMPRESS_THRESHOLD){ + zmq::message_t zmsg(msg.size()); + memcpy(zmsg.data(), msg.data(), msg.size()); + return zmsg; + } + + std::unique_lock 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 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; +} diff --git a/src/ServiceDiscovery/SlowControlCollection.h b/src/ServiceDiscovery/SlowControlCollection.h index c2a43b19..fb3abce8 100644 --- a/src/ServiceDiscovery/SlowControlCollection.h +++ b/src/ServiceDiscovery/SlowControlCollection.h @@ -5,6 +5,8 @@ #include #include "DAQUtilities.h" #include +#include +#include namespace ToolFramework{ @@ -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 T GetValue(std::string name){ if(!SC_vars.count(name)) return T{}; @@ -81,6 +85,14 @@ namespace ToolFramework{ std::map 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;