diff --git a/.github/workflows/cmake.yaml b/.github/workflows/cmake.yaml index b90214279..2b3512644 100644 --- a/.github/workflows/cmake.yaml +++ b/.github/workflows/cmake.yaml @@ -1,6 +1,6 @@ name: Native CMake Build -on: [push, pull_request] +on: [push, pull_request, workflow_dispatch] env: # Customize the CMake build type here (Release, Debug, RelWithDebInfo, etc.) 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/slsDetectorServers/ctbDetectorServer/Makefile b/slsDetectorServers/ctbDetectorServer/Makefile index 5cd463f62..f09990ae2 100755 --- a/slsDetectorServers/ctbDetectorServer/Makefile +++ b/slsDetectorServers/ctbDetectorServer/Makefile @@ -6,6 +6,7 @@ main_src = ../slsDetectorServer/src/ support_lib = ../../slsSupportLib/include/ det_lib = ../../slsDetectorSoftware/include/sls/ md5_dir = ../../slsSupportLib/src/ +script_dir = ../../etc/ CROSS = bfin-uclinux- CC = $(CROSS)gcc @@ -28,7 +29,7 @@ boot: $(OBJS) version_name=APICTB version_path=slsDetectorServers/ctbDetectorServer versioning: - cd ../../ && echo $(PWD) && echo `tput setaf 6; python etc/updateAPIVersion.py $(version_name) $(version_path); tput sgr0;` + echo `tput setaf 6; python $(script_dir)updateAPIVersion.py $(version_name) $(version_path); tput sgr0;` $(PROGS): $(OBJS) diff --git a/slsDetectorServers/eigerDetectorServer/Makefile b/slsDetectorServers/eigerDetectorServer/Makefile index 950547c7d..c582445f7 100755 --- a/slsDetectorServers/eigerDetectorServer/Makefile +++ b/slsDetectorServers/eigerDetectorServer/Makefile @@ -5,6 +5,7 @@ main_inc = ../slsDetectorServer/include/ main_src = ../slsDetectorServer/src/ support_lib = ../../slsSupportLib/include/ md5_dir = ../../slsSupportLib/src/ +script_dir = ../../etc/ BLACKFIN_CC = bfin-uclinux-gcc CROSS = powerpc-4xx-softfloat- @@ -28,7 +29,7 @@ boot: $(OBJS) version_name=APIEIGER version_path=slsDetectorServers/eigerDetectorServer versioning: - cd ../../ && echo $(PWD) && echo `tput setaf 6; python etc/updateAPIVersion.py $(version_name) $(version_path); tput sgr0;` + echo `tput setaf 6; python $(script_dir)updateAPIVersion.py $(version_name) $(version_path); tput sgr0;` $(PROGS): $(OBJS) diff --git a/slsDetectorServers/eigerDetectorServer/bin/eigerDetectorServer_developer b/slsDetectorServers/eigerDetectorServer/bin/eigerDetectorServer_developer index a089407f9..aa7de8654 100755 Binary files a/slsDetectorServers/eigerDetectorServer/bin/eigerDetectorServer_developer and b/slsDetectorServers/eigerDetectorServer/bin/eigerDetectorServer_developer differ diff --git a/slsDetectorServers/gotthard2DetectorServer/Makefile b/slsDetectorServers/gotthard2DetectorServer/Makefile index 30c71d318..47659ccbe 100755 --- a/slsDetectorServers/gotthard2DetectorServer/Makefile +++ b/slsDetectorServers/gotthard2DetectorServer/Makefile @@ -5,6 +5,7 @@ main_inc = ../slsDetectorServer/include/ main_src = ../slsDetectorServer/src/ support_lib = ../../slsSupportLib/include/ md5_dir = ../../slsSupportLib/src/ +script_dir = ../../etc/ CROSS = nios2-buildroot-linux-gnu- CC = $(CROSS)gcc @@ -27,7 +28,7 @@ boot: $(OBJS) version_name=APIGOTTHARD2 version_path=slsDetectorServers/gotthard2DetectorServer versioning: - cd ../../ && echo $(PWD) && echo `tput setaf 6; python etc/updateAPIVersion.py $(version_name) $(version_path); tput sgr0;` + echo `tput setaf 6; python $(script_dir)updateAPIVersion.py $(version_name) $(version_path); tput sgr0;` $(PROGS): $(OBJS) diff --git a/slsDetectorServers/jungfrauDetectorServer/Makefile b/slsDetectorServers/jungfrauDetectorServer/Makefile index cc92b99c6..b131b6340 100755 --- a/slsDetectorServers/jungfrauDetectorServer/Makefile +++ b/slsDetectorServers/jungfrauDetectorServer/Makefile @@ -5,6 +5,7 @@ main_inc = ../slsDetectorServer/include/ main_src = ../slsDetectorServer/src/ support_lib = ../../slsSupportLib/include/ md5_dir = ../../slsSupportLib/src/ +script_dir = ../../etc/ CROSS = bfin-uclinux- CC = $(CROSS)gcc @@ -27,7 +28,7 @@ boot: $(OBJS) version_name=APIJUNGFRAU version_path=slsDetectorServers/jungfrauDetectorServer versioning: - cd ../../ && echo $(PWD) && echo `tput setaf 6; python etc/updateAPIVersion.py $(version_name) $(version_path); tput sgr0;` + echo `tput setaf 6; python $(script_dir)updateAPIVersion.py $(version_name) $(version_path); tput sgr0;` $(PROGS): $(OBJS) diff --git a/slsDetectorServers/moenchDetectorServer/Makefile b/slsDetectorServers/moenchDetectorServer/Makefile index 2637c4e5d..9578f2139 100755 --- a/slsDetectorServers/moenchDetectorServer/Makefile +++ b/slsDetectorServers/moenchDetectorServer/Makefile @@ -5,6 +5,7 @@ main_inc = ../slsDetectorServer/include/ main_src = ../slsDetectorServer/src/ support_lib = ../../slsSupportLib/include/ md5_dir = ../../slsSupportLib/src/ +script_dir = ../../etc/ CROSS = bfin-uclinux- CC = $(CROSS)gcc @@ -27,7 +28,7 @@ boot: $(OBJS) version_name=APIMOENCH version_path=slsDetectorServers/moenchDetectorServer versioning: - cd ../../ && echo $(PWD) && echo `tput setaf 6; python etc/updateAPIVersion.py $(version_name) $(version_path); tput sgr0;` + echo `tput setaf 6; python $(script_dir)updateAPIVersion.py $(version_name) $(version_path); tput sgr0;` $(PROGS): $(OBJS) diff --git a/slsDetectorServers/mythen3DetectorServer/Makefile b/slsDetectorServers/mythen3DetectorServer/Makefile index 0b342191c..a08c71cde 100755 --- a/slsDetectorServers/mythen3DetectorServer/Makefile +++ b/slsDetectorServers/mythen3DetectorServer/Makefile @@ -6,6 +6,7 @@ main_src = ../slsDetectorServer/src/ support_lib = ../../slsSupportLib/include/ det_lib = ../../slsDetectorSoftware/include/sls/ md5_dir = ../../slsSupportLib/src/ +script_dir = ../../etc/ CROSS = nios2-buildroot-linux-gnu- CC = $(CROSS)gcc @@ -28,7 +29,7 @@ boot: $(OBJS) version_name=APIMYTHEN3 version_path=slsDetectorServers/mythen3DetectorServer versioning: - cd ../../ && echo $(PWD) && echo `tput setaf 6; python etc/updateAPIVersion.py $(version_name) $(version_path); tput sgr0;` + echo `tput setaf 6; python $(script_dir)updateAPIVersion.py $(version_name) $(version_path); tput sgr0;` $(PROGS): $(OBJS) diff --git a/slsDetectorServers/slsDetectorServer/src/slsDetectorServer_funcs.c b/slsDetectorServers/slsDetectorServer/src/slsDetectorServer_funcs.c index 0359bb467..0a5bb400a 100644 --- a/slsDetectorServers/slsDetectorServer/src/slsDetectorServer_funcs.c +++ b/slsDetectorServers/slsDetectorServer/src/slsDetectorServer_funcs.c @@ -1179,12 +1179,20 @@ int processDACEnums(enum dacIndex ind, int val, bool mV) { if (val != GET_FLAG) { ret = setDAC(serverDacIndex, val, mV, mess); // handle if set by user individually - if (serverDacIndex == E_VCMP_LL || serverDacIndex == E_VCMP_LR || - serverDacIndex == E_VCMP_RL || serverDacIndex == E_VCMP_RR || - serverDacIndex == E_VRPREAMP || serverDacIndex == E_VCP) { + switch (serverDacIndex) { + case E_VCMP_LL: + case E_VCMP_LR: + case E_VCMP_RL: + case E_VCMP_RR: + case E_VTHRESHOLD: + case E_VRPREAMP: + case E_VCP: setSettings(UNDEFINED, mess); LOG(logERROR, ("Settings has been changed " "to undefined (changed specific dacs)\n")); + break; + default: + break; } } // get diff --git a/slsDetectorServers/xilinx_ctbDetectorServer/Makefile b/slsDetectorServers/xilinx_ctbDetectorServer/Makefile index 68609daa0..591b63095 100755 --- a/slsDetectorServers/xilinx_ctbDetectorServer/Makefile +++ b/slsDetectorServers/xilinx_ctbDetectorServer/Makefile @@ -6,6 +6,7 @@ main_src = ../slsDetectorServer/src/ support_lib = ../../slsSupportLib/include/ det_lib = ../../slsDetectorSoftware/include/sls/ md5_dir = ../../slsSupportLib/src/ +script_dir = ../../etc/ ifeq ($(shell uname -m),aarch64) # no cross compilation needed when on aarch64 @@ -39,7 +40,7 @@ boot: $(OBJS) version_name=APIXILINXCTB version_path=slsDetectorServers/xilinx_ctbDetectorServer versioning: - cd ../../ && echo $(PWD) && echo `tput setaf 6; python etc/updateAPIVersion.py $(version_name) $(version_path); tput sgr0;` + echo `tput setaf 6; python $(script_dir)updateAPIVersion.py $(version_name) $(version_path); tput sgr0;` $(PROGS): $(OBJS) 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..6be175391 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, @@ -1695,7 +1670,9 @@ void DetectorImpl::updateRxUDPDatastreamMetadata() { } } - modules[0]->updateRxUDPPortDisableMetadata(disable); + if (modules[0]->getUseReceiverFlag()) { + modules[0]->updateRxUDPPortDisableMetadata(disable); + } } std::vector DetectorImpl::getRxDisabledUDPPortIndices() const { diff --git a/slsDetectorSoftware/src/Module.cpp b/slsDetectorSoftware/src/Module.cpp index 64be1d6e7..4c487d6a6 100644 --- a/slsDetectorSoftware/src/Module.cpp +++ b/slsDetectorSoftware/src/Module.cpp @@ -196,6 +196,7 @@ void Module::setSettings(detectorSettings isettings) { "Cannot set settings for Eiger. Use threshold energy."); } sendToDetector(F_SET_SETTINGS, isettings); + updateRxThresholdEnergyMetadata(); } int Module::getThresholdEnergy() const { @@ -414,9 +415,14 @@ void Module::setAllThresholdEnergy(std::array e_eV, throw RuntimeError("setThresholdEnergyAndSettings: Could not set " "settings in Module"); } +} +void Module::updateRxThresholdEnergyMetadata() { if (shm()->useReceiverFlag) { - sendToReceiver(F_RECEIVER_SET_ALL_THRESHOLD, e_eV, nullptr); + if (shm()->detType == MYTHEN3) { + auto e_eV = getAllThresholdEnergy(); + sendToReceiver(F_RECEIVER_SET_ALL_THRESHOLD, e_eV, nullptr); + } } } @@ -477,6 +483,7 @@ int Module::getAllTrimbits() const { void Module::setAllTrimbits(int val) { sendToDetector(F_SET_ALL_TRIMBITS, val); + updateRxThresholdEnergyMetadata(); } std::vector Module::getTrimEn() const { @@ -541,39 +548,15 @@ void Module::setSynchronization(const bool value) { } std::vector Module::getBadChannels() const { - auto client = DetectorSocket(shm()->hostname, shm()->controlPort); - client.Send(F_GET_BAD_CHANNELS); - client.setFnum(F_GET_BAD_CHANNELS); - if (client.Receive() == FAIL) { - throw DetectorError("Detector " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); + auto retval = sendToDetectorVarVector(F_GET_BAD_CHANNELS); + for (size_t i = 0; i < retval.size(); ++i) { + LOG(logDEBUG1) << i << ":" << retval[i]; } - // receive badchannels - auto nch = client.Receive(); - std::vector badchannels(nch); - if (nch > 0) { - client.Receive(badchannels); - for (size_t i = 0; i < badchannels.size(); ++i) { - LOG(logDEBUG1) << i << ":" << badchannels[i]; - } - } - return badchannels; + return retval; } void Module::setBadChannels(std::vector list) { - auto nch = static_cast(list.size()); - LOG(logDEBUG1) << "Sending bad channels to detector, nch:" << nch; - auto client = DetectorSocket(shm()->hostname, shm()->controlPort); - client.Send(F_SET_BAD_CHANNELS); - client.setFnum(F_SET_BAD_CHANNELS); - client.Send(nch); - if (nch > 0) { - client.Send(list); - } - if (client.Receive() == FAIL) { - throw DetectorError("Detector " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); - } + sendToDetectorVarVector(F_SET_BAD_CHANNELS, list); } int Module::getRow() const { return sendToDetector(F_GET_ROW); } @@ -808,6 +791,7 @@ void Module::setDAC(int val, dacIndex index, bool mV) { } int args[]{static_cast(index), static_cast(mV), val}; sendToDetector(F_SET_DAC, args); + updateRxThresholdEnergyMetadata(); } bool Module::getPowerChip() const { @@ -833,21 +817,17 @@ bool Module::isPowerEnabled(defs::powerIndex index) const { void Module::setPowerEnabled(const std::vector &indices, bool enable) { - auto client = DetectorSocket(shm()->hostname, shm()->controlPort); - client.Send(F_SET_POWER); - client.setFnum(F_SET_POWER); - int count = indices.size(); - client.Send(count); + int count = static_cast(indices.size()); std::vector indices_int(count); - for (size_t i = 0; i < indices.size(); ++i) { + for (int i = 0; i < count; ++i) { indices_int[i] = static_cast(indices[i]); } + // sends a variable vector and an enable + auto client = createDetectorSocket(); + client.sendCommand(F_SET_POWER, &count, sizeof(count)); client.Send(indices_int); client.Send(static_cast(enable)); - if (client.Receive() == FAIL) { - throw DetectorError("Detector " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); - } + client.readReply(nullptr, 0); } int Module::getPowerADC(defs::powerIndex index) const { @@ -1022,72 +1002,15 @@ double Module::getReceiverProgress() const { } std::vector Module::getFramesCaughtByReceiver() const { - // TODO!(Erik) Refactor - LOG(logDEBUG1) << "Getting frames caught"; - if (shm()->useReceiverFlag) { - auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort); - client.Send(F_GET_RECEIVER_FRAMES_CAUGHT); - client.setFnum(F_GET_RECEIVER_FRAMES_CAUGHT); - if (client.Receive() == FAIL) { - throw ReceiverError( - "Receiver " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); - } else { - auto nports = client.Receive(); - std::vector retval(nports); - client.Receive(retval); - LOG(logDEBUG1) << "Frames caught of Receiver" << moduleIndex << ": " - << ToString(retval); - return retval; - } - } - throw RuntimeError("No receiver to get frames caught."); + return sendToReceiverVarVector(F_GET_RECEIVER_FRAMES_CAUGHT); } std::vector Module::getNumMissingPackets() const { - // TODO!(Erik) Refactor - LOG(logDEBUG1) << "Getting num missing packets"; - if (shm()->useReceiverFlag) { - auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort); - client.Send(F_GET_NUM_MISSING_PACKETS); - client.setFnum(F_GET_NUM_MISSING_PACKETS); - if (client.Receive() == FAIL) { - throw ReceiverError( - "Receiver " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); - } else { - auto nports = client.Receive(); - std::vector retval(nports); - client.Receive(retval); - LOG(logDEBUG1) << "Missing packets of Receiver" << moduleIndex - << ": " << ToString(retval); - return retval; - } - } - throw RuntimeError("No receiver to get missing packets."); + return sendToReceiverVarVector(F_GET_NUM_MISSING_PACKETS); } std::vector Module::getReceiverCurrentFrameIndex() const { - // TODO!(Erik) Refactor - LOG(logDEBUG1) << "Getting frame index"; - if (shm()->useReceiverFlag) { - auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort); - client.Send(F_GET_RECEIVER_FRAME_INDEX); - client.setFnum(F_GET_RECEIVER_FRAME_INDEX); - if (client.Receive() == FAIL) { - throw ReceiverError( - "Receiver " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); - } else { - auto nports = client.Receive(); - std::vector retval(nports); - client.Receive(retval); - LOG(logDEBUG1) << "Frame index of Receiver" << moduleIndex << ": " - << ToString(retval); - return retval; - } - } - throw RuntimeError("No receiver to get frame index."); + return sendToReceiverVarVector(F_GET_RECEIVER_FRAME_INDEX); } uint64_t Module::getNextFrameNumber() const { @@ -1435,46 +1358,11 @@ void Module::setUDPDataStream(const portPosition port, const bool enable) { } void Module::updateRxUDPPortDisableMetadata(const std::vector &disable) { - if (!shm()->useReceiverFlag) { - return; - } - LOG(logDEBUG) << "Updating UDP port disable metadata in Receiver 0"; - auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort); - - client.Send(F_RECEIVER_SET_UDP_PORT_DISABLE_META); - client.setFnum(F_RECEIVER_SET_UDP_PORT_DISABLE_META); - - auto nports = static_cast(disable.size()); - client.Send(nports); - if (nports > 0) { - client.Send(disable); - } - if (client.Receive() == FAIL) { - throw ReceiverError("Receiver " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); - } + sendToReceiverVarVector(F_RECEIVER_SET_UDP_PORT_DISABLE_META, disable); } std::vector Module::getRxUDPPortDisableMetadata() const { - if (!shm()->useReceiverFlag) { - throw RuntimeError("No receiver to get disabled udp port indices."); - } - - LOG(logDEBUG) << "Getting UDP port disable metadata in Receiver 0"; - auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort); - - client.Send(F_RECEIVER_GET_UDP_PORT_DISABLE_META); - client.setFnum(F_RECEIVER_GET_UDP_PORT_DISABLE_META); - if (client.Receive() == FAIL) { - throw ReceiverError("Receiver " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); - } - auto nports = client.Receive(); - std::vector retval(nports); - if (nports > 0) { - client.Receive(retval); - } - return retval; + return sendToReceiverVarVector(F_RECEIVER_GET_UDP_PORT_DISABLE_META); } // Receiver Config @@ -1638,93 +1526,39 @@ void Module::setRxArping(bool enable) { } std::vector Module::getRxROI() const { - LOG(logDEBUG1) << "Getting receiver ROI for Module " << moduleIndex; - // check number of ports - if (!shm()->useReceiverFlag) { - throw RuntimeError("No receiver to get ROI."); - } - auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort); - client.Send(F_RECEIVER_GET_RECEIVER_ROI); - client.setFnum(F_RECEIVER_GET_RECEIVER_ROI); - auto nPorts = client.Receive(); - std::vector retval(nPorts); - if (nPorts > 0) - client.Receive(retval); - if (nPorts != shm()->numUDPInterfaces) { + auto retval = sendToReceiverVarVector(F_RECEIVER_GET_RECEIVER_ROI); + auto nPorts = retval.size(); + if (static_cast(nPorts) != shm()->numUDPInterfaces) { throw RuntimeError( "Invalid number of rois: " + std::to_string(nPorts) + ". Expected: " + std::to_string(shm()->numUDPInterfaces)); } - LOG(logDEBUG1) << "ROI of Receiver" << moduleIndex << ": " - << ToString(retval); return retval; } void Module::setRxROI(const std::vector &portRois) { - LOG(logDEBUG) << "Sending to receiver " << moduleIndex - << " [roi: " << ToString(portRois) << ']'; - if (!shm()->useReceiverFlag) { - throw RuntimeError("No receiver to set ROI."); - } if ((int)portRois.size() != shm()->numUDPInterfaces) { throw RuntimeError( "Invalid number of ROIs: " + std::to_string(portRois.size()) + ". Expected: " + std::to_string(shm()->numUDPInterfaces)); } - // check number of ports - auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort); - client.Send(F_RECEIVER_SET_RECEIVER_ROI); - client.setFnum(F_RECEIVER_SET_RECEIVER_ROI); - int size = static_cast(portRois.size()); - client.Send(size); - if (size > 0) - client.Send(portRois); - if (client.Receive() == FAIL) { - throw ReceiverError("Receiver " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); - } + sendToReceiverVarVector(F_RECEIVER_SET_RECEIVER_ROI, portRois); } std::vector Module::getRxROIMetadata() const { - LOG(logDEBUG1) << "Getting receiver ROI metadata for Module " - << moduleIndex; - // check number of ports - if (!shm()->useReceiverFlag) { - throw RuntimeError("No receiver to get ROI metadata."); - } - auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort); - client.Send(F_RECEIVER_GET_ROI_METADATA); - client.setFnum(F_RECEIVER_GET_ROI_METADATA); - auto size = client.Receive(); - std::vector retval(size); - if (size > 0) - client.Receive(retval); - if (size == 0) { - throw RuntimeError("Invalid number of ROI metadata: " + - std::to_string(size) + ". Min: 1."); - } + auto retval = sendToReceiverVarVector(F_RECEIVER_GET_ROI_METADATA); + if (retval.size() == 0) + throw RuntimeError("Invalid number of ROI metadata: 0. Min: 1."); LOG(logDEBUG1) << "ROI metadata of Receiver: " << ToString(retval); return retval; } void Module::setRxROIMetadata(const std::vector &args) { - LOG(logDEBUG) << "Sending to receiver " << moduleIndex - << " [roi metadata: " << ToString(args) << ']'; - auto receiver = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort); - receiver.Send(F_RECEIVER_SET_RECEIVER_ROI_METADATA); - receiver.setFnum(F_RECEIVER_SET_RECEIVER_ROI_METADATA); - int size = static_cast(args.size()); - receiver.Send(size); - if (size > 0) - receiver.Send(args); - if (size < 1) { + if (args.size() < 1) { throw RuntimeError("Invalid number of ROI metadata: " + - std::to_string(size) + ". Min: 1."); - } - if (receiver.Receive() == FAIL) { - throw ReceiverError("Receiver " + std::to_string(moduleIndex) + - " returned error: " + receiver.readErrorMessage()); + std::to_string(args.size()) + ". Min: 1."); } + sendToReceiverVarVector(F_RECEIVER_SET_RECEIVER_ROI_METADATA, args); } // File @@ -1935,17 +1769,7 @@ void Module::setRateCorrection(int64_t t) { } void Module::sendReceiverRateCorrections(const std::vector &t) { - LOG(logDEBUG) << "Sending to receiver 0 [rate corrections: " << ToString(t) - << ']'; - auto receiver = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort); - receiver.Send(F_SET_RECEIVER_RATE_CORRECT); - receiver.setFnum(F_SET_RECEIVER_RATE_CORRECT); - receiver.Send(static_cast(t.size())); - receiver.Send(t); - if (receiver.Receive() == FAIL) { - throw ReceiverError("Receiver " + std::to_string(moduleIndex) + - " returned error: " + receiver.readErrorMessage()); - } + sendToReceiverVarVector(F_SET_RECEIVER_RATE_CORRECT, t); } bool Module::getInterruptSubframe() const { @@ -2189,51 +2013,42 @@ void Module::sendVetoPhoton(const int chipIndex, const std::vector &gainIndices, const std::vector &values) { const int nch = gainIndices.size(); - if (gainIndices.size() != values.size()) { - throw RuntimeError("Number of Gain Indices and values do not match! " - "Gain Indices size: " + - std::to_string(gainIndices.size()) + - ", values size: " + std::to_string(values.size())); + if (nch != static_cast(values.size())) { + std::ostringstream oss; + oss << "Number of Gain Indices and values do not match! Gain Indices " + "size: " + << nch << ", values size: " << values.size(); + throw RuntimeError(oss.str()); } LOG(logDEBUG1) << "Sending veto photon/file to detector [chip:" << chipIndex << ", nch:" << nch << "]"; - const int args[]{chipIndex, nch}; - auto client = DetectorSocket(shm()->hostname, shm()->controlPort); - client.Send(F_SET_VETO_PHOTON); - client.setFnum(F_SET_VETO_PHOTON); - client.Send(args); + auto client = createDetectorSocket(); + client.sendCommand(F_SET_VETO_PHOTON, args, sizeof(args)); client.Send(gainIndices); client.Send(values); - if (client.Receive() == FAIL) { - throw DetectorError("Detector " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); - } + client.readReply(nullptr, 0); } void Module::getVetoPhoton(const int chipIndex, const std::string &fname) const { LOG(logDEBUG1) << "Getting veto photon [" << chipIndex << "]\n"; - auto client = DetectorSocket(shm()->hostname, shm()->controlPort); - client.Send(F_GET_VETO_PHOTON); - client.setFnum(F_GET_VETO_PHOTON); - client.Send(chipIndex); - if (client.Receive() == FAIL) { - throw DetectorError("Detector " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); - } - - auto nch = client.Receive(); - if (nch != shm()->nChan.x) { - throw DetectorError("Could not get veto photon. Expected " + - std::to_string(shm()->nChan.x) + " channels, got " + - std::to_string(nch)); - } + auto client = createDetectorSocket(); + client.sendCommand(F_GET_VETO_PHOTON, &chipIndex, sizeof(chipIndex)); + int nch = 0; + client.readReply(&nch, sizeof(nch)); std::vector gainIndices(nch); std::vector values(nch); client.Receive(gainIndices); client.Receive(values); + if (nch != shm()->nChan.x) { + std::ostringstream oss; + oss << "Could not get veto photon. Expected " << shm()->nChan.x + << " channels, got " << nch; + throw DetectorError(oss.str()); + } + // save to file std::ofstream outfile(fname); if (!outfile) { @@ -2313,10 +2128,10 @@ void Module::setVetoPhoton(const int chipIndex, const int numPhotons, } // check size if ((int)gainIndices.size() != shm()->nChan.x) { - throw RuntimeError("Could not set veto photon. Invalid number of " - "entries in file. Expected " + - std::to_string(shm()->nChan.x) + ", read " + - std::to_string(gainIndices.size())); + std::ostringstream oss; + oss << "Could not set veto photon. Invalid number of entries in file. " + << "Expected " << shm()->nChan.x << ", read " << gainIndices.size(); + throw RuntimeError(oss.str()); } sendVetoPhoton(chipIndex, gainIndices, values); @@ -2733,17 +2548,13 @@ std::string Module::getPatternFileName() const { } void Module::setPattern(const Pattern &pat, const std::string &fname) { - auto client = DetectorSocket(shm()->hostname, shm()->controlPort); - client.Send(F_SET_PATTERN); - client.setFnum(F_SET_PATTERN); - client.Send(pat.data(), pat.size()); char args[MAX_STR_LENGTH]{}; strcpy_safe(args, fname.c_str()); + + auto client = createDetectorSocket(); + client.sendCommand(F_SET_PATTERN, pat.data(), pat.size()); client.Send(args); - if (client.Receive() == FAIL) { - throw DetectorError("Detector " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); - } + client.readReply(nullptr, 0); } Pattern Module::getPattern() { @@ -2846,69 +2657,44 @@ void Module::startPattern() { sendToDetector(F_START_PATTERN); } // Json Header specific std::map Module::getAdditionalJsonHeader() const { - // TODO, refactor this function with a more robust sending. - // Now assuming whitespace separated key value - if (!shm()->useReceiverFlag) { - throw RuntimeError("Set rx_hostname first to use receiver parameters " - "(zmq json header)"); - } - auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort); - client.Send(F_GET_ADDITIONAL_JSON_HEADER); - client.setFnum(F_GET_ADDITIONAL_JSON_HEADER); - if (client.Receive() == FAIL) { - throw ReceiverError("Receiver " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); - } else { - auto size = client.Receive(); - std::string buff(size, '\0'); - std::map retval; - if (size > 0) { - client.Receive(&buff[0], buff.size()); - std::istringstream iss(buff); - std::string key, value; - while (iss >> key) { - iss >> value; - retval[key] = value; - } - } - LOG(logDEBUG) << "Getting additional json header " << ToString(retval); - return retval; + auto vec = sendToReceiverVarVector(F_GET_ADDITIONAL_JSON_HEADER); + + // convert vector to string + std::string longString(vec.begin(), vec.end()); + // convert string (space separated) to map of key-value pairs + std::map retval; + std::istringstream iss(longString); + std::string key, value; + while (iss >> key) { + iss >> value; + retval[key] = value; } + LOG(logDEBUG) << "Getting additional json header " << ToString(retval); + return retval; } void Module::setAdditionalJsonHeader( const std::map &jsonHeader) { - if (!shm()->useReceiverFlag) { - throw RuntimeError("Set rx_hostname first to use receiver parameters " - "(zmq json header)"); - } + // validate for (auto &it : jsonHeader) { - if (it.first.empty() || it.first.length() > SHORT_STR_LENGTH || - it.second.length() > SHORT_STR_LENGTH) { - throw RuntimeError( - it.first + " or " + it.second + - " pair has invalid size. " - "Key cannot be empty. Both can have max 20 characters"); + auto key = it.first; + auto value = it.second; + if (key.empty() || key.length() > SHORT_STR_LENGTH || + value.length() > SHORT_STR_LENGTH) { + throw RuntimeError(key + " or " + value + + " pair has invalid size. Key cannot be empty. " + "Both can have max 20 characters"); } } + // convert map to string (space separated) std::ostringstream oss; for (auto &it : jsonHeader) oss << it.first << ' ' << it.second << ' '; - auto buff = oss.str(); - const auto size = static_cast(buff.size()); - LOG(logDEBUG) << "Sending to receiver additional json header " - << ToString(jsonHeader); - auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort); - client.Send(F_SET_ADDITIONAL_JSON_HEADER); - client.setFnum(F_SET_ADDITIONAL_JSON_HEADER); - client.Send(size); - if (size > 0) - client.Send(&buff[0], buff.size()); + auto longString = oss.str(); + // convert string to vector to send over socket (variable vector impl) + std::vector args(longString.begin(), longString.end()); - if (client.Receive() == FAIL) { - throw ReceiverError("Receiver " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); - } + sendToReceiverVarVector(F_SET_ADDITIONAL_JSON_HEADER, args); } std::string Module::getAdditionalJsonParameter(const std::string &key) const { @@ -3090,23 +2876,12 @@ IpAddr Module::getLastClientIP() const { } std::string Module::executeCommand(const std::string &cmd) { - char arg[MAX_STR_LENGTH]{}; - char retval[MAX_STR_LENGTH]{}; - strcpy_safe(arg, cmd.c_str()); LOG(logINFO) << "Module " << moduleIndex << " (" << shm()->hostname << "): Sending command " << cmd; - auto client = DetectorSocket(shm()->hostname, shm()->controlPort); - client.Send(F_EXEC_COMMAND); - client.setFnum(F_EXEC_COMMAND); - client.Send(arg); - if (client.Receive() == FAIL) { - std::cout << '\n'; - std::ostringstream os; - os << "Module " << moduleIndex << " (" << shm()->hostname << ")" - << " returned error: " << client.readErrorMessage(); - throw DetectorError(os.str()); - } - client.Receive(retval); + char args[MAX_STR_LENGTH]{}; + char retval[MAX_STR_LENGTH]{}; + strcpy_safe(args, cmd.c_str()); + sendToDetector(F_EXEC_COMMAND, args, retval); LOG(logINFO) << "Module " << moduleIndex << " (" << shm()->hostname << "): command executed"; return retval; @@ -3126,14 +2901,18 @@ int64_t Module::getMeasurementTime() const { // private -void Module::checkArgs(const void *args, size_t args_size, void *retval, - size_t retval_size) const { +void Module::checkArgs(const void *args, size_t args_size) const { if (args == nullptr && args_size != 0) throw RuntimeError( "Passed nullptr as args to Send function but size is not 0"); if (args != nullptr && args_size == 0) throw RuntimeError( "Passed size 0 to Send function but args is not nullptr"); +} + +void Module::checkArgs(const void *args, size_t args_size, void *retval, + size_t retval_size) const { + checkArgs(args, args_size); if (retval == nullptr && retval_size != 0) throw RuntimeError( "Passed nullptr as retval to Send function but size is not 0"); @@ -3150,13 +2929,42 @@ void Module::checkArgs(const void *args, size_t args_size, void *retval, static_assert(!std::is_same::value, \ "nullptr_t type is incompatible with templated " DST); +DetectorSocket Module::createDetectorSocket() const { + return DetectorSocket(shm()->hostname, shm()->controlPort); +} + +template +void Module::sendToDetectorVarVector(int fnum, + const std::vector &args) const { + LOG(logDEBUG1) << "Sending to Detector: [" + << getFunctionNameFromEnum(static_cast(fnum)) + << ", std::vector<" << typeid(Arg).name() << ">, nullptr, 0," + << "]"; + auto client = createDetectorSocket(); + client.sendCommandVariableSize(fnum, args, nullptr, 0); + client.close(); +} + +template +std::vector Module::sendToDetectorVarVector(int fnum) const { + LOG(logDEBUG1) << "Sending to Detector: [" + << getFunctionNameFromEnum(static_cast(fnum)) + << ", nullptr, 0, std::vector<" << typeid(Ret).name() + << ">]"; + std::vector retval; + auto client = createDetectorSocket(); + client.sendCommandVariableSize(fnum, nullptr, 0, retval); + client.close(); + return retval; +} + void Module::sendToDetector(int fnum, const void *args, size_t args_size, void *retval, size_t retval_size) const { // This is the only function that actually sends data to the detector // the other versions use templates to deduce sizes and create // the return type checkArgs(args, args_size, retval, retval_size); - auto client = DetectorSocket(shm()->hostname, shm()->controlPort); + auto client = createDetectorSocket(); client.sendCommandThenRead(fnum, args, args_size, retval, retval_size); client.close(); } @@ -3379,6 +3187,47 @@ Ret Module::sendToDetectorStop(int fnum, const Arg &args) { //-------------------------------------------------------------- sendToReceiver +ReceiverSocket Module::createReceiverSocket() const { + return ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort); +} + +template +void Module::sendToReceiverVarVector(int fnum, + const std::vector &args) const { + LOG(logDEBUG1) << "Sending to Receiver: [" + << getFunctionNameFromEnum(static_cast(fnum)) + << ", std::vector<" << typeid(Arg).name() << ">, nullptr, 0," + << "]"; + if (!shm()->useReceiverFlag) { + std::ostringstream oss; + oss << "Set rx_hostname first to use receiver parameters, "; + oss << getFunctionNameFromEnum(static_cast(fnum)); + throw RuntimeError(oss.str()); + } + auto receiver = createReceiverSocket(); + receiver.sendCommandVariableSize(fnum, args, nullptr, 0); + receiver.close(); +} + +template +std::vector Module::sendToReceiverVarVector(int fnum) const { + LOG(logDEBUG1) << "Sending to Receiver: [" + << getFunctionNameFromEnum(static_cast(fnum)) + << ", nullptr, 0, std::vector<" << typeid(Ret).name() + << ">]"; + if (!shm()->useReceiverFlag) { + std::ostringstream oss; + oss << "Set rx_hostname first to use receiver parameters, "; + oss << getFunctionNameFromEnum(static_cast(fnum)); + throw RuntimeError(oss.str()); + } + std::vector retval; + auto receiver = createReceiverSocket(); + receiver.sendCommandVariableSize(fnum, nullptr, 0, retval); + receiver.close(); + return retval; +} + void Module::sendToReceiver(int fnum, const void *args, size_t args_size, void *retval, size_t retval_size) const { // This is the only function that actually sends data to the receiver @@ -3391,7 +3240,7 @@ void Module::sendToReceiver(int fnum, const void *args, size_t args_size, throw RuntimeError(oss.str()); } checkArgs(args, args_size, retval, retval_size); - auto receiver = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort); + auto receiver = createReceiverSocket(); receiver.sendCommandThenRead(fnum, args, args_size, retval, retval_size); receiver.close(); } @@ -3707,26 +3556,19 @@ void Module::setModule(sls_detector_module &module, bool trimbits) { "these have been replaced with 0/200 or 2800/2400."; } } - auto client = DetectorSocket(shm()->hostname, shm()->controlPort); - client.Send(F_SET_MODULE); - client.setFnum(F_SET_MODULE); + auto client = createDetectorSocket(); + client.sendCommand(F_SET_MODULE, nullptr, 0); sendModule(&module, client); - if (client.Receive() == FAIL) { - throw DetectorError("Module " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); - } + client.readReply(nullptr, 0); + updateRxThresholdEnergyMetadata(); } sls_detector_module Module::getModule() { LOG(logDEBUG1) << "Getting module"; sls_detector_module module(shm()->detType); - auto client = DetectorSocket(shm()->hostname, shm()->controlPort); - client.Send(F_GET_MODULE); - client.setFnum(F_GET_MODULE); - if (client.Receive() == FAIL) { - throw DetectorError("Module " + std::to_string(moduleIndex) + - " returned error: " + client.readErrorMessage()); - } + auto client = createDetectorSocket(); + client.sendCommand(F_GET_MODULE, nullptr, 0); + client.readReply(nullptr, 0); receiveModule(&module, client); return module; } @@ -4063,7 +3905,7 @@ void Module::sendProgram(bool blackfin, std::vector buffer, << "): Sending " << functionType; // send fnum and filesize - auto client = DetectorSocket(shm()->hostname, shm()->controlPort); + auto client = createDetectorSocket(); client.Send(functionEnum); uint64_t filesize = buffer.size(); client.Send(filesize); @@ -4187,47 +4029,27 @@ void Module::simulatingActivityinDetector(const std::string &functionType, std::vector Module::readSpi(int chip_id, int register_id, int n_bytes) const { - auto client = DetectorSocket(shm()->hostname, shm()->controlPort); - client.Send(F_SPI_READ); - client.setFnum(F_SPI_READ); - client.Send(chip_id); - client.Send(register_id); - client.Send(n_bytes); + auto client = createDetectorSocket(); + int args[] = {chip_id, register_id, n_bytes}; + client.sendCommand(F_SPI_READ, &args, sizeof(args)); - if (client.Receive() == FAIL) { - std::ostringstream os; - os << "Module " << moduleIndex << " (" << shm()->hostname << ")" - << " returned error: " << client.readErrorMessage(); - throw DetectorError(os.str()); - } - - std::vector data(n_bytes); - client.Receive(data); - return data; + std::vector retval(n_bytes); + client.readReply(retval.data(), retval.size()); + return retval; } std::vector Module::writeSpi(int chip_id, int register_id, const std::vector &data) { - auto client = DetectorSocket(shm()->hostname, shm()->controlPort); - client.Send(F_SPI_WRITE); - client.setFnum(F_SPI_WRITE); - client.Send(chip_id); - client.Send(register_id); - client.Send(static_cast(data.size())); + int count = static_cast(data.size()); + auto client = createDetectorSocket(); + int args[] = {chip_id, register_id, count}; + client.sendCommand(F_SPI_WRITE, &args, sizeof(args)); client.Send(data); - - if (client.Receive() == FAIL) { - std::ostringstream os; - os << "Module " << moduleIndex << " (" << shm()->hostname << ")" - << " returned error: " << client.readErrorMessage(); - throw DetectorError(os.str()); - } - // Read the output from the SPI write. This contains the data before the - // write. - std::vector ret(data.size()); - client.Receive(ret); - return ret; + // write + std::vector retval(data.size()); + client.readReply(retval.data(), retval.size()); + return retval; } } // namespace sls diff --git a/slsDetectorSoftware/src/Module.h b/slsDetectorSoftware/src/Module.h index 1f5097530..49d7812ac 100644 --- a/slsDetectorSoftware/src/Module.h +++ b/slsDetectorSoftware/src/Module.h @@ -111,6 +111,7 @@ class Module : public virtual slsDetectorDefs { void setAllThresholdEnergy(std::array e_eV, detectorSettings isettings, bool trimbits); std::string getSettingsDir() const; + void updateRxThresholdEnergyMetadata(); std::string setSettingsDir(const std::string &dir); void loadTrimbits(const std::string &fname); void saveTrimbits(const std::string &fname); @@ -626,9 +627,18 @@ class Module : public virtual slsDetectorDefs { private: std::string getReceiverLongVersion() const; + void checkArgs(const void *args, size_t args_size) const; void checkArgs(const void *args, size_t args_size, void *retval, size_t retval_size) const; + DetectorSocket createDetectorSocket() const; + + template + void sendToDetectorVarVector(int fnum, const std::vector &args) const; + + template + std::vector sendToDetectorVarVector(int fnum) const; + /** * Send function parameters to detector (control server) * @param fnum function enum @@ -674,6 +684,9 @@ class Module : public virtual slsDetectorDefs { Ret sendToDetector(int fnum, const Arg &args) const; /** Send function parameters to detector (stop server) */ + + DetectorSocket createDetectorStopSocket() const; + void sendToDetectorStop(int fnum, const void *args, size_t args_size, void *retval, size_t retval_size); @@ -713,6 +726,15 @@ class Module : public virtual slsDetectorDefs { Ret sendToDetectorStop(int fnum, const Arg &args) const; /** Send function parameters to receiver */ + + ReceiverSocket createReceiverSocket() const; + + template + void sendToReceiverVarVector(int fnum, const std::vector &args) const; + + template + std::vector sendToReceiverVarVector(int fnum) const; + void sendToReceiver(int fnum, const void *args, size_t args_size, void *retval, size_t retval_size); diff --git a/slsDetectorSoftware/tests/Caller/test-Caller-rx-running.cpp b/slsDetectorSoftware/tests/Caller/test-Caller-rx-running.cpp index 85ab76599..68fade955 100644 --- a/slsDetectorSoftware/tests/Caller/test-Caller-rx-running.cpp +++ b/slsDetectorSoftware/tests/Caller/test-Caller-rx-running.cpp @@ -236,7 +236,6 @@ TEST_CASE("general - cant put if receiver is not idle", det.setAcquisitionIndex(prev_findex[i], {i}); det.setFileWrite(prev_fwrite[i], {i}); det.setRxFifoDepth(prev_fifodepth[i], {i}); - det.setRxHostname(rx_hostname[i], {i}); } } } diff --git a/slsReceiverSoftware/src/ClientInterface.cpp b/slsReceiverSoftware/src/ClientInterface.cpp index 4b14004c7..9cc94ef3b 100644 --- a/slsReceiverSoftware/src/ClientInterface.cpp +++ b/slsReceiverSoftware/src/ClientInterface.cpp @@ -831,32 +831,17 @@ int ClientInterface::get_file_index(Interface &socket) { int ClientInterface::get_frame_index(Interface &socket) { auto retval = impl()->getCurrentFrameIndex(); - LOG(logDEBUG1) << "frames index:" << ToString(retval); - auto size = static_cast(retval.size()); - socket.Send(OK); - socket.Send(size); - socket.Send(retval); - return OK; + return socket.sendVariableResult(retval); } int ClientInterface::get_missing_packets(Interface &socket) { auto missing_packets = impl()->getNumMissingPackets(); - LOG(logDEBUG1) << "missing packets:" << ToString(missing_packets); - auto size = static_cast(missing_packets.size()); - socket.Send(OK); - socket.Send(size); - socket.Send(missing_packets); - return OK; + return socket.sendVariableResult(missing_packets); } int ClientInterface::get_frames_caught(Interface &socket) { auto retval = impl()->getFramesCaught(); - LOG(logDEBUG1) << "frames caught:" << ToString(retval); - auto size = static_cast(retval.size()); - socket.Send(OK); - socket.Send(size); - socket.Send(retval); - return OK; + return socket.sendVariableResult(retval); } int ClientInterface::set_file_write(Interface &socket) { @@ -1113,37 +1098,39 @@ int ClientInterface::stream_rx_dummy_header(Interface &socket) { } int ClientInterface::set_additional_json_header(Interface &socket) { - std::map json; - auto size = socket.Receive(); - if (size > 0) { - std::string buff(size, '\0'); - socket.Receive(&buff[0], buff.size()); - std::istringstream iss(buff); - std::string key, value; - while (iss >> key) { - iss >> value; - json[key] = value; - } - } // verifyIdle(socket); allowing it to be set on the fly - LOG(logDEBUG1) << "Setting additional json header: " << ToString(json); - impl()->setAdditionalJsonHeader(json); + + auto vec = socket.receiveVariableArgs>(); + // convert vector to string (space separated) + std::string longString(vec.begin(), vec.end()); + // convert string to map of key-value pairs + std::map args; + std::istringstream iss(longString); + std::string key, value; + while (iss >> key) { + iss >> value; + args[key] = value; + } + LOG(logDEBUG1) << "Setting additional json header: " << ToString(args); + + impl()->setAdditionalJsonHeader(args); return socket.Send(OK); } int ClientInterface::get_additional_json_header(Interface &socket) { std::map json = impl()->getAdditionalJsonHeader(); LOG(logDEBUG1) << "additional json header:" << ToString(json); + + // convert map to string (space separated) to send over socket std::ostringstream oss; for (auto &it : json) { oss << it.first << ' ' << it.second << ' '; } - auto buff = oss.str(); - auto size = static_cast(buff.size()); - socket.sendResult(size); - if (size > 0) - socket.Send(buff); - return OK; + auto longString = oss.str(); + + // conver to vector to send over socket (variable vector length impl) + std::vector retval(longString.begin(), longString.end()); + return socket.sendVariableResult(retval); } int ClientInterface::set_udp_socket_buffer_size(Interface &socket) { @@ -1612,16 +1599,13 @@ int ClientInterface::set_streaming_start_fnum(Interface &socket) { } int ClientInterface::set_rate_correct(Interface &socket) { - auto index = socket.Receive(); - if (index <= 0) { - throw RuntimeError("Invalid number of rate correction values: " + - std::to_string(index)); - } - LOG(logDEBUG) << "Number of detectors for rate correction: " << index; - std::vector t(index); - socket.Receive(t); verifyIdle(socket); - LOG(logINFO) << "Setting rate corrections[" << index << ']'; + + auto t = socket.receiveVariableArgs>(); + if (t.size() <= 0) + throw RuntimeError("Invalid number of rate correction values: " + + std::to_string(t.size())); + LOG(logINFO) << "Setting rate corrections[" << t.size() << ']'; impl()->setRateCorrections(t); return socket.Send(OK); } @@ -1748,58 +1732,41 @@ int ClientInterface::set_arping(Interface &socket) { int ClientInterface::get_receiver_roi(Interface &socket) { auto retvals = impl()->getPortROIs(); LOG(logDEBUG1) << "Receiver roi retval:" << ToString(retvals); - auto size = static_cast(retvals.size()); - if (size != impl()->getNumberofUDPInterfaces()) { - throw RuntimeError("Invalid number of ROIs received: " + - std::to_string(size) + ". Expected: " + - std::to_string(impl()->getNumberofUDPInterfaces())); - } - socket.Send(size); - if (size > 0) - socket.Send(retvals); - return OK; + return socket.sendVariableResult(retvals); } int ClientInterface::set_receiver_roi(Interface &socket) { - auto roiSize = socket.Receive(); - std::vector args(roiSize); - if (roiSize > 0) { - socket.Receive(args); - } - if (roiSize != impl()->getNumberofUDPInterfaces()) { - throw RuntimeError("Invalid number of ROIs received: " + - std::to_string(roiSize) + ". Expected: " + - std::to_string(impl()->getNumberofUDPInterfaces())); - } if (detType == CHIPTESTBOARD || detType == XILINX_CHIPTESTBOARD) functionNotImplemented(); - LOG(logDEBUG1) << "Set Receiver ROI: " << ToString(args); verifyIdle(socket); + + auto args = socket.receiveVariableArgs>(); + auto numInterfaces = impl()->getNumberofUDPInterfaces(); + if (static_cast(args.size()) != numInterfaces) { + std::ostringstream oss; + oss << "Invalid number of ROIs received: " << args.size() + << ". Expected: " << numInterfaces; + throw RuntimeError(oss.str()); + } + LOG(logDEBUG1) << "Set Receiver ROI: " << ToString(args); try { impl()->setPortROIs(args); } catch (const std::exception &e) { throw RuntimeError("Could not set Receiver ROI [" + std::string(e.what()) + ']'); } - return socket.Send(OK); } int ClientInterface::set_receiver_roi_metadata(Interface &socket) { - auto roiSize = socket.Receive(); - LOG(logDEBUG1) << "Number of ReceiverROI metadata: " << roiSize; - if (roiSize < 1) { - throw RuntimeError("Invalid number of ROIs received: " + - std::to_string(roiSize) + ". Min: 1."); - } - std::vector rois(roiSize); - if (roiSize > 0) { - socket.Receive(rois); - } if (detType == CHIPTESTBOARD || detType == XILINX_CHIPTESTBOARD) functionNotImplemented(); verifyIdle(socket); - LOG(logINFO) << "Setting ReceiverROI metadata[" << roiSize << ']'; + + auto rois = socket.receiveVariableArgs>(); + if (rois.size() < 1) + throw RuntimeError("Invalid number of ROI metadata: " + + std::to_string(rois.size()) + ". Min: 1."); try { impl()->setMultiROIMetadata(rois); } catch (const std::exception &e) { @@ -1896,12 +1863,7 @@ int ClientInterface::get_roi_metadata(Interface &socket) { if (detType == CHIPTESTBOARD || detType == XILINX_CHIPTESTBOARD) functionNotImplemented(); auto retvals = impl()->getMultiROIMetadata(); - LOG(logDEBUG1) << "Receiver ROI metadata retval:" << ToString(retvals); - auto size = static_cast(retvals.size()); - socket.Send(size); - if (size > 0) - socket.Send(retvals); - return OK; + return socket.sendVariableResult(retvals); } int ClientInterface::set_readout_speed(Interface &socket) { @@ -1934,16 +1896,9 @@ int ClientInterface::set_readout_speed(Interface &socket) { } int ClientInterface::set_udp_port_disable_meta(Interface &socket) { - auto nports = socket.Receive(); - std::vector portsDisabled; - if (nports > 0) { - portsDisabled.resize(nports); - socket.Receive(portsDisabled); - LOG(logDEBUG1) << "Disabled Ports Metadata:" << ToString(portsDisabled); - } verifyIdle(socket); + auto portsDisabled = socket.receiveVariableArgs>(); try { - impl()->setUDPPortsDisabledMetadata(portsDisabled); } catch (const std::exception &e) { throw RuntimeError("Could not update UDP ports disabled metadata [" + @@ -1954,14 +1909,7 @@ int ClientInterface::set_udp_port_disable_meta(Interface &socket) { int ClientInterface::get_udp_port_disable_meta(Interface &socket) { auto retvals = impl()->getUDPPortsDisabledMetadata(); - LOG(logDEBUG1) << "Receiver disabled udp ports retval:" - << ToString(retvals); - socket.Send(OK); - auto size = static_cast(retvals.size()); - socket.Send(size); - if (size > 0) - socket.Send(retvals); - return OK; + return socket.sendVariableResult(retvals); } } // namespace sls 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/ClientSocket.h b/slsSupportLib/include/sls/ClientSocket.h index 4ef8e5da4..1e9bedd8f 100644 --- a/slsSupportLib/include/sls/ClientSocket.h +++ b/slsSupportLib/include/sls/ClientSocket.h @@ -14,13 +14,36 @@ class ClientSocket : public DataSocket { ClientSocket(std::string stype, const std::string &hostname, uint16_t port_number); ClientSocket(std::string stype, struct sockaddr_in addr); - int sendCommandThenRead(int fnum, const void *args, size_t args_size, - void *retval, size_t retval_size); - + void sendCommandThenRead(int fnum, const void *args, size_t args_size, + void *retval, size_t retval_size); std::string readErrorMessage(); + template + void sendCommandVariableSize(int fnum, const std::vector &args, + void *retval, size_t retval_size) { + int count = args.size(); + sendCommand(fnum, &count, sizeof(count)); + if (count > 0) { + Send(args); + } + readReply(retval, retval_size); + } + + template + void sendCommandVariableSize(int fnum, const void *args, size_t args_size, + std::vector &retval) { + sendCommand(fnum, args, args_size); + int count = 0; + readReply(&count, sizeof(count)); + retval.resize(count); + if (count > 0) { + Receive(retval); + } + } + void sendCommand(int fnum, const void *args, size_t args_size); + void readReply(void *retval, size_t retval_size); + private: - void readReply(int &ret, void *retval, size_t retval_size); [[noreturn]] void throwError(const std::string &msg) const; struct sockaddr_in serverAddr {}; std::string socketType; diff --git a/slsSupportLib/include/sls/ServerInterface.h b/slsSupportLib/include/sls/ServerInterface.h index d4ed04576..70f30debc 100644 --- a/slsSupportLib/include/sls/ServerInterface.h +++ b/slsSupportLib/include/sls/ServerInterface.h @@ -30,6 +30,25 @@ class ServerInterface : public DataSocket { Send(retval); return defs::OK; } + + template int sendVariableResult(T &&retval) { + int count = static_cast(retval.size()); + Send(defs::OK); + Send(count); + if (count > 0) + Send(retval); + return defs::OK; + } + + template T receiveVariableArgs() { + int count = 0; + Receive(count); + T retval(static_cast(count)); + if (count > 0) { + Receive(retval); + } + return retval; + } }; } // namespace sls \ No newline at end of file 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/include/sls/versionAPI.h b/slsSupportLib/include/sls/versionAPI.h index 4da99e33a..1bf107977 100644 --- a/slsSupportLib/include/sls/versionAPI.h +++ b/slsSupportLib/include/sls/versionAPI.h @@ -6,7 +6,7 @@ #define APICTB "0.0.0 0x260506" #define APIGOTTHARD2 "0.0.0 0x260427" #define APIMOENCH "0.0.0 0x260424" -#define APIEIGER "0.0.0 0x260424" +#define APIEIGER "0.0.0 0x260807" #define APIXILINXCTB "0.0.0 0x260506" #define APIJUNGFRAU "0.0.0 0x260424" #define APIMYTHEN3 "0.0.0 0x260506" diff --git a/slsSupportLib/src/ClientSocket.cpp b/slsSupportLib/src/ClientSocket.cpp index fe62d8208..62dfb2d19 100644 --- a/slsSupportLib/src/ClientSocket.cpp +++ b/slsSupportLib/src/ClientSocket.cpp @@ -73,20 +73,23 @@ void ClientSocket::throwError(const std::string &msg) const { } } -int ClientSocket::sendCommandThenRead(int fnum, const void *args, - size_t args_size, void *retval, - size_t retval_size) { - int ret = slsDetectorDefs::FAIL; - Send(&fnum, sizeof(fnum)); - setFnum(fnum); - Send(args, args_size); - readReply(ret, retval, retval_size); - return ret; +void ClientSocket::sendCommandThenRead(int fnum, const void *args, + size_t args_size, void *retval, + size_t retval_size) { + sendCommand(fnum, args, args_size); + readReply(retval, retval_size); } -void ClientSocket::readReply(int &ret, void *retval, size_t retval_size) { +void ClientSocket::sendCommand(int fnum, const void *args, size_t args_size) { + Send(fnum); + setFnum(fnum); + Send(args, args_size); +} + +void ClientSocket::readReply(void *retval, size_t retval_size) { try { + int ret = slsDetectorDefs::FAIL; Receive(&ret, sizeof(ret)); if (ret == slsDetectorDefs::FAIL) { std::string mess = readErrorMessage(); 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/slsSupportLib/tests/test-Sockets.cpp b/slsSupportLib/tests/test-Sockets.cpp index 897ed4e65..01c53877a 100644 --- a/slsSupportLib/tests/test-Sockets.cpp +++ b/slsSupportLib/tests/test-Sockets.cpp @@ -231,14 +231,13 @@ TEST_CASE("Using DetectorSocket to talk to a Server Socket", "[support]") { auto client = DetectorSocket("localhost", port); int retval = 0; - int ret = client.sendCommandThenRead(fnum, &arg, sizeof(arg), &retval, - sizeof(retval)); + CHECK_NOTHROW(client.sendCommandThenRead(fnum, &arg, sizeof(arg), &retval, + sizeof(retval))); client.close(); auto server_received = s.get(); // Client got OK and the expected return value back from the server - CHECK(ret == slsDetectorDefs::OK); CHECK(retval == arg * 2); // Server received the function number and argument we sent CHECK(server_received.first == fnum); 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