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

esrf changes: Listener, DataProcessor, DataStreamer, Fifo, ZmqSocket constructors return an exception when it fails

Browse Source
This commit is contained in:
maliakal_d
2018-05-02 16:49:27 +02:00
parent fdd19bc316
commit d3a0319863
16 changed files with 120 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
slsReceiverSoftware/src/DataProcessor.cpp
View File

@ -23,7 +23,7 @@ using namespace std;
const string DataProcessor::TypeName = "DataProcessor";
DataProcessor::DataProcessor(int& ret, int ind, Fifo*& f, fileFormat* ftype, bool fwenable, bool* dsEnable, bool* gpEnable, uint32_t* dr,
DataProcessor::DataProcessor(int ind, Fifo*& f, fileFormat* ftype, bool fwenable, bool* dsEnable, bool* gpEnable, uint32_t* dr,
uint32_t* freq, uint32_t* timer,
void (*dataReadycb)(uint64_t, uint32_t, uint32_t, uint64_t, uint64_t,
uint16_t, uint16_t, uint16_t, uint16_t, uint32_t, uint16_t, uint8_t, uint8_t,
@ -61,9 +61,8 @@ DataProcessor::DataProcessor(int& ret, int ind, Fifo*& f, fileFormat* ftype, boo
rawDataModifyReadyCallBack(dataModifyReadycb),
pRawDataReady(pDataReadycb)
{
ret = FAIL;
if(ThreadObject::CreateThread() == OK)
ret = OK;
if(ThreadObject::CreateThread() == FAIL)
throw std::exception();
FILE_LOG(logDEBUG) << "DataProcessor " << ind << " created";

19
slsReceiverSoftware/src/DataStreamer.cpp
View File

@ -16,7 +16,7 @@ using namespace std;
const string DataStreamer::TypeName = "DataStreamer";
DataStreamer::DataStreamer(int& ret, int ind, Fifo*& f, uint32_t* dr, int* sEnable, uint64_t* fi, int* fd, char* ajh) :
DataStreamer::DataStreamer(int ind, Fifo*& f, uint32_t* dr, int* sEnable, uint64_t* fi, int* fd, char* ajh) :
ThreadObject(ind),
runningFlag(0),
generalData(0),
@ -34,13 +34,12 @@ DataStreamer::DataStreamer(int& ret, int ind, Fifo*& f, uint32_t* dr, int* sEnab
additionJsonHeader(ajh),
silentMode(false)
{
ret = FAIL;
if(ThreadObject::CreateThread() == OK)
ret = OK;
if(ThreadObject::CreateThread() == FAIL)
throw std::exception();
FILE_LOG(logDEBUG) << "DataStreamer " << ind << " created";
strcpy(fileNametoStream, "");
memset(fileNametoStream, 0, MAX_STR_LENGTH);
}
@ -129,16 +128,16 @@ int DataStreamer::SetThreadPriority(int priority) {
}
int DataStreamer::CreateZmqSockets(int* nunits, uint32_t port, const char* srcip) {
void DataStreamer::CreateZmqSockets(int* nunits, uint32_t port, const char* srcip) {
uint32_t portnum = port + index;
zmqSocket = new ZmqSocket(portnum, (strlen(srcip)?srcip:NULL));
if (zmqSocket->IsError()) {
try {
zmqSocket = new ZmqSocket(portnum, (strlen(srcip)?srcip:NULL));
} catch (...) {
cprintf(RED, "Error: Could not create Zmq socket on port %d for Streamer %d\n", portnum, index);
return FAIL;
throw;
}
FILE_LOG(logINFO) << index << " Streamer: Zmq Server started at " << zmqSocket->GetZmqServerAddress();
return OK;
}

5
slsReceiverSoftware/src/Fifo.cpp
View File

@ -13,7 +13,7 @@
using namespace std;
Fifo::Fifo(int ind, uint32_t fifoItemSize, uint32_t depth, bool &success):
Fifo::Fifo(int ind, uint32_t fifoItemSize, uint32_t depth):
index(ind),
memory(0),
fifoBound(0),
@ -23,9 +23,8 @@ Fifo::Fifo(int ind, uint32_t fifoItemSize, uint32_t depth, bool &success):
status_fifoBound(0),
status_fifoFree(depth){
FILE_LOG(logDEBUG) << __AT__ << " called";
success = true;
if(CreateFifos(fifoItemSize) == FAIL)
success = false;
throw std::exception();
}

12
slsReceiverSoftware/src/Listener.cpp
View File

@ -19,7 +19,7 @@ using namespace std;
const string Listener::TypeName = "Listener";
Listener::Listener(int& ret, int ind, detectorType dtype, Fifo*& f, runStatus* s,
Listener::Listener(int ind, detectorType dtype, Fifo*& f, runStatus* s,
uint32_t* portno, char* e, int* act, uint64_t* nf, uint32_t* dr,
uint32_t* us, uint32_t* as) :
ThreadObject(ind),
@ -49,9 +49,8 @@ Listener::Listener(int& ret, int ind, detectorType dtype, Fifo*& f, runStatus* s
udpSocketBufferSize(us),
actualUDPSocketBufferSize(as)
{
ret = FAIL;
if(ThreadObject::CreateThread() == OK)
ret = OK;
if(ThreadObject::CreateThread() == FAIL)
throw std::exception();
FILE_LOG(logDEBUG) << "Listener " << ind << " created";
}
@ -235,6 +234,7 @@ void Listener::SetSilentMode(bool mode) {
int Listener::CreateDummySocketForUDPSocketBufferSize(uint32_t s) {
FILE_LOG(logINFO) << "Testing UDP Socket Buffer size with test port " << *udpPortNumber;
uint32_t temp = *udpSocketBufferSize;
*udpSocketBufferSize = s;
@ -263,11 +263,11 @@ int Listener::CreateDummySocketForUDPSocketBufferSize(uint32_t s) {
FILE_LOG(logERROR) << "Could not create a test UDP socket on port " << *udpPortNumber << " error: " << iret;
return FAIL;
}
// doubled due to kernel bookkeeping (could also be less due to permissions)
*actualUDPSocketBufferSize = udpSocket->getActualUDPSocketBufferSize();
if (*actualUDPSocketBufferSize != (s*2))
if (*actualUDPSocketBufferSize != (s*2)) {
*udpSocketBufferSize = temp;
}
// shutdown socket

120
slsReceiverSoftware/src/UDPStandardImplementation.cpp
View File

@ -217,32 +217,23 @@ int UDPStandardImplementation::setDataStreamEnable(const bool enable) {
dataStreamer.clear();
if (enable) {
for ( int i = 0; i < numThreads; ++i ) {
int ret = FAIL;
DataStreamer* s = new DataStreamer(ret, i, fifo[i], &dynamicRange, &shortFrameEnable, &fileIndex, flippedData, additionalJsonHeader);
if (ret == FAIL)
cprintf(RED,"Error: Could not create data callback threads\n");
else {
dataStreamer.push_back(s);
dataStreamer[i]->SetGeneralData(generalData);
if (dataStreamer[i]->CreateZmqSockets(&numThreads, streamingPort, streamingSrcIP) == FAIL) {
cprintf(RED,"Error: Could not create zmq sockets\n");
ret = FAIL;
}
}
// error in creating threads or zmq sockets
if (ret == FAIL) {
for (vector<DataStreamer*>::const_iterator it = dataStreamer.begin(); it != dataStreamer.end(); ++it)
delete(*it);
dataStreamer.clear();
dataStreamEnable = false;
return FAIL;
}
}
SetThreadPriorities();
for ( int i = 0; i < numThreads; ++i ) {
try {
DataStreamer* s = new DataStreamer(i, fifo[i], &dynamicRange,
&shortFrameEnable, &fileIndex, flippedData, additionalJsonHeader);
dataStreamer.push_back(s);
dataStreamer[i]->SetGeneralData(generalData);
dataStreamer[i]->CreateZmqSockets(&numThreads, streamingPort, streamingSrcIP);
}
catch(...) {
for (vector<DataStreamer*>::const_iterator it = dataStreamer.begin(); it != dataStreamer.end(); ++it)
delete(*it);
dataStreamer.clear();
dataStreamEnable = false;
return FAIL;
}
}
SetThreadPriorities();
}
}
FILE_LOG(logINFO) << "Data Send to Gui: " << dataStreamEnable;
@ -375,31 +366,28 @@ int UDPStandardImplementation::setDetectorType(const detectorType d) {
//create threads
for ( int i = 0; i < numThreads; ++i ) {
int ret = FAIL;
Listener* l = new Listener(ret, i, myDetectorType, fifo[i], &status,
&udpPortNum[i], eth, &activated, &numberOfFrames, &dynamicRange,
&udpSocketBufferSize, &actualUDPSocketBufferSize);
DataProcessor* p = NULL;
if (ret == OK)
p = new DataProcessor(ret, i, fifo[i], &fileFormatType,
try {
Listener* l = new Listener(i, myDetectorType, fifo[i], &status,
&udpPortNum[i], eth, &activated, &numberOfFrames, &dynamicRange,
&udpSocketBufferSize, &actualUDPSocketBufferSize);
listener.push_back(l);
DataProcessor* p = new DataProcessor(i, fifo[i], &fileFormatType,
fileWriteEnable, &dataStreamEnable, &gapPixelsEnable,
&dynamicRange, &frameToGuiFrequency, &frameToGuiTimerinMS,
rawDataReadyCallBack, rawDataModifyReadyCallBack, pRawDataReady);
// error in creating threads
if (ret == FAIL) {
FILE_LOG(logERROR) << "Could not create listener/dataprocessor threads (index:" << i << ")";
for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it)
delete(*it);
listener.clear();
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it)
delete(*it);
dataProcessor.clear();
return FAIL;
}
listener.push_back(l);
dataProcessor.push_back(p);
dataProcessor.push_back(p);
}
catch (...) {
FILE_LOG(logERROR) << "Could not create listener/dataprocessor threads (index:" << i << ")";
for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it)
delete(*it);
listener.clear();
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it)
delete(*it);
dataProcessor.clear();
return FAIL;
}
}
//set up writer and callbacks
@ -410,8 +398,8 @@ int UDPStandardImplementation::setDetectorType(const detectorType d) {
SetThreadPriorities();
// check udp socket buffer size
setUDPSocketBufferSize(udpSocketBufferSize);
// check udp socket buffer size
setUDPSocketBufferSize(udpSocketBufferSize);
FILE_LOG(logDEBUG) << " Detector type set to " << getDetectorType(d);
return OK;
@ -673,11 +661,11 @@ void UDPStandardImplementation::SetLocalNetworkParameters() {
if (proc_file.good()) {
proc_file << MAX_SOCKET_INPUT_PACKET_QUEUE << endl;
cprintf(GREEN, "Max length of input packet queue "
"(/proc/sys/net/core/netdev_max_backlog) modified to %d\n",
"[/proc/sys/net/core/netdev_max_backlog] modified to %d\n",
MAX_SOCKET_INPUT_PACKET_QUEUE);
} else {
const char *msg = "Could not change max length of"
"input packet queue (net.core.netdev_max_backlog). No Root Privileges?";
"input packet queue [net.core.netdev_max_backlog]. (No Root Privileges?)";
FILE_LOG(logWARNING) << msg;
}
}
@ -689,7 +677,7 @@ void UDPStandardImplementation::SetThreadPriorities() {
for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it){
if ((*it)->SetThreadPriority(LISTENER_PRIORITY) == FAIL) {
FILE_LOG(logWARNING) << "Could not prioritize listener threads. No Root Privileges?";
FILE_LOG(logWARNING) << "Could not prioritize listener threads. (No Root Privileges?)";
return;
}
}
@ -711,20 +699,18 @@ int UDPStandardImplementation::SetupFifoStructure() {
for ( int i = 0; i < numThreads; i++ ) {
//create fifo structure
bool success = true;
Fifo* f = new Fifo (i,
(generalData->imageSize) * numberofJobs + (generalData->fifoBufferHeaderSize),
fifoDepth, success);
//error
if (!success) {
cprintf(RED,"Error: Could not allocate memory for fifo structure of index %d\n", i);
for (vector<Fifo*>::const_iterator it = fifo.begin(); it != fifo.end(); ++it)
delete(*it);
fifo.clear();
return FAIL;
}
fifo.push_back(f);
try {
Fifo* f = new Fifo (i,
(generalData->imageSize) * numberofJobs + (generalData->fifoBufferHeaderSize),
fifoDepth);
fifo.push_back(f);
} catch (...) {
cprintf(RED,"Error: Could not allocate memory for fifo structure of index %d\n", i);
for (vector<Fifo*>::const_iterator it = fifo.begin(); it != fifo.end(); ++it)
delete(*it);
fifo.clear();
return FAIL;
}
//set the listener & dataprocessor threads to point to the right fifo
if(listener.size())listener[i]->SetFifo(fifo[i]);
if(dataProcessor.size())dataProcessor[i]->SetFifo(fifo[i]);

2
slsReceiverSoftware/src/slsReceiverTCPIPInterface.cpp
View File

@ -2649,7 +2649,7 @@ int slsReceiverTCPIPInterface::set_udp_socket_buffer_size() {
}
//get
retval=receiverBase->getUDPSocketBufferSize();
if(index >= 0 && retval != index) {
if(index >= 0 && ((retval != index) || (receiverBase->getActualUDPSocketBufferSize() != (index*2)))) {
ret = FAIL;
strcpy(mess, "Could not set UDP Socket buffer size (No CAP_NET_ADMIN privileges?)\n");
FILE_LOG(logERROR) << mess;