detectors/slsDetectorPackage
25
0
Fork 0
mirror of https://github.com/slsdetectorgroup/slsDetectorPackage.git synced 2025-06-12 21:07:13 +02:00
Code Issues Actions Packages Releases Activity

merge from 7.0.0

Browse Source
This commit is contained in:
maliakal_d
2023-02-24 10:39:51 +01:00
parent 48a684b95f e942678139
commit dc5db905d4
101 changed files with 4009 additions and 2128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
slsReceiverSoftware/src/Arping.cpp
View File

@ -4,6 +4,8 @@
#include "Arping.h"
#include <chrono>
#include <signal.h>
#include <thread>
#include <unistd.h>
namespace sls {
@ -18,7 +20,7 @@ Arping::Arping() {}
Arping::~Arping() {
if (IsRunning()) {
StopThread();
StopProcess();
}
}
@ -33,59 +35,66 @@ void Arping::SetInterfacesAndIps(const int index, const std::string &interface,
// create commands to arping
std::ostringstream os;
os << "arping -c 1 -U -I " << interface << " " << ip;
// to read error messages
os << " 2>&1";
std::string cmd = os.str();
commands[index] = cmd;
}
pid_t Arping::GetThreadId() const { return threadId; }
pid_t Arping::GetProcessId() const { return childPid; }
bool Arping::IsRunning() const { return runningFlag; }
void Arping::StartThread() {
void Arping::StartProcess() {
TestCommands();
try {
t = std::thread(&Arping::ThreadExecution, this);
} catch (...) {
throw RuntimeError("Could not start arping thread");
// to prevent zombies from child processes being killed
signal(SIGCHLD, SIG_IGN);
// Needs to be a fork and udp socket deleted after Listening threads
// done running to prevent udp socket cannot bind because of popen
// that forks
childPid = fork();
// child process
if (childPid == 0) {
LOG(logINFOBLUE) << "Created [ Arping Process, Tid: " << gettid()
<< " ]";
ProcessExecution();
}
// parent process
else if (childPid > 0) {
runningFlag = true;
}
// error
else {
throw RuntimeError("Could not start arping Process");
}
runningFlag = true;
}
void Arping::StopThread() {
void Arping::StopProcess() {
LOG(logINFOBLUE) << "Exiting [ Arping Process ]";
if (kill(childPid, SIGTERM)) {
throw RuntimeError("Could not kill the arping Process");
}
runningFlag = false;
t.join();
}
void Arping::ThreadExecution() {
threadId = gettid();
LOG(logINFOBLUE) << "Created [ Arping Thread, Tid: " << threadId << " ]";
while (runningFlag) {
void Arping::ProcessExecution() {
while (true) {
std::string error = ExecuteCommands();
// just print (was already tested at thread start)
// just print (was already tested at Process start)
if (!error.empty()) {
LOG(logERROR) << error;
}
// wait for 60s as long as thread not killed
int nsecs = 0;
while (runningFlag && nsecs != 60) {
std::this_thread::sleep_for(std::chrono::seconds(1));
++nsecs;
}
const auto interval = std::chrono::seconds(60);
std::this_thread::sleep_for(interval);
}
LOG(logINFOBLUE) << "Exiting [ Arping Thread, Tid: " << threadId << " ]";
threadId = 0;
}
void Arping::TestCommands() {
// atleast one interface must be set up
if (commands[0].empty()) {
throw RuntimeError(
"Could not arping. Interface not set up in apring thread");
"Could not arping. Interface not set up in arping Process");
}
// test if arping commands throw an error
std::string error = ExecuteCommands();
@ -101,7 +110,7 @@ std::string Arping::ExecuteCommands() {
if (cmd.empty())
continue;
LOG(logDEBUG) << "Executing Arping Command: " << cmd;
LOG(logDEBUG1) << "Executing Arping Command: " << cmd;
// execute command
FILE *sysFile = popen(cmd.c_str(), "r");

17
slsReceiverSoftware/src/Arping.h
View File

@ -2,15 +2,15 @@
// Copyright (C) 2021 Contributors to the SLS Detector Package
#pragma once
/**
*@short creates/destroys an ARPing thread to arping the interfaces slsReceiver
is listening to.
*@short creates/destroys an ARPing child process to arping the interfaces
slsReceiver is listening to.
*/
#include "receiver_defs.h"
#include "sls/logger.h"
#include <atomic>
#include <thread>
#include <unistd.h>
namespace sls {
@ -22,21 +22,20 @@ class Arping {
void SetInterfacesAndIps(const int index, const std::string &interface,
const std::string &ip);
pid_t GetThreadId() const;
pid_t GetProcessId() const;
bool IsRunning() const;
void StartThread();
void StopThread();
void StartProcess();
void StopProcess();
private:
void TestCommands();
std::string ExecuteCommands();
void ThreadExecution();
void ProcessExecution();
std::vector<std::string> commands =
std::vector<std::string>(MAX_NUMBER_OF_LISTENING_THREADS);
std::atomic<bool> runningFlag{false};
std::thread t;
std::atomic<pid_t> threadId{0};
std::atomic<pid_t> childPid{0};
};
} // namespace sls

393
slsReceiverSoftware/src/ClientInterface.cpp
View File

@ -234,8 +234,7 @@ int ClientInterface::decodeFunction(Interface &socket) {
socket.Receive(fnum);
socket.setFnum(fnum);
if (fnum <= NUM_DET_FUNCTIONS || fnum >= NUM_REC_FUNCTIONS) {
throw RuntimeError("Unrecognized Function enum " +
std::to_string(fnum) + "\n");
throw RuntimeError(UNRECOGNIZED_FNUM_ENUM + std::to_string(fnum));
} else {
LOG(logDEBUG1) << "calling function fnum: " << fnum << " ("
<< getFunctionNameFromEnum((enum detFuncs)fnum) << ")";
@ -320,164 +319,121 @@ int ClientInterface::setup_receiver(Interface &socket) {
auto arg = socket.Receive<rxParameters>();
LOG(logDEBUG) << ToString(arg);
// if object exists, verify unlocked and idle, else only verify lock
// (connecting first time)
if (receiver != nullptr) {
verifyIdle(socket);
}
// basic setup
setDetectorType(arg.detType);
impl()->setDetectorSize(arg.numberOfModule);
impl()->setModulePositionId(arg.moduleIndex);
impl()->setDetectorHostname(arg.hostname);
// udp setup
// update retvals only if detmac is not the same as in detector
MacAddr retvals[2];
if (arg.udp_dstip != 0) {
MacAddr r = setUdpIp(IpAddr(arg.udp_dstip));
MacAddr detMac{arg.udp_dstmac};
if (detMac != r) {
retvals[0] = r;
try {
// if object exists, verify unlocked and idle, else only verify lock
// (connecting first time)
if (receiver != nullptr) {
verifyIdle(socket);
}
}
if (arg.udp_dstip2 != 0) {
MacAddr r = setUdpIp2(IpAddr(arg.udp_dstip2));
MacAddr detMac{arg.udp_dstmac2};
if (detMac != r) {
retvals[1] = r;
}
}
impl()->setUDPPortNumber(arg.udp_dstport);
impl()->setUDPPortNumber2(arg.udp_dstport2);
if (detType == JUNGFRAU || detType == MOENCH || detType == GOTTHARD2) {
try {
impl()->setNumberofUDPInterfaces(arg.udpInterfaces);
} catch (const RuntimeError &e) {
throw RuntimeError("Failed to set number of interfaces to " +
std::to_string(arg.udpInterfaces));
}
}
impl()->setUDPSocketBufferSize(0);
// acquisition parameters
impl()->setNumberOfFrames(arg.frames);
impl()->setNumberOfTriggers(arg.triggers);
if (detType == GOTTHARD2) {
impl()->setNumberOfBursts(arg.bursts);
}
if (detType == JUNGFRAU) {
impl()->setNumberOfAdditionalStorageCells(arg.additionalStorageCells);
}
if (detType == CHIPTESTBOARD) {
try {
// basic setup
setDetectorType(arg.detType);
impl()->setDetectorSize(arg.numberOfModule);
impl()->setModulePositionId(arg.moduleIndex);
impl()->setDetectorHostname(arg.hostname);
// udp setup
// update retvals only if detmac is not the same as in detector
if (arg.udp_dstip != 0) {
MacAddr r = setUdpIp(IpAddr(arg.udp_dstip));
MacAddr detMac{arg.udp_dstmac};
if (detMac != r) {
retvals[0] = r;
}
}
if (arg.udp_dstip2 != 0) {
MacAddr r = setUdpIp2(IpAddr(arg.udp_dstip2));
MacAddr detMac{arg.udp_dstmac2};
if (detMac != r) {
retvals[1] = r;
}
}
impl()->setUDPPortNumber(arg.udp_dstport);
impl()->setUDPPortNumber2(arg.udp_dstport2);
if (detType == JUNGFRAU || detType == MOENCH || detType == GOTTHARD2) {
impl()->setNumberofUDPInterfaces(arg.udpInterfaces);
}
impl()->setUDPSocketBufferSize(0);
// acquisition parameters
impl()->setNumberOfFrames(arg.frames);
impl()->setNumberOfTriggers(arg.triggers);
if (detType == GOTTHARD2) {
impl()->setNumberOfBursts(arg.bursts);
}
if (detType == JUNGFRAU) {
impl()->setNumberOfAdditionalStorageCells(
arg.additionalStorageCells);
}
if (detType == CHIPTESTBOARD) {
impl()->setNumberofAnalogSamples(arg.analogSamples);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not set num analog samples to " +
std::to_string(arg.analogSamples) +
" due to fifo structure memory allocation.");
}
}
if (detType == CHIPTESTBOARD) {
try {
if (detType == CHIPTESTBOARD) {
impl()->setNumberofDigitalSamples(arg.digitalSamples);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not set num digital samples to " +
std::to_string(arg.analogSamples) +
" due to fifo structure memory allocation.");
}
}
if (detType != MYTHEN3) {
impl()->setAcquisitionTime(std::chrono::nanoseconds(arg.expTimeNs));
}
impl()->setAcquisitionPeriod(std::chrono::nanoseconds(arg.periodNs));
if (detType == EIGER) {
impl()->setSubExpTime(std::chrono::nanoseconds(arg.subExpTimeNs));
impl()->setSubPeriod(std::chrono::nanoseconds(arg.subExpTimeNs) +
std::chrono::nanoseconds(arg.subDeadTimeNs));
impl()->setActivate(static_cast<bool>(arg.activate));
impl()->setDetectorDataStream(LEFT, arg.dataStreamLeft);
impl()->setDetectorDataStream(RIGHT, arg.dataStreamRight);
try {
if (detType != MYTHEN3) {
impl()->setAcquisitionTime(std::chrono::nanoseconds(arg.expTimeNs));
}
impl()->setAcquisitionPeriod(std::chrono::nanoseconds(arg.periodNs));
if (detType == EIGER) {
impl()->setSubExpTime(std::chrono::nanoseconds(arg.subExpTimeNs));
impl()->setSubPeriod(std::chrono::nanoseconds(arg.subExpTimeNs) +
std::chrono::nanoseconds(arg.subDeadTimeNs));
impl()->setActivate(static_cast<bool>(arg.activate));
impl()->setDetectorDataStream(LEFT, arg.dataStreamLeft);
impl()->setDetectorDataStream(RIGHT, arg.dataStreamRight);
impl()->setQuad(arg.quad == 0 ? false : true);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not set quad to " +
std::to_string(arg.quad) +
" due to fifo strucutre memory allocation");
impl()->setThresholdEnergy(arg.thresholdEnergyeV[0]);
}
impl()->setThresholdEnergy(arg.thresholdEnergyeV[0]);
}
if (detType == EIGER || detType == JUNGFRAU || detType == MOENCH) {
impl()->setReadNRows(arg.readNRows);
}
if (detType == MYTHEN3) {
std::array<int, 3> val;
for (int i = 0; i < 3; ++i) {
val[i] = arg.thresholdEnergyeV[i];
if (detType == EIGER || detType == JUNGFRAU || detType == MOENCH) {
impl()->setReadNRows(arg.readNRows);
}
impl()->setThresholdEnergy(val);
}
if (detType == EIGER || detType == MYTHEN3) {
try {
if (detType == MYTHEN3) {
std::array<int, 3> val;
for (int i = 0; i < 3; ++i) {
val[i] = arg.thresholdEnergyeV[i];
}
impl()->setThresholdEnergy(val);
}
if (detType == EIGER || detType == MYTHEN3) {
impl()->setDynamicRange(arg.dynamicRange);
} catch (const RuntimeError &e) {
throw RuntimeError(
"Could not set dynamic range. Could not allocate "
"memory for fifo or could not start listening/writing threads");
}
}
impl()->setTimingMode(arg.timMode);
if (detType == EIGER || detType == CHIPTESTBOARD || detType == MYTHEN3) {
try {
impl()->setTimingMode(arg.timMode);
if (detType == EIGER || detType == CHIPTESTBOARD || detType == MYTHEN3) {
impl()->setTenGigaEnable(arg.tenGiga);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not set 10GbE.");
}
}
if (detType == CHIPTESTBOARD) {
try {
if (detType == CHIPTESTBOARD) {
impl()->setReadoutMode(arg.roMode);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not set read out mode "
"due to fifo memory allocation.");
}
}
if (detType == CHIPTESTBOARD) {
try {
impl()->setADCEnableMask(arg.adcMask);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not set adc enable mask "
"due to fifo memory allcoation");
}
try {
impl()->setTenGigaADCEnableMask(arg.adc10gMask);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not set 10Gb adc enable mask "
"due to fifo memory allcoation");
}
}
if (detType == GOTTHARD) {
try {
if (detType == GOTTHARD) {
impl()->setDetectorROI(arg.roi);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not set ROI");
}
if (detType == MYTHEN3) {
impl()->setCounterMask(arg.countermask);
impl()->setAcquisitionTime1(
std::chrono::nanoseconds(arg.expTime1Ns));
impl()->setAcquisitionTime2(
std::chrono::nanoseconds(arg.expTime2Ns));
impl()->setAcquisitionTime3(
std::chrono::nanoseconds(arg.expTime3Ns));
impl()->setGateDelay1(std::chrono::nanoseconds(arg.gateDelay1Ns));
impl()->setGateDelay2(std::chrono::nanoseconds(arg.gateDelay2Ns));
impl()->setGateDelay3(std::chrono::nanoseconds(arg.gateDelay3Ns));
impl()->setNumberOfGates(arg.gates);
}
if (detType == GOTTHARD2) {
impl()->setBurstMode(arg.burstType);
}
impl()->setScan(arg.scanParams);
} catch (std::exception &e) {
throw RuntimeError("Could not setup receiver [" +
std::string(e.what()) + ']');
}
if (detType == MYTHEN3) {
impl()->setCounterMask(arg.countermask);
impl()->setAcquisitionTime1(std::chrono::nanoseconds(arg.expTime1Ns));
impl()->setAcquisitionTime2(std::chrono::nanoseconds(arg.expTime2Ns));
impl()->setAcquisitionTime3(std::chrono::nanoseconds(arg.expTime3Ns));
impl()->setGateDelay1(std::chrono::nanoseconds(arg.gateDelay1Ns));
impl()->setGateDelay2(std::chrono::nanoseconds(arg.gateDelay2Ns));
impl()->setGateDelay3(std::chrono::nanoseconds(arg.gateDelay3Ns));
impl()->setNumberOfGates(arg.gates);
}
if (detType == GOTTHARD2) {
impl()->setBurstMode(arg.burstType);
}
impl()->setScan(arg.scanParams);
return socket.sendResult(retvals);
}
@ -501,13 +457,10 @@ void ClientInterface::setDetectorType(detectorType arg) {
detType = GENERIC;
receiver = make_unique<Implementation>(arg);
detType = arg;
} catch (std::exception &e) {
std::ostringstream os;
os << "Could not set detector type in the receiver. ";
os << e.what();
throw RuntimeError(os.str());
} catch (const std::exception &e) {
throw RuntimeError("Could not set detector type in the receiver. [" +
std::string(e.what()) + ']');
}
// callbacks after (in setdetectortype, the object is reinitialized)
if (startAcquisitionCallBack != nullptr)
impl()->registerCallBackStartAcquisition(startAcquisitionCallBack,
@ -535,8 +488,8 @@ int ClientInterface::set_detector_roi(Interface &socket) {
verifyIdle(socket);
try {
impl()->setDetectorROI(arg);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not set ROI");
} catch (const std::exception &e) {
throw RuntimeError("Could not set ROI [" + std::string(e.what()) + ']');
}
return socket.Send(OK);
}
@ -615,10 +568,10 @@ int ClientInterface::set_num_analog_samples(Interface &socket) {
}
try {
impl()->setNumberofAnalogSamples(value);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not set num analog samples to " +
std::to_string(value) +
" due to fifo structure memory allocation.");
} catch (const std::exception &e) {
throw RuntimeError("Could not set number of analog samples to " +
std::to_string(value) + " [" +
std::string(e.what()) + ']');
}
return socket.Send(OK);
}
@ -631,11 +584,12 @@ int ClientInterface::set_num_digital_samples(Interface &socket) {
}
try {
impl()->setNumberofDigitalSamples(value);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not set num digital samples to " +
std::to_string(value) +
" due to fifo structure memory allocation.");
} catch (const std::exception &e) {
throw RuntimeError("Could not set number of digital samples to " +
std::to_string(value) + " [" +
std::string(e.what()) + ']');
}
return socket.Send(OK);
}
@ -742,9 +696,9 @@ int ClientInterface::set_dynamic_range(Interface &socket) {
} else {
try {
impl()->setDynamicRange(dr);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not allocate memory for fifo or "
"could not start listening/writing threads");
} catch (const std::exception &e) {
throw RuntimeError("Could not set dynamic range [" +
std::string(e.what()) + ']');
}
}
}
@ -780,7 +734,12 @@ int ClientInterface::get_status(Interface &socket) {
int ClientInterface::start_receiver(Interface &socket) {
if (impl()->getStatus() == IDLE) {
LOG(logDEBUG1) << "Starting Receiver";
impl()->startReceiver();
try {
impl()->startReceiver();
} catch (const std::exception &e) {
throw RuntimeError("Could not start reciever [" +
std::string(e.what()) + ']');
}
}
return socket.Send(OK);
}
@ -790,12 +749,16 @@ int ClientInterface::stop_receiver(Interface &socket) {
if (impl()->getStatus() == RUNNING) {
LOG(logDEBUG1) << "Stopping Receiver";
impl()->setStoppedFlag(static_cast<bool>(arg));
impl()->stopReceiver();
try {
impl()->stopReceiver();
} catch (const std::exception &e) {
throw RuntimeError("Could not stop receiver [" +
std::string(e.what()) + ']');
}
}
auto s = impl()->getStatus();
if (s != IDLE)
throw RuntimeError("Could not stop receiver. It as it is: " +
ToString(s));
throw RuntimeError("Could not stop receiver. Status: " + ToString(s));
return socket.Send(OK);
}
@ -810,7 +773,12 @@ int ClientInterface::set_file_dir(Interface &socket) {
throw RuntimeError("Receiver path needs to be absolute path");
LOG(logDEBUG1) << "Setting file path: " << fpath;
impl()->setFilePath(fpath);
try {
impl()->setFilePath(fpath);
} catch (const std::exception &e) {
throw RuntimeError("Could not set file path [" + std::string(e.what()) +
']');
}
return socket.Send(OK);
}
@ -892,7 +860,12 @@ int ClientInterface::set_file_write(Interface &socket) {
}
verifyIdle(socket);
LOG(logDEBUG1) << "Setting File write enable:" << enable;
impl()->setFileWriteEnable(enable);
try {
impl()->setFileWriteEnable(enable);
} catch (const std::exception &e) {
throw RuntimeError("Could not enable/disable file write [" +
std::string(e.what()) + ']');
}
return socket.Send(OK);
}
@ -946,8 +919,9 @@ int ClientInterface::enable_tengiga(Interface &socket) {
LOG(logDEBUG1) << "Setting 10GbE:" << val;
try {
impl()->setTenGigaEnable(val);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not set 10GbE.");
} catch (const std::exception &e) {
throw RuntimeError("Could not set 10GbE. [" +
std::string(e.what()) + ']');
}
}
int retval = impl()->getTenGigaEnable();
@ -963,9 +937,9 @@ int ClientInterface::set_fifo_depth(Interface &socket) {
LOG(logDEBUG1) << "Setting fifo depth:" << value;
try {
impl()->setFifoDepth(value);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not set fifo depth due to fifo structure "
"memory allocation.");
} catch (const std::exception &e) {
throw RuntimeError("Could not set fifo depth [" +
std::string(e.what()) + ']');
}
}
int retval = impl()->getFifoDepth();
@ -1000,10 +974,12 @@ int ClientInterface::set_streaming(Interface &socket) {
LOG(logDEBUG1) << "Setting data stream enable:" << index;
try {
impl()->setDataStreamEnable(index);
} catch (const RuntimeError &e) {
} catch (const std::exception &e) {
throw RuntimeError("Could not set data stream enable to " +
std::to_string(index));
std::to_string(index) + " [" +
std::string(e.what()) + ']');
}
return socket.Send(OK);
}
@ -1062,7 +1038,12 @@ int ClientInterface::set_file_format(Interface &socket) {
}
verifyIdle(socket);
LOG(logDEBUG1) << "Setting file format:" << f;
impl()->setFileFormat(f);
try {
impl()->setFileFormat(f);
} catch (const std::exception &e) {
throw RuntimeError("Could not set file format to " + ToString(f) +
" [" + std::string(e.what()) + ']');
}
auto retval = impl()->getFileFormat();
validate(f, retval, "set file format", DEC);
@ -1183,7 +1164,13 @@ int ClientInterface::set_udp_socket_buffer_size(Interface &socket) {
"Receiver socket buffer size exceeded max (INT_MAX/2)");
}
LOG(logDEBUG1) << "Setting UDP Socket Buffer size: " << size;
impl()->setUDPSocketBufferSize(size);
try {
impl()->setUDPSocketBufferSize(size);
} catch (const std::exception &e) {
throw RuntimeError("Could not set udp socket buffer size to " +
std::to_string(size) + " [" +
std::string(e.what()) + ']');
}
}
int retval = impl()->getUDPSocketBufferSize();
if (size != 0)
@ -1262,9 +1249,9 @@ int ClientInterface::set_readout_mode(Interface &socket) {
LOG(logDEBUG1) << "Setting readout mode: " << arg;
try {
impl()->setReadoutMode(arg);
} catch (const RuntimeError &e) {
throw RuntimeError(
"Could not set read out mode due to fifo memory allocation.");
} catch (const std::exception &e) {
throw RuntimeError("Could not set read out mode [" +
std::string(e.what()) + ']');
}
}
auto retval = impl()->getReadoutMode();
@ -1280,10 +1267,11 @@ int ClientInterface::set_adc_mask(Interface &socket) {
LOG(logDEBUG1) << "Setting 1Gb ADC enable mask: " << arg;
try {
impl()->setADCEnableMask(arg);
} catch (const RuntimeError &e) {
throw RuntimeError(
"Could not set adc enable mask due to fifo memory allcoation");
} catch (const std::exception &e) {
throw RuntimeError("Could not set adc enable mask [" +
std::string(e.what()) + ']');
}
auto retval = impl()->getADCEnableMask();
if (retval != arg) {
std::ostringstream os;
@ -1347,10 +1335,10 @@ int ClientInterface::set_quad_type(Interface &socket) {
LOG(logDEBUG1) << "Setting quad:" << quadEnable;
try {
impl()->setQuad(quadEnable == 0 ? false : true);
} catch (const RuntimeError &e) {
} catch (const std::exception &e) {
throw RuntimeError("Could not set quad to " +
std::to_string(quadEnable) +
" due to fifo strucutre memory allocation");
std::to_string(quadEnable) + " [" +
std::string(e.what()) + ']');
}
}
int retval = impl()->getQuad() ? 1 : 0;
@ -1486,10 +1474,12 @@ int ClientInterface::set_num_interfaces(Interface &socket) {
LOG(logDEBUG1) << "Setting Number of UDP Interfaces:" << arg;
try {
impl()->setNumberofUDPInterfaces(arg);
} catch (const RuntimeError &e) {
throw RuntimeError("Failed to set number of interfaces to " +
std::to_string(arg));
} catch (const std::exception &e) {
throw RuntimeError("Could not set number of interfaces to " +
std::to_string(arg) + " [" + std::string(e.what()) +
']');
}
return socket.Send(OK);
}
@ -1499,10 +1489,11 @@ int ClientInterface::set_adc_mask_10g(Interface &socket) {
LOG(logDEBUG1) << "Setting 10Gb ADC enable mask: " << arg;
try {
impl()->setTenGigaADCEnableMask(arg);
} catch (const RuntimeError &e) {
throw RuntimeError(
"Could not set 10Gb adc enable mask due to fifo memory allcoation");
} catch (const std::exception &e) {
throw RuntimeError("Could not set 10Gb adc enable mask [" +
std::string(e.what()) + ']');
}
auto retval = impl()->getTenGigaADCEnableMask();
if (retval != arg) {
std::ostringstream os;
@ -1518,7 +1509,12 @@ int ClientInterface::set_counter_mask(Interface &socket) {
auto arg = socket.Receive<uint32_t>();
verifyIdle(socket);
LOG(logDEBUG1) << "Setting counters: " << arg;
impl()->setCounterMask(arg);
try {
impl()->setCounterMask(arg);
} catch (const std::exception &e) {
throw RuntimeError("Could not set counter mask [" +
std::string(e.what()) + ']');
}
return socket.Send(OK);
}
@ -1714,7 +1710,12 @@ int ClientInterface::set_arping(Interface &socket) {
}
verifyIdle(socket);
LOG(logDEBUG1) << "Starting/ Killing arping thread:" << value;
impl()->setArping(value, udpips);
try {
impl()->setArping(value, udpips);
} catch (const std::exception &e) {
throw RuntimeError("Could not start/kill arping thread [" +
std::string(e.what()) + ']');
}
return socket.Send(OK);
}
@ -1732,9 +1733,11 @@ int ClientInterface::set_receiver_roi(Interface &socket) {
verifyIdle(socket);
try {
impl()->setReceiverROI(arg);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not set ReceiverROI");
} catch (const std::exception &e) {
throw RuntimeError("Could not set Receiver ROI [" +
std::string(e.what()) + ']');
}
return socket.Send(OK);
}
@ -1746,9 +1749,11 @@ int ClientInterface::set_receiver_roi_metadata(Interface &socket) {
verifyIdle(socket);
try {
impl()->setReceiverROIMetadata(arg);
} catch (const RuntimeError &e) {
throw RuntimeError("Could not set ReceiverROI metadata");
} catch (const std::exception &e) {
throw RuntimeError("Could not set ReceiverROI metadata [" +
std::string(e.what()) + ']');
}
return socket.Send(OK);
}

17
slsReceiverSoftware/src/DataStreamer.cpp
View File

@ -87,19 +87,18 @@ void DataStreamer::CreateZmqSockets(uint32_t port, const IpAddr ip, int hwm) {
std::string sip = ip.str();
try {
zmqSocket = new ZmqSocket(portnum, (ip != 0 ? sip.c_str() : nullptr));
// set if custom
if (hwm >= 0) {
zmqSocket->SetSendHighWaterMark(hwm);
if (zmqSocket->GetSendHighWaterMark() != hwm) {
throw RuntimeError(
"Could not set zmq send high water mark to " +
std::to_string(hwm));
}
// needed, or HWL is not taken
zmqSocket->Rebind();
}
} catch (...) {
LOG(logERROR) << "Could not create Zmq socket on port " << portnum
<< " for Streamer " << index;
throw;
} catch (std::exception &e) {
std::ostringstream oss;
oss << "Could not create zmq pub socket on port " << portnum;
oss << " [" << e.what() << ']';
throw RuntimeError(oss.str());
}
LOG(logINFO) << index << " Streamer: Zmq Server started at "
<< zmqSocket->GetZmqServerAddress()

65
slsReceiverSoftware/src/Implementation.cpp
View File

@ -78,11 +78,12 @@ void Implementation::SetupFifoStructure() {
try {
fifo.push_back(
sls::make_unique<Fifo>(i, datasize, generalData->fifoDepth));
} catch (...) {
} catch (const std::exception &e) {
fifo.clear();
generalData->fifoDepth = 0;
throw RuntimeError("Could not allocate memory for fifo structure " +
std::to_string(i) + ". FifoDepth is now 0.");
std::ostringstream oss;
oss << e.what() << ". Fifo depth is now 0";
throw RuntimeError(oss.str());
}
// set the listener & dataprocessor threads to point to the right fifo
if (listener.size())
@ -165,12 +166,10 @@ void Implementation::setDetectorType(const detectorType d) {
SetupListener(i);
dataProcessor.push_back(sls::make_unique<DataProcessor>(i));
SetupDataProcessor(i);
} catch (...) {
} catch (const std::exception &e) {
listener.clear();
dataProcessor.clear();
throw RuntimeError(
"Could not create listener/dataprocessor threads (index:" +
std::to_string(i) + ")");
throw;
}
}
@ -212,6 +211,7 @@ void Implementation::SetupDataStreamer(int i) {
streamingHwm);
dataStreamer[i]->SetAdditionalJsonHeader(additionalJsonHeader);
dataStreamer[i]->SetFileIndex(fileIndex);
dataStreamer[i]->SetQuadEnable(quadEnable);
dataStreamer[i]->SetFlipRows(flipRows);
dataStreamer[i]->SetNumberofPorts(numPorts);
dataStreamer[i]->SetQuadEnable(quadEnable);
@ -351,19 +351,21 @@ std::array<pid_t, NUM_RX_THREAD_IDS> Implementation::getThreadIds() const {
retval[id++] = 0;
}
}
retval[NUM_RX_THREAD_IDS - 1] = arping.GetThreadId();
retval[NUM_RX_THREAD_IDS - 1] = arping.GetProcessId();
return retval;
}
bool Implementation::getArping() const { return arping.IsRunning(); }
pid_t Implementation::getArpingThreadId() const { return arping.GetThreadId(); }
pid_t Implementation::getArpingProcessId() const {
return arping.GetProcessId();
}
void Implementation::setArping(const bool i,
const std::vector<std::string> ips) {
if (i != arping.IsRunning()) {
if (!i) {
arping.StopThread();
arping.StopProcess();
} else {
// setup interface
for (int i = 0; i != generalData->numUDPInterfaces; ++i) {
@ -374,7 +376,7 @@ void Implementation::setArping(const bool i,
}
arping.SetInterfacesAndIps(i, eth[i], ips[i]);
}
arping.StartThread();
arping.StartProcess();
}
}
}
@ -665,8 +667,9 @@ void Implementation::startReceiver() {
startAcquisitionCallBack(filePath, fileName, fileIndex, imageSize,
pStartAcquisition);
} catch (const std::exception &e) {
throw RuntimeError("Start Acquisition Callback Error: " +
std::string(e.what()));
std::ostringstream oss;
oss << "Start Acquisition Callback Error: " << e.what();
throw RuntimeError(oss.str());
}
if (rawDataReadyCallBack != nullptr) {
LOG(logINFO) << "Data Write has been defined externally";
@ -713,6 +716,10 @@ void Implementation::stopReceiver() {
std::this_thread::sleep_for(std::chrono::milliseconds(5));
}
// delete the udp sockets
for (const auto &it : listener)
it->DeleteUDPSocket();
if (fileWriteEnable && modulePos == 0) {
// master and virtual file (hdf5)
StartMasterWriter();
@ -777,8 +784,9 @@ void Implementation::stopReceiver() {
status = IDLE;
LOG(logINFO) << "Receiver Stopped";
LOG(logINFO) << "Status: " << ToString(status);
throw RuntimeError("Acquisition Finished Callback Error: " +
std::string(e.what()));
std::ostringstream oss;
oss << "Acquisition Finished Callback Error: " << e.what();
throw RuntimeError(oss.str());
}
}
}
@ -859,7 +867,7 @@ void Implementation::CreateUDPSockets() {
}
} catch (const RuntimeError &e) {
shutDownUDPSockets();
throw RuntimeError("Could not create UDP Socket(s).");
throw;
}
LOG(logDEBUG) << "UDP socket(s) created successfully.";
}
@ -879,7 +887,9 @@ void Implementation::SetupWriter() {
shutDownUDPSockets();
for (const auto &it : dataProcessor)
it->CloseFiles();
throw RuntimeError("Could not create first data file.");
std::ostringstream oss;
oss << "Could not set up writer: " << e.what();
throw RuntimeError(oss.str());
}
}
@ -968,10 +978,9 @@ void Implementation::StartMasterWriter() {
}
}
#endif
} catch (std::exception &e) {
} catch (const std::exception &e) {
// ignore it and just print it
LOG(logWARNING) << "Caught exception when handling virtual hdf5 file ["
<< e.what() << "]";
LOG(logWARNING) << "Error creating master/virtualfiles: " << e.what();
}
}
@ -1032,12 +1041,10 @@ void Implementation::setNumberofUDPInterfaces(const int n) {
SetupListener(i);
dataProcessor.push_back(sls::make_unique<DataProcessor>(i));
SetupDataProcessor(i);
} catch (...) {
} catch (const std::exception &e) {
listener.clear();
dataProcessor.clear();
throw RuntimeError(
"Could not create listener/dataprocessor threads (index:" +
std::to_string(i) + ")");
throw;
}
// streamer threads
@ -1045,16 +1052,14 @@ void Implementation::setNumberofUDPInterfaces(const int n) {
try {
dataStreamer.push_back(sls::make_unique<DataStreamer>(i));
SetupDataStreamer(i);
} catch (...) {
} catch (const std::exception &e) {
if (dataStreamEnable) {
dataStreamer.clear();
dataStreamEnable = false;
for (const auto &it : dataProcessor)
it->SetDataStreamEnable(dataStreamEnable);
}
throw RuntimeError(
"Could not create datastreamer threads (index:" +
std::to_string(i) + ")");
throw;
}
}
}
@ -1165,12 +1170,12 @@ void Implementation::setDataStreamEnable(const bool enable) {
try {
dataStreamer.push_back(sls::make_unique<DataStreamer>(i));
SetupDataStreamer(i);
} catch (...) {
} catch (const std::exception &e) {
dataStreamer.clear();
dataStreamEnable = false;
for (const auto &it : dataProcessor)
it->SetDataStreamEnable(dataStreamEnable);
throw RuntimeError("Could not set data stream enable.");
throw;
}
}
SetThreadPriorities();

