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

in the process of streamer

Browse Source
This commit is contained in:
Dhanya Maliakal 2017-02-27 15:38:46 +01:00
parent 936dfea8a7
commit 3b07afe3fc
16 changed files with 654 additions and 171 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
slsReceiverSoftware/Makefile
View File

@ -16,6 +16,9 @@ DFLAGS= -g -DDACS_INT -DSLS_RECEIVER_UDP_FUNCTIONS #-DVERBOSE
INCLUDES?= $(INCLUDESRXR) -I include/ -I ../slsDetectorCalibration INCLUDES?= $(INCLUDESRXR) -I include/ -I ../slsDetectorCalibration
##############################################################
# ZMQ specific: in this case, you need ZMQ libraries already included in this package
###########################################################
LIBZMQDIR = include LIBZMQDIR = include
LIBZMQ = -L$(LIBZMQDIR) -Wl,-rpath=$(LIBZMQDIR) -lzmq LIBZMQ = -L$(LIBZMQDIR) -Wl,-rpath=$(LIBZMQDIR) -lzmq

27
slsReceiverSoftware/include/DataProcessor.h
View File

@ -14,6 +14,7 @@
class GeneralData; class GeneralData;
class Fifo; class Fifo;
class File; class File;
class DataStreamer;
#include <vector> #include <vector>
@ -24,15 +25,14 @@ class DataProcessor : private virtual slsReceiverDefs, public ThreadObject {
* Constructor * Constructor
* Calls Base Class CreateThread(), sets ErrorMask if error and increments NumberofDataProcessors * Calls Base Class CreateThread(), sets ErrorMask if error and increments NumberofDataProcessors
* @param f address of Fifo pointer * @param f address of Fifo pointer
* @param s pointer to receiver status
* @param m pointer to mutex for status
* @param ftype pointer to file format type * @param ftype pointer to file format type
* @param fwenable pointer to file writer enable * @param fwenable pointer to file writer enable
* @param dsEnable pointer to data stream enable
* @param cbaction pointer to call back action * @param cbaction pointer to call back action
* @param dataReadycb pointer to data ready call back function * @param dataReadycb pointer to data ready call back function
* @param pDataReadycb pointer to arguments of data ready call back function * @param pDataReadycb pointer to arguments of data ready call back function
*/ */
DataProcessor(Fifo*& f, runStatus* s, pthread_mutex_t* m, fileFormat* ftype, bool* fwenable, DataProcessor(Fifo*& f, fileFormat* ftype, bool* fwenable, bool* dsEnable,
int* cbaction, int* cbaction,
void (*dataReadycb)(int, char*, int, FILE*, char*, void*), void (*dataReadycb)(int, char*, int, FILE*, char*, void*),
void *pDataReadycb); void *pDataReadycb);
@ -57,6 +57,10 @@ class DataProcessor : private virtual slsReceiverDefs, public ThreadObject {
*/ */
static uint64_t GetRunningMask(); static uint64_t GetRunningMask();
/**
* Reset RunningMask
*/
static void ResetRunningMask();
//*** non static functions *** //*** non static functions ***
//*** getters *** //*** getters ***
@ -124,6 +128,13 @@ class DataProcessor : private virtual slsReceiverDefs, public ThreadObject {
*/ */
void SetGeneralData(GeneralData* g); void SetGeneralData(GeneralData* g);
/**
* Set thread priority
* @priority priority
* @returns OK or FAIL
*/
int SetThreadPriority(int priority);
/** /**
* Set File Format * Set File Format
* @param f file format * @param f file format
@ -235,8 +246,8 @@ class DataProcessor : private virtual slsReceiverDefs, public ThreadObject {
/** Fifo structure */ /** Fifo structure */
Fifo* fifo; Fifo* fifo;
/** Data Stream Enable */
// individual members bool* dataStreamEnable;
/** Aquisition Started flag */ /** Aquisition Started flag */
bool acquisitionStartedFlag; bool acquisitionStartedFlag;
@ -244,12 +255,6 @@ class DataProcessor : private virtual slsReceiverDefs, public ThreadObject {
/** Measurement Started flag */ /** Measurement Started flag */
bool measurementStartedFlag; bool measurementStartedFlag;
/** Receiver Status */
runStatus* status;
/** Status mutex */
pthread_mutex_t* statusMutex;
/**Number of complete frames caught for an entire acquisition (including all scans) */ /**Number of complete frames caught for an entire acquisition (including all scans) */
uint64_t numTotalFramesCaught; uint64_t numTotalFramesCaught;

153
slsReceiverSoftware/include/DataStreamer.h
View File

@ -9,14 +9,23 @@
#include "ThreadObject.h" #include "ThreadObject.h"
class GeneralData;
class Fifo;
class DataStreamer;
class ZmqSocket;
class DataStreamer : private virtual slsReceiverDefs, public ThreadObject { class DataStreamer : private virtual slsReceiverDefs, public ThreadObject {
public: public:
/** /**
* Constructor * Constructor
* Calls Base Class CreateThread(), sets ErrorMask if error and increments NumberofDataStreamers * Calls Base Class CreateThread(), sets ErrorMask if error and increments NumberofDataStreamers
* @param f address of Fifo pointer
* @param dr pointer to dynamic range
* @param freq poiner to streaming frequency
* @param timer poiner to timer if streaming frequency is random
*/ */
DataStreamer(); DataStreamer(Fifo*& f, uint32_t* dr, uint32_t* freq, uint32_t* timer);
/** /**
* Destructor * Destructor
@ -25,18 +34,30 @@ class DataStreamer : private virtual slsReceiverDefs, public ThreadObject {
~DataStreamer(); ~DataStreamer();
//*** static functions ***
/** /**
* Get RunningMask * Get RunningMask
* @return RunningMask * @return RunningMask
*/ */
static uint64_t GetErrorMask(); static uint64_t GetErrorMask();
/**
* Get RunningMask
* @return RunningMask
*/
static uint64_t GetRunningMask();
/** /**
* Reset RunningMask * Reset RunningMask
*/ */
static void ResetRunningMask(); static void ResetRunningMask();
//*** non static functions ***
//*** getters ***
//*** setters ***
/** /**
* Set bit in RunningMask to allow thread to run * Set bit in RunningMask to allow thread to run
*/ */
@ -47,6 +68,52 @@ class DataStreamer : private virtual slsReceiverDefs, public ThreadObject {
*/ */
void StopRunning(); void StopRunning();
/**
* Set Fifo pointer to the one given
* @param f address of Fifo pointer
*/
void SetFifo(Fifo*& f);
/**
* Reset parameters for new acquisition (including all scans)
*/
void ResetParametersforNewAcquisition();
/**
* Reset parameters for new measurement (eg. for each scan)
*/
void ResetParametersforNewMeasurement();
/**
* Create Part1 of Json Header which includes common attributes in an acquisition
*/
void CreateHeaderPart1();
/**
* Set GeneralData pointer to the one given
* @param g address of GeneralData (Detector Data) pointer
*/
void SetGeneralData(GeneralData* g);
/**
* Set thread priority
* @priority priority
* @returns OK or FAIL
*/
int SetThreadPriority(int priority);
/**
* Creates Zmq Sockets
* @param dindex pointer to detector index
* @param nunits pointer to number of theads/ units per detector
* @return OK or FAIL
*/
int CreateZmqSockets(int* dindex, int* nunits);
/**
* Shuts down and deletes Zmq Sockets
*/
void CloseZmqSocket();
private: private:
@ -62,13 +129,52 @@ class DataStreamer : private virtual slsReceiverDefs, public ThreadObject {
*/ */
bool IsRunning(); bool IsRunning();
/**
* Record First Indices (firstAcquisitionIndex, firstMeasurementIndex)
* @param fnum frame index to record
*/
void RecordFirstIndices(uint64_t fnum);
/** /**
* Thread Exeution for DataStreamer Class * Thread Exeution for DataStreamer Class
* Stream an image via zmq * Stream an image via zmq
*/ */
void ThreadExecution(); void ThreadExecution();
/**
* Frees dummy buffer,
* reset running mask by calling StopRunning()
* @param buf address of pointer
*/
void StopProcessing(char* buf);
/**
* Process an image popped from fifo,
* write to file if fw enabled & update parameters
* @param buffer
*/
void ProcessAnImage(char* buf);
/**
* This function should be called only in random frequency mode
* Checks if timer is done and ready to send data
* @returns true if ready to send data, else false
*/
bool CheckTimer();
/**
* This function should be called only in non random frequency mode
* Checks if count is done and ready to send data
* @returns true if ready to send data, else false
*/
bool CheckCount();
/**
* Create and send Json Header
* @param fnum frame number
* @returns 0 if error, else 1
*/
int SendHeader(uint64_t fnum);
/** type of thread */ /** type of thread */
static const std::string TypeName; static const std::string TypeName;
@ -84,5 +190,50 @@ class DataStreamer : private virtual slsReceiverDefs, public ThreadObject {
/** mutex to update static items among objects (threads)*/ /** mutex to update static items among objects (threads)*/
static pthread_mutex_t Mutex; static pthread_mutex_t Mutex;
/** Json Header Format for each measurement part */
static const char *jsonHeaderFormat_part1;
/** Json Header Format */
static const char *jsonHeaderFormat;
/** GeneralData (Detector Data) object */
const GeneralData* generalData;
/** Fifo structure */
Fifo* fifo;
/** ZMQ Socket - Receiver to Client */
ZmqSocket* zmqSocket;
/** Pointer to dynamic range */
uint32_t* dynamicRange;
/** Pointer to Streaming frequency, if 0, sending random images with a timer */
uint32_t* streamingFrequency;
/** Pointer to the timer if Streaming frequency is random */
uint32_t* streamingTimerInMs;
/** Current frequency count */
uint32_t currentFreqCount;
/** timer beginning stamp for random streaming */
struct timespec timerBegin;
/** Current Json Header prefix*/
char* currentHeader;
/** Aquisition Started flag */
bool acquisitionStartedFlag;
/** Measurement Started flag */
bool measurementStartedFlag;
/** Frame Number of First Frame of an entire Acquisition (including all scans) */
uint64_t firstAcquisitionIndex;
/** Frame Number of First Frame for each real time acquisition (eg. for each scan) */
uint64_t firstMeasurementIndex;
}; };

22
slsReceiverSoftware/include/Fifo.h
View File

@ -31,16 +31,16 @@ class Fifo : private virtual slsReceiverDefs {
*/ */
~Fifo(); ~Fifo();
/**
* Pops free address from fifoFree
*/
void GetNewAddress(char*& address);
/** /**
* Frees the bound address by pushing into fifoFree * Frees the bound address by pushing into fifoFree
*/ */
void FreeAddress(char*& address); void FreeAddress(char*& address);
/**
* Pops free address from fifoFree
*/
void GetNewAddress(char*& address);
/** /**
* Pushes bound address into fifoBound * Pushes bound address into fifoBound
*/ */
@ -51,6 +51,16 @@ class Fifo : private virtual slsReceiverDefs {
*/ */
void PopAddress(char*& address); void PopAddress(char*& address);
/**
* Pushes bound address into fifoStream
*/
void PushAddressToStream(char*& address);
/**
* Pops bound address from fifoStream to stream data
*/
void PopAddressToStream(char*& address);
private: private:
/** /**
@ -82,4 +92,6 @@ class Fifo : private virtual slsReceiverDefs {
/** Circular Fifo pointing to addresses of freed data in memory */ /** Circular Fifo pointing to addresses of freed data in memory */
CircularFifo<char>* fifoFree; CircularFifo<char>* fifoFree;
/** Circular Fifo pointing to addresses of to be streamed data in memory */
CircularFifo<char>* fifoStream;
}; };

2
slsReceiverSoftware/include/GeneralData.h
View File

@ -428,7 +428,7 @@ private:
packetsPerFrame = 256; packetsPerFrame = 256;
imageSize = dataSize*packetsPerFrame; imageSize = dataSize*packetsPerFrame;
frameIndexMask = 0xffffff; frameIndexMask = 0xffffff;
maxFramesPerFile = 5;//EIGER_MAX_FRAMES_PER_FILE; maxFramesPerFile = EIGER_MAX_FRAMES_PER_FILE;
fifoBufferSize = imageSize; fifoBufferSize = imageSize;
fifoBufferHeaderSize= FIFO_HEADER_NUMBYTES + FILE_FRAME_HEADER_SIZE; fifoBufferHeaderSize= FIFO_HEADER_NUMBYTES + FILE_FRAME_HEADER_SIZE;
defaultFifoDepth = 100; defaultFifoDepth = 100;

9
slsReceiverSoftware/include/HDF5FileStatic.h
View File

@ -447,10 +447,10 @@ public:
hid_t vdsdataset = H5Dcreate2 (fd, virtualDatasetname.c_str(), dtype, vdsDataspace, H5P_DEFAULT, dcpl, H5P_DEFAULT); hid_t vdsdataset = H5Dcreate2 (fd, virtualDatasetname.c_str(), dtype, vdsDataspace, H5P_DEFAULT, dcpl, H5P_DEFAULT);
if (vdsdataset >= 0){ if (vdsdataset >= 0){
H5Sclose(vdsDataspace); H5Sclose(vdsDataspace);vdsDataspace = 0;
H5Sclose(srcDataspace); H5Sclose(srcDataspace);srcDataspace = 0;
H5Dclose(vdsdataset); H5Dclose(vdsdataset);vdsdataset = 0;
H5Fclose(fd); H5Fclose(fd);fd = 0;
return 0; return 0;
} else cprintf(RED, "could not create virtual dataset in virtual file %s\n", virtualfname.c_str()); } else cprintf(RED, "could not create virtual dataset in virtual file %s\n", virtualfname.c_str());
@ -468,6 +468,7 @@ public:
} else cprintf(RED, "could not create dfal for virtual file %s\n", virtualfname.c_str()); } else cprintf(RED, "could not create dfal for virtual file %s\n", virtualfname.c_str());
H5Fclose(fd); H5Fclose(fd);
fd = 0;
return 1; return 1;
} }

39
slsReceiverSoftware/include/Listener.h
View File

@ -25,8 +25,10 @@ class Listener : private virtual slsReceiverDefs, public ThreadObject {
* @param s pointer to receiver status * @param s pointer to receiver status
* @param portno pointer to udp port number * @param portno pointer to udp port number
* @param e ethernet interface * @param e ethernet interface
* @param act pointer to activated
* @param nf pointer to number of images to catch
*/ */
Listener(Fifo*& f, runStatus* s, uint32_t* portno, char* e); Listener(Fifo*& f, runStatus* s, uint32_t* portno, char* e, int* act, uint64_t* nf);
/** /**
* Destructor * Destructor
@ -49,11 +51,9 @@ class Listener : private virtual slsReceiverDefs, public ThreadObject {
static uint64_t GetRunningMask(); static uint64_t GetRunningMask();
/** /**
* Set GeneralData pointer to the one given * Reset RunningMask
* @param g address of GeneralData (Detector Data) pointer
*/ */
static void SetGeneralData(GeneralData*& g); static void ResetRunningMask();
//*** non static functions *** //*** non static functions ***
@ -94,7 +94,6 @@ class Listener : private virtual slsReceiverDefs, public ThreadObject {
*/ */
void StopRunning(); void StopRunning();
/** /**
* Set Fifo pointer to the one given * Set Fifo pointer to the one given
* @param f address of Fifo pointer * @param f address of Fifo pointer
@ -111,6 +110,19 @@ class Listener : private virtual slsReceiverDefs, public ThreadObject {
*/ */
void ResetParametersforNewMeasurement(); void ResetParametersforNewMeasurement();
/**
* Set GeneralData pointer to the one given
* @param g address of GeneralData (Detector Data) pointer
*/
void SetGeneralData(GeneralData*& g);
/**
* Set thread priority
* @priority priority
* @returns OK or FAIL
*/
int SetThreadPriority(int priority);
/** /**
* Creates UDP Sockets * Creates UDP Sockets
* @return OK or FAIL * @return OK or FAIL
@ -168,6 +180,13 @@ class Listener : private virtual slsReceiverDefs, public ThreadObject {
*/ */
uint32_t ListenToAnImage(char* buf); uint32_t ListenToAnImage(char* buf);
/**
* Create an image (for deactivated detectors),
* @param buffer
* @returns image size or 0
*/
uint32_t CreateAnImage(char* buf);
/** type of thread */ /** type of thread */
@ -202,7 +221,7 @@ class Listener : private virtual slsReceiverDefs, public ThreadObject {
/** Receiver Status */ /** Receiver Status */
runStatus* status; runStatus* status;
/** UDP Sockets - Detector to Receiver */ /** UDP Socket - Detector to Receiver */
genericSocket* udpSocket; genericSocket* udpSocket;
/** UDP Port Number */ /** UDP Port Number */
@ -211,6 +230,12 @@ class Listener : private virtual slsReceiverDefs, public ThreadObject {
/** ethernet interface */ /** ethernet interface */
char* eth; char* eth;
/** if the detector is activated */
int* activated;
/** Number of Images to catch */
uint64_t* numImages;
/**Number of complete Packets caught for an entire acquisition (including all scans) */ /**Number of complete Packets caught for an entire acquisition (including all scans) */
uint64_t numTotalPacketsCaught; uint64_t numTotalPacketsCaught;

5
slsReceiverSoftware/include/UDPStandardImplementation.h
View File

@ -206,6 +206,11 @@ private:
*/ */
void SetLocalNetworkParameters(); void SetLocalNetworkParameters();
/**
* Set Thread Priorities
*/
void SetThreadPriorities();
/** /**
* Set up the Fifo Structure for processing buffers * Set up the Fifo Structure for processing buffers
* between listening and dataprocessor threads * between listening and dataprocessor threads

14
slsReceiverSoftware/include/genericSocket.h
View File

@ -62,7 +62,6 @@ class sockaddr_in;
#include <stdio.h> #include <stdio.h>
using namespace std; using namespace std;
#define DEFAULT_PACKET_SIZE 1286 #define DEFAULT_PACKET_SIZE 1286
@ -72,7 +71,7 @@ using namespace std;
#define DEFAULT_BACKLOG 5 #define DEFAULT_BACKLOG 5
#define DEFAULT_UDP_PORTNO 50001 #define DEFAULT_UDP_PORTNO 50001
#define DEFAULT_GUI_PORTNO 65000 #define DEFAULT_GUI_PORTNO 65000
#define DEFAULT_ZMQ_PORTNO 70001 //#define DEFAULT_ZMQ_PORTNO 70001
class genericSocket{ class genericSocket{
@ -102,13 +101,10 @@ enum communicationProtocol{
{ {
memset(&serverAddress, 0, sizeof(serverAddress)); memset(&serverAddress, 0, sizeof(serverAddress));
memset(&clientAddress, 0, sizeof(clientAddress)); memset(&clientAddress, 0, sizeof(clientAddress));
// strcpy(hostname,host_ip_or_name);
strcpy(lastClientIP,"none"); strcpy(lastClientIP,"none");
strcpy(thisClientIP,"none1"); strcpy(thisClientIP,"none1");
strcpy(dummyClientIP,"dummy"); strcpy(dummyClientIP,"dummy");
differentClients = 0; differentClients = 0;
struct hostent *hostInfo = gethostbyname(host_ip_or_name); struct hostent *hostInfo = gethostbyname(host_ip_or_name);
if (hostInfo == NULL){ if (hostInfo == NULL){
cerr << "Exiting: Problem interpreting host: " << host_ip_or_name << "\n"; cerr << "Exiting: Problem interpreting host: " << host_ip_or_name << "\n";
@ -170,13 +166,13 @@ enum communicationProtocol{
/* // you can specify an IP address: */ /* // you can specify an IP address: */
/* // or you can let it automatically select one: */ /* // or you can let it automatically select one: */
/* myaddr.sin_addr.s_addr = INADDR_ANY; */ /* myaddr.sin_addr.s_addr = INADDR_ANY; */
strcpy(lastClientIP,"none"); strcpy(lastClientIP,"none");
strcpy(thisClientIP,"none1"); strcpy(thisClientIP,"none1");
strcpy(dummyClientIP,"dummy"); strcpy(dummyClientIP,"dummy");
differentClients = 0; differentClients = 0;
if(serverAddress.sin_port == htons(port_number)){ if(serverAddress.sin_port == htons(port_number)){
socketDescriptor = -10; socketDescriptor = -10;
return; return;
@ -209,8 +205,7 @@ enum communicationProtocol{
if (string(ip)!=string("0.0.0.0")) { if (string(ip)!=string("0.0.0.0")) {
if (inet_pton(AF_INET, ip, &(serverAddress.sin_addr))) if (inet_pton(AF_INET, ip, &(serverAddress.sin_addr)));
;
else else
serverAddress.sin_addr.s_addr = htonl(INADDR_ANY); serverAddress.sin_addr.s_addr = htonl(INADDR_ANY);
} }
@ -719,6 +714,5 @@ enum communicationProtocol{
int total_sent; int total_sent;
int header_packet_size; int header_packet_size;
// pthread_mutex_t mp; // pthread_mutex_t mp;
}; };

4
slsReceiverSoftware/include/receiver_defs.h
View File

@ -35,3 +35,7 @@
#define DUMMY_PACKET_VALUE 0xFFFFFFFF #define DUMMY_PACKET_VALUE 0xFFFFFFFF
#define LISTENER_PRIORITY 99
#define PROCESSOR_PRIORITY 90
#define STREAMER_PRIORITY 80
#define TCP_PRIORITY 50

43
slsReceiverSoftware/src/DataProcessor.cpp
View File

@ -13,8 +13,10 @@
#ifdef HDF5C #ifdef HDF5C
#include "HDF5File.h" #include "HDF5File.h"
#endif #endif
#include "DataStreamer.h"
#include <iostream> #include <iostream>
#include <errno.h>
#include <cstring> #include <cstring>
using namespace std; using namespace std;
@ -29,17 +31,17 @@ uint64_t DataProcessor::RunningMask(0x0);
pthread_mutex_t DataProcessor::Mutex = PTHREAD_MUTEX_INITIALIZER; pthread_mutex_t DataProcessor::Mutex = PTHREAD_MUTEX_INITIALIZER;
DataProcessor::DataProcessor(Fifo*& f, runStatus* s, pthread_mutex_t* m, fileFormat* ftype, bool* fwenable, DataProcessor::DataProcessor(Fifo*& f, fileFormat* ftype, bool* fwenable, bool* dsEnable,
int* cbaction, int* cbaction,
void (*dataReadycb)(int, char*, int, FILE*, char*, void*), void (*dataReadycb)(int, char*, int, FILE*, char*, void*),
void *pDataReadycb) : void *pDataReadycb) :
ThreadObject(NumberofDataProcessors), ThreadObject(NumberofDataProcessors),
generalData(0), generalData(0),
fifo(f), fifo(f),
dataStreamEnable(dsEnable),
acquisitionStartedFlag(false), acquisitionStartedFlag(false),
measurementStartedFlag(false), measurementStartedFlag(false),
status(s),
statusMutex(m),
numTotalFramesCaught(0), numTotalFramesCaught(0),
numFramesCaught(0), numFramesCaught(0),
firstAcquisitionIndex(0), firstAcquisitionIndex(0),
@ -59,7 +61,7 @@ DataProcessor::DataProcessor(Fifo*& f, runStatus* s, pthread_mutex_t* m, fileFor
} }
NumberofDataProcessors++; NumberofDataProcessors++;
FILE_LOG (logDEBUG) << "Number of DataProcessors: " << NumberofDataProcessors << endl; FILE_LOG (logDEBUG) << "Number of DataProcessors: " << NumberofDataProcessors;
} }
@ -79,6 +81,10 @@ uint64_t DataProcessor::GetRunningMask() {
return RunningMask; return RunningMask;
} }
void DataProcessor::ResetRunningMask() {
RunningMask = 0x0;
}
/** non static functions */ /** non static functions */
/** getters */ /** getters */
string DataProcessor::GetType(){ string DataProcessor::GetType(){
@ -126,7 +132,6 @@ void DataProcessor::StopRunning() {
pthread_mutex_unlock(&Mutex); pthread_mutex_unlock(&Mutex);
} }
void DataProcessor::SetFifo(Fifo*& f) { void DataProcessor::SetFifo(Fifo*& f) {
fifo = f; fifo = f;
} }
@ -142,11 +147,6 @@ void DataProcessor::ResetParametersforNewMeasurement(){
numFramesCaught = 0; numFramesCaught = 0;
firstMeasurementIndex = 0; firstMeasurementIndex = 0;
measurementStartedFlag = false; measurementStartedFlag = false;
if(RunningMask){
pthread_mutex_lock(&Mutex);
RunningMask = 0x0;
pthread_mutex_unlock(&Mutex);
}
} }
@ -185,6 +185,15 @@ void DataProcessor::SetGeneralData(GeneralData* g) {
} }
int DataProcessor::SetThreadPriority(int priority) {
struct sched_param param;
param.sched_priority = priority;
if (pthread_setschedparam(thread, SCHED_RR, &param) == EPERM)
return FAIL;
return OK;
}
void DataProcessor::SetFileFormat(const fileFormat f) { void DataProcessor::SetFileFormat(const fileFormat f) {
if (file->GetFileType() != f) { if (file->GetFileType() != f) {
//remember the pointer values before they are destroyed //remember the pointer values before they are destroyed
@ -257,14 +266,22 @@ void DataProcessor::ThreadExecution() {
ProcessAnImage(buffer + FIFO_HEADER_NUMBYTES); ProcessAnImage(buffer + FIFO_HEADER_NUMBYTES);
//free //stream or free
fifo->FreeAddress(buffer); if (*dataStreamEnable)
fifo->PushAddressToStream(buffer);
else
fifo->FreeAddress(buffer);
} }
void DataProcessor::StopProcessing(char* buf) { void DataProcessor::StopProcessing(char* buf) {
fifo->FreeAddress(buf); //stream or free
if (*dataStreamEnable)
fifo->PushAddressToStream(buf);
else
fifo->FreeAddress(buf);
file->CloseCurrentFile(); file->CloseCurrentFile();
StopRunning(); StopRunning();
cprintf(BLUE,"%d: Processing Completed\n", index); cprintf(BLUE,"%d: Processing Completed\n", index);

246
slsReceiverSoftware/src/DataStreamer.cpp
View File

@ -5,8 +5,12 @@
#include "DataStreamer.h" #include "DataStreamer.h"
#include "GeneralData.h"
#include "Fifo.h"
#include "ZmqSocket.h"
#include <iostream> #include <iostream>
#include <errno.h>
using namespace std; using namespace std;
const string DataStreamer::TypeName = "DataStreamer"; const string DataStreamer::TypeName = "DataStreamer";
@ -19,12 +23,36 @@ uint64_t DataStreamer::RunningMask(0x0);
pthread_mutex_t DataStreamer::Mutex = PTHREAD_MUTEX_INITIALIZER; pthread_mutex_t DataStreamer::Mutex = PTHREAD_MUTEX_INITIALIZER;
const char* DataStreamer::jsonHeaderFormat_part1 =
"{"
"\"htype\":[\"chunk-1.0\"], "
"\"type\":\"%s\", "
"\"shape\":%s, ";
DataStreamer::DataStreamer() : const char* DataStreamer::jsonHeaderFormat =
ThreadObject(NumberofDataStreamers) "%s"
"\"acqIndex\":%lld, "
"\"fIndex\":%lld, "
"\"subfnum\":%lld, "
"\"fname\":\"%s\"}";
DataStreamer::DataStreamer(Fifo*& f, uint32_t* dr, uint32_t* freq, uint32_t* timer) :
ThreadObject(NumberofDataStreamers),
generalData(0),
fifo(f),
zmqSocket(0),
dynamicRange(dr),
streamingFrequency(freq),
streamingTimerInMs(timer),
currentFreqCount(0),
currentHeader(0),
acquisitionStartedFlag(false),
measurementStartedFlag(false),
firstAcquisitionIndex(0),
firstMeasurementIndex(0)
{ {
FILE_LOG (logDEBUG) << __AT__ << " called"; memset(timerBegin, 0xFF, sizeof(timespec));
if(ThreadObject::CreateThread()){ if(ThreadObject::CreateThread()){
pthread_mutex_lock(&Mutex); pthread_mutex_lock(&Mutex);
ErrorMask ^= (1<<index); ErrorMask ^= (1<<index);
@ -32,49 +60,44 @@ DataStreamer::DataStreamer() :
} }
NumberofDataStreamers++; NumberofDataStreamers++;
FILE_LOG (logDEBUG) << "Number of DataStreamers: " << NumberofDataStreamers << endl; FILE_LOG (logDEBUG) << "Number of DataStreamers: " << NumberofDataStreamers;
} }
DataStreamer::~DataStreamer() { DataStreamer::~DataStreamer() {
FILE_LOG (logDEBUG) << __AT__ << " called"; CloseZmqSocket();
ThreadObject::DestroyThread(); ThreadObject::DestroyThread();
NumberofDataStreamers--; NumberofDataStreamers--;
} }
/** static functions */ /** static functions */
uint64_t DataStreamer::GetErrorMask() { uint64_t DataStreamer::GetErrorMask() {
FILE_LOG (logDEBUG) << __AT__ << " called";
return ErrorMask; return ErrorMask;
} }
uint64_t DataStreamer::GetRunningMask() {
void DataStreamer::ResetRunningMask() { return RunningMask;
FILE_LOG (logDEBUG) << __AT__ << " called";
pthread_mutex_lock(&Mutex);
RunningMask = 0x0;
pthread_mutex_unlock(&Mutex);
} }
void DataStreamer::ResetRunningMask() {
RunningMask = 0x0;
}
/** non static functions */ /** non static functions */
/** getters */
string DataStreamer::GetType(){ string DataStreamer::GetType(){
return TypeName; return TypeName;
} }
bool DataStreamer::IsRunning() { bool DataStreamer::IsRunning() {
FILE_LOG (logDEBUG) << __AT__ << " called";
return ((1 << index) & RunningMask); return ((1 << index) & RunningMask);
} }
/** setters */
void DataStreamer::StartRunning() { void DataStreamer::StartRunning() {
FILE_LOG (logDEBUG) << __AT__ << " called";
pthread_mutex_lock(&Mutex); pthread_mutex_lock(&Mutex);
RunningMask |= (1<<index); RunningMask |= (1<<index);
pthread_mutex_unlock(&Mutex); pthread_mutex_unlock(&Mutex);
@ -82,15 +105,196 @@ void DataStreamer::StartRunning() {
void DataStreamer::StopRunning() { void DataStreamer::StopRunning() {
FILE_LOG (logDEBUG) << __AT__ << " called";
pthread_mutex_lock(&Mutex); pthread_mutex_lock(&Mutex);
RunningMask ^= (1<<index); RunningMask ^= (1<<index);
pthread_mutex_unlock(&Mutex); pthread_mutex_unlock(&Mutex);
} }
void DataStreamer::SetFifo(Fifo*& f) {
fifo = f;
}
void DataStreamer::ResetParametersforNewAcquisition() {
firstAcquisitionIndex = 0;
acquisitionStartedFlag = false;
}
void DataStreamer::ResetParametersforNewMeasurement(){
firstMeasurementIndex = 0;
measurementStartedFlag = false;
CreateHeaderPart1();
}
void DataStreamer::CreateHeaderPart1() {
char *type;
switch (*dynamicRange) {
case 4: strcpy(type, "uint8"); break;
case 8: strcpy(type, "uint8"); break;
case 16: strcpy(type, "uint16"); break;
case 32: strcpy(type, "uint32"); break;
default:
strcpy(type, "unknown");
cprintf(RED," Unknown datatype in json format\n");
break;
}
char *shape= "[%d, %d]";
sprintf(shape, shape, generalData->nPixelsX, generalData->nPixelsY);
sprintf(currentHeader, jsonHeaderFormat_part1, type, shape);
}
void DataStreamer::RecordFirstIndices(uint64_t fnum) {
measurementStartedFlag = true;
firstMeasurementIndex = fnum;
//start of entire acquisition
if (!acquisitionStartedFlag) {
acquisitionStartedFlag = true;
firstAcquisitionIndex = fnum;
}
#ifdef VERBOSE
cprintf(BLUE,"%d First Acquisition Index:%lld\tFirst Measurement Index:%lld\n",
index, (long long int)firstAcquisitionIndex, (long long int)firstMeasurementIndex);
#endif
}
void DataStreamer::SetGeneralData(GeneralData* g) {
generalData = g;
#ifdef VERY_VERBOSE
generalData->Print();
#endif
}
int DataStreamer::SetThreadPriority(int priority) {
struct sched_param param;
param.sched_priority = priority;
if (pthread_setschedparam(thread, SCHED_RR, &param) == EPERM)
return FAIL;
return OK;
}
int DataStreamer::CreateZmqSockets(int* dindex, int* nunits) {
uint32_t portnum = DEFAULT_ZMQ_PORTNO + ((*dindex) * (*nunits) + index);
printf("%d Streamer: Port number: %d\n", index, portnum);
zmqSocket = new ZmqSocket(portnum);
if (zmqSocket->GetErrorStatus()) {
cprintf(RED, "Error: Could not create Zmq socket on port %d for Streamer %d\n", portnum, index);
return FAIL;
}
printf("%d Streamer: Zmq Server started at %s\n",zmqSocket->GetZmqServerAddress());
return OK;
}
void DataStreamer::CloseZmqSocket() {
if (zmqSocket) {
delete zmqSocket;
zmqSocket = 0;
}
}
void DataStreamer::ThreadExecution() { void DataStreamer::ThreadExecution() {
FILE_LOG (logDEBUG) << __AT__ << " called"; char* buffer=0;
fifo->PopAddressToStream(buffer);
#ifdef FIFODEBUG
if (!index) cprintf(BLUE,"DataProcessor %d, pop 0x%p buffer:%s\n", index,(void*)(buffer),buffer);
#endif
//check dummy
uint32_t numBytes = (uint32_t)(*((uint32_t*)buffer));
if (numBytes == DUMMY_PACKET_VALUE) {
StopProcessing(buffer);
return;
}
ProcessAnImage(buffer + FIFO_HEADER_NUMBYTES);
//free
fifo->FreeAddress(buffer);
} }
void DataStreamer::StopProcessing(char* buf) {
fifo->FreeAddress(buf);
StopRunning();
cprintf(MAGENTA,"%d: Streaming Completed\n", index);
}
void DataStreamer::ProcessAnImage(char* buf) {
uint64_t fnum = (*((uint64_t*)buf));
#ifdef VERBOSE
if (!index) cprintf(MAGENTA,"DataStreamer %d: fnum:%lld\n", index, (long long int)fnum);
#endif
if (!measurementStartedFlag) {
#ifdef VERBOSE
if (!index) cprintf(MAGENTA,"DataStreamer %d: fnum:%lld\n", index, (long long int)fnum);
#endif
RecordFirstIndices(fnum);
//restart timer
clock_gettime(CLOCK_REALTIME, &timerBegin);
//to send first image
currentFreqCount = *streamingFrequency;
}
//skip
if (!(*streamingFrequency)) {
if (!CheckTimer())
return;
} else {
if (!CheckCount())
return;
}
if(!SendHeader(fnum))
cprintf(RED,"Error: Could not send zmq header for fnum %lld and streamer %d\n",
(long long int) fnum, index);
Send Datat();
}
bool DataStreamer::CheckTimer() {
struct timespec end;
clock_gettime(CLOCK_REALTIME, &end);
#ifdef VERBOSE
cprintf(BLUE,"%d Timer elapsed time:%f seconds\n", index, ( end.tv_sec - timerBegin.tv_sec ) + ( end.tv_nsec - timerBegin.tv_nsec ) / 1000000000.0);
#endif
//still less than streaming timer, keep waiting
if((( end.tv_sec - timerBegin.tv_sec ) + ( end.tv_nsec - timerBegin.tv_nsec ) / 1000000000.0) < (streamingTimerInMs/1000))
return false;
//restart timer
clock_gettime(CLOCK_REALTIME, &timerBegin);
return true;
}
bool DataStreamer::CheckCount() {
if (currentFreqCount == *streamingFrequency ) {
currentFreqCount = 1;
return true;
}
currentFreqCount++;
return false;
}
int DataStreamer::SendHeader(uint64_t fnum) {
uint64_t frameIndex = fnum - firstMeasurementIndex;
uint64_t acquisitionIndex = fnum - firstAcquisitionIndex;
uint64_t subframeIndex = -1; /* subframe to be included in fifo buffer? */
char buf[1000];
int len = sprintf(buf, jsonHeaderFormat, jsonHeaderFormat_part1, acquisitionIndex, frameIndex, subframeIndex,completeFileName[ithread]);
return zmqSocket->SendDataOnly(buf, len);
}

23
slsReceiverSoftware/src/Fifo.cpp
View File

@ -16,7 +16,8 @@ int Fifo::NumberofFifoClassObjects(0);
Fifo::Fifo(uint32_t fifoItemSize, uint32_t fifoDepth, bool &success): Fifo::Fifo(uint32_t fifoItemSize, uint32_t fifoDepth, bool &success):
memory(0), memory(0),
fifoBound(0), fifoBound(0),
fifoFree(0) { fifoFree(0),
fifoStream(0){
FILE_LOG (logDEBUG) << __AT__ << " called"; FILE_LOG (logDEBUG) << __AT__ << " called";
index = NumberofFifoClassObjects++; index = NumberofFifoClassObjects++;
if(CreateFifos(fifoItemSize, fifoDepth) == FAIL) if(CreateFifos(fifoItemSize, fifoDepth) == FAIL)
@ -41,6 +42,7 @@ int Fifo::CreateFifos(uint32_t fifoItemSize, uint32_t fifoDepth) {
//create fifos //create fifos
fifoBound = new CircularFifo<char>(fifoDepth); fifoBound = new CircularFifo<char>(fifoDepth);
fifoFree = new CircularFifo<char>(fifoDepth); fifoFree = new CircularFifo<char>(fifoDepth);
fifoStream = new CircularFifo<char>(fifoDepth);
//allocate memory //allocate memory
memory = (char*) calloc (fifoItemSize * fifoDepth, sizeof(char)); memory = (char*) calloc (fifoItemSize * fifoDepth, sizeof(char));
if (memory == NULL){ if (memory == NULL){
@ -75,6 +77,10 @@ void Fifo::DestroyFifos(){
delete fifoFree; delete fifoFree;
fifoFree = 0; fifoFree = 0;
} }
if (fifoStream) {
delete fifoStream;
fifoStream = 0;
}
if(memory) { if(memory) {
free(memory); free(memory);
memory = 0; memory = 0;
@ -82,15 +88,14 @@ void Fifo::DestroyFifos(){
} }
void Fifo::FreeAddress(char*& address) {
while(!fifoFree->push(address));
}
void Fifo::GetNewAddress(char*& address) { void Fifo::GetNewAddress(char*& address) {
fifoFree->pop(address); fifoFree->pop(address);
} }
void Fifo::FreeAddress(char*& address) {
while(!fifoFree->push(address));
}
void Fifo::PushAddress(char*& address) { void Fifo::PushAddress(char*& address) {
while(!fifoBound->push(address)); while(!fifoBound->push(address));
} }
@ -99,3 +104,11 @@ void Fifo::PopAddress(char*& address) {
fifoBound->pop(address); fifoBound->pop(address);
} }
void Fifo::PushAddressToStream(char*& address) {
while(!fifoStream->push(address));
}
void Fifo::PopAddressToStream(char*& address) {
fifoStream->pop(address);
}

80
slsReceiverSoftware/src/Listener.cpp
View File

@ -12,6 +12,7 @@
#include "genericSocket.h" #include "genericSocket.h"
#include <iostream> #include <iostream>
#include <errno.h>
#include <cstring> #include <cstring>
using namespace std; using namespace std;
@ -28,7 +29,7 @@ pthread_mutex_t Listener::Mutex = PTHREAD_MUTEX_INITIALIZER;
const GeneralData* Listener::generalData(0); const GeneralData* Listener::generalData(0);
Listener::Listener(Fifo*& f, runStatus* s, uint32_t* portno, char* e) : Listener::Listener(Fifo*& f, runStatus* s, uint32_t* portno, char* e, int* act, uint64_t* nf) :
ThreadObject(NumberofListeners), ThreadObject(NumberofListeners),
fifo(f), fifo(f),
acquisitionStartedFlag(false), acquisitionStartedFlag(false),
@ -37,6 +38,8 @@ Listener::Listener(Fifo*& f, runStatus* s, uint32_t* portno, char* e) :
udpSocket(0), udpSocket(0),
udpPortNumber(portno), udpPortNumber(portno),
eth(e), eth(e),
activated(act),
numImages(nf),
numTotalPacketsCaught(0), numTotalPacketsCaught(0),
numPacketsCaught(0), numPacketsCaught(0),
firstAcquisitionIndex(0), firstAcquisitionIndex(0),
@ -53,7 +56,7 @@ Listener::Listener(Fifo*& f, runStatus* s, uint32_t* portno, char* e) :
pthread_mutex_unlock(&Mutex); pthread_mutex_unlock(&Mutex);
} }
NumberofListeners++; NumberofListeners++;
FILE_LOG (logDEBUG) << "Number of Listeners: " << NumberofListeners << endl; FILE_LOG (logDEBUG) << "Number of Listeners: " << NumberofListeners;
} }
@ -74,16 +77,11 @@ uint64_t Listener::GetRunningMask() {
return RunningMask; return RunningMask;
} }
void Listener::SetGeneralData(GeneralData*& g) { void Listener::ResetRunningMask() {
FILE_LOG (logDEBUG) << __AT__ << " called"; RunningMask = 0x0;
generalData = g;
#ifdef VERY_VERBOSE
generalData->Print();
#endif
} }
/** non static functions */ /** non static functions */
/** getters */ /** getters */
string Listener::GetType(){ string Listener::GetType(){
@ -150,12 +148,6 @@ void Listener::ResetParametersforNewMeasurement(){
if (listeningPacket) if (listeningPacket)
delete listeningPacket; delete listeningPacket;
listeningPacket = new char[generalData->packetSize]; listeningPacket = new char[generalData->packetSize];
if(RunningMask){
pthread_mutex_lock(&Mutex);
RunningMask = 0x0;
pthread_mutex_unlock(&Mutex);
}
} }
@ -172,17 +164,35 @@ void Listener::RecordFirstIndices(uint64_t fnum) {
acquisitionStartedFlag = true; acquisitionStartedFlag = true;
firstAcquisitionIndex = fnum; firstAcquisitionIndex = fnum;
} }
if (!index) cprintf(MAGENTA,"%d First Acquisition Index:%lld\n" if (!index) cprintf(GREEN,"%d First Acquisition Index:%lld\n"
"%d First Measurement Index:%lld\n", "%d First Measurement Index:%lld\n",
index, (long long int)firstAcquisitionIndex, index, (long long int)firstAcquisitionIndex,
index, (long long int)firstMeasurementIndex); index, (long long int)firstMeasurementIndex);
} }
void Listener::SetGeneralData(GeneralData*& g) {
generalData = g;
#ifdef VERY_VERBOSE
generalData->Print();
#endif
}
int Listener::SetThreadPriority(int priority) {
struct sched_param param;
param.sched_priority = priority;
if (pthread_setschedparam(thread, SCHED_RR, &param) == EPERM)
return FAIL;
return OK;
}
int Listener::CreateUDPSockets() { int Listener::CreateUDPSockets() {
ShutDownUDPSocket(); ShutDownUDPSocket();
if (!(*activated))
return OK;
//if eth is mistaken with ip address //if eth is mistaken with ip address
if (strchr(eth,'.') != NULL){ if (strchr(eth,'.') != NULL){
strcpy(eth,""); strcpy(eth,"");
@ -226,7 +236,7 @@ void Listener::ThreadExecution() {
#endif #endif
//udpsocket doesnt exist //udpsocket doesnt exist
if (!udpSocket) { if (*activated && !udpSocket) {
FILE_LOG(logERROR) << "Listening_Thread " << index << ": UDP Socket not created or shut down earlier"; FILE_LOG(logERROR) << "Listening_Thread " << index << ": UDP Socket not created or shut down earlier";
(*((uint32_t*)buffer)) = 0; (*((uint32_t*)buffer)) = 0;
StopListening(buffer); StopListening(buffer);
@ -234,17 +244,25 @@ void Listener::ThreadExecution() {
} }
//get data //get data
if (*status != TRANSMITTING) if (*status != TRANSMITTING) {
rc = ListenToAnImage(buffer + generalData->fifoBufferHeaderSize); if (*activated)
rc = ListenToAnImage(buffer + generalData->fifoBufferHeaderSize);
else
rc = CreateAnImage(buffer + generalData->fifoBufferHeaderSize);
}
//done acquiring //done acquiring
if (*status == TRANSMITTING) { if (*status == TRANSMITTING || ((!(*activated)) && (rc == 0))) {
StopListening(buffer); StopListening(buffer);
return; return;
} }
//error check //error check
if (rc <= 0) cprintf(BG_RED,"Error:(Weird), UDP Sockets not shut down, but received nothing\n"); if (rc <= 0) {
cprintf(BG_RED,"Error:(Weird), UDP Sockets not shut down, but received nothing\n");
fifo->FreeAddress(buffer);
return;
}
(*((uint32_t*)buffer)) = rc; (*((uint32_t*)buffer)) = rc;
(*((uint64_t*)(buffer + FIFO_HEADER_NUMBYTES ))) = currentFrameIndex; (*((uint64_t*)(buffer + FIFO_HEADER_NUMBYTES ))) = currentFrameIndex;
@ -311,22 +329,34 @@ uint32_t Listener::ListenToAnImage(char* buf) {
//future packet //future packet
if(fnum != currentFrameIndex) { if (fnum != currentFrameIndex) {
carryOverFlag = true; carryOverFlag = true;
memcpy(carryOverPacket,listeningPacket, generalData->packetSize); memcpy(carryOverPacket,listeningPacket, generalData->packetSize);
return generalData->imageSize; return generalData->imageSize;
} }
//copy packet and update fnum //copy packet
memcpy(buf + (pnum * dsize), listeningPacket + generalData->headerSizeinPacket, dsize); memcpy(buf + (pnum * dsize), listeningPacket + generalData->headerSizeinPacket, dsize);
(*((uint64_t*)(buf - FILE_FRAME_HEADER_SIZE))) = fnum;
} }
return generalData->imageSize; return generalData->imageSize;
} }
uint32_t Listener::CreateAnImage(char* buf) {
if (!measurementStartedFlag)
RecordFirstIndices(0);
if (currentFrameIndex == *numImages)
return 0;
//update parameters
numPacketsCaught++; //record immediately to get more time before socket shutdown
numTotalPacketsCaught++;
//reset data to -1
memset(buf, 0xFF, generalData->dataSize);
return generalData->imageSize;
}

4
slsReceiverSoftware/src/ThreadObject.cpp
View File

@ -80,7 +80,7 @@ void ThreadObject::RunningThread() {
ThreadExecution(); ThreadExecution();
}/*--end of inner loop */ }//end of inner loop
//wait till the next acquisition //wait till the next acquisition
@ -91,7 +91,7 @@ void ThreadObject::RunningThread() {
pthread_exit(NULL); pthread_exit(NULL);
} }
}/*--end of loop for each acquisition (outer loop) */ }//end of outer loop
} }

151
slsReceiverSoftware/src/UDPStandardImplementation.cpp
View File

@ -13,27 +13,24 @@
#include <cstdlib> //system #include <cstdlib> //system
#include <cstring> //strcpy #include <cstring> //strcpy
#include <errno.h> //eperm
using namespace std; using namespace std;
/** cosntructor & destructor */ /** cosntructor & destructor */
UDPStandardImplementation::UDPStandardImplementation() { UDPStandardImplementation::UDPStandardImplementation() {
InitializeMembers(); InitializeMembers();
} }
UDPStandardImplementation::~UDPStandardImplementation() { UDPStandardImplementation::~UDPStandardImplementation() {
DeleteMembers(); DeleteMembers();
} }
void UDPStandardImplementation::DeleteMembers() { void UDPStandardImplementation::DeleteMembers() {
if (generalData) { delete generalData; generalData=0;} if (generalData) { delete generalData; generalData=0;}
for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it) for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it)
delete(*it); delete(*it);
@ -51,8 +48,6 @@ void UDPStandardImplementation::DeleteMembers() {
void UDPStandardImplementation::InitializeMembers() { void UDPStandardImplementation::InitializeMembers() {
UDPBaseImplementation::initializeMembers(); UDPBaseImplementation::initializeMembers();
acquisitionPeriod = SAMPLE_TIME_IN_NS; acquisitionPeriod = SAMPLE_TIME_IN_NS;
@ -73,8 +68,6 @@ void UDPStandardImplementation::InitializeMembers() {
/*** Overloaded Functions called by TCP Interface ***/ /*** Overloaded Functions called by TCP Interface ***/
uint64_t UDPStandardImplementation::getTotalFramesCaught() const { uint64_t UDPStandardImplementation::getTotalFramesCaught() const {
uint64_t sum = 0; uint64_t sum = 0;
uint32_t flagsum = 0; uint32_t flagsum = 0;
@ -91,8 +84,6 @@ uint64_t UDPStandardImplementation::getTotalFramesCaught() const {
} }
uint64_t UDPStandardImplementation::getFramesCaught() const { uint64_t UDPStandardImplementation::getFramesCaught() const {
uint64_t sum = 0; uint64_t sum = 0;
uint32_t flagsum = 0; uint32_t flagsum = 0;
@ -108,8 +99,6 @@ uint64_t UDPStandardImplementation::getFramesCaught() const {
} }
int64_t UDPStandardImplementation::getAcquisitionIndex() const { int64_t UDPStandardImplementation::getAcquisitionIndex() const {
uint64_t sum = 0; uint64_t sum = 0;
uint32_t flagsum = 0; uint32_t flagsum = 0;
@ -126,8 +115,6 @@ int64_t UDPStandardImplementation::getAcquisitionIndex() const {
void UDPStandardImplementation::setFileFormat(const fileFormat f){ void UDPStandardImplementation::setFileFormat(const fileFormat f){
FILE_LOG(logDEBUG) << __AT__ << " starting";
switch(f){ switch(f){
#ifdef HDF5C #ifdef HDF5C
case HDF5: case HDF5:
@ -147,8 +134,6 @@ void UDPStandardImplementation::setFileFormat(const fileFormat f){
void UDPStandardImplementation::setFileName(const char c[]) { void UDPStandardImplementation::setFileName(const char c[]) {
if (strlen(c)) { if (strlen(c)) {
strcpy(fileName, c); //automatically update fileName in Filewriter (pointer) strcpy(fileName, c); //automatically update fileName in Filewriter (pointer)
/*int detindex = -1; /*int detindex = -1;
@ -169,8 +154,6 @@ void UDPStandardImplementation::setFileName(const char c[]) {
int UDPStandardImplementation::setShortFrameEnable(const int i) { int UDPStandardImplementation::setShortFrameEnable(const int i) {
if (myDetectorType != GOTTHARD) { if (myDetectorType != GOTTHARD) {
cprintf(RED, "Error: Can not set short frame for this detector\n"); cprintf(RED, "Error: Can not set short frame for this detector\n");
return FAIL; return FAIL;
@ -193,6 +176,8 @@ int UDPStandardImplementation::setShortFrameEnable(const int i) {
(*it)->SetGeneralData(generalData); (*it)->SetGeneralData(generalData);
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it) for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it)
(*it)->SetGeneralData(generalData); (*it)->SetGeneralData(generalData);
for (vector<DataStreamer*>::const_iterator it = dataStreamer.begin(); it != dataStreamer.end(); ++it)
(*it)->SetGeneralData(generalData);
} }
FILE_LOG (logINFO) << "Short Frame Enable: " << shortFrameEnable; FILE_LOG (logINFO) << "Short Frame Enable: " << shortFrameEnable;
return OK; return OK;
@ -200,8 +185,6 @@ int UDPStandardImplementation::setShortFrameEnable(const int i) {
int UDPStandardImplementation::setFrameToGuiFrequency(const uint32_t freq) { int UDPStandardImplementation::setFrameToGuiFrequency(const uint32_t freq) {
if (frameToGuiFrequency != freq) { if (frameToGuiFrequency != freq) {
frameToGuiFrequency = freq; frameToGuiFrequency = freq;
@ -222,8 +205,6 @@ int UDPStandardImplementation::setFrameToGuiFrequency(const uint32_t freq) {
int UDPStandardImplementation::setDataStreamEnable(const bool enable) { int UDPStandardImplementation::setDataStreamEnable(const bool enable) {
if (dataStreamEnable != enable) { if (dataStreamEnable != enable) {
dataStreamEnable = enable; dataStreamEnable = enable;
@ -233,16 +214,27 @@ int UDPStandardImplementation::setDataStreamEnable(const bool enable) {
dataStreamer.clear(); dataStreamer.clear();
if (enable) { if (enable) {
for ( int i=0; i < numThreads; ++i ) { bool error = false;
dataStreamer.push_back(new DataStreamer()); for ( int i = 0; i < numThreads; ++i ) {
if (DataStreamer::GetErrorMask()) { dataStreamer.push_back(new DataStreamer(fifo[i], &frameToGuiFrequency, &frameToGuiTimerinMS, &dynamicRange));
cprintf(BG_RED,"Error: Could not create data callback threads\n"); dataStreamer[i]->SetGeneralData(generalData);
for (vector<DataStreamer*>::const_iterator it = dataStreamer.begin(); it != dataStreamer.end(); ++it) if (dataStreamer[i]->CreateZmqSockets() == FAIL) {
delete(*it); error = true;
dataStreamer.clear(); break;
return FAIL;
} }
} }
if (DataStreamer::GetErrorMask() || error) {
if (DataStreamer::GetErrorMask())
cprintf(BG_RED,"Error: Could not create data callback threads\n");
else
cprintf(BG_RED,"Error: Could not create zmq sockets\n");
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; FILE_LOG (logINFO) << "Data Send to Gui: " << dataStreamEnable;
@ -251,8 +243,6 @@ int UDPStandardImplementation::setDataStreamEnable(const bool enable) {
int UDPStandardImplementation::setAcquisitionPeriod(const uint64_t i) { int UDPStandardImplementation::setAcquisitionPeriod(const uint64_t i) {
if (acquisitionPeriod != i) { if (acquisitionPeriod != i) {
acquisitionPeriod = i; acquisitionPeriod = i;
@ -273,8 +263,6 @@ int UDPStandardImplementation::setAcquisitionPeriod(const uint64_t i) {
int UDPStandardImplementation::setAcquisitionTime(const uint64_t i) { int UDPStandardImplementation::setAcquisitionTime(const uint64_t i) {
if (acquisitionTime != i) { if (acquisitionTime != i) {
acquisitionTime = i; acquisitionTime = i;
@ -295,8 +283,6 @@ int UDPStandardImplementation::setAcquisitionTime(const uint64_t i) {
int UDPStandardImplementation::setNumberOfFrames(const uint64_t i) { int UDPStandardImplementation::setNumberOfFrames(const uint64_t i) {
if (numberOfFrames != i) { if (numberOfFrames != i) {
numberOfFrames = i; numberOfFrames = i;
@ -317,8 +303,6 @@ int UDPStandardImplementation::setNumberOfFrames(const uint64_t i) {
int UDPStandardImplementation::setDynamicRange(const uint32_t i) { int UDPStandardImplementation::setDynamicRange(const uint32_t i) {
if (dynamicRange != i) { if (dynamicRange != i) {
dynamicRange = i; dynamicRange = i;
@ -335,8 +319,6 @@ int UDPStandardImplementation::setDynamicRange(const uint32_t i) {
int UDPStandardImplementation::setTenGigaEnable(const bool b) { int UDPStandardImplementation::setTenGigaEnable(const bool b) {
if (tengigaEnable != b) { if (tengigaEnable != b) {
tengigaEnable = b; tengigaEnable = b;
//side effects //side effects
@ -352,8 +334,6 @@ int UDPStandardImplementation::setTenGigaEnable(const bool b) {
int UDPStandardImplementation::setFifoDepth(const uint32_t i) { int UDPStandardImplementation::setFifoDepth(const uint32_t i) {
if (fifoDepth != i) { if (fifoDepth != i) {
fifoDepth = i; fifoDepth = i;
@ -368,8 +348,6 @@ int UDPStandardImplementation::setFifoDepth(const uint32_t i) {
int UDPStandardImplementation::setDetectorType(const detectorType d) { int UDPStandardImplementation::setDetectorType(const detectorType d) {
FILE_LOG (logDEBUG) << "Setting receiver type"; FILE_LOG (logDEBUG) << "Setting receiver type";
DeleteMembers(); DeleteMembers();
@ -415,8 +393,8 @@ int UDPStandardImplementation::setDetectorType(const detectorType d) {
//create threads //create threads
for ( int i=0; i < numThreads; ++i ) { for ( int i=0; i < numThreads; ++i ) {
listener.push_back(new Listener(fifo[i], &status, &udpPortNum[i], eth)); listener.push_back(new Listener(fifo[i], &status, &udpPortNum[i], eth, &activated, &numberOfFrames));
dataProcessor.push_back(new DataProcessor(fifo[i], &status, &statusMutex, &fileFormatType, &fileWriteEnable, dataProcessor.push_back(new DataProcessor(fifo[i], &fileFormatType, &fileWriteEnable, &dataStreamEnable,
&callbackAction, rawDataReadyCallBack,pRawDataReady)); &callbackAction, rawDataReadyCallBack,pRawDataReady));
if (Listener::GetErrorMask() || DataProcessor::GetErrorMask()) { if (Listener::GetErrorMask() || DataProcessor::GetErrorMask()) {
FILE_LOG (logERROR) << "Error: Could not creates listener/dataprocessor threads (index:" << i << ")"; FILE_LOG (logERROR) << "Error: Could not creates listener/dataprocessor threads (index:" << i << ")";
@ -436,6 +414,9 @@ int UDPStandardImplementation::setDetectorType(const detectorType d) {
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it) { for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it) {
(*it)->SetGeneralData(generalData); (*it)->SetGeneralData(generalData);
} }
SetThreadPriorities();
FILE_LOG (logDEBUG) << " Detector type set to " << getDetectorType(d); FILE_LOG (logDEBUG) << " Detector type set to " << getDetectorType(d);
return OK; return OK;
} }
@ -444,7 +425,6 @@ int UDPStandardImplementation::setDetectorType(const detectorType d) {
void UDPStandardImplementation::setDetectorPositionId(const int i){ void UDPStandardImplementation::setDetectorPositionId(const int i){
FILE_LOG(logDEBUG) << __AT__ << " starting";
detID = i; detID = i;
FILE_LOG(logINFO) << "Detector Position Id:" << detID; FILE_LOG(logINFO) << "Detector Position Id:" << detID;
@ -456,17 +436,19 @@ void UDPStandardImplementation::setDetectorPositionId(const int i){
void UDPStandardImplementation::resetAcquisitionCount() { void UDPStandardImplementation::resetAcquisitionCount() {
for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it) for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it)
(*it)->ResetParametersforNewAcquisition(); (*it)->ResetParametersforNewAcquisition();
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it) for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it)
(*it)->ResetParametersforNewAcquisition(); (*it)->ResetParametersforNewAcquisition();
for (vector<DataStreamer*>::const_iterator it = dataStreamer.begin(); it != dataStreamer.end(); ++it)
(*it)->ResetParametersforNewAcquisition();
FILE_LOG (logINFO) << "Acquisition Count has been reset"; FILE_LOG (logINFO) << "Acquisition Count has been reset";
} }
int UDPStandardImplementation::VerifyCallBackAction() { int UDPStandardImplementation::VerifyCallBackAction() {
/** file path and file index not required?? or need to include detector index? do they need the datasize? its given for write data anyway */ /** file path and file index not required?? or need to include detector index? do they need the datasize? its given for write data anyway */
@ -496,7 +478,6 @@ int UDPStandardImplementation::VerifyCallBackAction() {
} }
int UDPStandardImplementation::startReceiver(char *c) { int UDPStandardImplementation::startReceiver(char *c) {
ResetParametersforNewMeasurement(); ResetParametersforNewMeasurement();
//listener //listener
@ -541,7 +522,6 @@ int UDPStandardImplementation::startReceiver(char *c) {
void UDPStandardImplementation::stopReceiver(){ void UDPStandardImplementation::stopReceiver(){
FILE_LOG(logDEBUG) << __AT__ << " called";
FILE_LOG(logINFO) << "Stopping Receiver"; FILE_LOG(logINFO) << "Stopping Receiver";
//set status to transmitting //set status to transmitting
@ -554,7 +534,9 @@ void UDPStandardImplementation::stopReceiver(){
while(DataProcessor::GetRunningMask()){ while(DataProcessor::GetRunningMask()){
usleep(5000); usleep(5000);
} }
while(DataStreamer::GetRunningMask()){
usleep(5000);
}
pthread_mutex_lock(&statusMutex); pthread_mutex_lock(&statusMutex);
status = RUN_FINISHED; status = RUN_FINISHED;
@ -579,6 +561,8 @@ void UDPStandardImplementation::stopReceiver(){
cprintf(GREEN, "Last Frame Number Caught :%lld\n",(long long int)listener[i]->GetLastFrameIndexCaught()); cprintf(GREEN, "Last Frame Number Caught :%lld\n",(long long int)listener[i]->GetLastFrameIndexCaught());
} }
} }
if(!activated)
cprintf(RED,"Note: Deactivated Receiver\n");
//callback //callback
if (acquisitionFinishedCallBack) if (acquisitionFinishedCallBack)
acquisitionFinishedCallBack((int)(tot/numThreads), pAcquisitionFinished); acquisitionFinishedCallBack((int)(tot/numThreads), pAcquisitionFinished);
@ -597,8 +581,6 @@ void UDPStandardImplementation::stopReceiver(){
void UDPStandardImplementation::startReadout(){ void UDPStandardImplementation::startReadout(){
FILE_LOG(logDEBUG) << __AT__ << " called";
if(status == RUNNING){ if(status == RUNNING){
//needs to wait for packets only if activated //needs to wait for packets only if activated
@ -647,7 +629,6 @@ void UDPStandardImplementation::startReadout(){
void UDPStandardImplementation::shutDownUDPSockets() { void UDPStandardImplementation::shutDownUDPSockets() {
for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it) for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it)
(*it)->ShutDownUDPSocket(); (*it)->ShutDownUDPSocket();
} }
@ -655,7 +636,6 @@ void UDPStandardImplementation::shutDownUDPSockets() {
void UDPStandardImplementation::closeFiles() { void UDPStandardImplementation::closeFiles() {
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it) for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it)
(*it)->CloseFiles(); (*it)->CloseFiles();
} }
@ -663,8 +643,6 @@ void UDPStandardImplementation::closeFiles() {
void UDPStandardImplementation::SetLocalNetworkParameters() { void UDPStandardImplementation::SetLocalNetworkParameters() {
//to increase socket receiver buffer size and max length of input queue by changing kernel settings //to increase socket receiver buffer size and max length of input queue by changing kernel settings
if (myDetectorType == EIGER) if (myDetectorType == EIGER)
return; return;
@ -691,10 +669,46 @@ void UDPStandardImplementation::SetLocalNetworkParameters() {
void UDPStandardImplementation::SetThreadPriorities() {
for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it){
if ((*it)->SetThreadPriority(LISTENER_PRIORITY) == FAIL) {
FILE_LOG(logWARNING) << "Unable to prioritize threads. Root privileges required for this option.";
return;
}
}
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it){
if ((*it)->SetThreadPriority(PROCESSOR_PRIORITY) == FAIL) {
FILE_LOG(logWARNING) << "Unable to prioritize threads. Root privileges required for this option.";
return;
}
}
for (vector<DataStreamer*>::const_iterator it = dataStreamer.begin(); it != dataStreamer.end(); ++it){
if ((*it)->SetThreadPriority(STREAMER_PRIORITY) == FAIL) {
FILE_LOG(logWARNING) << "Unable to prioritize threads. Root privileges required for this option.";
return;
}
}
struct sched_param tcp_param;
tcp_param.sched_priority = TCP_PRIORITY;
if (pthread_setschedparam(pthread_self(),5 , &tcp_param) != EPERM) {
FILE_LOG(logWARNING) << "Unable to prioritize threads. Root privileges required for this option.";
return;
}
ostringstream osfn;
osfn << "Priorities set - "
"TCP:"<< TCP_PRIORITY <<
", Listener:" << LISTENER_PRIORITY <<
", Processor:" << PROCESSOR_PRIORITY;
if (dataStreamEnable)
osfn << ", Streamer:" << STREAMER_PRIORITY;
FILE_LOG(logINFO) << osfn.str();
}
int UDPStandardImplementation::SetupFifoStructure() { int UDPStandardImplementation::SetupFifoStructure() {
//recalculate number of jobs & fifodepth, return if no change //recalculate number of jobs & fifodepth, return if no change
if ((myDetectorType == GOTTHARD) || (myDetectorType == PROPIX)) { if ((myDetectorType == GOTTHARD) || (myDetectorType == PROPIX)) {
@ -747,6 +761,7 @@ int UDPStandardImplementation::SetupFifoStructure() {
//set the listener & dataprocessor threads to point to the right fifo //set the listener & dataprocessor threads to point to the right fifo
if(listener.size())listener[i]->SetFifo(fifo[i]); if(listener.size())listener[i]->SetFifo(fifo[i]);
if(dataProcessor.size())dataProcessor[i]->SetFifo(fifo[i]); if(dataProcessor.size())dataProcessor[i]->SetFifo(fifo[i]);
if(dataStreamer.size())dataStreamer[i]->SetFifo(fifo[i]);
} }
FILE_LOG (logINFO) << "Fifo structure(s) reconstructed"; FILE_LOG (logINFO) << "Fifo structure(s) reconstructed";
@ -756,18 +771,21 @@ int UDPStandardImplementation::SetupFifoStructure() {
void UDPStandardImplementation::ResetParametersforNewMeasurement() { void UDPStandardImplementation::ResetParametersforNewMeasurement() {
Listener::ResetRunningMask();
DataProcessor::ResetRunningMask();
DataStreamer::ResetRunningMask();
for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it) for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it)
(*it)->ResetParametersforNewMeasurement(); (*it)->ResetParametersforNewMeasurement();
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it) for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it)
(*it)->ResetParametersforNewMeasurement(); (*it)->ResetParametersforNewMeasurement();
for (vector<DataStreamer*>::const_iterator it = dataStreamer.begin(); it != dataStreamer.end(); ++it)
(*it)->ResetParametersforNewMeasurement();
} }
int UDPStandardImplementation::CreateUDPSockets() { int UDPStandardImplementation::CreateUDPSockets() {
bool error = false; bool error = false;
for (unsigned int i = 0; i < listener.size(); ++i) for (unsigned int i = 0; i < listener.size(); ++i)
if (listener[i]->CreateUDPSockets() == FAIL) { if (listener[i]->CreateUDPSockets() == FAIL) {
@ -785,7 +803,6 @@ int UDPStandardImplementation::CreateUDPSockets() {
int UDPStandardImplementation::SetupWriter() { int UDPStandardImplementation::SetupWriter() {
bool error = false; bool error = false;
for (unsigned int i = 0; i < dataProcessor.size(); ++i) for (unsigned int i = 0; i < dataProcessor.size(); ++i)
if (dataProcessor[i]->CreateNewFile(tengigaEnable, if (dataProcessor[i]->CreateNewFile(tengigaEnable,
@ -805,8 +822,6 @@ int UDPStandardImplementation::SetupWriter() {
void UDPStandardImplementation::StartRunning() { void UDPStandardImplementation::StartRunning() {
//set running mask and post semaphore to start the inner loop in execution thread //set running mask and post semaphore to start the inner loop in execution thread
for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it) { for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it) {
(*it)->StartRunning(); (*it)->StartRunning();
@ -816,4 +831,8 @@ void UDPStandardImplementation::StartRunning() {
(*it)->StartRunning(); (*it)->StartRunning();
(*it)->Continue(); (*it)->Continue();
} }
for (vector<DataStreamer*>::const_iterator it = dataStreamer.begin(); it != dataStreamer.end(); ++it){
(*it)->StartRunning();
(*it)->Continue();
}
} }