detectors/slsDetectorPackage
24
1
Fork 0
mirror of https://github.com/slsdetectorgroup/slsDetectorPackage.git synced 2026-01-18 16:22:19 +01:00
Code Issues Actions Packages Releases Activity

Dev/zmq stream all ipv6 adn remove rx_zmqip (#958)

Browse Source 
* enable ipv6 in zmq socket

* removed rx_zmqip API and field in gui, changed client updaterxrzip to updateclientzmqip to have the rx_hostname ip if 0. 

* updated command line for rx_zmqip to give a warning. 

* Replaced 'depreciated' to 'deprecated' everywhere

* switching from * to 0.0.0.0 works for rebinding zmq sockets

* fixed help in command line for rx_zmqip * to 0.0.0.0 and removed cmd in python

* remove publisher zmq socket ip also for moench post processing

* fixed tests

* publisher zmq ip macros to be reused
This commit is contained in:
maliakal_d
2024-09-10 15:19:08 +02:00
committed by GitHub
parent 15e8c0d9f1
commit e848293916
35 changed files with 168 additions and 426 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
slsSupportLib/include/sls/ZmqSocket.h
View File

@@ -24,7 +24,7 @@
#include <rapidjson/document.h> //json header in zmq stream
#pragma GCC diagnostic pop
//#include <zmq.h>
// #include <zmq.h>
class zmq_msg_t;
namespace sls {
@@ -38,6 +38,8 @@ namespace sls {
#define DEFAULT_LOW_ZMQ_HWM_BUFFERSIZE (1024 * 1024) // 1MB
#define DEFAULT_ZMQ_BUFFERSIZE (-1) // os default
#define ZMQ_PUBLISHER_IP "0.0.0.0"
/** zmq header structure */
struct zmqHeader {
/** true if incoming data, false if end of acquisition */
@@ -98,22 +100,11 @@ class ZmqSocket {
// use this to optimize if optimizing required eg. int value = -1; if
// (zmq_setsockopt(socketDescriptor, ZMQ_LINGER, &value,sizeof(value))) {
// Close();
/**
* Constructor for a client
* Creates socket, context and connects to server
* @param hostname_or_ip hostname or ip of server
* @param portnumber port number
*/
/** Constructor for a subscriber socket */
ZmqSocket(const char *const hostname_or_ip, const uint16_t portnumber);
/**
* Constructor for a server
* Creates socket, context and connects to server
* socket option: keep alive added
* @param portnumber port number
* @param ethip is the ip of the ethernet interface to stream zmq from
*/
ZmqSocket(const uint16_t portnumber, const char *ethip);
/** Constructor for a publisher socket */
ZmqSocket(const uint16_t portnumber);
/** Returns high water mark for outbound messages */
int GetSendHighWaterMark();

6
slsSupportLib/include/sls/sls_detector_defs.h
View File

@@ -181,9 +181,9 @@ class slsDetectorDefs {
int ymin{-1};
int ymax{-1};
ROI() = default;
ROI(int xmin, int xmax) : xmin(xmin), xmax(xmax){};
ROI(int xmin, int xmax) : xmin(xmin), xmax(xmax) {};
ROI(int xmin, int xmax, int ymin, int ymax)
: xmin(xmin), xmax(xmax), ymin(ymin), ymax(ymax){};
: xmin(xmin), xmax(xmax), ymin(ymin), ymax(ymax) {};
constexpr std::array<int, 4> getIntArray() const {
return std::array<int, 4>({xmin, xmax, ymin, ymax});
}
@@ -234,7 +234,7 @@ typedef struct {
int x{0};
int y{0};
xy() = default;
xy(int x, int y) : x(x), y(y){};
xy(int x, int y) : x(x), y(y) {};
} __attribute__((packed));
#endif

4
slsSupportLib/include/sls/sls_detector_funcs.h
View File

@@ -748,8 +748,8 @@ const char* getFunctionNameFromEnum(enum detFuncs func) {
case F_GET_RECEIVER_FILE_FORMAT: return "F_GET_RECEIVER_FILE_FORMAT";
case F_SET_RECEIVER_STREAMING_PORT: return "F_SET_RECEIVER_STREAMING_PORT";
case F_GET_RECEIVER_STREAMING_PORT: return "F_GET_RECEIVER_STREAMING_PORT";
case F_SET_RECEIVER_STREAMING_SRC_IP: return "F_SET_RECEIVER_STREAMING_SRC_IP";
case F_GET_RECEIVER_STREAMING_SRC_IP: return "F_GET_RECEIVER_STREAMING_SRC_IP";
case F_SET_RECEIVER_STREAMING_SRC_IP: return "F_SET_RECEIVER_STREAMING_SRC_IP - obsolete";
case F_GET_RECEIVER_STREAMING_SRC_IP: return "F_GET_RECEIVER_STREAMING_SRC_IP - obsolete";
case F_SET_RECEIVER_SILENT_MODE: return "F_SET_RECEIVER_SILENT_MODE";
case F_GET_RECEIVER_SILENT_MODE: return "F_GET_RECEIVER_SILENT_MODE";
case F_RESTREAM_STOP_FROM_RECEIVER: return "F_RESTREAM_STOP_FROM_RECEIVER";

45
slsSupportLib/src/ZmqSocket.cpp
View File

@@ -53,10 +53,19 @@ ZmqSocket::ZmqSocket(const char *const hostname_or_ip,
}
LOG(logDEBUG) << "Default receive high water mark:"
<< GetReceiveHighWaterMark();
// enable IPv6 addresses
int ipv6 = 1;
if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_IPV6, &ipv6,
sizeof(ipv6))) {
PrintError();
throw ZmqSocketError("Could not set ZMQ_IPV6");
}
}
ZmqSocket::ZmqSocket(const uint16_t portnumber, const char *ethip)
ZmqSocket::ZmqSocket(const uint16_t portnumber)
: portno(portnumber), sockfd(true) {
// create context
sockfd.contextDescriptor = zmq_ctx_new();
if (sockfd.contextDescriptor == nullptr)
@@ -72,32 +81,44 @@ ZmqSocket::ZmqSocket(const uint16_t portnumber, const char *ethip)
// construct address, can be refactored with libfmt
std::ostringstream oss;
oss << "tcp://" << ethip << ":" << portno;
oss << "tcp://" << ZMQ_PUBLISHER_IP << ":" << portno;
sockfd.serverAddress = oss.str();
LOG(logDEBUG) << "zmq address: " << sockfd.serverAddress;
// enable IPv6 addresses
int ipv6 = 1;
if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_IPV6, &ipv6,
sizeof(ipv6))) {
PrintError();
throw ZmqSocketError("Could not set ZMQ_IPV6");
}
// Socket Options for keepalive
// enable TCP keepalive
int keepalive = 1;
if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE, &keepalive, sizeof(keepalive))) {
if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE, &keepalive,
sizeof(keepalive))) {
PrintError();
throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE");
}
// set the number of keepalives before death
keepalive = 10;
if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE_CNT, &keepalive, sizeof(keepalive))) {
if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE_CNT,
&keepalive, sizeof(keepalive))) {
PrintError();
throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE_CNT");
}
// set the time before the first keepalive
keepalive = 60;
if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE_IDLE, &keepalive, sizeof(keepalive))) {
if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE_IDLE,
&keepalive, sizeof(keepalive))) {
PrintError();
throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE_IDLE");
}
// set the interval between keepalives
keepalive = 1;
if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE_INTVL, &keepalive, sizeof(keepalive))) {
if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE_INTVL,
&keepalive, sizeof(keepalive))) {
PrintError();
throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE_INTVL");
}
@@ -109,7 +130,7 @@ ZmqSocket::ZmqSocket(const uint16_t portnumber, const char *ethip)
}
// sleep to allow a slow-joiner
std::this_thread::sleep_for(std::chrono::milliseconds(200));
};
}
int ZmqSocket::GetSendHighWaterMark() {
int value = 0;
@@ -214,8 +235,9 @@ void ZmqSocket::SetReceiveBuffer(int limit) {
}
}
void ZmqSocket::Rebind() { // the purpose is to apply HWL changes, which are
// frozen at bind, which is in the constructor.
void ZmqSocket::Rebind() {
// the purpose is to apply HWL changes, which are
// frozen at bind, which is in the constructor.
// unbbind
if (zmq_unbind(sockfd.socketDescriptor, sockfd.serverAddress.c_str())) {
@@ -498,8 +520,11 @@ void ZmqSocket::PrintError() {
LOG(logERROR)
<< "No I/O thread is available to accomplish the task (zmq)";
break;
case ENOENT:
LOG(logERROR) << "The requested endpoint does not exist (zmq)";
break;
default:
LOG(logERROR) << "Unknown socket error (zmq)";
LOG(logERROR) << "Unknown socket error (zmq). Error code: " << errno;
break;
}
}

12
slsSupportLib/tests/test-ZmqSocket.cpp
View File

@@ -17,14 +17,14 @@ TEST_CASE("Get port number for sub") {
TEST_CASE("Get port number for pub") {
constexpr int port = 50001;
ZmqSocket pub(port, "*");
ZmqSocket pub(port);
REQUIRE(pub.GetPortNumber() == port);
}
TEST_CASE("Server address") {
constexpr int port = 50001;
ZmqSocket pub(port, "*");
REQUIRE(pub.GetZmqServerAddress() == std::string("tcp://*:50001"));
ZmqSocket pub(port);
REQUIRE(pub.GetZmqServerAddress() == std::string("tcp://0.0.0.0:50001"));
}
TEST_CASE("Send header on localhost") {
@@ -32,7 +32,7 @@ TEST_CASE("Send header on localhost") {
ZmqSocket sub("localhost", port);
sub.Connect();
ZmqSocket pub(port, "*");
ZmqSocket pub(port);
// Header to send
zmqHeader header;
@@ -67,7 +67,7 @@ TEST_CASE("Send serveral headers of different length") {
ZmqSocket sub("localhost", port);
sub.Connect();
ZmqSocket pub(port, "*");
ZmqSocket pub(port);
zmqHeader header;
header.data = false; // if true we wait for the data
@@ -95,7 +95,7 @@ TEST_CASE("Send header and data") {
ZmqSocket sub("localhost", port);
sub.Connect();
ZmqSocket pub(port, "*");
ZmqSocket pub(port);
std::vector<int> data{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
const int nbytes = data.size() * sizeof(decltype(data)::value_type);