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

Merge branch 'eiger_receiver' of gitorious.psi.ch:sls_det_software/sls_receiver_software into eiger_receiver

Browse Source 
Conflicts:
	include/UDPInterface.h
This commit is contained in:
sala 2014-09-22 14:46:34 +02:00
commit 546d43dedb
16 changed files with 685 additions and 607 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
slsReceiverSoftware/.gitignore vendored
View File

@ -1,6 +1,6 @@
*~ *~
*.o *.o
build build/*
GPATH GPATH
GRTAGS GRTAGS
GSYMS GSYMS

20
slsReceiverSoftware/Makefile
View File

@ -15,17 +15,31 @@ CFLAGS= -g -DC_ONLY -fPIC
DFLAGS= -g -DDACS_INT -DSLS_RECEIVER_UDP_FUNCTIONS DFLAGS= -g -DDACS_INT -DSLS_RECEIVER_UDP_FUNCTIONS
INCLUDES?= -Iinclude -I../slsDetectorCalibration -I$(ASM) INCLUDES?= $(INCLUDESRXR) -I include/ -I ../slsDetectorCalibration
#-Iinclude -I../slsDetectorCalibration -I$(ASM)
SRC_CLNT = MySocketTCP.cpp UDPInterface.cpp UDPBaseImplementation.cpp UDPStandardImplementation.cpp slsReceiverTCPIPInterface.cpp slsReceiver.cpp slsReceiverUsers.cpp utilities.cpp SRC_CLNT = MySocketTCP.cpp UDPInterface.cpp UDPBaseImplementation.cpp UDPStandardImplementation.cpp slsReceiverTCPIPInterface.cpp slsReceiver.cpp slsReceiverUsers.cpp utilities.cpp
ifeq ($(REST), yes)
SRC_CLNT += UDPRESTImplementation.cpp
endif
MAIN_SRC = main.cpp MAIN_SRC = main.cpp
OBJS=$(SRC_CLNT:%.cpp=$(BUILDDIR)/%.o) OBJS=$(SRC_CLNT:%.cpp=$(BUILDDIR)/%.o)
$(info )
$(info #######################################)
$(info # Compiling slsReceiverSoftware #)
$(info #######################################)
$(info )
.PHONY: all intdoc package eigerReceiver clean .PHONY: all intdoc package eigerReceiver clean
all: lib receiver all: builddir lib receiver
intdoc: $(SRC_H) $(SRC_CLNT) intdoc: $(SRC_H) $(SRC_CLNT)
doxygen doxy.config doxygen doxy.config
@ -69,6 +83,8 @@ clean: buildclean
rm $(DESTDIR)/libSlsReceiver.a $(DESTDIR)/libSlsReceiver.so rm $(DESTDIR)/libSlsReceiver.a $(DESTDIR)/libSlsReceiver.so
rm $(PROGS) rm $(PROGS)
builddir:
if [ ! -d $(BUILDDIR) ]; then mkdir $(BUILDDIR); fi
buildclean: buildclean:
rm -rf $(OBJS) rm -rf $(OBJS)

29
slsReceiverSoftware/include/RestHelper.h
View File

@ -19,17 +19,13 @@
#include "JsonBox/Value.h" #include "JsonBox/Value.h"
//#include "logger.h"
#include <iostream> #include <iostream>
#include <sstream> #include <sstream>
#include <string> #include <string>
#include <exception> #include <exception>
#define EIGER_DEBUG
#ifdef EIGER_DEBUG
#define DEBUG(x) do { std::cerr << "[DEBUG] " << x << std::endl; } while (0)
#else
#define DEBUG(x)
#endif
using namespace Poco::Net; using namespace Poco::Net;
@ -39,7 +35,7 @@ using namespace std;
class RestHelper { class RestHelper {
public: public:
RestHelper(int timeout=10, int n_tries=3){ RestHelper(int timeout=10, int n_tries=10){
/** /**
* *
* *
@ -78,9 +74,14 @@ class RestHelper {
*/ */
//Check for http:// string //Check for http:// string
FILE_LOG(logDEBUG) << __func__ << " starting";
string proto_str = "http://"; string proto_str = "http://";
if( size_t found = hostname.find(proto_str) != string::npos ){ if( size_t found = hostname.find(proto_str) != string::npos ){
cout << hostname << endl;
char c1[hostname.size()-found-1]; char c1[hostname.size()-found-1];
cout << c1 << endl;
size_t length1 = hostname.copy(c1, hostname.size()-found-1, proto_str.size()); size_t length1 = hostname.copy(c1, hostname.size()-found-1, proto_str.size());
c1[length1]='\0'; c1[length1]='\0';
hostname = c1; hostname = c1;
@ -89,7 +90,6 @@ class RestHelper {
full_hostname = "http://"+hostname; full_hostname = "http://"+hostname;
session = new HTTPClientSession(hostname, port ); session = new HTTPClientSession(hostname, port );
session->setKeepAliveTimeout( Timespan( http_timeout,0) ); session->setKeepAliveTimeout( Timespan( http_timeout,0) );
}; };
@ -168,7 +168,7 @@ class RestHelper {
string answer; string answer;
int code = send_request(session, req, &answer); int code = send_request(session, req, &answer);
if(code == 0 ) { if(code == 0 ) {
DEBUG("ANSWER " << answer ); FILE_LOG(logDEBUG) << "ANSWER " << answer;
json_value->loadFromString(answer); json_value->loadFromString(answer);
} }
delete uri; delete uri;
@ -176,7 +176,7 @@ class RestHelper {
}; };
int post_json(string request, string *answer, string request_body=""){ int post_json(string request, string *answer, string request_body="{}"){
/** /**
* *
* *
@ -192,15 +192,16 @@ class RestHelper {
if (path.empty()) path = "/"; if (path.empty()) path = "/";
HTTPRequest req(HTTPRequest::HTTP_POST, path, HTTPMessage::HTTP_1_1 ); HTTPRequest req(HTTPRequest::HTTP_POST, path, HTTPMessage::HTTP_1_1 );
req.setContentType("application/json\r\n"); req.setContentType("application/json\r\n");
req.setContentLength( request.length() ); cout << "REQUEST BODY " << request_body << endl;
req.setContentLength( request_body.length() );
int code = send_request(session, req, answer, request_body); int code = send_request(session, req, answer, request_body);
delete uri; delete uri;
return code; return code;
} }
int post_json(string request, JsonBox::Value* json_value, string request_body=""){ int post_json(string request, JsonBox::Value* json_value, string request_body="{}"){
/** /**
* *
* *
@ -283,7 +284,7 @@ class RestHelper {
return code; return code;
} }
catch (exception& e){ catch (exception& e){
cout << "Exception connecting to "<< full_hostname << ": "<< e.what() << ", sleeping 5 seconds (" << n << "/"<<n_connection_tries << ")" << endl; FILE_LOG(logERROR) << "Exception connecting to "<< full_hostname << ": "<< e.what() << ", sleeping 5 seconds (" << n << "/"<<n_connection_tries << ")";
sleep(5); sleep(5);
} }
n+=1; n+=1;

12
slsReceiverSoftware/include/UDPBaseImplementation.h
View File

@ -32,7 +32,7 @@
* @short does all the functions for a receiver, set/get parameters, start/stop etc. * @short does all the functions for a receiver, set/get parameters, start/stop etc.
*/ */
class UDPBaseImplementation : private virtual slsReceiverDefs, public UDPInterface { class UDPBaseImplementation : protected virtual slsReceiverDefs, public UDPInterface {
public: public:
/** /**
@ -45,6 +45,7 @@ class UDPBaseImplementation : private virtual slsReceiverDefs, public UDPInterfa
*/ */
virtual ~UDPBaseImplementation(); virtual ~UDPBaseImplementation();
void configure(map<string, string> config_map);
/** /**
@ -319,7 +320,7 @@ class UDPBaseImplementation : private virtual slsReceiverDefs, public UDPInterfa
*/ */
int shutDownUDPSockets(); int shutDownUDPSockets();
private: protected:
/* /*
void not_implemented(string method_name){ void not_implemented(string method_name){
@ -466,6 +467,10 @@ private:
void handleDataCompression(int ithread, char* wbuffer[], int &npackets, char* data, int xmax, int ymax, int &nf); void handleDataCompression(int ithread, char* wbuffer[], int &npackets, char* data, int xmax, int ymax, int &nf);
//// Could be done more fine-grained... TODO
// private:
protected:
/** structure of an eiger image header*/ /** structure of an eiger image header*/
typedef struct typedef struct
{ {
@ -683,6 +688,9 @@ private:
/** 10Gbe enable*/ /** 10Gbe enable*/
int tengigaEnable; int tengigaEnable;
// TODO: not properly sure where to put these...
/** structure of an eiger image header*/

11
slsReceiverSoftware/include/UDPInterface.h
View File

@ -13,6 +13,7 @@
* @short Base class with all the functions for the UDP inteface of the receiver * @short Base class with all the functions for the UDP inteface of the receiver
*/ */
#include <exception>
#include "sls_receiver_defs.h" #include "sls_receiver_defs.h"
#include "receiver_defs.h" #include "receiver_defs.h"
@ -24,6 +25,7 @@
class UDPInterface { class UDPInterface {
/* abstract class that defines the UDP interface of an sls detector data receiver. /* abstract class that defines the UDP interface of an sls detector data receiver.
* *
* Use the factory method UDPInterface::create() to get an instance: * Use the factory method UDPInterface::create() to get an instance:
@ -59,11 +61,6 @@ class UDPInterface {
public: public:
/**
* constructor
*/
//UDPInterface(){};
/** /**
* Destructor * Destructor
*/ */
@ -74,11 +71,11 @@ class UDPInterface {
*/ */
static UDPInterface *create(string receiver_type = "standard"); static UDPInterface *create(string receiver_type = "standard");
virtual void configure(map<string, string> config_map) = 0;
public: public:
/** /**
* Initialize the Receiver * Initialize the Receiver
@param detectorHostName detector hostname @param detectorHostName detector hostname

40
slsReceiverSoftware/include/UDPRESTImplementation.h
View File

@ -34,7 +34,7 @@
* @short does all the functions for a receiver, set/get parameters, start/stop etc. * @short does all the functions for a receiver, set/get parameters, start/stop etc.
*/ */
class UDPRESTImplementation : private virtual slsReceiverDefs, public UDPBaseImplementation { class UDPRESTImplementation : protected virtual slsReceiverDefs, public UDPBaseImplementation {
public: public:
/** /**
@ -48,6 +48,11 @@ class UDPRESTImplementation : private virtual slsReceiverDefs, public UDPBaseImp
virtual ~UDPRESTImplementation(); virtual ~UDPRESTImplementation();
protected:
void initialize_REST();
public:
void configure(map<string, string> config_map);
/** /**
* delete and free member parameters * delete and free member parameters
@ -57,14 +62,28 @@ class UDPRESTImplementation : private virtual slsReceiverDefs, public UDPBaseImp
/** /**
* initialize member parameters * initialize member parameters
*/ */
void initializeMembers(); //void initializeMembers();
/**
* Set detector hostname
* @param c hostname
*/
//void initialize(const char *detectorHostName);
/* Returns detector hostname
/returns hostname
* caller needs to deallocate the returned char array.
* if uninitialized, it must return NULL
*/
//char *getDetectorHostname() const;
/** /**
* Set receiver type * Set receiver type
* @param det detector type * @param det detector type
* Returns success or FAIL * Returns success or FAIL
*/ */
int setDetectorType(detectorType det); //int setDetectorType(detectorType det);
//Frame indices and numbers caught //Frame indices and numbers caught
@ -192,18 +211,6 @@ class UDPRESTImplementation : private virtual slsReceiverDefs, public UDPBaseImp
*/ */
runStatus getStatus() const; runStatus getStatus() const;
/**
* Set detector hostname
* @param c hostname
*/
void initialize(const char *detectorHostName);
/* Returns detector hostname
/returns hostname
* caller needs to deallocate the returned char array.
* if uninitialized, it must return NULL
*/
//char *getDetectorHostname() const;
/** /**
* Set Ethernet Interface or IP to listen to * Set Ethernet Interface or IP to listen to
@ -811,6 +818,9 @@ public:
//REST specific //REST specific
bool isInitialized; bool isInitialized;
RestHelper * rest ; RestHelper * rest ;
int rest_port; // receiver backend port
string rest_hostname; // receiver hostname
}; };

302
slsReceiverSoftware/include/UDPStandardImplementation.h
View File

@ -466,308 +466,6 @@ private:
void handleDataCompression(int ithread, char* wbuffer[], int &npackets, char* data, int xmax, int ymax, int &nf); void handleDataCompression(int ithread, char* wbuffer[], int &npackets, char* data, int xmax, int ymax, int &nf);
/** structure of an eiger image header*/
typedef struct
{
unsigned char header_before[20];
unsigned char fnum[4];
unsigned char header_after[24];
} eiger_image_header;
/** structure of an eiger image header*/
typedef struct
{
unsigned char num1[4];
unsigned char num2[4];
} eiger_packet_header;
/** max number of listening threads */
const static int MAX_NUM_LISTENING_THREADS = EIGER_MAX_PORTS;
/** max number of writer threads */
const static int MAX_NUM_WRITER_THREADS = 15;
/** detector type */
detectorType myDetectorType;
/** detector hostname */
char detHostname[MAX_STR_LENGTH];
/** status of receiver */
runStatus status;
/** UDP Socket between Receiver and Detector */
genericSocket* udpSocket[MAX_NUM_LISTENING_THREADS];
/** Server UDP Port*/
int server_port[MAX_NUM_LISTENING_THREADS];
/** ethernet interface or IP to listen to */
char *eth;
/** max packets per file **/
int maxPacketsPerFile;
/** File write enable */
int enableFileWrite;
/** File over write enable */
int overwrite;
/** Complete File name */
char savefilename[MAX_STR_LENGTH];
/** File Name without frame index, file index and extension*/
char fileName[MAX_STR_LENGTH];
/** File Path */
char filePath[MAX_STR_LENGTH];
/** File Index */
int fileIndex;
/** scan tag */
int scanTag;
/** if frame index required in file name */
int frameIndexNeeded;
/* Acquisition started */
bool acqStarted;
/* Measurement started */
bool measurementStarted;
/** Frame index at start of each real time acquisition (eg. for each scan) */
uint32_t startFrameIndex;
/** Actual current frame index of each time acquisition (eg. for each scan) */
uint32_t frameIndex;
/** Frames Caught for each real time acquisition (eg. for each scan) */
int packetsCaught;
/** Total packets caught for an entire acquisition (including all scans) */
int totalPacketsCaught;
/** Pckets currently in current file, starts new file when it reaches max */
int packetsInFile;
/** Frame index at start of an entire acquisition (including all scans) */
uint32_t startAcquisitionIndex;
/** Actual current frame index of an entire acquisition (including all scans) */
uint32_t acquisitionIndex;
/** number of packets per frame*/
int packetsPerFrame;
/** frame index mask */
uint32_t frameIndexMask;
/** packet index mask */
uint32_t packetIndexMask;
/** frame index offset */
int frameIndexOffset;
/** acquisition period */
int64_t acquisitionPeriod;
/** frame number */
int32_t numberOfFrames;
/** dynamic range */
int dynamicRange;
/** short frames */
int shortFrame;
/** current frame number */
uint32_t currframenum;
/** Previous Frame number from buffer */
uint32_t prevframenum;
/** size of one frame */
int frameSize;
/** buffer size. different from framesize as we wait for one packet instead of frame for eiger */
int bufferSize;
/** oen buffer size */
int onePacketSize;
/** latest data */
char* latestData;
/** gui data ready */
int guiDataReady;
/** points to the data to send to gui */
char* guiData;
/** points to the filename to send to gui */
char* guiFileName;
/** temporary number for eiger frame number as its not included in the packet */
uint32_t guiFrameNumber;
/** send every nth frame to gui or only upon gui request*/
int nFrameToGui;
/** fifo size */
unsigned int fifosize;
/** number of jobs per thread for data compression */
int numJobsPerThread;
/** datacompression - save only hits */
bool dataCompression;
/** memory allocated for the buffer */
char *mem0[MAX_NUM_LISTENING_THREADS];
/** circular fifo to store addresses of data read */
CircularFifo<char>* fifo[MAX_NUM_LISTENING_THREADS];
/** circular fifo to store addresses of data already written and ready to be resued*/
CircularFifo<char>* fifoFree[MAX_NUM_LISTENING_THREADS];
/** Receiver buffer */
char *buffer[MAX_NUM_LISTENING_THREADS];
/** number of writer threads */
int numListeningThreads;
/** number of writer threads */
int numWriterThreads;
/** to know if listening and writer threads created properly */
int thread_started;
/** current listening thread index*/
int currentListeningThreadIndex;
/** current writer thread index*/
int currentWriterThreadIndex;
/** thread listening to packets */
pthread_t listening_thread[MAX_NUM_LISTENING_THREADS];
/** thread writing packets */
pthread_t writing_thread[MAX_NUM_WRITER_THREADS];
/** total frame count the listening thread has listened to */
int totalListeningFrameCount[MAX_NUM_LISTENING_THREADS];
/** mask showing which listening threads are running */
volatile uint32_t listeningthreads_mask;
/** mask showing which writer threads are running */
volatile uint32_t writerthreads_mask;
/** mask showing which threads have created files*/
volatile uint32_t createfile_mask;
/** OK if file created was successful */
int ret_createfile;
/** variable used to self terminate threads waiting for semaphores */
int killAllListeningThreads;
/** variable used to self terminate threads waiting for semaphores */
int killAllWritingThreads;
/** 10Gbe enable*/
int tengigaEnable;
//semaphores
/** semaphore to synchronize writer and guireader threads */
sem_t smp;
/** semaphore to synchronize listener threads */
sem_t listensmp[MAX_NUM_LISTENING_THREADS];
/** semaphore to synchronize writer threads */
sem_t writersmp[MAX_NUM_WRITER_THREADS];
//mutex
/** guiDataReady mutex */
pthread_mutex_t dataReadyMutex;
/** mutex for status */
pthread_mutex_t status_mutex;
/** mutex for progress variable currframenum */
pthread_mutex_t progress_mutex;
/** mutex for writing data to file */
pthread_mutex_t write_mutex;
/** File Descriptor */
FILE *sfilefd;
//filter
singlePhotonDetector<uint16_t> *singlePhotonDet[MAX_NUM_WRITER_THREADS];
slsReceiverData<uint16_t> *receiverdata[MAX_NUM_WRITER_THREADS];
moenchCommonMode *cmSub;
bool commonModeSubtractionEnable;
#ifdef MYROOT1
/** Tree where the hits are stored */
TTree *myTree[MAX_NUM_WRITER_THREADS];
/** File where the tree is saved */
TFile *myFile[MAX_NUM_WRITER_THREADS];
#endif
/**
callback arguments are
filepath
filename
fileindex
data size
return value is
0 callback takes care of open,close,write file
1 callback writes file, we have to open, close it
2 we open, close, write file, callback does not do anything
*/
int (*startAcquisitionCallBack)(char*, char*,int, int, void*);
void *pStartAcquisition;
/**
args to acquisition finished callback
total frames caught
*/
void (*acquisitionFinishedCallBack)(int, void*);
void *pAcquisitionFinished;
/**
args to raw data ready callback are
framenum
datapointer
datasize in bytes
file descriptor
guidatapointer (NULL, no data required)
*/
void (*rawDataReadyCallBack)(int, char*, int, FILE*, char*, void*);
void *pRawDataReady;
/** The action which decides what the user and default responsibilites to save data are
* 0 raw data ready callback takes care of open,close,write file
* 1 callback writes file, we have to open, close it
* 2 we open, close, write file, callback does not do anything */
int cbAction;
public: public:

13
slsReceiverSoftware/include/logger.h
View File

@ -5,6 +5,19 @@
#include <string> #include <string>
#include <stdio.h> #include <stdio.h>
#define STRINGIFY(x) #x
#define TOSTRING(x) STRINGIFY(x)
#define MYCONCAT(x,y)
#define __AT__ string(__FILE__) + string("::") + string(__func__) + string("(): ")
//":" TOSTRING(__LINE__)
/*
void error(const char *location, const char *msg){
printf("Error at %s: %s\n", location, msg);
}
*/
inline std::string NowTime(); inline std::string NowTime();
enum TLogLevel {logERROR, logWARNING, logINFO, logDEBUG, logDEBUG1, logDEBUG2, logDEBUG3, logDEBUG4}; enum TLogLevel {logERROR, logWARNING, logINFO, logDEBUG, logDEBUG1, logDEBUG2, logDEBUG3, logDEBUG4};

4
slsReceiverSoftware/include/utilities.h
View File

@ -2,6 +2,8 @@
#include <string> #include <string>
#include <sstream> #include <sstream>
#include <iostream> #include <iostream>
#include <map>
using namespace std; using namespace std;
#include "sls_receiver_defs.h" #include "sls_receiver_defs.h"
@ -9,5 +11,5 @@ using namespace std;
//#define EIGER_DEBUG //#define EIGER_DEBUG
int read_config_file(string fname, int *tcpip_port_no); int read_config_file(string fname, int *tcpip_port_no, map<string, string> * configuration_map);

157
slsReceiverSoftware/src/UDPBaseImplementation.cpp
View File

@ -35,6 +35,11 @@ UDPBaseImplementation::UDPBaseImplementation(){}
UDPBaseImplementation::~UDPBaseImplementation(){} UDPBaseImplementation::~UDPBaseImplementation(){}
void UDPBaseImplementation::configure(map<string, string> config_map){
FILE_LOG(logWARNING) << __AT__ << "doing nothing...";
};
void UDPBaseImplementation::deleteMembers(){ void UDPBaseImplementation::deleteMembers(){
FILE_LOG(logWARNING) << "[WARNING] This is a base implementation, " << __func__ << " could have no effects."; FILE_LOG(logWARNING) << "[WARNING] This is a base implementation, " << __func__ << " could have no effects.";
} }
@ -47,6 +52,92 @@ void UDPBaseImplementation::initializeMembers(){
int UDPBaseImplementation::setDetectorType(detectorType det){ int UDPBaseImplementation::setDetectorType(detectorType det){
cout << "[WARNING] This is a base implementation, " << __func__ << " not correctly implemented" << endl; cout << "[WARNING] This is a base implementation, " << __func__ << " not correctly implemented" << endl;
cout << "Setting Receiver Type " << endl;
deleteMembers();
initializeMembers();
myDetectorType = det;
switch(myDetectorType){
case GOTTHARD:
cout << endl << "***** This is a GOTTHARD Receiver *****" << endl << endl;
break;
case MOENCH:
cout << endl << "***** This is a MOENCH Receiver *****" << endl << endl;
break;
case EIGER:
cout << endl << "***** This is a EIGER Receiver *****" << endl << endl;
break;
default:
cout << endl << "***** Unknown Receiver *****" << endl << endl;
return FAIL;
break;
}
//moench variables
if(myDetectorType == GOTTHARD){
fifosize = GOTTHARD_FIFO_SIZE;
packetsPerFrame = GOTTHARD_PACKETS_PER_FRAME;
onePacketSize = GOTTHARD_ONE_PACKET_SIZE;
frameSize = GOTTHARD_BUFFER_SIZE;
bufferSize = GOTTHARD_BUFFER_SIZE;
maxPacketsPerFile = MAX_FRAMES_PER_FILE * GOTTHARD_PACKETS_PER_FRAME;
frameIndexMask = GOTTHARD_FRAME_INDEX_MASK;
frameIndexOffset = GOTTHARD_FRAME_INDEX_OFFSET;
packetIndexMask = GOTTHARD_PACKET_INDEX_MASK;
}else if(myDetectorType == MOENCH){
fifosize = MOENCH_FIFO_SIZE;
packetsPerFrame = MOENCH_PACKETS_PER_FRAME;
onePacketSize = MOENCH_ONE_PACKET_SIZE;
frameSize = MOENCH_BUFFER_SIZE;
bufferSize = MOENCH_BUFFER_SIZE;
maxPacketsPerFile = MOENCH_MAX_FRAMES_PER_FILE * MOENCH_PACKETS_PER_FRAME;
frameIndexMask = MOENCH_FRAME_INDEX_MASK;
frameIndexOffset = MOENCH_FRAME_INDEX_OFFSET;
packetIndexMask = MOENCH_PACKET_INDEX_MASK;
}
else if(myDetectorType == EIGER){
fifosize = EIGER_FIFO_SIZE;
packetsPerFrame = EIGER_ONE_GIGA_CONSTANT * dynamicRange * EIGER_MAX_PORTS;
onePacketSize = EIGER_ONE_GIGA_ONE_PACKET_SIZE;
frameSize = onePacketSize * packetsPerFrame;
bufferSize = (frameSize/EIGER_MAX_PORTS) + EIGER_HEADER_LENGTH;//everything one port gets (img header plus packets)
maxPacketsPerFile = EIGER_MAX_FRAMES_PER_FILE * packetsPerFrame;
frameIndexMask = EIGER_FRAME_INDEX_MASK;
frameIndexOffset = EIGER_FRAME_INDEX_OFFSET;
packetIndexMask = EIGER_PACKET_INDEX_MASK;
pthread_mutex_lock(&status_mutex);
listeningthreads_mask = 0x0;
pthread_mutex_unlock(&(status_mutex));
if(thread_started)
createListeningThreads(true);
numListeningThreads = MAX_NUM_LISTENING_THREADS;
}
latestData = new char[frameSize];
setupFifoStructure();
if(createListeningThreads() == FAIL){
cout << "ERROR: Could not create listening thread" << endl;
exit (-1);
}
if(createWriterThreads() == FAIL){
cout << "ERROR: Could not create writer threads" << endl;
exit (-1);
}
setThreadPriorities();
cout << "Ready..." << endl;
return OK;
return OK; return OK;
} }
@ -94,11 +185,11 @@ void UDPBaseImplementation::resetTotalFramesCaught(){
/*file parameters*/ /*file parameters*/
char* UDPBaseImplementation::getFilePath() const{ char* UDPBaseImplementation::getFilePath() const{
FILE_LOG(logWARNING) << "[WARNING] This is a base implementation, " << __func__ << " could have no effects.";
return (char*)filePath; return (char*)filePath;
} }
char* UDPBaseImplementation::setFilePath(const char c[]){ inline char* UDPBaseImplementation::setFilePath(const char c[]){
FILE_LOG(logDEBUG) << __AT__ << "called";
if(strlen(c)){ if(strlen(c)){
//check if filepath exists //check if filepath exists
struct stat st; struct stat st;
@ -109,6 +200,8 @@ char* UDPBaseImplementation::setFilePath(const char c[]){
FILE_LOG(logWARNING) << "FilePath does not exist:" << filePath; FILE_LOG(logWARNING) << "FilePath does not exist:" << filePath;
} }
} }
FILE_LOG(logDEBUG) << __AT__ << getFilePath();
cout << getFilePath() << " " << filePath << endl;
return getFilePath(); return getFilePath();
} }
@ -117,7 +210,7 @@ char* UDPBaseImplementation::getFileName() const{
return (char*)fileName; return (char*)fileName;
} }
char* UDPBaseImplementation::setFileName(const char c[]){ inline char* UDPBaseImplementation::setFileName(const char c[]){
//cout << "[WARNING] This is a base implementation, " << __func__ << " could have no effects." << endl; //cout << "[WARNING] This is a base implementation, " << __func__ << " could have no effects." << endl;
if(strlen(c)) if(strlen(c))
@ -132,17 +225,15 @@ int UDPBaseImplementation::getFileIndex(){
} }
int UDPBaseImplementation::setFileIndex(int i){ int UDPBaseImplementation::setFileIndex(int i){
cout << "[WARNING] This is a base implementation, " << __func__ << " could have no effects." << endl; //cout << "[WARNING] This is a base implementation, " << __func__ << " could have no effects." << endl;
/*
if(i>=0) if(i>=0)
fileIndex = i; fileIndex = i;
*/
return getFileIndex(); return getFileIndex();
} }
int UDPBaseImplementation::setFrameIndexNeeded(int i){ int UDPBaseImplementation::setFrameIndexNeeded(int i){
cout << "[WARNING] This is a base implementation, " << __func__ << " could have no effects." << endl; //cout << "[WARNING] This is a base implementation, " << __func__ << " could have no effects." << endl;
frameIndexNeeded = i; frameIndexNeeded = i;
return frameIndexNeeded; return frameIndexNeeded;
} }
@ -628,6 +719,7 @@ void UDPBaseImplementation::copyFrameToGui(char* startbuf[], uint32_t fnum, char
int UDPBaseImplementation::createUDPSockets(){ int UDPBaseImplementation::createUDPSockets(){
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " starting";
//if eth is mistaken with ip address //if eth is mistaken with ip address
@ -871,8 +963,10 @@ int UDPBaseImplementation::setupWriter(){
//acquisition start call back returns enable write //acquisition start call back returns enable write
if (startAcquisitionCallBack) if (startAcquisitionCallBack){
cout << filePath << " - " << fileName << endl;
cbAction=startAcquisitionCallBack(filePath,fileName,fileIndex,bufferSize,pStartAcquisition); cbAction=startAcquisitionCallBack(filePath,fileName,fileIndex,bufferSize,pStartAcquisition);
}
if(cbAction < DO_EVERYTHING) if(cbAction < DO_EVERYTHING)
cout << endl << "Note: Call back activated. Data saving must be taken care of by user in call back." << endl; cout << endl << "Note: Call back activated. Data saving must be taken care of by user in call back." << endl;
@ -952,6 +1046,11 @@ int UDPBaseImplementation::createCompressionFile(int ithr, int iframe){
int UDPBaseImplementation::createNewFile(){ int UDPBaseImplementation::createNewFile(){
cout << "[WARNING] This is a base implementation, " << __func__ << " not correctly implemented" << endl;
/*
int gt = getFrameIndex(); int gt = getFrameIndex();
if(gt==-1) gt=0; if(gt==-1) gt=0;
//create file name //create file name
@ -960,6 +1059,8 @@ int UDPBaseImplementation::createNewFile(){
else else
sprintf(savefilename, "%s/%s_f%012d_%d.raw", filePath,fileName,(packetsCaught/packetsPerFrame),fileIndex); sprintf(savefilename, "%s/%s_f%012d_%d.raw", filePath,fileName,(packetsCaught/packetsPerFrame),fileIndex);
cout << filePath << " + " << fileName << endl;
//if filewrite and we are allowed to write //if filewrite and we are allowed to write
if(enableFileWrite && cbAction > DO_NOTHING){ if(enableFileWrite && cbAction > DO_NOTHING){
//close //close
@ -974,7 +1075,7 @@ int UDPBaseImplementation::createNewFile(){
return FAIL; return FAIL;
} }
}else if (NULL == (sfilefd = fopen((const char *) (savefilename), "w"))){ }else if (NULL == (sfilefd = fopen((const char *) (savefilename), "w"))){
cout << "Error: Could not create file " << savefilename << endl; cout << "Error: Could not creat dsdasdserwe file " << savefilename << endl;
return FAIL; return FAIL;
} }
//setting buffer //setting buffer
@ -1001,21 +1102,16 @@ int UDPBaseImplementation::createNewFile(){
prevframenum = currframenum; prevframenum = currframenum;
packetsInFile = 0; packetsInFile = 0;
} }
*/
return OK; return OK;
} }
// This is actually called on CTRL-C
void UDPBaseImplementation::closeFile(int ithr)
{
FILE_LOG(logDEBUG) << __AT__ << "called";
void UDPBaseImplementation::closeFile(int ithr){
#ifdef VERBOSE
cout << "In closeFile for thread " << ithr << endl;
#endif
if(!dataCompression){ if(!dataCompression){
if(sfilefd){ if(sfilefd){
@ -1063,6 +1159,9 @@ void UDPBaseImplementation::closeFile(int ithr){
#endif #endif
} }
FILE_LOG(logDEBUG) << __AT__ << "exited";
} }
@ -2022,12 +2121,24 @@ int UDPBaseImplementation::enableTenGiga(int enable){
createWriterThreads(true); createWriterThreads(true);
} }
for(int i=0;i<numListeningThreads;i++){ for(int i=0;i<numListeningThreads;i++){
if(mem0[i]) {free(mem0[i]); mem0[i] = NULL;} if(mem0[i]){
if(fifo[i]) {delete fifo[i]; fifo[i] = NULL;} free(mem0[i]);
if(fifoFree[i]) {delete fifoFree[i]; fifoFree[i] = NULL;} mem0[i] = NULL;
}
if(fifo[i]){
delete fifo[i];
fifo[i] = NULL;
}
if(fifoFree[i]) {
delete fifoFree[i];
fifoFree[i] = NULL;
}
buffer[i] = NULL; buffer[i] = NULL;
} }
if(latestData) {delete [] latestData; latestData = NULL;} if(latestData){
delete [] latestData;
latestData = NULL;
}
latestData = new char[frameSize]; latestData = new char[frameSize];
numJobsPerThread = -1; numJobsPerThread = -1;

19
slsReceiverSoftware/src/UDPInterface.cpp
View File

@ -6,16 +6,6 @@
/*
#include <signal.h> // SIGINT
#include <sys/stat.h> // stat
#include <sys/socket.h> // socket(), bind(), listen(), accept(), shut down
#include <arpa/inet.h> // sock_addr_in, htonl, INADDR_ANY
#include <stdlib.h> // exit()
#include <iomanip> //set precision
#include <sys/mman.h> //munmap
*/
#include <string.h> #include <string.h>
#include <iostream> #include <iostream>
using namespace std; using namespace std;
@ -23,6 +13,7 @@ using namespace std;
#include "UDPInterface.h" #include "UDPInterface.h"
#include "UDPBaseImplementation.h" #include "UDPBaseImplementation.h"
#include "UDPStandardImplementation.h" #include "UDPStandardImplementation.h"
#include "UDPRESTImplementation.h"
@ -34,11 +25,11 @@ UDPInterface * UDPInterface::create(string receiver_type){
cout << "Starting " << receiver_type << endl; cout << "Starting " << receiver_type << endl;
return new UDPStandardImplementation(); return new UDPStandardImplementation();
} }
//#ifdef REST
#ifdef REST else if (receiver_type == "REST"){
else if (receiver_type == "REST")
return new UDPRESTImplementation(); return new UDPRESTImplementation();
#endif }
//#endif
else{ else{
FILE_LOG(logWARNING) << "[ERROR] UDP interface not supported, using standard implementation"; FILE_LOG(logWARNING) << "[ERROR] UDP interface not supported, using standard implementation";
return new UDPBaseImplementation(); return new UDPBaseImplementation();

352
slsReceiverSoftware/src/UDPRESTImplementation.cpp
View File

@ -22,48 +22,110 @@
#include <string.h> #include <string.h>
#include <iostream> #include <iostream>
#include <sstream>
//#include "utilities.h" //#include "utilities.h"
using namespace std; using namespace std;
/*
TODO
+ filePath != getFilePath
+ better state handling. Now it is only IDLE - RUNNING - IDLE
*/
UDPRESTImplementation::UDPRESTImplementation() : isInitialized(false), status(slsReceiverDefs::ERROR) {} UDPRESTImplementation::UDPRESTImplementation(){
FILE_LOG(logDEBUG) << __AT__ << " called";
// Default values
rest_hostname = "localhost";
rest_port = 8081;
}
UDPRESTImplementation::~UDPRESTImplementation(){} UDPRESTImplementation::~UDPRESTImplementation(){
delete rest;
}
void UDPRESTImplementation::initialize(const char *detectorHostName){ void UDPRESTImplementation::configure(map<string, string> config_map){
FILE_LOG(logWARNING) << __AT__ << " called";
string name; map<string, string>::const_iterator pos;
if (detectorHostName != NULL)
name = detectorHostName;
if (name.empty()) { pos = config_map.find("rest_hostname");
FILE_LOG(logDEBUG) << "initialize(): can't initialize with empty string or NULL for detectorHostname"; if (pos != config_map.end() ){
} else if (isInitialized == true) { string host_port_str = pos->second;
FILE_LOG(logDEBUG) << "initialize(): already initialized, can't initialize several times"; std::size_t pos = host_port_str.find(":"); // position of "live" in str
} else { if(pos != string::npos){
FILE_LOG(logDEBUG) << "initialize(): initialize() with: detectorHostName=" << name; istringstream (host_port_str.substr (pos)) >> rest_port;
strcpy(detHostname,detectorHostName); rest_hostname = host_port_str.substr(0, pos);
//init_config.detectorHostname = name; }
}
//REST call - hardcoded /*
//RestHelper rest ; for(map<string, string>::const_iterator i=config_map.begin(); i != config_map.end(); i++){
rest->init(detHostname, 8080); std::cout << i->first << " " << i->second<< std::endl;
}
*/
};
void UDPRESTImplementation::initialize_REST(){
FILE_LOG(logDEBUG) << __AT__ << " called";
if (rest_hostname.empty()) {
FILE_LOG(logDEBUG) << __AT__ <<"can't initialize with empty string or NULL for detectorHostname";
}
else if (isInitialized == true) {
FILE_LOG(logDEBUG) << __AT__ << "already initialized, can't initialize several times";
}
else {
FILE_LOG(logDEBUG) << __AT__ << "with receiverHostName=" << rest_hostname << ":" << rest_port;
rest = new RestHelper() ;
std::string answer; std::string answer;
int code = rest->get_json("status", &answer); int code;
try{
rest->init(rest_hostname, rest_port);
code = rest->get_json("state", &answer);
if (code != 0){ if (code != 0){
//throw -1; throw answer;
std::cout << "I SHOULD THROW AN EXCEPTION!!!" << std::endl;
} }
else{ else{
isInitialized = true; isInitialized = true;
status = slsReceiverDefs::IDLE; status = slsReceiverDefs::IDLE;
} }
std::cout << "Answer: " << answer << std::endl; FILE_LOG(logDEBUG) << __func__ << "Answer: " << answer;
}
catch(std::string e){
FILE_LOG(logERROR) << __func__ << ": " << e;
throw;
}
//JsonBox::Object json_object;
//json_object["configfile"] = JsonBox::Value("FILENAME");
JsonBox::Value json_request;
//json_request["configfile"] = "config.py";
json_request["path"] = filePath;
stringstream ss;
string test;
std::cout << "GetSTring: " << json_request << std::endl;
json_request.writeToStream(ss, false);
//ss << json_request;
ss >> test;
test = "{\"path\":\"" + string( getFilePath() ) + "\"}";
code = rest->post_json("state/initialize", &answer, test);
FILE_LOG(logDEBUG) << __AT__ << "state/configure got " << code;
code = rest->get_json("state", &answer);
FILE_LOG(logDEBUG) << __AT__ << "state got " << code << " " << answer;
/* /*
@ -73,29 +135,54 @@ void UDPRESTImplementation::initialize(const char *detectorHostName){
std::cout << "JSON " << json_value["status"] << std::endl; std::cout << "JSON " << json_value["status"] << std::endl;
*/ */
} }
FILE_LOG(logDEBUG) << __func__ << ": initialize() done";
} }
/*
int UDPRESTImplementation::setDetectorType(detectorType det){ int UDPRESTImplementation::setDetectorType(detectorType det){
cout << "[WARNING] This is a base implementation, " << __func__ << " not correctly implemented" << endl; cout << "[WARNING] This is a base implementation, " << __func__ << " not correctly implemented" << endl;
return OK; return OK;
} }
*/
/*Frame indices and numbers caught*/ /*Frame indices and numbers caught*/
bool UDPRESTImplementation::getAcquistionStarted(){return acqStarted;}; bool UDPRESTImplementation::getAcquistionStarted(){
FILE_LOG(logDEBUG) << __AT__ << " called";
return acqStarted;
};
bool UDPRESTImplementation::getMeasurementStarted(){return measurementStarted;}; bool UDPRESTImplementation::getMeasurementStarted(){
FILE_LOG(logDEBUG) << __AT__ << " called";
return measurementStarted;
};
int UDPRESTImplementation::getFramesCaught(){return (packetsCaught/packetsPerFrame);} int UDPRESTImplementation::getFramesCaught(){
FILE_LOG(logDEBUG) << __AT__ << " called";
return (packetsCaught/packetsPerFrame);
}
int UDPRESTImplementation::getTotalFramesCaught(){return (totalPacketsCaught/packetsPerFrame);} int UDPRESTImplementation::getTotalFramesCaught(){
FILE_LOG(logDEBUG) << __AT__ << " called";
if (packetsPerFrame == 0){
FILE_LOG(logWARNING) << __AT__ << " packetsPerFrame is 0!!!";
return 0;
}
return (totalPacketsCaught/packetsPerFrame);
}
uint32_t UDPRESTImplementation::getStartFrameIndex(){return startFrameIndex;} uint32_t UDPRESTImplementation::getStartFrameIndex(){
FILE_LOG(logDEBUG) << __AT__ << " called";
return startFrameIndex;
}
uint32_t UDPRESTImplementation::getFrameIndex(){ uint32_t UDPRESTImplementation::getFrameIndex(){
FILE_LOG(logDEBUG) << __AT__ << " called";
if(!packetsCaught) if(!packetsCaught)
frameIndex=-1; frameIndex=-1;
else else
@ -105,15 +192,22 @@ uint32_t UDPRESTImplementation::getFrameIndex(){
uint32_t UDPRESTImplementation::getAcquisitionIndex(){ uint32_t UDPRESTImplementation::getAcquisitionIndex(){
//FILE_LOG(logDEBUG) << __AT__ << " called, idx: " << acquisitionIndex;
if(!totalPacketsCaught) if(!totalPacketsCaught)
acquisitionIndex = -1; acquisitionIndex = -1;
else else
acquisitionIndex = currframenum - startAcquisitionIndex; acquisitionIndex = currframenum - startAcquisitionIndex;
//FILE_LOG(logDEBUG) << __AT__ << " idx: " << acquisitionIndex
// << " currframenum: " << currframenum
// << " startAcqIdx: " << startAcquisitionIndex;
return acquisitionIndex; return acquisitionIndex;
} }
void UDPRESTImplementation::resetTotalFramesCaught(){ void UDPRESTImplementation::resetTotalFramesCaught(){
FILE_LOG(logDEBUG) << __AT__ << " called";
acqStarted = false; acqStarted = false;
startAcquisitionIndex = 0; startAcquisitionIndex = 0;
totalPacketsCaught = 0; totalPacketsCaught = 0;
@ -122,21 +216,22 @@ void UDPRESTImplementation::resetTotalFramesCaught(){
/*file parameters*/ /*file parameters*/
int UDPRESTImplementation::getFileIndex(){ int UDPRESTImplementation::getFileIndex(){
FILE_LOG(logDEBUG) << __AT__ << " called";
return fileIndex; return fileIndex;
} }
int UDPRESTImplementation::setFileIndex(int i){ int UDPRESTImplementation::setFileIndex(int i){
cout << "[WARNING] This is a base implementation, " << __func__ << " could have no effects." << endl; FILE_LOG(logDEBUG) << __AT__ << " called";
/*
if(i>=0) if(i>=0)
fileIndex = i; fileIndex = i;
*/
return getFileIndex(); return getFileIndex();
} }
int UDPRESTImplementation::setFrameIndexNeeded(int i){ int UDPRESTImplementation::setFrameIndexNeeded(int i){
cout << "[WARNING] This is a base implementation, " << __func__ << " could have no effects." << endl; FILE_LOG(logDEBUG) << __AT__ << " called";
frameIndexNeeded = i; frameIndexNeeded = i;
return frameIndexNeeded; return frameIndexNeeded;
} }
@ -168,6 +263,7 @@ int UDPRESTImplementation::setEnableOverwrite(int i){
/*other parameters*/ /*other parameters*/
slsReceiverDefs::runStatus UDPRESTImplementation::getStatus() const{ slsReceiverDefs::runStatus UDPRESTImplementation::getStatus() const{
FILE_LOG(logDEBUG) << __AT__ << " called, status: " << status;
return status; return status;
} }
@ -180,11 +276,16 @@ char *UDPRESTImplementation::getDetectorHostname() const{
*/ */
void UDPRESTImplementation::setEthernetInterface(char* c){ void UDPRESTImplementation::setEthernetInterface(char* c){
strcpy(eth,c); FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " starting";
// TODO: this segfaults
//strcpy(eth,c);
//FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " done";
} }
void UDPRESTImplementation::setUDPPortNo(int p){ void UDPRESTImplementation::setUDPPortNo(int p){
FILE_LOG(logDEBUG) << __AT__ << " called";
for(int i=0;i<numListeningThreads;i++){ for(int i=0;i<numListeningThreads;i++){
server_port[i] = p+i; server_port[i] = p+i;
} }
@ -219,19 +320,23 @@ int32_t UDPRESTImplementation::setScanTag(int32_t stag){
*/ */
int32_t UDPRESTImplementation::setDynamicRange(int32_t dr){ int32_t UDPRESTImplementation::setDynamicRange(int32_t dr){
cout << "Setting Dynamic Range" << endl; FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " starting";
int olddr = dynamicRange; int olddr = dynamicRange;
if(dr >= 0){ if(dr >= 0){
dynamicRange = dr; dynamicRange = dr;
} }
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " " << getDynamicRange();
return getDynamicRange(); return getDynamicRange();
} }
int UDPRESTImplementation::setShortFrame(int i){ int UDPRESTImplementation::setShortFrame(int i){
FILE_LOG(logDEBUG) << __AT__ << " called";
shortFrame=i; shortFrame=i;
if(shortFrame!=-1){ if(shortFrame!=-1){
@ -262,6 +367,7 @@ int UDPRESTImplementation::setShortFrame(int i){
int UDPRESTImplementation::setNFrameToGui(int i){ int UDPRESTImplementation::setNFrameToGui(int i){
FILE_LOG(logDEBUG) << __AT__ << " called";
if(i>=0){ if(i>=0){
nFrameToGui = i; nFrameToGui = i;
setupFifoStructure(); setupFifoStructure();
@ -272,7 +378,7 @@ int UDPRESTImplementation::setNFrameToGui(int i){
int64_t UDPRESTImplementation::setAcquisitionPeriod(int64_t index){ int64_t UDPRESTImplementation::setAcquisitionPeriod(int64_t index){
FILE_LOG(logDEBUG) << __AT__ << " called";
if(index >= 0){ if(index >= 0){
if(index != acquisitionPeriod){ if(index != acquisitionPeriod){
acquisitionPeriod = index; acquisitionPeriod = index;
@ -283,62 +389,23 @@ int64_t UDPRESTImplementation::setAcquisitionPeriod(int64_t index){
} }
bool UDPRESTImplementation::getDataCompression(){return dataCompression;} bool UDPRESTImplementation::getDataCompression(){
FILE_LOG(logDEBUG) << __AT__ << " called";
return dataCompression;
}
int UDPRESTImplementation::enableDataCompression(bool enable){ int UDPRESTImplementation::enableDataCompression(bool enable){
cout << "Data compression "; FILE_LOG(logDEBUG) << __AT__ << " called, doing nothing";
if(enable)
cout << "enabled" << endl;
else
cout << "disabled" << endl;
#ifdef MYROOT1
cout << " WITH ROOT" << endl;
#else
cout << " WITHOUT ROOT" << endl;
#endif
//delete filter for the current number of threads
deleteFilter();
dataCompression = enable;
pthread_mutex_lock(&status_mutex);
writerthreads_mask = 0x0;
pthread_mutex_unlock(&(status_mutex));
createWriterThreads(true);
if(enable)
numWriterThreads = MAX_NUM_WRITER_THREADS;
else
numWriterThreads = 1;
if(createWriterThreads() == FAIL){
cout << "ERROR: Could not create writer threads" << endl;
return FAIL;
}
setThreadPriorities();
if(enable)
setupFilter();
return OK; return OK;
} }
/*other functions*/ /*other functions*/
void UDPRESTImplementation::deleteFilter(){ void UDPRESTImplementation::deleteFilter(){
FILE_LOG(logDEBUG) << __AT__ << " called";
int i; int i;
cmSub=NULL; cmSub=NULL;
@ -356,6 +423,9 @@ void UDPRESTImplementation::deleteFilter(){
void UDPRESTImplementation::setupFilter(){ void UDPRESTImplementation::setupFilter(){
//LEO: check
FILE_LOG(logDEBUG) << __AT__ << " called";
double hc = 0; double hc = 0;
double sigma = 5; double sigma = 5;
int sign = 1; int sign = 1;
@ -392,6 +462,11 @@ void UDPRESTImplementation::setupFilter(){
//LEO: it is not clear to me.. //LEO: it is not clear to me..
void UDPRESTImplementation::setupFifoStructure(){ void UDPRESTImplementation::setupFifoStructure(){
FILE_LOG(logDEBUG) << __AT__ << " called, doing nothing";
}
/*
void UDPRESTImplementation::setupFifoStructure(){
FILE_LOG(logDEBUG) << __AT__ << " called";
int64_t i; int64_t i;
int oldn = numJobsPerThread; int oldn = numJobsPerThread;
@ -437,10 +512,6 @@ void UDPRESTImplementation::setupFifoStructure(){
cout << "Number of Frames per buffer:" << numJobsPerThread << endl; cout << "Number of Frames per buffer:" << numJobsPerThread << endl;
cout << "Fifo Size:" << fifosize << endl; cout << "Fifo Size:" << fifosize << endl;
/*
//for testing
numJobsPerThread = 3; fifosize = 11;
*/
for(int i=0;i<numListeningThreads;i++){ for(int i=0;i<numListeningThreads;i++){
//deleting old structure and creating fifo structure //deleting old structure and creating fifo structure
@ -457,7 +528,7 @@ void UDPRESTImplementation::setupFifoStructure(){
//allocate memory //allocate memory
mem0[i]=(char*)malloc((bufferSize * numJobsPerThread + HEADER_SIZE_NUM_TOT_PACKETS)*fifosize); mem0[i]=(char*)malloc((bufferSize * numJobsPerThread + HEADER_SIZE_NUM_TOT_PACKETS)*fifosize);
/** shud let the client know about this */ // shud let the client know about this
if (mem0[i]==NULL){ if (mem0[i]==NULL){
cout<<"++++++++++++++++++++++ COULD NOT ALLOCATE MEMORY FOR LISTENING !!!!!!!+++++++++++++++++++++" << endl; cout<<"++++++++++++++++++++++ COULD NOT ALLOCATE MEMORY FOR LISTENING !!!!!!!+++++++++++++++++++++" << endl;
exit(-1); exit(-1);
@ -471,7 +542,7 @@ void UDPRESTImplementation::setupFifoStructure(){
} }
cout << "Fifo structure(s) reconstructed" << endl; cout << "Fifo structure(s) reconstructed" << endl;
} }
*/
@ -481,6 +552,7 @@ void UDPRESTImplementation::setupFifoStructure(){
/** acquisition functions */ /** acquisition functions */
void UDPRESTImplementation::readFrame(char* c,char** raw, uint32_t &fnum){ void UDPRESTImplementation::readFrame(char* c,char** raw, uint32_t &fnum){
FILE_LOG(logDEBUG) << __AT__ << " called";
//point to gui data //point to gui data
if (guiData == NULL) if (guiData == NULL)
guiData = latestData; guiData = latestData;
@ -515,6 +587,7 @@ void UDPRESTImplementation::readFrame(char* c,char** raw, uint32_t &fnum){
void UDPRESTImplementation::copyFrameToGui(char* startbuf[], uint32_t fnum, char* buf){ void UDPRESTImplementation::copyFrameToGui(char* startbuf[], uint32_t fnum, char* buf){
FILE_LOG(logDEBUG) << __AT__ << " called";
//random read when gui not ready //random read when gui not ready
if((!nFrameToGui) && (!guiData)){ if((!nFrameToGui) && (!guiData)){
@ -571,6 +644,7 @@ void UDPRESTImplementation::copyFrameToGui(char* startbuf[], uint32_t fnum, char
int UDPRESTImplementation::createUDPSockets(){ int UDPRESTImplementation::createUDPSockets(){
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " starting";
//if eth is mistaken with ip address //if eth is mistaken with ip address
if (strchr(eth,'.')!=NULL) if (strchr(eth,'.')!=NULL)
@ -615,13 +689,38 @@ int UDPRESTImplementation::createUDPSockets(){
int UDPRESTImplementation::shutDownUDPSockets(){ int UDPRESTImplementation::shutDownUDPSockets(){
FILE_LOG(logDEBUG) << __AT__ << "called";
std::string answer;
int code = rest->get_json("state", &answer);
std::cout << answer << std::endl;
code = rest->post_json("state/close", &answer);
std::cout << answer << std::endl;
code = rest->post_json("state/reset", &answer);
std::cout << answer << std::endl;
code = rest->get_json("state", &answer);
std::cout << answer << std::endl;
status = slsReceiverDefs::RUN_FINISHED;
/*
for(int i=0;i<numListeningThreads;i++){ for(int i=0;i<numListeningThreads;i++){
if(udpSocket[i]){ if(udpSocket[i]){
FILE_LOG(logDEBUG) << __AT__ << " UDP socket " << i;
udpSocket[i]->ShutDownSocket(); udpSocket[i]->ShutDownSocket();
delete udpSocket[i]; delete udpSocket[i];
udpSocket[i] = NULL; udpSocket[i] = NULL;
} }
} }
*/
FILE_LOG(logDEBUG) << __AT__ << "finished";
return OK; return OK;
} }
@ -630,6 +729,8 @@ int UDPRESTImplementation::shutDownUDPSockets(){
int UDPRESTImplementation::createListeningThreads(bool destroy){ int UDPRESTImplementation::createListeningThreads(bool destroy){
FILE_LOG(logDEBUG) << __AT__ << " called";
int i; int i;
void* status; void* status;
@ -685,6 +786,7 @@ int UDPRESTImplementation::createListeningThreads(bool destroy){
int UDPRESTImplementation::createWriterThreads(bool destroy){ int UDPRESTImplementation::createWriterThreads(bool destroy){
FILE_LOG(logDEBUG) << __AT__ << " called";
int i; int i;
void* status; void* status;
@ -746,6 +848,7 @@ int UDPRESTImplementation::createWriterThreads(bool destroy){
void UDPRESTImplementation::setThreadPriorities(){ void UDPRESTImplementation::setThreadPriorities(){
FILE_LOG(logDEBUG) << __AT__ << " called";
int i; int i;
//assign priorities //assign priorities
struct sched_param tcp_param, listen_param, write_param; struct sched_param tcp_param, listen_param, write_param;
@ -783,7 +886,7 @@ void UDPRESTImplementation::setThreadPriorities(){
int UDPRESTImplementation::setupWriter(){ int UDPRESTImplementation::setupWriter(){
FILE_LOG(logDEBUG) << __AT__ << " called";
//reset writing thread variables //reset writing thread variables
packetsInFile=0; packetsInFile=0;
packetsCaught=0; packetsCaught=0;
@ -866,6 +969,8 @@ int UDPRESTImplementation::setupWriter(){
int UDPRESTImplementation::createCompressionFile(int ithr, int iframe){ int UDPRESTImplementation::createCompressionFile(int ithr, int iframe){
FILE_LOG(logDEBUG) << __AT__ << " called";
#ifdef MYROOT1 #ifdef MYROOT1
char temp[MAX_STR_LENGTH]; char temp[MAX_STR_LENGTH];
//create file name for gui purposes, and set up acquistion parameters //create file name for gui purposes, and set up acquistion parameters
@ -894,6 +999,7 @@ int UDPRESTImplementation::createCompressionFile(int ithr, int iframe){
int UDPRESTImplementation::createNewFile(){ int UDPRESTImplementation::createNewFile(){
FILE_LOG(logDEBUG) << __AT__ << " called";
int gt = getFrameIndex(); int gt = getFrameIndex();
if(gt==-1) gt=0; if(gt==-1) gt=0;
//create file name //create file name
@ -955,9 +1061,7 @@ int UDPRESTImplementation::createNewFile(){
void UDPRESTImplementation::closeFile(int ithr){ void UDPRESTImplementation::closeFile(int ithr){
#ifdef VERBOSE FILE_LOG(logDEBUG) << __AT__ << "called for thread " << ithr;
cout << "In closeFile for thread " << ithr << endl;
#endif
if(!dataCompression){ if(!dataCompression){
if(sfilefd){ if(sfilefd){
@ -1005,6 +1109,8 @@ void UDPRESTImplementation::closeFile(int ithr){
#endif #endif
} }
FILE_LOG(logDEBUG) << __AT__ << "exited for thread " << ithr;
} }
@ -1014,13 +1120,33 @@ void UDPRESTImplementation::closeFile(int ithr){
int UDPRESTImplementation::startReceiver(char message[]){ int UDPRESTImplementation::startReceiver(char message[]){
int i; int i;
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " starting";
initialize_REST();
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " initialized";
// #ifdef VERBOSE // #ifdef VERBOSE
cout << "Starting Receiver" << endl; cout << "Starting Receiver" << endl;
//#endif //#endif
std::string answer;
int code;
//test = "{\"configfile\":\"config.pu\", \"path\":\"patto\"}";
code = rest->post_json("state/configure", &answer);
std::cout << answer << std::endl;
code = rest->get_json("state", &answer);
std::cout << answer << std::endl;
code = rest->post_json("state/open", &answer);
std::cout << answer << std::endl;
code = rest->get_json("state", &answer);
std::cout << answer << std::endl;
status = slsReceiverDefs::RUNNING;
//reset listening thread variables //reset listening thread variables
/*
measurementStarted = false; measurementStarted = false;
//should be set to zero as its added to get next start frame indices for scans for eiger //should be set to zero as its added to get next start frame indices for scans for eiger
if(!acqStarted) currframenum = 0; if(!acqStarted) currframenum = 0;
@ -1028,8 +1154,9 @@ int UDPRESTImplementation::startReceiver(char message[]){
for(int i = 0; i < numListeningThreads; ++i) for(int i = 0; i < numListeningThreads; ++i)
totalListeningFrameCount[i] = 0; totalListeningFrameCount[i] = 0;
*/
//udp socket //udp socket
/*
if(createUDPSockets() == FAIL){ if(createUDPSockets() == FAIL){
strcpy(message,"Could not create UDP Socket(s).\n"); strcpy(message,"Could not create UDP Socket(s).\n");
cout << endl << message << endl; cout << endl << message << endl;
@ -1037,7 +1164,8 @@ int UDPRESTImplementation::startReceiver(char message[]){
} }
cout << "UDP socket(s) created successfully. 1st port " << server_port[0] << endl; cout << "UDP socket(s) created successfully. 1st port " << server_port[0] << endl;
*/
/*
if(setupWriter() == FAIL){ if(setupWriter() == FAIL){
//stop udp socket //stop udp socket
shutDownUDPSockets(); shutDownUDPSockets();
@ -1069,9 +1197,7 @@ int UDPRESTImplementation::startReceiver(char message[]){
sem_post(&listensmp[i]); sem_post(&listensmp[i]);
for(i=0; i < numWriterThreads; ++i) for(i=0; i < numWriterThreads; ++i)
sem_post(&writersmp[i]); sem_post(&writersmp[i]);
*/
cout << "Receiver Started.\nStatus:" << status << endl;
return OK; return OK;
} }
@ -1081,10 +1207,7 @@ int UDPRESTImplementation::startReceiver(char message[]){
int UDPRESTImplementation::stopReceiver(){ int UDPRESTImplementation::stopReceiver(){
FILE_LOG(logDEBUG) << __AT__ << "called";
//#ifdef VERBOSE
cout << "Stopping Receiver" << endl;
//#endif
if(status == RUNNING) if(status == RUNNING)
startReadout(); startReadout();
@ -1101,7 +1224,8 @@ int UDPRESTImplementation::stopReceiver(){
status = IDLE; status = IDLE;
pthread_mutex_unlock(&(status_mutex)); pthread_mutex_unlock(&(status_mutex));
cout << "Receiver Stopped.\nStatus:" << status << endl << endl; FILE_LOG(logDEBUG) << __AT__ << "exited, status " << endl;
return OK; return OK;
} }
@ -1110,10 +1234,7 @@ int UDPRESTImplementation::stopReceiver(){
void UDPRESTImplementation::startReadout(){ void UDPRESTImplementation::startReadout(){
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " starting";
//#ifdef VERBOSE
cout << "Start Receiver Readout" << endl;
//#endif
//wait so that all packets which take time has arrived //wait so that all packets which take time has arrived
usleep(50000); usleep(50000);
@ -1127,12 +1248,15 @@ void UDPRESTImplementation::startReadout(){
//kill udp socket to tell the listening thread to push last packet //kill udp socket to tell the listening thread to push last packet
shutDownUDPSockets(); shutDownUDPSockets();
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " done";
} }
void* UDPRESTImplementation::startListeningThread(void* this_pointer){ void* UDPRESTImplementation::startListeningThread(void* this_pointer){
FILE_LOG(logDEBUG) << __AT__ << " called";
((UDPRESTImplementation*)this_pointer)->startListening(); ((UDPRESTImplementation*)this_pointer)->startListening();
return this_pointer; return this_pointer;
@ -1141,6 +1265,7 @@ void* UDPRESTImplementation::startListeningThread(void* this_pointer){
void* UDPRESTImplementation::startWritingThread(void* this_pointer){ void* UDPRESTImplementation::startWritingThread(void* this_pointer){
FILE_LOG(logDEBUG) << __AT__ << " called";
((UDPRESTImplementation*)this_pointer)->startWriting(); ((UDPRESTImplementation*)this_pointer)->startWriting();
return this_pointer; return this_pointer;
} }
@ -1151,6 +1276,8 @@ void* UDPRESTImplementation::startWritingThread(void* this_pointer){
int UDPRESTImplementation::startListening(){ int UDPRESTImplementation::startListening(){
FILE_LOG(logDEBUG) << __AT__ << " called";
int ithread = currentListeningThreadIndex; int ithread = currentListeningThreadIndex;
#ifdef VERYVERBOSE #ifdef VERYVERBOSE
cout << "In startListening() " << endl; cout << "In startListening() " << endl;
@ -1352,6 +1479,7 @@ int UDPRESTImplementation::startListening(){
int UDPRESTImplementation::startWriting(){ int UDPRESTImplementation::startWriting(){
FILE_LOG(logDEBUG) << __AT__ << " called";
int ithread = currentWriterThreadIndex; int ithread = currentWriterThreadIndex;
#ifdef VERYVERBOSE #ifdef VERYVERBOSE
cout << ithread << "In startWriting()" <<endl; cout << ithread << "In startWriting()" <<endl;
@ -1545,6 +1673,7 @@ int loop;
void UDPRESTImplementation::startFrameIndices(int ithread){ void UDPRESTImplementation::startFrameIndices(int ithread){
FILE_LOG(logDEBUG) << __AT__ << " called";
if (myDetectorType == EIGER) if (myDetectorType == EIGER)
//add currframenum later in this method for scans //add currframenum later in this method for scans
@ -1577,15 +1706,21 @@ void UDPRESTImplementation::startFrameIndices(int ithread){
} }
void UDPRESTImplementation::stopListening(int ithread, int rc, int &pc, int &t){ void UDPRESTImplementation::stopListening(int ithread, int rc, int &pc, int &t){
FILE_LOG(logDEBUG) << __AT__ << " called, doing nothing";
};
/*
void UDPRESTImplementation::stopListening(int ithread, int rc, int &pc, int &t){
FILE_LOG(logDEBUG) << __AT__ << " called";
int i; int i;
#ifdef VERYVERBOSE #ifdef VERYVERBOSE
cerr << ithread << " recvfrom() failed:"<<endl; cerr << ithread << " recvfrom() failed:"<<endl;
#endif #endif
if(status != TRANSMITTING){ if(status != TRANSMITTING){
cout << ithread << " *** shoule never be here********* status not transmitting***********************"<<endl;/**/ cout << ithread << " *** shoule never be here********* status not transmitting***********************"<<endl;
fifoFree[ithread]->push(buffer[ithread]); fifoFree[ithread]->push(buffer[ithread]);
exit(-1); exit(-1);
} }
@ -1646,7 +1781,7 @@ int i;
} }
} }
*/
@ -1657,6 +1792,7 @@ int i;
void UDPRESTImplementation::stopWriting(int ithread, char* wbuffer[]){ void UDPRESTImplementation::stopWriting(int ithread, char* wbuffer[]){
FILE_LOG(logDEBUG) << __AT__ << " called";
int i,j; int i,j;
#ifdef VERYDEBUG #ifdef VERYDEBUG
cout << ithread << " **********************popped last dummy frame:" << (void*)wbuffer[wIndex] << endl; cout << ithread << " **********************popped last dummy frame:" << (void*)wbuffer[wIndex] << endl;
@ -1724,6 +1860,7 @@ void UDPRESTImplementation::stopWriting(int ithread, char* wbuffer[]){
void UDPRESTImplementation::writeToFile_withoutCompression(char* buf,int numpackets, uint32_t framenum){ void UDPRESTImplementation::writeToFile_withoutCompression(char* buf,int numpackets, uint32_t framenum){
FILE_LOG(logDEBUG) << __AT__ << " called";
int packetsToSave, offset,lastpacket; int packetsToSave, offset,lastpacket;
uint32_t tempframenum = framenum; uint32_t tempframenum = framenum;
@ -1827,6 +1964,8 @@ void UDPRESTImplementation::writeToFile_withoutCompression(char* buf,int numpack
void UDPRESTImplementation::handleDataCompression(int ithread, char* wbuffer[], int &npackets, char* data, int xmax, int ymax, int &nf){ void UDPRESTImplementation::handleDataCompression(int ithread, char* wbuffer[], int &npackets, char* data, int xmax, int ymax, int &nf){
FILE_LOG(logDEBUG) << __AT__ << " called";
#if defined(MYROOT1) && defined(ALLFILE_DEBUG) #if defined(MYROOT1) && defined(ALLFILE_DEBUG)
writeToFile_withoutCompression(wbuf[0], numpackets,currframenum); writeToFile_withoutCompression(wbuf[0], numpackets,currframenum);
#endif #endif
@ -1925,6 +2064,7 @@ void UDPRESTImplementation::handleDataCompression(int ithread, char* wbuffer[],
int UDPRESTImplementation::enableTenGiga(int enable){ int UDPRESTImplementation::enableTenGiga(int enable){
FILE_LOG(logDEBUG) << __AT__ << " called";
cout << "Enabling 10Gbe to" << enable << endl; cout << "Enabling 10Gbe to" << enable << endl;

120
slsReceiverSoftware/src/UDPStandardImplementation.cpp
View File

@ -110,8 +110,11 @@ void UDPStandardImplementation::initializeMembers(){
strcpy(detHostname,""); strcpy(detHostname,"");
strcpy(guiFileName,""); strcpy(guiFileName,"");
strcpy(savefilename,""); strcpy(savefilename,"");
strcpy(filePath,"");
strcpy(fileName,"run"); setFileName("run");
setFilePath("");
//strcpy(filePath,"");
//strcpy(fileName,"run");
//status //status
@ -122,15 +125,17 @@ void UDPStandardImplementation::initializeMembers(){
} }
UDPStandardImplementation::UDPStandardImplementation(): UDPStandardImplementation::UDPStandardImplementation(){ FILE_LOG(logDEBUG) << __AT__ << " called";
thread_started(0),
eth(NULL),
latestData(NULL),
guiFileName(NULL),
guiFrameNumber(0),
tengigaEnable(0){
cout << "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAa" << endl;
FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " starting" ;
thread_started = 0;
eth = NULL;
latestData = NULL;
guiFileName = NULL;
guiFrameNumber = 0;
tengigaEnable = 0;
for(int i=0;i<MAX_NUM_LISTENING_THREADS;i++){ for(int i=0;i<MAX_NUM_LISTENING_THREADS;i++){
udpSocket[i] = NULL; udpSocket[i] = NULL;
@ -178,7 +183,8 @@ UDPStandardImplementation::UDPStandardImplementation():
UDPStandardImplementation::~UDPStandardImplementation(){ UDPStandardImplementation::~UDPStandardImplementation(){ FILE_LOG(logDEBUG) << __AT__ << " called";
createListeningThreads(true); createListeningThreads(true);
createWriterThreads(true); createWriterThreads(true);
deleteMembers(); deleteMembers();
@ -187,7 +193,8 @@ UDPStandardImplementation::~UDPStandardImplementation(){
void UDPStandardImplementation::deleteMembers(){ void UDPStandardImplementation::deleteMembers(){ FILE_LOG(logDEBUG) << __AT__ << " called";
//kill threads //kill threads
if(thread_started){ if(thread_started){
createListeningThreads(true); createListeningThreads(true);
@ -220,7 +227,8 @@ void UDPStandardImplementation::deleteMembers(){
int UDPStandardImplementation::setDetectorType(detectorType det){ int UDPStandardImplementation::setDetectorType(detectorType det){ FILE_LOG(logDEBUG) << __AT__ << " called";
cout << "Setting Receiver Type " << endl; cout << "Setting Receiver Type " << endl;
deleteMembers(); deleteMembers();
@ -364,7 +372,6 @@ char* UDPStandardImplementation::getFilePath() const{
return (char*)filePath; return (char*)filePath;
} }
*/ */
/* /*
char* UDPStandardImplementation::setFilePath(const char c[]){ char* UDPStandardImplementation::setFilePath(const char c[]){
if(strlen(c)){ if(strlen(c)){
@ -448,6 +455,9 @@ int UDPStandardImplementation::setEnableOverwrite(int i){
/*other parameters*/ /*other parameters*/
slsReceiverDefs::runStatus UDPStandardImplementation::getStatus() const{ slsReceiverDefs::runStatus UDPStandardImplementation::getStatus() const{
FILE_LOG(logDEBUG) << __AT__ << " called, status: " << status;
return status; return status;
} }
@ -462,12 +472,14 @@ char *UDPStandardImplementation::getDetectorHostname() const{
return (char*)detHostname; return (char*)detHostname;
} }
void UDPStandardImplementation::setEthernetInterface(char* c){ void UDPStandardImplementation::setEthernetInterface(char* c){ FILE_LOG(logDEBUG) << __AT__ << " called";
strcpy(eth,c); strcpy(eth,c);
} }
void UDPStandardImplementation::setUDPPortNo(int p){ void UDPStandardImplementation::setUDPPortNo(int p){ FILE_LOG(logDEBUG) << __AT__ << " called";
for(int i=0;i<numListeningThreads;i++){ for(int i=0;i<numListeningThreads;i++){
server_port[i] = p+i; server_port[i] = p+i;
} }
@ -479,7 +491,8 @@ int UDPStandardImplementation::getNumberOfFrames() const {
} }
int32_t UDPStandardImplementation::setNumberOfFrames(int32_t fnum){ int32_t UDPStandardImplementation::setNumberOfFrames(int32_t fnum){ FILE_LOG(logDEBUG) << __AT__ << " called";
if(fnum >= 0) if(fnum >= 0)
numberOfFrames = fnum; numberOfFrames = fnum;
@ -491,7 +504,8 @@ int UDPStandardImplementation::getScanTag() const{
} }
int32_t UDPStandardImplementation::setScanTag(int32_t stag){ int32_t UDPStandardImplementation::setScanTag(int32_t stag){ FILE_LOG(logDEBUG) << __AT__ << " called";
if(stag >= 0) if(stag >= 0)
scanTag = stag; scanTag = stag;
@ -503,7 +517,8 @@ int UDPStandardImplementation::getDynamicRange() const{
return dynamicRange; return dynamicRange;
} }
int32_t UDPStandardImplementation::setDynamicRange(int32_t dr){ int32_t UDPStandardImplementation::setDynamicRange(int32_t dr){ FILE_LOG(logDEBUG) << __AT__ << " called";
cout << "Setting Dynamic Range" << endl; cout << "Setting Dynamic Range" << endl;
int olddr = dynamicRange; int olddr = dynamicRange;
@ -563,7 +578,8 @@ int32_t UDPStandardImplementation::setDynamicRange(int32_t dr){
int UDPStandardImplementation::setShortFrame(int i){ int UDPStandardImplementation::setShortFrame(int i){ FILE_LOG(logDEBUG) << __AT__ << " called";
shortFrame=i; shortFrame=i;
if(shortFrame!=-1){ if(shortFrame!=-1){
@ -593,7 +609,8 @@ int UDPStandardImplementation::setShortFrame(int i){
} }
int UDPStandardImplementation::setNFrameToGui(int i){ int UDPStandardImplementation::setNFrameToGui(int i){ FILE_LOG(logDEBUG) << __AT__ << " called";
if(i>=0){ if(i>=0){
nFrameToGui = i; nFrameToGui = i;
setupFifoStructure(); setupFifoStructure();
@ -603,7 +620,8 @@ int UDPStandardImplementation::setNFrameToGui(int i){
int64_t UDPStandardImplementation::setAcquisitionPeriod(int64_t index){ int64_t UDPStandardImplementation::setAcquisitionPeriod(int64_t index){ FILE_LOG(logDEBUG) << __AT__ << " called";
if(index >= 0){ if(index >= 0){
if(index != acquisitionPeriod){ if(index != acquisitionPeriod){
@ -615,9 +633,11 @@ int64_t UDPStandardImplementation::setAcquisitionPeriod(int64_t index){
} }
bool UDPStandardImplementation::getDataCompression(){return dataCompression;} bool UDPStandardImplementation::getDataCompression(){ FILE_LOG(logDEBUG) << __AT__ << " called";
return dataCompression;}
int UDPStandardImplementation::enableDataCompression(bool enable){ FILE_LOG(logDEBUG) << __AT__ << " called";
int UDPStandardImplementation::enableDataCompression(bool enable){
cout << "Data compression "; cout << "Data compression ";
if(enable) if(enable)
cout << "enabled" << endl; cout << "enabled" << endl;
@ -670,7 +690,8 @@ int UDPStandardImplementation::enableDataCompression(bool enable){
/*other functions*/ /*other functions*/
void UDPStandardImplementation::deleteFilter(){ void UDPStandardImplementation::deleteFilter(){ FILE_LOG(logDEBUG) << __AT__ << " called";
int i; int i;
cmSub=NULL; cmSub=NULL;
@ -688,6 +709,8 @@ void UDPStandardImplementation::deleteFilter(){
void UDPStandardImplementation::setupFilter(){ void UDPStandardImplementation::setupFilter(){
FILE_LOG(logDEBUG) << __AT__ << " called";
double hc = 0; double hc = 0;
double sigma = 5; double sigma = 5;
int sign = 1; int sign = 1;
@ -724,6 +747,7 @@ void UDPStandardImplementation::setupFilter(){
//LEO: it is not clear to me.. //LEO: it is not clear to me..
void UDPStandardImplementation::setupFifoStructure(){ void UDPStandardImplementation::setupFifoStructure(){
FILE_LOG(logDEBUG) << __AT__ << " called";
int64_t i; int64_t i;
int oldn = numJobsPerThread; int oldn = numJobsPerThread;
@ -813,6 +837,8 @@ void UDPStandardImplementation::setupFifoStructure(){
/** acquisition functions */ /** acquisition functions */
void UDPStandardImplementation::readFrame(char* c,char** raw, uint32_t &fnum){ void UDPStandardImplementation::readFrame(char* c,char** raw, uint32_t &fnum){
FILE_LOG(logDEBUG) << __AT__ << " called";
//point to gui data //point to gui data
if (guiData == NULL) if (guiData == NULL)
guiData = latestData; guiData = latestData;
@ -847,6 +873,8 @@ void UDPStandardImplementation::readFrame(char* c,char** raw, uint32_t &fnum){
void UDPStandardImplementation::copyFrameToGui(char* startbuf[], uint32_t fnum, char* buf){ void UDPStandardImplementation::copyFrameToGui(char* startbuf[], uint32_t fnum, char* buf){
FILE_LOG(logDEBUG) << __AT__ << " called";
//random read when gui not ready //random read when gui not ready
if((!nFrameToGui) && (!guiData)){ if((!nFrameToGui) && (!guiData)){
@ -902,6 +930,7 @@ void UDPStandardImplementation::copyFrameToGui(char* startbuf[], uint32_t fnum,
int UDPStandardImplementation::createUDPSockets(){ int UDPStandardImplementation::createUDPSockets(){
FILE_LOG(logDEBUG) << __AT__ << " called";
//if eth is mistaken with ip address //if eth is mistaken with ip address
@ -947,6 +976,8 @@ int UDPStandardImplementation::createUDPSockets(){
int UDPStandardImplementation::shutDownUDPSockets(){ int UDPStandardImplementation::shutDownUDPSockets(){
FILE_LOG(logDEBUG) << __AT__ << " called";
for(int i=0;i<numListeningThreads;i++){ for(int i=0;i<numListeningThreads;i++){
if(udpSocket[i]){ if(udpSocket[i]){
udpSocket[i]->ShutDownSocket(); udpSocket[i]->ShutDownSocket();
@ -960,8 +991,10 @@ int UDPStandardImplementation::shutDownUDPSockets(){
// TODO: add a destroyListeningThreads
int UDPStandardImplementation::createListeningThreads(bool destroy){ int UDPStandardImplementation::createListeningThreads(bool destroy){
FILE_LOG(logDEBUG) << __AT__ << " called";
int i; int i;
void* status; void* status;
@ -971,6 +1004,8 @@ int UDPStandardImplementation::createListeningThreads(bool destroy){
listeningthreads_mask = 0x0; listeningthreads_mask = 0x0;
pthread_mutex_unlock(&(status_mutex)); pthread_mutex_unlock(&(status_mutex));
FILE_LOG(logDEBUG) << "Starting " << __func__ << endl;
if(!destroy){ if(!destroy){
//start listening threads //start listening threads
@ -1017,6 +1052,8 @@ int UDPStandardImplementation::createListeningThreads(bool destroy){
int UDPStandardImplementation::createWriterThreads(bool destroy){ int UDPStandardImplementation::createWriterThreads(bool destroy){
FILE_LOG(logDEBUG) << __AT__ << " called";
int i; int i;
void* status; void* status;
@ -1078,6 +1115,8 @@ int UDPStandardImplementation::createWriterThreads(bool destroy){
void UDPStandardImplementation::setThreadPriorities(){ void UDPStandardImplementation::setThreadPriorities(){
FILE_LOG(logDEBUG) << __AT__ << " called";
int i; int i;
//assign priorities //assign priorities
struct sched_param tcp_param, listen_param, write_param; struct sched_param tcp_param, listen_param, write_param;
@ -1115,6 +1154,7 @@ void UDPStandardImplementation::setThreadPriorities(){
int UDPStandardImplementation::setupWriter(){ int UDPStandardImplementation::setupWriter(){
FILE_LOG(logDEBUG) << __AT__ << " called";
//reset writing thread variables //reset writing thread variables
packetsInFile=0; packetsInFile=0;
@ -1198,6 +1238,8 @@ int UDPStandardImplementation::setupWriter(){
int UDPStandardImplementation::createCompressionFile(int ithr, int iframe){ int UDPStandardImplementation::createCompressionFile(int ithr, int iframe){
FILE_LOG(logDEBUG) << __AT__ << " called";
#ifdef MYROOT1 #ifdef MYROOT1
char temp[MAX_STR_LENGTH]; char temp[MAX_STR_LENGTH];
//create file name for gui purposes, and set up acquistion parameters //create file name for gui purposes, and set up acquistion parameters
@ -1226,6 +1268,8 @@ int UDPStandardImplementation::createCompressionFile(int ithr, int iframe){
int UDPStandardImplementation::createNewFile(){ int UDPStandardImplementation::createNewFile(){
FILE_LOG(logDEBUG) << __AT__ << " called";
int gt = getFrameIndex(); int gt = getFrameIndex();
if(gt==-1) gt=0; if(gt==-1) gt=0;
//create file name //create file name
@ -1234,6 +1278,9 @@ int UDPStandardImplementation::createNewFile(){
else else
sprintf(savefilename, "%s/%s_f%012d_%d.raw", filePath,fileName,(packetsCaught/packetsPerFrame),fileIndex); sprintf(savefilename, "%s/%s_f%012d_%d.raw", filePath,fileName,(packetsCaught/packetsPerFrame),fileIndex);
cout << filePath << " + " << fileName << endl;
//if filewrite and we are allowed to write //if filewrite and we are allowed to write
if(enableFileWrite && cbAction > DO_NOTHING){ if(enableFileWrite && cbAction > DO_NOTHING){
//close //close
@ -1287,6 +1334,8 @@ int UDPStandardImplementation::createNewFile(){
void UDPStandardImplementation::closeFile(int ithr){ void UDPStandardImplementation::closeFile(int ithr){
FILE_LOG(logDEBUG) << __AT__ << " called";
#ifdef VERBOSE #ifdef VERBOSE
cout << "In closeFile for thread " << ithr << endl; cout << "In closeFile for thread " << ithr << endl;
#endif #endif
@ -1344,6 +1393,8 @@ void UDPStandardImplementation::closeFile(int ithr){
int UDPStandardImplementation::startReceiver(char message[]){ int UDPStandardImplementation::startReceiver(char message[]){
FILE_LOG(logDEBUG) << __AT__ << " called";
int i; int i;
@ -1412,6 +1463,7 @@ int UDPStandardImplementation::startReceiver(char message[]){
int UDPStandardImplementation::stopReceiver(){ int UDPStandardImplementation::stopReceiver(){
FILE_LOG(logDEBUG) << __AT__ << " called";
//#ifdef VERBOSE //#ifdef VERBOSE
@ -1442,6 +1494,7 @@ int UDPStandardImplementation::stopReceiver(){
void UDPStandardImplementation::startReadout(){ void UDPStandardImplementation::startReadout(){
FILE_LOG(logDEBUG) << __AT__ << " called";
//#ifdef VERBOSE //#ifdef VERBOSE
cout << "Start Receiver Readout" << endl; cout << "Start Receiver Readout" << endl;
@ -1465,6 +1518,7 @@ void UDPStandardImplementation::startReadout(){
void* UDPStandardImplementation::startListeningThread(void* this_pointer){ void* UDPStandardImplementation::startListeningThread(void* this_pointer){
FILE_LOG(logDEBUG) << __AT__ << " called";
((UDPStandardImplementation*)this_pointer)->startListening(); ((UDPStandardImplementation*)this_pointer)->startListening();
return this_pointer; return this_pointer;
@ -1473,6 +1527,7 @@ void* UDPStandardImplementation::startListeningThread(void* this_pointer){
void* UDPStandardImplementation::startWritingThread(void* this_pointer){ void* UDPStandardImplementation::startWritingThread(void* this_pointer){
FILE_LOG(logDEBUG) << __AT__ << " called";
((UDPStandardImplementation*)this_pointer)->startWriting(); ((UDPStandardImplementation*)this_pointer)->startWriting();
return this_pointer; return this_pointer;
} }
@ -1483,6 +1538,7 @@ void* UDPStandardImplementation::startWritingThread(void* this_pointer){
int UDPStandardImplementation::startListening(){ int UDPStandardImplementation::startListening(){
FILE_LOG(logDEBUG) << __AT__ << " called";
int ithread = currentListeningThreadIndex; int ithread = currentListeningThreadIndex;
#ifdef VERYVERBOSE #ifdef VERYVERBOSE
cout << "In startListening() " << endl; cout << "In startListening() " << endl;
@ -1684,6 +1740,8 @@ int UDPStandardImplementation::startListening(){
int UDPStandardImplementation::startWriting(){ int UDPStandardImplementation::startWriting(){
FILE_LOG(logDEBUG) << __AT__ << " called";
int ithread = currentWriterThreadIndex; int ithread = currentWriterThreadIndex;
#ifdef VERYVERBOSE #ifdef VERYVERBOSE
cout << ithread << "In startWriting()" <<endl; cout << ithread << "In startWriting()" <<endl;
@ -1877,6 +1935,7 @@ int loop;
void UDPStandardImplementation::startFrameIndices(int ithread){ void UDPStandardImplementation::startFrameIndices(int ithread){
FILE_LOG(logDEBUG) << __AT__ << " called";
if (myDetectorType == EIGER) if (myDetectorType == EIGER)
//add currframenum later in this method for scans //add currframenum later in this method for scans
@ -1911,6 +1970,9 @@ void UDPStandardImplementation::startFrameIndices(int ithread){
void UDPStandardImplementation::stopListening(int ithread, int rc, int &pc, int &t){ void UDPStandardImplementation::stopListening(int ithread, int rc, int &pc, int &t){
FILE_LOG(logDEBUG) << __AT__ << " called";
int i; int i;
#ifdef VERYVERBOSE #ifdef VERYVERBOSE
@ -1989,6 +2051,8 @@ int i;
void UDPStandardImplementation::stopWriting(int ithread, char* wbuffer[]){ void UDPStandardImplementation::stopWriting(int ithread, char* wbuffer[]){
FILE_LOG(logDEBUG) << __AT__ << " called";
int i,j; int i,j;
#ifdef VERYDEBUG #ifdef VERYDEBUG
cout << ithread << " **********************popped last dummy frame:" << (void*)wbuffer[wIndex] << endl; cout << ithread << " **********************popped last dummy frame:" << (void*)wbuffer[wIndex] << endl;
@ -2056,6 +2120,8 @@ void UDPStandardImplementation::stopWriting(int ithread, char* wbuffer[]){
void UDPStandardImplementation::writeToFile_withoutCompression(char* buf,int numpackets, uint32_t framenum){ void UDPStandardImplementation::writeToFile_withoutCompression(char* buf,int numpackets, uint32_t framenum){
FILE_LOG(logDEBUG) << __AT__ << " called";
int packetsToSave, offset,lastpacket; int packetsToSave, offset,lastpacket;
uint32_t tempframenum = framenum; uint32_t tempframenum = framenum;
@ -2158,6 +2224,7 @@ void UDPStandardImplementation::writeToFile_withoutCompression(char* buf,int num
void UDPStandardImplementation::handleDataCompression(int ithread, char* wbuffer[], int &npackets, char* data, int xmax, int ymax, int &nf){ void UDPStandardImplementation::handleDataCompression(int ithread, char* wbuffer[], int &npackets, char* data, int xmax, int ymax, int &nf){
FILE_LOG(logDEBUG) << __AT__ << " called";
#if defined(MYROOT1) && defined(ALLFILE_DEBUG) #if defined(MYROOT1) && defined(ALLFILE_DEBUG)
writeToFile_withoutCompression(wbuf[0], numpackets,currframenum); writeToFile_withoutCompression(wbuf[0], numpackets,currframenum);
@ -2257,6 +2324,7 @@ void UDPStandardImplementation::handleDataCompression(int ithread, char* wbuffer
int UDPStandardImplementation::enableTenGiga(int enable){ int UDPStandardImplementation::enableTenGiga(int enable){
FILE_LOG(logDEBUG) << __AT__ << " called";
cout << "Enabling 10Gbe to" << enable << endl; cout << "Enabling 10Gbe to" << enable << endl;

31
slsReceiverSoftware/src/slsReceiver.cpp
View File

@ -8,7 +8,7 @@
#include <sstream> #include <sstream>
#include <fstream> #include <fstream>
#include <stdlib.h> #include <stdlib.h>
#include <map>
#include <getopt.h> #include <getopt.h>
#include "slsReceiver.h" #include "slsReceiver.h"
@ -30,10 +30,12 @@ slsReceiver::slsReceiver(int argc, char *argv[], int &success){
*/ */
//creating base receiver //creating base receiver
int tcpip_port_no = 1984; map<string, string> configuration_map;
int tcpip_port_no = 1954;
success=OK; success=OK;
string fname = ""; string fname = "";
string udp_interface_type = "standard"; string udp_interface_type = "standard";
string rest_hostname = "localhost:8081";
//parse command line for config //parse command line for config
static struct option long_options[] = { static struct option long_options[] = {
@ -44,6 +46,7 @@ slsReceiver::slsReceiver(int argc, char *argv[], int &success){
{"type", required_argument, 0, 't'}, {"type", required_argument, 0, 't'},
{"config", required_argument, 0, 'f'}, {"config", required_argument, 0, 'f'},
{"rx_tcpport", required_argument, 0, 'b'}, {"rx_tcpport", required_argument, 0, 'b'},
{"rest_hostname", required_argument, 0, 'r'},
{"help", no_argument, 0, 'h'}, {"help", no_argument, 0, 'h'},
{0, 0, 0, 0} {0, 0, 0, 0}
}; };
@ -52,7 +55,7 @@ slsReceiver::slsReceiver(int argc, char *argv[], int &success){
int c; int c;
while ( c != -1 ){ while ( c != -1 ){
c = getopt_long (argc, argv, "bfht", long_options, &option_index); c = getopt_long (argc, argv, "bfhtr", long_options, &option_index);
/* Detect the end of the options. */ /* Detect the end of the options. */
if (c == -1) if (c == -1)
@ -72,12 +75,19 @@ slsReceiver::slsReceiver(int argc, char *argv[], int &success){
udp_interface_type = optarg; udp_interface_type = optarg;
break; break;
case 'r':
rest_hostname = optarg;
break;
case 'h': case 'h':
string help_message = """\nSLS Receiver Server\n\n"""; string help_message = """\nSLS Receiver Server\n\n""";
help_message += """usage: slsReceiver --config config_fname [--rx_tcpport port]\n\n"""; help_message += """usage: slsReceiver --config config_fname [--rx_tcpport port]\n\n""";
help_message += """\t--config:\t configuration filename for SLS Detector receiver\n"""; help_message += """\t--config:\t configuration filename for SLS Detector receiver\n""";
help_message += """\t--rx_tcpport:\t TCP Communication Port with the client. Default: 1954.\n\n"""; help_message += """\t--rx_tcpport:\t TCP Communication Port with the client. Default: 1954.\n\n""";
help_message += """\t--rest_hostname:\t Receiver hostname:port. It applies only to REST receivers, and indicates the hostname of the REST backend. Default: localhost:8081.\n\n""";
help_message += """\t--type:\t Type of the receiver. Possible arguments are: standard, REST. Default: standard.\n\n"""; help_message += """\t--type:\t Type of the receiver. Possible arguments are: standard, REST. Default: standard.\n\n""";
cout << help_message << endl; cout << help_message << endl;
break; break;
@ -85,18 +95,20 @@ slsReceiver::slsReceiver(int argc, char *argv[], int &success){
} }
// if required fname parameter not available, fail // if required fname parameter not available, fail
if (fname == "") //if (fname == "")
success = FAIL; // success = FAIL;
if((!fname.empty()) && (success == OK)){ if( !fname.empty() ){
try{
FILE_LOG(logINFO) << "config file name " << fname; FILE_LOG(logINFO) << "config file name " << fname;
success = read_config_file(fname, &tcpip_port_no); success = read_config_file(fname, &tcpip_port_no, &configuration_map);
//VERBOSE_PRINT("Read configuration file of " + iline + " lines"); //VERBOSE_PRINT("Read configuration file of " + iline + " lines");
} }
else { catch(...){
FILE_LOG(logERROR) << "Error opening configuration file " << fname ; FILE_LOG(logERROR) << "Error opening configuration file " << fname ;
success = FAIL; success = FAIL;
} }
}
if(success != OK){ if(success != OK){
@ -104,8 +116,9 @@ slsReceiver::slsReceiver(int argc, char *argv[], int &success){
} }
if (success==OK){ if (success==OK){
cout << "SLS Receiver starting " << udp_interface_type << endl; FILE_LOG(logINFO) << "SLS Receiver starting " << udp_interface_type << " on port " << tcpip_port_no << endl;
udp_interface = UDPInterface::create(udp_interface_type); udp_interface = UDPInterface::create(udp_interface_type);
udp_interface->configure(configuration_map);
tcpipInterface = new slsReceiverTCPIPInterface(success, udp_interface, tcpip_port_no); tcpipInterface = new slsReceiverTCPIPInterface(success, udp_interface, tcpip_port_no);
//tcp ip interface //tcp ip interface
} }

6
slsReceiverSoftware/src/slsReceiverTCPIPInterface.cpp
View File

@ -69,12 +69,12 @@ slsReceiverTCPIPInterface::slsReceiverTCPIPInterface(int &success, UDPInterface*
//Catch signal SIGINT to close files properly //Catch signal SIGINT to close files properly
signal(SIGINT,staticCloseFile); signal(SIGINT,staticCloseFile);
} }
} }
} }
int slsReceiverTCPIPInterface::setPortNumber(int pn){ int slsReceiverTCPIPInterface::setPortNumber(int pn){
int p_number; int p_number;
MySocketTCP *oldsocket=NULL;; MySocketTCP *oldsocket=NULL;;
@ -144,9 +144,11 @@ void slsReceiverTCPIPInterface::stop(){
cout<<"Shutting down TCP Socket and TCP thread"<<endl; cout<<"Shutting down TCP Socket and TCP thread"<<endl;
killTCPServerThread = 1; killTCPServerThread = 1;
socket->ShutDownSocket(); socket->ShutDownSocket();
cout<<"Socket closed"<<endl;
void* status; void* status;
pthread_join(TCPServer_thread, &status); pthread_join(TCPServer_thread, &status);
killTCPServerThread = 0; killTCPServerThread = 0;
cout<<"Threads joined"<<endl;
} }
@ -656,7 +658,7 @@ int slsReceiverTCPIPInterface::setup_udp(){
strcpy(eth,""); strcpy(eth,"");
ret = FAIL; ret = FAIL;
} }
cout<<"eth:"<<eth<<endl; FILE_LOG(logDEBUG) << __FILE__ << "::" << __func__ << " " << eth;
receiverBase->setEthernetInterface(eth); receiverBase->setEthernetInterface(eth);
//get mac address from ethernet interface //get mac address from ethernet interface

26
slsReceiverSoftware/src/utilities.cpp
View File

@ -5,19 +5,23 @@
#include <stdio.h> #include <stdio.h>
#include <fstream> #include <fstream>
#include <map>
#include "utilities.h" #include "utilities.h"
using namespace std; using namespace std;
int read_config_file(string fname, int *tcpip_port_no){ int read_config_file(string fname, int *tcpip_port_no, map<string, string> * configuration_map ){
ifstream infile; ifstream infile;
string sLine,sargname; string sLine,sargname, sargvalue;
int iline = 0; int iline = 0;
int success = slsReceiverDefs::OK; int success = slsReceiverDefs::OK;
infile.open(fname.c_str(), ios_base::in); infile.open(fname.c_str(), ios_base::in);
if (infile.is_open()) { if (infile.is_open()) {
while(infile.good()){ while(infile.good()){
@ -26,21 +30,25 @@ int read_config_file(string fname, int *tcpip_port_no){
//VERBOSE_PRINT(sLine); //VERBOSE_PRINT(sLine);
if(sLine.find('#')!=string::npos){ if(sLine.find('#') != string::npos)
//VERBOSE_PRINT( "Line is a comment ");
continue; continue;
}
else if(sLine.length()<2){ else if(sLine.length()<2)
//VERBOSE_PRINT("Empty line ");
continue; continue;
}
else{ else{
istringstream sstr(sLine); istringstream sstr(sLine);
//parameter name //parameter name
if(sstr.good()) if(sstr.good()){
sstr >> sargname; sstr >> sargname;
if (! sstr.good())
continue;
sstr >> sargvalue;
(*configuration_map)[sargname] = sargvalue;
}
//tcp port //tcp port
if(sargname=="rx_tcpport"){ if(sargname=="rx_tcpport"){
if(sstr.good()) { if(sstr.good()) {