4
slsReceiverSoftware/src/Implementation.h
View File

@ -54,7 +54,7 @@ class Implementation : private virtual slsDetectorDefs {
void setThreadIds(const pid_t parentTid, const pid_t tcpTid);
std::array<pid_t, NUM_RX_THREAD_IDS> getThreadIds() const;
bool getArping() const;
pid_t getArpingThreadId() const;
pid_t getArpingProcessId() const;
void setArping(const bool i, const std::vector<std::string> ips);
ROI getReceiverROI() const;
void setReceiverROI(const ROI arg);
@ -323,7 +323,7 @@ class Implementation : private virtual slsDetectorDefs {
// acquisition
std::atomic<runStatus> status{IDLE};
bool stoppedFlag{false};
std::atomic<bool> stoppedFlag{false};
scanParameters scanParams{};
// network configuration (UDP)

43
slsReceiverSoftware/src/Listener.cpp
View File

@ -148,30 +148,34 @@ void Listener::RecordFirstIndex(uint64_t fnum) {
}
void Listener::CreateUDPSocket(int &actualSize) {
if (disabledPort) {
return;
}
uint32_t packetSize = generalData->packetSize;
if (generalData->detType == GOTTHARD2 && index != 0) {
packetSize = generalData->vetoPacketSize;
}
try {
if (disabledPort) {
return;
}
uint32_t packetSize = generalData->packetSize;
if (generalData->detType == GOTTHARD2 && index != 0) {
packetSize = generalData->vetoPacketSize;
}
udpSocket = nullptr;
udpSocket = make_unique<UdpRxSocket>(
udpPortNumber, packetSize,
(eth.length() ? InterfaceNameToIp(eth).str().c_str() : nullptr),
generalData->udpSocketBufferSize);
LOG(logINFO) << index << ": UDP port opened at port " << udpPortNumber;
} catch (...) {
throw RuntimeError("Could not create UDP socket on port " +
std::to_string(udpPortNumber));
udpSocketAlive = true;
// doubled due to kernel bookkeeping (could also be less due to
// permissions)
actualSize = udpSocket->getBufferSize();
} catch (std::exception &e) {
std::ostringstream oss;
oss << "Could not create UDP socket on port " << udpPortNumber << " ["
<< e.what() << ']';
throw RuntimeError(oss.str());
}
udpSocketAlive = true;
// doubled due to kernel bookkeeping (could also be less due to permissions)
actualSize = udpSocket->getBufferSize();
}
void Listener::ShutDownUDPSocket() {
@ -184,6 +188,13 @@ void Listener::ShutDownUDPSocket() {
}
}
void Listener::DeleteUDPSocket() {
if (udpSocket) {
udpSocket.reset();
LOG(logINFO) << "Closed UDP port " << udpPortNumber;
}
}
void Listener::CreateDummySocketForUDPSocketBufferSize(int s, int &actualSize) {
// custom setup (s != 0)
// default setup at startup (s = 0)

1
slsReceiverSoftware/src/Listener.h
View File

@ -50,6 +50,7 @@ class Listener : private virtual slsDetectorDefs, public ThreadObject {
void ResetParametersforNewAcquisition();
void CreateUDPSocket(int &actualSize);
void ShutDownUDPSocket();
void DeleteUDPSocket();
/** to set & get actual buffer size */
void CreateDummySocketForUDPSocketBufferSize(int s, int &actualSize);