diff --git a/RELEASE.md b/RELEASE.md index 11563b7af..e5179cee4 100644 --- a/RELEASE.md +++ b/RELEASE.md @@ -71,6 +71,8 @@ allow disabling one UDP interface in the receiver. setting number of UDP interfaces can only be set at detector level and not at module level. (individual modules) +zmqsocket - allowing hwm to be set already at the constructor level. As before, can still be set after construction with socket reconnection. + 2 On-board Detector Server Compatibility ========================================== diff --git a/python/tests/test_free.py b/python/tests/test_free.py index 91e959ef4..2e0b9b0e8 100644 --- a/python/tests/test_free.py +++ b/python/tests/test_free.py @@ -21,7 +21,7 @@ from slsdet import Detector, Ctb, freeSharedMemory from utils_for_test import ( Log, LogLevel, - SERVER_START_PORTNO + DET_START_TCP_PORTNO ) from conftest import session_simulator @@ -33,7 +33,7 @@ def test_exptime_after_free_should_raise(session_simulator): d = Ctb() # creates multi shm (assuming no shm exists) - d.hostname = f"localhost:{SERVER_START_PORTNO}" # hostname command creates mod shm, d maps to it + d.hostname = f"localhost:{DET_START_TCP_PORTNO}" # hostname command creates mod shm, d maps to it d.free() # frees the shm, d should not map to it anymore @@ -47,7 +47,7 @@ def test_exptime_after_free_should_raise(session_simulator): def free_and_create_shm(): k = Ctb() # opens existing shm if it exists - k.hostname = f"localhost:{SERVER_START_PORTNO}" # free and recreate shm, maps to local shm struct + k.hostname = f"localhost:{DET_START_TCP_PORTNO}" # free and recreate shm, maps to local shm struct @pytest.mark.detectorintegration @pytest.mark.parametrize("session_simulator",[("ctb", 1, 1)],indirect=True) @@ -56,7 +56,7 @@ def test_exptime_after_not_passing_var_should_raise(session_simulator): d = Ctb() # creates multi shm (assuming no shm exists) - d.hostname = f"localhost:{SERVER_START_PORTNO}" # hostname command creates mod shm, d maps to it + d.hostname = f"localhost:{DET_START_TCP_PORTNO}" # hostname command creates mod shm, d maps to it free_and_create_shm() # ctb() opens multi shm, hostname command frees and recreates mod shm but shm struct is local. d still maps to old shm struct @@ -72,7 +72,7 @@ def test_exptime_after_not_passing_var_should_raise(session_simulator): def free_and_create_shm_passing_ctb_var(k): k = Ctb() # opens existing shm if it exists (disregards k as its new Ctb only local to this function) - k.hostname = f"localhost:{SERVER_START_PORTNO}" # free and recreate shm, maps to local shm struct + k.hostname = f"localhost:{DET_START_TCP_PORTNO}" # free and recreate shm, maps to local shm struct @pytest.mark.detectorintegration @pytest.mark.parametrize("session_simulator",[("ctb", 1, 1)],indirect=True) @@ -80,7 +80,7 @@ def test_exptime_after_passing_ctb_var_should_raise(session_simulator): Log(LogLevel.INFOBLUE, f'\nRunning test_exptime_after_passing_ctb_var_should_raise') d = Ctb() # creates multi shm (assuming no shm exists) - d.hostname = f"localhost:{SERVER_START_PORTNO}" # hostname command creates mod shm, d maps to it + d.hostname = f"localhost:{DET_START_TCP_PORTNO}" # hostname command creates mod shm, d maps to it free_and_create_shm_passing_ctb_var(d) # ctb() opens multi shm, hostname command frees and recreates mod shm but shm struct is local. d still maps to old shm struct @@ -95,7 +95,7 @@ def test_exptime_after_passing_ctb_var_should_raise(session_simulator): def free_and_create_shm_returning_ctb(): k = Ctb() # opens existing shm if it exists (disregards k as its new Ctb only local to this function) - k.hostname = f"localhost:{SERVER_START_PORTNO}" # free and recreate shm, maps to local shm struct + k.hostname = f"localhost:{DET_START_TCP_PORTNO}" # free and recreate shm, maps to local shm struct return k @pytest.mark.detectorintegration @@ -128,8 +128,8 @@ def test_hostname_twice_acess_old_should_raise(session_simulator): Log(LogLevel.INFOBLUE, f'\nRunning test_hostname_twice_acess_old_should_raise') d = Ctb() # creates multi shm (assuming no shm exists) - d.hostname = f"localhost:{SERVER_START_PORTNO}" # hostname command creates mod shm, d maps to it - d.hostname = f"localhost:{SERVER_START_PORTNO}" # Freeing and recreating shm while mapping d to it (old shm is out of scope) + d.hostname = f"localhost:{DET_START_TCP_PORTNO}" # hostname command creates mod shm, d maps to it + d.hostname = f"localhost:{DET_START_TCP_PORTNO}" # Freeing and recreating shm while mapping d to it (old shm is out of scope) # this should not throw exptime_val = d.exptime diff --git a/python/tests/test_zmq.py b/python/tests/test_zmq.py new file mode 100644 index 000000000..0301a0cbe --- /dev/null +++ b/python/tests/test_zmq.py @@ -0,0 +1,51 @@ +import pytest +import sys +import time + +from conftest import session_simulator + +from slsdet import Detector +from slsdet.utils import element_if_equal + +from slsdet._slsdet import slsDetectorDefs +from utils_for_test import ( + Log, + LogLevel, +) + +detectorType = slsDetectorDefs.detectorType + +@pytest.mark.detectorintegration +@pytest.mark.parametrize( + "session_simulator", + [ + ("eiger", 1, 20) + ], + indirect=True, +) +def test_zmq_reconnect(session_simulator, request): + """ Test changing zmq ports and zmq hwm with zmq sockets connected (reconnect). """ + det_type, num_interfaces, num_mods, d = session_simulator + assert d is not None + assert d.type == detectorType.EIGER + + + d.rx_zmqstream = True + assert d.rx_zmqstream == True + + port = 14000 + hwm = 2 + + for _ in range(10): + d.rx_zmqport = port + d.rx_zmqhwm = hwm + Log(LogLevel.INFOGREEN, f"Set zmqport={port}, zmqhwm={hwm}") + + port += 1000 + hwm += 5 + time.sleep(1) + + + Log(LogLevel.INFOGREEN, f"✅ {request.node.name} passed") + + diff --git a/slsDetectorSoftware/src/Detector.cpp b/slsDetectorSoftware/src/Detector.cpp index 121b144d6..dde062e27 100644 --- a/slsDetectorSoftware/src/Detector.cpp +++ b/slsDetectorSoftware/src/Detector.cpp @@ -1096,28 +1096,35 @@ void Detector::setNumberofUDPInterfaces_(int n) { if (!size()) { throw RuntimeError("No modules added."); } - bool previouslyClientStreaming = pimpl->getDataStreamingToClient(); - uint16_t clientStartingPort = getClientZmqPort({0}).squash(0); + + // get starting ports and disable zmq streaming + // rx bool useReceiver = getUseReceiverFlag().squash(false); bool previouslyReceiverStreaming = false; uint16_t rxStartingPort = 0; if (useReceiver) { previouslyReceiverStreaming = getRxZmqDataStream().squash(true); + setRxZmqDataStream(false); rxStartingPort = getRxZmqPort({0}).squash(0); } + // client + bool previouslyClientStreaming = pimpl->getDataStreamingToClient(); + uint16_t clientStartingPort = getClientZmqPort({0}).squash(0); + pimpl->setDataStreamingToClient(false); + pimpl->Parallel(&Module::setNumberofUDPInterfaces, {}, n); + // ensure receiver zmq socket ports are multiplied by 2 (2 interfaces) setClientZmqPort(clientStartingPort, -1); - if (getUseReceiverFlag().squash(false)) { + if (useReceiver) { setRxZmqPort(rxStartingPort, -1); } + // redo the zmq sockets if enabled if (previouslyClientStreaming) { - pimpl->setDataStreamingToClient(false); pimpl->setDataStreamingToClient(true); } - if (previouslyReceiverStreaming) { - setRxZmqDataStream(false); + if (useReceiver && previouslyReceiverStreaming) { setRxZmqDataStream(true); } } @@ -1638,7 +1645,12 @@ void Detector::setClientZmqIp(const IpAddr ip, Positions pos) { int Detector::getClientZmqHwm() const { return pimpl->getClientStreamingHwm(); } void Detector::setClientZmqHwm(const int limit) { + bool previouslyClientStreaming = pimpl->getDataStreamingToClient(); pimpl->setClientStreamingHwm(limit); + if (previouslyClientStreaming) { + pimpl->setDataStreamingToClient(false); + pimpl->setDataStreamingToClient(true); + } } Result Detector::getRxZmqHwm(Positions pos) const { diff --git a/slsDetectorSoftware/src/DetectorImpl.cpp b/slsDetectorSoftware/src/DetectorImpl.cpp index fd08b8bf5..5ff20d219 100644 --- a/slsDetectorSoftware/src/DetectorImpl.cpp +++ b/slsDetectorSoftware/src/DetectorImpl.cpp @@ -470,22 +470,13 @@ void DetectorImpl::createReceivingDataSockets() { size_t numSockets = modules.size() * numUDPInterfaces; for (size_t iSocket = 0; iSocket < numSockets; ++iSocket) { - uint32_t portnum = - (modules[iSocket / numUDPInterfaces]->getClientStreamingPort()); + auto imod = iSocket / numUDPInterfaces; + uint32_t portnum = modules[imod]->getClientStreamingPort(); portnum += (iSocket % numUDPInterfaces); try { + auto ip = modules[imod]->getClientStreamingIP().str(); zmqSocket.push_back( - make_unique(modules[iSocket / numUDPInterfaces] - ->getClientStreamingIP() - .str() - .c_str(), - portnum)); - // set high water mark - int hwm = shm()->zmqHwm; - if (hwm >= 0) { - zmqSocket[iSocket]->SetReceiveHighWaterMark(hwm); - // need not reconnect. cannot be connected (detector idle) - } + make_unique(ip.c_str(), portnum, shm()->zmqHwm)); LOG(logINFO) << "Zmq Client[" << iSocket << "] at " << zmqSocket.back()->GetZmqServerAddress() << "[hwm: " << zmqSocket.back()->GetReceiveHighWaterMark() << "]"; @@ -1064,23 +1055,7 @@ void DetectorImpl::setClientStreamingHwm(const int limit) { } // update shm shm()->zmqHwm = limit; - - // streaming enabled - if (client_downstream) { - // custom limit, set it directly - if (limit >= 0) { - for (auto &it : zmqSocket) { - it->SetReceiveHighWaterMark(limit); - // need not reconnect. cannot be connected (detector idle) - } - LOG(logINFO) << "Setting Client Zmq socket rcv hwm to " << limit; - } - // default, disable and enable to get default - else { - setDataStreamingToClient(false); - setDataStreamingToClient(true); - } - } + LOG(logINFO) << "Setting Client Zmq socket rcv hwm to " << limit; } void DetectorImpl::registerAcquisitionFinishedCallback(void (*func)(double, int, diff --git a/slsReceiverSoftware/src/DataStreamer.cpp b/slsReceiverSoftware/src/DataStreamer.cpp index 7e2826f4a..78c8a44d8 100644 --- a/slsReceiverSoftware/src/DataStreamer.cpp +++ b/slsReceiverSoftware/src/DataStreamer.cpp @@ -78,16 +78,20 @@ void DataStreamer::RecordFirstIndex(uint64_t fnum, size_t firstImageIndex) { << ", First Streamer Index:" << fnum; } +int DataStreamer::GetZmqHwm() const { + if (zmqSocket) { + return zmqSocket->GetSendHighWaterMark(); + } + return -1; +} + void DataStreamer::CreateZmqSockets(uint16_t port, int hwm) { uint16_t portnum = port + index; try { - zmqSocket = new ZmqSocket(portnum); - - // set if custom if (hwm >= 0) { - zmqSocket->SetSendHighWaterMark(hwm); - // needed, or HWL is not taken - zmqSocket->Rebind(); + zmqSocket = new ZmqSocket(portnum, hwm); + } else { + zmqSocket = new ZmqSocket(portnum); } } catch (std::exception &e) { std::ostringstream oss; diff --git a/slsReceiverSoftware/src/DataStreamer.h b/slsReceiverSoftware/src/DataStreamer.h index f2f8dd1f9..53d72431e 100644 --- a/slsReceiverSoftware/src/DataStreamer.h +++ b/slsReceiverSoftware/src/DataStreamer.h @@ -47,11 +47,13 @@ class DataStreamer : private virtual slsDetectorDefs, public ThreadObject { * Creates Zmq Sockets * (throws an exception if it couldnt create zmq sockets) * @param port streaming port start index - * @param hwm streaming high water mark + * @param hwm high water mark for zmq socket */ void CreateZmqSockets(uint16_t port, int hwm); void CloseZmqSocket(); void StreamRxDummyHeader(); + void RestreamStop(); + int GetZmqHwm() const; private: /** diff --git a/slsReceiverSoftware/src/Implementation.cpp b/slsReceiverSoftware/src/Implementation.cpp index e0a61199b..6a78b764e 100644 --- a/slsReceiverSoftware/src/Implementation.cpp +++ b/slsReceiverSoftware/src/Implementation.cpp @@ -1307,7 +1307,26 @@ void Implementation::setStreamingPort(const uint16_t i) { LOG(logINFO) << "Streaming Port: " << streamingPort; } -int Implementation::getStreamingHwm() const { return streamingHwm; } +int Implementation::getStreamingHwm() const { + switch (dataStreamer.size()) { + case 0: + return streamingHwm; + case 2: + if (dataStreamer[0]->GetZmqHwm() != dataStreamer[1]->GetZmqHwm()) { + throw RuntimeError( + "Streaming Hwm is not same for all data streamers: " + + std::to_string(dataStreamer[0]->GetZmqHwm()) + ", " + + std::to_string(dataStreamer[1]->GetZmqHwm())); + } + [[fallthrough]]; + case 1: + return dataStreamer[0]->GetZmqHwm(); + break; + default: + throw RuntimeError("Invalid number of data streamers: " + + std::to_string(dataStreamer.size())); + } +} void Implementation::setStreamingHwm(const int i) { streamingHwm = i; diff --git a/slsSupportLib/include/sls/ZmqSocket.h b/slsSupportLib/include/sls/ZmqSocket.h index ba65d59ce..b270a53b1 100644 --- a/slsSupportLib/include/sls/ZmqSocket.h +++ b/slsSupportLib/include/sls/ZmqSocket.h @@ -95,22 +95,23 @@ class ZmqSocket { public: // Socket Options for optimization - // ZMQ_LINGER default is already -1 means no messages discarded. use this - // options if optimizing required ZMQ_SNDHWM default is 0 means no limit. - // use this to optimize if optimizing required eg. int value = -1; if - // (zmq_setsockopt(socketDescriptor, ZMQ_LINGER, &value,sizeof(value))) { - // Close(); + // ZMQ_LINGER default is already -1 means no messages discarded. + // ZMQ_RCVHWM default is from zmqlib (1000). If not -1, calls + // SetReceiveHighWaterMark, also setting receive buffer size accordingly /** Constructor for a subscriber socket */ - ZmqSocket(const char *const hostname_or_ip, const uint16_t portnumber); + ZmqSocket(const char *const hostname_or_ip, const uint16_t portnumber, + int hwm = -1); - /** Constructor for a publisher socket */ - ZmqSocket(const uint16_t portnumber); + /** Constructor for a publisher socket with high water mark as -1 to mean + * default from zmqlib (1000). If hwm is not -1, it calls + * SetSendHighWaterMark, also setting send buffer size accordingly*/ + ZmqSocket(const uint16_t portnumber, int hwm = -1); /** Returns high water mark for outbound messages */ int GetSendHighWaterMark(); /** Sets high water mark for outbound messages. Default 1000 (zmqlib). Also - * changes send buffer size depending on hwm. Must rebind. */ + * changes send buffer size depending on hwm. Must rebind or reconnect. */ void SetSendHighWaterMark(int limit); /** Returns high water mark for inbound messages */ diff --git a/slsSupportLib/src/ZmqSocket.cpp b/slsSupportLib/src/ZmqSocket.cpp index 2fafe4799..9632ca61e 100644 --- a/slsSupportLib/src/ZmqSocket.cpp +++ b/slsSupportLib/src/ZmqSocket.cpp @@ -15,10 +15,11 @@ namespace sls { using namespace rapidjson; ZmqSocket::ZmqSocket(const char *const hostname_or_ip, - const uint16_t portnumber) + const uint16_t portnumber, int hwm) : portno(portnumber), sockfd(false) { // Extra check that throws if conversion fails, could be removed auto ipstr = HostnameToIp(hostname_or_ip).str(); + std::ostringstream oss; oss << "tcp://" << ipstr << ":" << portno; sockfd.serverAddress = oss.str(); @@ -27,20 +28,23 @@ ZmqSocket::ZmqSocket(const char *const hostname_or_ip, // create context sockfd.contextDescriptor = zmq_ctx_new(); if (sockfd.contextDescriptor == nullptr) - throw ZmqSocketError("Could not create contextDescriptor"); + throw ZmqSocketError("Could not create contextDescriptor for " + + sockfd.serverAddress); // create subscriber sockfd.socketDescriptor = zmq_socket(sockfd.contextDescriptor, ZMQ_SUB); if (sockfd.socketDescriptor == nullptr) { PrintError(); - throw ZmqSocketError("Could not create socket"); + throw ZmqSocketError("Could not create socket for " + + sockfd.serverAddress); } // Socket Options provided above // an empty string implies receiving any messages if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_SUBSCRIBE, "", 0)) { PrintError(); - throw ZmqSocketError("Could set socket opt"); + throw ZmqSocketError("Could set socket opt for " + + sockfd.serverAddress); } // ZMQ_LINGER default is already -1 means no messages discarded. use this // options if optimizing required ZMQ_SNDHWM default is 0 means no limit. @@ -49,7 +53,8 @@ ZmqSocket::ZmqSocket(const char *const hostname_or_ip, if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_LINGER, &value, sizeof(value))) { PrintError(); - throw ZmqSocketError("Could not set ZMQ_LINGER"); + throw ZmqSocketError("Could not set ZMQ_LINGER for " + + sockfd.serverAddress); } LOG(logDEBUG) << "Default receive high water mark:" << GetReceiveHighWaterMark(); @@ -59,38 +64,47 @@ ZmqSocket::ZmqSocket(const char *const hostname_or_ip, if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_IPV6, &ipv6, sizeof(ipv6))) { PrintError(); - throw ZmqSocketError("Could not set ZMQ_IPV6"); + throw ZmqSocketError("Could not set ZMQ_IPV6 for " + + sockfd.serverAddress); + } + + // set hwm if not default + if (hwm >= 0) { + SetReceiveHighWaterMark(hwm); } } -ZmqSocket::ZmqSocket(const uint16_t portnumber) +ZmqSocket::ZmqSocket(const uint16_t portnumber, int hwm) : portno(portnumber), sockfd(true) { - // create context - sockfd.contextDescriptor = zmq_ctx_new(); - if (sockfd.contextDescriptor == nullptr) - throw ZmqSocketError("Could not create contextDescriptor"); - - // create publisher - sockfd.socketDescriptor = zmq_socket(sockfd.contextDescriptor, ZMQ_PUB); - if (sockfd.socketDescriptor == nullptr) { - PrintError(); - throw ZmqSocketError("Could not create socket"); - } - LOG(logDEBUG) << "Default send high water mark:" << GetSendHighWaterMark(); - // construct address, can be refactored with libfmt std::ostringstream oss; oss << "tcp://" << ZMQ_PUBLISHER_IP << ":" << portno; sockfd.serverAddress = oss.str(); LOG(logDEBUG) << "zmq address: " << sockfd.serverAddress; + // create context + sockfd.contextDescriptor = zmq_ctx_new(); + if (sockfd.contextDescriptor == nullptr) + throw ZmqSocketError("Could not create contextDescriptor for " + + sockfd.serverAddress); + + // create publisher + sockfd.socketDescriptor = zmq_socket(sockfd.contextDescriptor, ZMQ_PUB); + if (sockfd.socketDescriptor == nullptr) { + PrintError(); + throw ZmqSocketError("Could not create socket for " + + sockfd.serverAddress); + } + LOG(logDEBUG) << "Default send high water mark:" << GetSendHighWaterMark(); + // enable IPv6 addresses int ipv6 = 1; if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_IPV6, &ipv6, sizeof(ipv6))) { PrintError(); - throw ZmqSocketError("Could not set ZMQ_IPV6"); + throw ZmqSocketError("Could not set ZMQ_IPV6 for " + + sockfd.serverAddress); } // Socket Options for keepalive @@ -99,34 +113,45 @@ ZmqSocket::ZmqSocket(const uint16_t portnumber) if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE, &keepalive, sizeof(keepalive))) { PrintError(); - throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE"); + throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE for " + + sockfd.serverAddress); } // set the number of keepalives before death keepalive = 10; if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE_CNT, &keepalive, sizeof(keepalive))) { PrintError(); - throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE_CNT"); + throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE_CNT for " + + sockfd.serverAddress); } // set the time before the first keepalive keepalive = 60; if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE_IDLE, &keepalive, sizeof(keepalive))) { PrintError(); - throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE_IDLE"); + throw ZmqSocketError( + "Could set socket opt ZMQ_TCP_KEEPALIVE_IDLE for " + + sockfd.serverAddress); } // set the interval between keepalives keepalive = 1; if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE_INTVL, &keepalive, sizeof(keepalive))) { PrintError(); - throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE_INTVL"); + throw ZmqSocketError( + "Could set socket opt ZMQ_TCP_KEEPALIVE_INTVL for " + + sockfd.serverAddress); + } + // set hwm if not default + if (hwm >= 0) { + SetSendHighWaterMark(hwm); } // bind address if (zmq_bind(sockfd.socketDescriptor, sockfd.serverAddress.c_str())) { PrintError(); - throw ZmqSocketError("Could not bind socket"); + throw ZmqSocketError("Could not bind socket for " + + sockfd.serverAddress); } // sleep to allow a slow-joiner std::this_thread::sleep_for(std::chrono::milliseconds(200)); @@ -138,7 +163,8 @@ int ZmqSocket::GetSendHighWaterMark() { if (zmq_getsockopt(sockfd.socketDescriptor, ZMQ_SNDHWM, &value, &value_size)) { PrintError(); - throw ZmqSocketError("Could not get ZMQ_SNDHWM"); + throw ZmqSocketError("Could not get ZMQ_SNDHWM for " + + sockfd.serverAddress); } return value; } @@ -147,10 +173,13 @@ void ZmqSocket::SetSendHighWaterMark(int limit) { if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_SNDHWM, &limit, sizeof(limit))) { PrintError(); - throw ZmqSocketError("Could not set ZMQ_SNDHWM"); + throw ZmqSocketError("Could not set ZMQ_SNDHWM for " + + sockfd.serverAddress + " to " + + std::to_string(limit)); } if (GetSendHighWaterMark() != limit) { - throw ZmqSocketError("Could not set ZMQ_SNDHWM to " + + throw ZmqSocketError("Could not set ZMQ_SNDHWM for " + + sockfd.serverAddress + " to " + std::to_string(limit)); } @@ -167,7 +196,8 @@ int ZmqSocket::GetReceiveHighWaterMark() { if (zmq_getsockopt(sockfd.socketDescriptor, ZMQ_RCVHWM, &value, &value_size)) { PrintError(); - throw ZmqSocketError("Could not get ZMQ_RCVHWM"); + throw ZmqSocketError("Could not get ZMQ_RCVHWM for " + + sockfd.serverAddress); } return value; } @@ -176,10 +206,13 @@ void ZmqSocket::SetReceiveHighWaterMark(int limit) { if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_RCVHWM, &limit, sizeof(limit))) { PrintError(); - throw ZmqSocketError("Could not set ZMQ_RCVHWM"); + throw ZmqSocketError("Could not set ZMQ_RCVHWM for " + + sockfd.serverAddress + " to " + + std::to_string(limit)); } if (GetReceiveHighWaterMark() != limit) { - throw ZmqSocketError("Could not set ZMQ_RCVHWM to " + + throw ZmqSocketError("Could not set ZMQ_RCVHWM for " + + sockfd.serverAddress + " to " + std::to_string(limit)); } int bufsize = DEFAULT_ZMQ_BUFFERSIZE; @@ -195,7 +228,8 @@ int ZmqSocket::GetSendBuffer() { if (zmq_getsockopt(sockfd.socketDescriptor, ZMQ_SNDBUF, &value, &value_size)) { PrintError(); - throw ZmqSocketError("Could not get ZMQ_SNDBUF"); + throw ZmqSocketError("Could not get ZMQ_SNDBUF for " + + sockfd.serverAddress); } return value; } @@ -204,10 +238,13 @@ void ZmqSocket::SetSendBuffer(int limit) { if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_SNDBUF, &limit, sizeof(limit))) { PrintError(); - throw ZmqSocketError("Could not set ZMQ_SNDBUF"); + throw ZmqSocketError("Could not set ZMQ_SNDBUF for " + + sockfd.serverAddress + " to " + + std::to_string(limit)); } if (GetSendBuffer() != limit) { - throw ZmqSocketError("Could not set ZMQ_SNDBUF to " + + throw ZmqSocketError("Could not set ZMQ_SNDBUF for " + + sockfd.serverAddress + " to " + std::to_string(limit)); } } @@ -218,7 +255,8 @@ int ZmqSocket::GetReceiveBuffer() { if (zmq_getsockopt(sockfd.socketDescriptor, ZMQ_RCVBUF, &value, &value_size)) { PrintError(); - throw ZmqSocketError("Could not get ZMQ_RCVBUF"); + throw ZmqSocketError("Could not get ZMQ_RCVBUF for " + + sockfd.serverAddress); } return value; } @@ -227,10 +265,13 @@ void ZmqSocket::SetReceiveBuffer(int limit) { if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_RCVBUF, &limit, sizeof(limit))) { PrintError(); - throw ZmqSocketError("Could not set ZMQ_RCVBUF"); + throw ZmqSocketError("Could not set ZMQ_RCVBUF for " + + sockfd.serverAddress + " to " + + std::to_string(limit)); } if (GetReceiveBuffer() != limit) { - throw ZmqSocketError("Could not set ZMQ_RCVBUF to " + + throw ZmqSocketError("Could not set ZMQ_RCVBUF for " + + sockfd.serverAddress + " to " + std::to_string(limit)); } } @@ -242,12 +283,14 @@ void ZmqSocket::Rebind() { // unbbind if (zmq_unbind(sockfd.socketDescriptor, sockfd.serverAddress.c_str())) { PrintError(); - throw ZmqSocketError("Could not unbind socket"); + throw ZmqSocketError("Could not unbind socket for " + + sockfd.serverAddress); } // bind address if (zmq_bind(sockfd.socketDescriptor, sockfd.serverAddress.c_str())) { PrintError(); - throw ZmqSocketError("Could not bind socket"); + throw ZmqSocketError("Could not bind socket for " + + sockfd.serverAddress); } } diff --git a/tests/scripts/test_frame_synchronizer.py b/tests/scripts/test_frame_synchronizer.py index 281d3a587..b1ef3cf63 100644 --- a/tests/scripts/test_frame_synchronizer.py +++ b/tests/scripts/test_frame_synchronizer.py @@ -8,7 +8,6 @@ import sys, time import traceback, json from slsdet import Detector -from slsdet.defines import DEFAULT_TCP_RX_PORTNO from utils_for_test import ( Log, @@ -22,7 +21,8 @@ from utils_for_test import ( loadBasicSettings, ParseArguments, build_dir, - optional_file + optional_file, + RX_START_TCP_PORTNO ) LOG_PREFIX_FNAME = '/tmp/slsFrameSynchronizer_test' @@ -44,9 +44,7 @@ def startFrameSynchronizerPullSocket(name, fp, no_log_file = False, quiet_mode=F def startFrameSynchronizer(num_mods, fp, no_log_file = False, quiet_mode=False): - cmd = [str(build_dir / 'slsFrameSynchronizer'), str(DEFAULT_TCP_RX_PORTNO), str(num_mods)] - # in 10.0.0 - #cmd = ['slsFrameSynchronizer', '-p', str(DEFAULT_TCP_RX_PORTNO), '-n', str(num_mods)] + cmd = [str(build_dir / 'slsFrameSynchronizer'), '-p', str(RX_START_TCP_PORTNO), '-n', str(num_mods)] fname = SYNCHRONIZER_SUFFIX_FNAME if no_log_file: fname = None diff --git a/tests/scripts/test_simulators.py b/tests/scripts/test_simulators.py index a7f211d12..7fbf0671e 100644 --- a/tests/scripts/test_simulators.py +++ b/tests/scripts/test_simulators.py @@ -14,7 +14,6 @@ import sys, subprocess, time, traceback from contextlib import contextmanager from slsdet import Detector -from slsdet.defines import DEFAULT_TCP_RX_PORTNO from utils_for_test import ( diff --git a/tests/scripts/utils_for_test.py b/tests/scripts/utils_for_test.py index 14c90e196..6ca2847d4 100644 --- a/tests/scripts/utils_for_test.py +++ b/tests/scripts/utils_for_test.py @@ -12,8 +12,10 @@ from datetime import timedelta from contextlib import contextmanager from slsdet import Detector, Ctb, detectorSettings, burstMode -from slsdet.defines import DEFAULT_TCP_RX_PORTNO, DEFAULT_UDP_DST_PORTNO -SERVER_START_PORTNO=1900 +from slsdet.defines import DEFAULT_UDP_DST_PORTNO +DET_START_TCP_PORTNO=1900 +RX_START_TCP_PORTNO=2000 +START_ZMQ_PORTNO=8000 LOG_PREFIX_FNAME = "/tmp/slsDetectorPackage_" @@ -230,9 +232,9 @@ def runProcess(name, cmd, fp, log_file_name = None, quiet_mode=False): def startDetectorVirtualServer(name :str, num_mods, fp, no_log_file = False, quiet_mode=False): for i in range(num_mods): - port_no = SERVER_START_PORTNO + (i * 2) + port_no = DET_START_TCP_PORTNO + (i * 2) cmd = [str(build_dir / (name + 'DetectorServer_virtual')), '-p', str(port_no)] - fname = LOG_PREFIX_FNAME + "virtual_det_" + name + "_" + str(SERVER_START_PORTNO) + ".txt" + fname = LOG_PREFIX_FNAME + "virtual_det_" + name + "_" + str(DET_START_TCP_PORTNO) + ".txt" if no_log_file: fname = None startProcessInBackground(cmd, fp, fname, quiet_mode) @@ -257,7 +259,7 @@ def connectToVirtualServers(name, num_mods, ctb_object=False): counts_sec = 5 while (counts_sec != 0): try: - d.virtual = [num_mods, SERVER_START_PORTNO] # sets the hostnames + d.virtual = [num_mods, DET_START_TCP_PORTNO] # sets the hostnames break except Exception as e: # stop server still not up, wait a bit longer @@ -272,13 +274,11 @@ def connectToVirtualServers(name, num_mods, ctb_object=False): def startReceiver(num_mods, fp, no_log_file = False, quiet_mode=False): if num_mods == 1: - cmd = [str(build_dir / 'slsReceiver')] + cmd = [str(build_dir / 'slsReceiver'), '-p', str(RX_START_TCP_PORTNO)] fname = LOG_PREFIX_FNAME + "slsReceiver.txt" else: - cmd = [str(build_dir / 'slsMultiReceiver'), str(DEFAULT_TCP_RX_PORTNO), str(num_mods)] + cmd = [str(build_dir / 'slsMultiReceiver'), '-p', str(RX_START_TCP_PORTNO), '-n', str(num_mods)] fname = LOG_PREFIX_FNAME + "slsMultiReceiver.txt" - # in 10.0.0 - #cmd = ['slsMultiReceiver', '-p', str(DEFAULT_TCP_RX_PORTNO), '-n', str(num_mods)] if no_log_file: fname = None @@ -300,6 +300,7 @@ def loadConfig(name, rx_hostname = 'localhost', settingsdir = None, log_file_fp if d.numinterfaces == 2: d.udp_dstport2 = DEFAULT_UDP_DST_PORTNO + 1 + d.rx_tcpport = RX_START_TCP_PORTNO d.rx_hostname = rx_hostname d.udp_dstip = 'auto' if name != "eiger": @@ -308,6 +309,9 @@ def loadConfig(name, rx_hostname = 'localhost', settingsdir = None, log_file_fp if num_interfaces == 2: d.udp_dstip2 = 'auto' + d.zmqport = START_ZMQ_PORTNO + d.rx_zmqport = START_ZMQ_PORTNO + if name == "jungfrau" or name == "moench": d.powerchip = 1