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

somewhere

Browse Source
This commit is contained in:
Dhanya Maliakal
2017-02-10 10:08:00 +01:00
parent c89f6e649c
commit b260d08225
18 changed files with 1076 additions and 356 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
slsReceiverSoftware/src/BinaryFileWriter.cpp
View File

@ -7,16 +7,64 @@
#include "BinaryFileWriter.h"
#include <iostream>
#include <iomanip>
using namespace std;
BinaryFileWriter::BinaryFileWriter(char* fname):
FileWriter(fname) {
BinaryFileWriter::BinaryFileWriter(int ind, char* fname, char* fpath, uint64_t* findex,
bool* frindexenable, bool* owenable, int* dindex, int* nunits, uint32_t maxf):
FileWriter(ind, fname, fpath, findex, frindexenable, owenable, dindex, nunits),
maxFramesPerFile(maxf)
{
printf("%d BinaryFileWriter constructor\n",index);
PrintMembers();
}
BinaryFileWriter::~BinaryFileWriter() {
printf("%d BinaryFileWriter destructor\n",index);
}
void BinaryFileWriter::PrintMembers() {
FileWriter::PrintMembers();
printf("Max Frames Per File: %d\n",maxFramesPerFile);
}
slsReceiverDefs::fileFormat BinaryFileWriter::GetType() {
return BINARY;
}
void BinaryFileWriter::SetMaxFramesPerFile(uint32_t maxf) {
maxFramesPerFile = maxf;
}
int BinaryFileWriter::CreateFile(uint64_t fnum) {
currentFileName = CreateFileName(filePath, fileNamePrefix, *fileIndex,
*frameIndexEnable, fnum, *detIndex, *numUnitsPerDetector, index);
printf("%d Binary File: %s\n", index, currentFileName.c_str());
return OK;
}
void BinaryFileWriter::CloseFile() {
printf("%d Closing File: %s\n", index, currentFileName.c_str());
}
string BinaryFileWriter::CreateFileName(char* fpath, char* fnameprefix, uint64_t findex,
bool frindexenable, uint64_t fnum, int dindex, int numunits, int unitindex) {
ostringstream osfn;
osfn << fpath << "/" << fnameprefix;
if (dindex >= 0) osfn << "_d" << (dindex * numunits + unitindex);
if (frindexenable) osfn << "_f" << setfill('0') << setw(12) << fnum;
osfn << "_" << findex;
osfn << ".raw";
return osfn.str();
}
int BinaryFileWriter::CreateDataFile(bool owenable, char* fname) {
return OK;
}
void BinaryFileWriter::CloseDataFile() {
}

227
slsReceiverSoftware/src/DataProcessor.cpp
View File

@ -30,24 +30,29 @@ pthread_mutex_t DataProcessor::Mutex = PTHREAD_MUTEX_INITIALIZER;
const GeneralData* DataProcessor::generalData(0);
bool DataProcessor::acquisitionStartedFlag(false);
bool DataProcessor::measurementStartedFlag(false);
DataProcessor::DataProcessor(Fifo*& f, runStatus* s, pthread_mutex_t* m) :
DataProcessor::DataProcessor(Fifo*& f, runStatus* s, pthread_mutex_t* m, fileFormat* ftype, bool* fwenable,
int* cbaction,
void (*dataReadycb)(int, char*, int, FILE*, char*, void*),
void *pDataReadycb) :
ThreadObject(NumberofDataProcessors),
fifo(f),
acquisitionStartedFlag(false),
measurementStartedFlag(false),
status(s),
statusMutex(m),
numTotalFramesCaught(0),
numFramesCaught(0),
firstAcquisitionIndex(0),
firstMeasurementIndex(0),
currentFrameIndex(0)
currentFrameIndex(0),
fileWriter(0),
fileFormatType(ftype),
fileWriteEnable(fwenable),
callbackAction(cbaction),
rawDataReadyCallBack(dataReadycb),
pRawDataReady(pDataReadycb)
{
FILE_LOG (logDEBUG) << __AT__ << " called";
if(ThreadObject::CreateThread()){
pthread_mutex_lock(&Mutex);
ErrorMask ^= (1<<index);
@ -59,40 +64,22 @@ DataProcessor::DataProcessor(Fifo*& f, runStatus* s, pthread_mutex_t* m) :
DataProcessor::~DataProcessor() {
FILE_LOG (logDEBUG) << __AT__ << " called";
for (vector<FileWriter*>::const_iterator it = fileWriter.begin(); it != fileWriter.end(); ++it)
delete(*it);
fileWriter.clear();
if (fileWriter) delete fileWriter;
ThreadObject::DestroyThread();
NumberofDataProcessors--;
}
/** static functions */
uint64_t DataProcessor::GetErrorMask() {
FILE_LOG (logDEBUG) << __AT__ << " called";
return ErrorMask;
}
bool DataProcessor::GetAcquisitionStartedFlag(){
FILE_LOG (logDEBUG) << __AT__ << " called";
return acquisitionStartedFlag;
uint64_t DataProcessor::GetRunningMask() {
return RunningMask;
}
bool DataProcessor::GetMeasurementStartedFlag(){
FILE_LOG (logDEBUG) << __AT__ << " called";
return measurementStartedFlag;
}
void DataProcessor::SetGeneralData(GeneralData*& g) {
FILE_LOG (logDEBUG) << __AT__ << " called";
generalData = g;
#ifdef VERY_VERBOSE
generalData->Print();
@ -101,37 +88,40 @@ void DataProcessor::SetGeneralData(GeneralData*& g) {
/** non static functions */
/** getters */
string DataProcessor::GetType(){
return TypeName;
}
bool DataProcessor::IsRunning() {
return ((1 << index) & RunningMask);
}
bool DataProcessor::GetAcquisitionStartedFlag(){
return acquisitionStartedFlag;
}
bool DataProcessor::GetMeasurementStartedFlag(){
return measurementStartedFlag;
}
uint64_t DataProcessor::GetNumTotalFramesCaught() {
FILE_LOG (logDEBUG) << __AT__ << " called";
return numTotalFramesCaught;
}
uint64_t DataProcessor::GetNumFramesCaught() {
FILE_LOG (logDEBUG) << __AT__ << " called";
return numFramesCaught;
}
uint64_t DataProcessor::GetProcessedAcquisitionIndex() {
FILE_LOG (logDEBUG) << __AT__ << " called";
return currentFrameIndex - firstAcquisitionIndex;
}
bool DataProcessor::IsRunning() {
FILE_LOG (logDEBUG) << __AT__ << " called";
return ((1 << index) & RunningMask);
}
/** setters */
void DataProcessor::StartRunning() {
FILE_LOG (logDEBUG) << __AT__ << " called";
pthread_mutex_lock(&Mutex);
RunningMask |= (1<<index);
pthread_mutex_unlock(&Mutex);
@ -139,7 +129,6 @@ void DataProcessor::StartRunning() {
void DataProcessor::StopRunning() {
FILE_LOG (logDEBUG) << __AT__ << " called";
pthread_mutex_lock(&Mutex);
RunningMask ^= (1<<index);
pthread_mutex_unlock(&Mutex);
@ -147,32 +136,22 @@ void DataProcessor::StopRunning() {
void DataProcessor::SetFifo(Fifo*& f) {
FILE_LOG (logDEBUG) << __AT__ << " called";
fifo = f;
}
void DataProcessor::ResetParametersforNewAcquisition() {
FILE_LOG (logDEBUG) << __AT__ << " called";
numTotalFramesCaught = 0;
firstAcquisitionIndex = 0;
currentFrameIndex = 0;
if(acquisitionStartedFlag){
pthread_mutex_lock(&Mutex);
acquisitionStartedFlag = false;
pthread_mutex_unlock(&Mutex);
}
acquisitionStartedFlag = false;
}
void DataProcessor::ResetParametersforNewMeasurement(){
FILE_LOG (logDEBUG) << __AT__ << " called";
numFramesCaught = 0;
firstMeasurementIndex = 0;
if(measurementStartedFlag){
pthread_mutex_lock(&Mutex);
measurementStartedFlag = false;
pthread_mutex_unlock(&Mutex);
}
measurementStartedFlag = false;
if(RunningMask){
pthread_mutex_lock(&Mutex);
RunningMask = 0x0;
@ -181,57 +160,133 @@ void DataProcessor::ResetParametersforNewMeasurement(){
}
void DataProcessor::ThreadExecution() {
FILE_LOG (logDEBUG) << __AT__ << " called";
void DataProcessor::RecordFirstIndices(uint64_t fnum) {
//listen to this fnum, later +1
currentFrameIndex = fnum;
char* buffer=0;
fifo->PopAddress(buffer);
#ifdef FIFODEBUG
if (!index) cprintf(BLUE,"DataProcessor %d, pop 0x%p buffer:%s\n", index,(void*)(buffer),buffer);
#endif
uint32_t numPackets = (uint32_t)(*((uint32_t*)buffer));
if (numPackets == DUMMY_PACKET_VALUE) {
StopProcessing(buffer);
return;
measurementStartedFlag = true;
firstMeasurementIndex = fnum;
//start of entire acquisition
if (!acquisitionStartedFlag) {
acquisitionStartedFlag = true;
firstAcquisitionIndex = fnum;
}
uint64_t fnum; uint32_t pnum; uint32_t snum; uint64_t bcid;
generalData->GetHeaderInfo(index,buffer+generalData->fifoBufferHeaderSize,16,fnum,pnum,snum,bcid);
if (!index) cprintf(BLUE,"DataProcessing %d: fnum:%lld, pnum:%d\n", index, (long long int)fnum, pnum);
#ifdef VERBOSE
cprintf(BLUE,"%d First Acquisition Index:%lld\tFirst Measurement Index:%lld\n",
index, (long long int)firstAcquisitionIndex, (long long int)firstMeasurementIndex);
#endif
}
fifo->FreeAddress(buffer);
void DataProcessor::SetMaxFramesPerFile() {
if (fileWriter->GetType() == BINARY)
fileWriter->SetMaxFramesPerFile(generalData->maxFramesPerFile);
}
void DataProcessor::SetFileFormat(const fileFormat f) {
if (*fileFormatType != f) {
switch(f){
#ifdef HDF5C
case HDF5:
*fileFormatType = f;
#endif
default:
*fileFormatType = f;
break;
}
//remember the pointer values before they are destroyed
char* fname=0; char* fpath=0; uint64_t* findex=0; bool* frindexenable=0;
bool* fwenable=0; bool* owenable=0; int* dindex=0; int* nunits=0;
fileWriter->GetMemberPointerValues(fname, fpath, findex, frindexenable, owenable, dindex, nunits);
SetupFileWriter(fname, fpath, findex, frindexenable, owenable, dindex, nunits);
}
}
void DataProcessor::StopProcessing(char* buf) {
FILE_LOG (logDEBUG) << __AT__ << " called";
void DataProcessor::SetupFileWriter(char* fname, char* fpath, uint64_t* findex,
bool* frindexenable, bool* owenable, int* dindex, int* nunits)
{
cprintf(BLUE,"%d: End of Processing\n", index);
if (fileWriter)
delete fileWriter;
fifo->FreeAddress(buf);
StopRunning();
if (!index) {
while (RunningMask)
usleep(5000);
pthread_mutex_lock(statusMutex);
*status = RUN_FINISHED;
pthread_mutex_unlock((statusMutex));
FILE_LOG(logINFO) << "Status: " << runStatusType(*status);
switch(*fileFormatType){
#ifdef HDF5C
case HDF5:
fileWriter = new HDF5FileWriter(index, fname, fpath, findex,
frindexenable, owenable, dindex, nunits);
break;
#endif
default:
fileWriter = new BinaryFileWriter(index, fname, fpath, findex,
frindexenable, owenable, dindex, nunits, generalData->maxFramesPerFile);
break;
}
}
int DataProcessor::CreateNewFile() {
FILE_LOG (logDEBUG) << __AT__ << " called";
//create file fileWriter.push_back(new BinaryFileWriter(fileName))
if (!fileWriter)
return FAIL;
if (fileWriter->CreateFile(currentFrameIndex) == FAIL)
return FAIL;
return OK;
}
void DataProcessor::CloseFile() {
FILE_LOG (logDEBUG) << __AT__ << " called";
if (fileWriter)
fileWriter->CloseFile();
}
void DataProcessor::ThreadExecution() {
char* buffer=0;
fifo->PopAddress(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 DataProcessor::StopProcessing(char* buf) {
fifo->FreeAddress(buf);
CloseFile();
StopRunning();
cprintf(BLUE,"%d: Processing Completed\n", index);
}
void DataProcessor::ProcessAnImage(char* buf) {
numFramesCaught++;
numTotalFramesCaught++;
uint64_t fnum = (*((uint64_t*)buf));
//#ifdef VERBOSE
if (!index) cprintf(BLUE,"DataProcessing %d: fnum:%lld\n", index, (long long int)fnum);
//#endif
if (!measurementStartedFlag)
RecordFirstIndices(fnum);
}

54
slsReceiverSoftware/src/FileWriter.cpp
View File

@ -10,15 +10,57 @@
using namespace std;
FileWriter::FileWriter(char* fname):
fileName(fname) {
cout<<"fileName:"<<fileName<<endl;
FileWriter::FileWriter(int ind, char* fname, char* fpath, uint64_t* findex,
bool* frindexenable, bool* owenable, int* dindex, int* nunits):
index(ind),
fileNamePrefix(fname),
filePath(fpath),
fileIndex(findex),
frameIndexEnable(frindexenable),
overWriteEnable(owenable),
detIndex(dindex),
numUnitsPerDetector(nunits)
{
printf("%d FileWriter constructor\n",index);
}
FileWriter::~FileWriter() {
printf("%d FileWriter Destructor\n", index);
}
char* FileWriter::GetFileName() {
return fileName;
string FileWriter::GetCurrentFileName() {
return currentFileName;
}
void FileWriter::PrintMembers() {
printf("\nGeneral Writer Variables:"
"Index: %d\n"
"File Name Prefix: %s\n"
"File Path: %s\n"
"File Index: %lld\n"
"Frame Index Enable: %d\n"
"Over Write Enable: %d\n"
"Detector Index: %d\n"
"Number of Units Per Detector: %d\n",
index,
fileNamePrefix,
filePath,
(long long int)*fileIndex,
*frameIndexEnable,
*overWriteEnable,
*detIndex,
*numUnitsPerDetector);
}
void FileWriter::GetMemberPointerValues(char* fname, char* fpath, uint64_t* findex,
bool* frindexenable, bool* owenable, int* dindex, int* nunits)
{
fname = fileNamePrefix;
fpath = filePath;
findex = fileIndex;
frindexenable = frameIndexEnable;
owenable = overWriteEnable;
dindex = detIndex;
nunits = numUnitsPerDetector;
}

48
slsReceiverSoftware/src/HDF5FileWriter.cpp
View File

@ -7,17 +7,59 @@
#include "HDF5FileWriter.h"
#include <iostream>
#include <iomanip>
using namespace std;
HDF5FileWriter::HDF5FileWriter(char* fname):
FileWriter(fname) {
HDF5FileWriter::HDF5FileWriter(int ind, char* fname, char* fpath, uint64_t* findex,
bool* frindexenable, bool* owenable, int* dindex, int* nunits):
FileWriter(ind, fname, fpath, findex, frindexenable, owenable, dindex, nunits)
{
printf("%d HDF5FileWriter constructor\n",index);
PrintMembers();
}
HDF5FileWriter::~HDF5FileWriter() {
printf("%d HDF5FileWriter destructor\n",index);
}
void HDF5FileWriter::PrintMembers() {
FileWriter::PrintMembers();
printf("\nHDF5 Print Members \n");
}
slsReceiverDefs::fileFormat HDF5FileWriter::GetType() {
return HDF5;
}
int HDF5FileWriter::CreateFile(uint64_t fnum) {
currentFileName = CreateFileName(filePath, fileNamePrefix, *fileIndex,
*frameIndexEnable, fnum, *detIndex, *numUnitsPerDetector, index);
printf("%d HDF5 File: %s\n", index, currentFileName.c_str());
return OK;
}
void HDF5FileWriter::CloseFile() {
}
string HDF5FileWriter::CreateFileName(char* fpath, char* fnameprefix, uint64_t findex,
bool frindexenable, uint64_t fnum, int dindex, int numunits, int unitindex) {
ostringstream osfn;
osfn << fpath << "/" << fnameprefix;
if (dindex >= 0) osfn << "_d" << (dindex * numunits + unitindex);
if (frindexenable) osfn << "_f" << setfill('0') << setw(12) << fnum;
osfn << "_" << findex;
osfn << ".h5";
return osfn.str();
}
int HDF5FileWriter::CreateDataFile(bool owenable, char* fname) {
return OK;
}
void HDF5FileWriter::CloseDataFile() {
}

303
slsReceiverSoftware/src/Listener.cpp
View File

@ -27,23 +27,25 @@ pthread_mutex_t Listener::Mutex = PTHREAD_MUTEX_INITIALIZER;
const GeneralData* Listener::generalData(0);
bool Listener::acquisitionStartedFlag(false);
bool Listener::measurementStartedFlag(false);
Listener::Listener(Fifo*& f, runStatus* s, uint32_t* portno) :
Listener::Listener(Fifo*& f, runStatus* s, uint32_t* portno, char* e) :
ThreadObject(NumberofListeners),
fifo(f),
acquisitionStartedFlag(false),
measurementStartedFlag(false),
status(s),
udpSocket(0),
udpPortNumber(portno),
eth(e),
numTotalPacketsCaught(0),
numPacketsCaught(0),
firstAcquisitionIndex(0),
firstMeasurementIndex(0)
firstMeasurementIndex(0),
currentFrameIndex(0),
lastCaughtFrameIndex(0),
carryOverFlag(0),
carryOverPacket(0)
{
FILE_LOG (logDEBUG) << __AT__ << " called";
if(ThreadObject::CreateThread()){
pthread_mutex_lock(&Mutex);
ErrorMask ^= (1<<index);
@ -55,67 +57,59 @@ Listener::Listener(Fifo*& f, runStatus* s, uint32_t* portno) :
Listener::~Listener() {
FILE_LOG (logDEBUG) << __AT__ << " called";
if (carryOverPacket) delete carryOverPacket;
ThreadObject::DestroyThread();
NumberofListeners--;
}
/** static functions */
uint64_t Listener::GetErrorMask() {
FILE_LOG (logDEBUG) << __AT__ << " called";
return ErrorMask;
}
bool Listener::GetAcquisitionStartedFlag(){
FILE_LOG (logDEBUG) << __AT__ << " called";
return acquisitionStartedFlag;
}
bool Listener::GetMeasurementStartedFlag(){
FILE_LOG (logDEBUG) << __AT__ << " called";
return measurementStartedFlag;
uint64_t Listener::GetRunningMask() {
return RunningMask;
}
void Listener::SetGeneralData(GeneralData*& g) {
FILE_LOG (logDEBUG) << __AT__ << " called";
generalData = g;
//#ifdef VERY_VERBOSE
#ifdef VERY_VERBOSE
generalData->Print();
//#endif
#endif
}
/** non static functions */
/** non static functions */
/** getters */
string Listener::GetType(){
return TypeName;
}
uint64_t Listener::GetTotalPacketsCaught() {
FILE_LOG (logDEBUG) << __AT__ << " called";
return numTotalPacketsCaught;
}
uint64_t Listener::GetNumReceivedinUDPBuffer() {
FILE_LOG (logDEBUG) << __AT__ << " called";
if(!udpSocket)
return 0;
return udpSocket->getCurrentTotalReceived();
}
bool Listener::IsRunning() {
FILE_LOG (logDEBUG) << __AT__ << " called";
return ((1 << index) & RunningMask);
}
bool Listener::GetAcquisitionStartedFlag(){
return acquisitionStartedFlag;
}
bool Listener::GetMeasurementStartedFlag(){
return measurementStartedFlag;
}
uint64_t Listener::GetTotalPacketsCaught() {
return numTotalPacketsCaught;
}
uint64_t Listener::GetLastFrameIndexCaught() {
return lastCaughtFrameIndex;
}
/** setters */
void Listener::StartRunning() {
FILE_LOG (logDEBUG) << __AT__ << " called";
pthread_mutex_lock(&Mutex);
RunningMask |= (1<<index);
pthread_mutex_unlock(&Mutex);
@ -123,7 +117,6 @@ void Listener::StartRunning() {
void Listener::StopRunning() {
FILE_LOG (logDEBUG) << __AT__ << " called";
pthread_mutex_lock(&Mutex);
RunningMask ^= (1<<index);
pthread_mutex_unlock(&Mutex);
@ -131,32 +124,28 @@ void Listener::StopRunning() {
void Listener::SetFifo(Fifo*& f) {
FILE_LOG (logDEBUG) << __AT__ << " called";
fifo = f;
}
void Listener::ResetParametersforNewAcquisition() {
FILE_LOG (logDEBUG) << __AT__ << " called";
acquisitionStartedFlag = false;
numTotalPacketsCaught = 0;
firstAcquisitionIndex = 0;
if(acquisitionStartedFlag){
pthread_mutex_lock(&Mutex);
acquisitionStartedFlag = false;
pthread_mutex_unlock(&Mutex);
}
currentFrameIndex = 0;
lastCaughtFrameIndex = 0;
}
void Listener::ResetParametersforNewMeasurement(){
FILE_LOG (logDEBUG) << __AT__ << " called";
measurementStartedFlag = false;
numPacketsCaught = 0;
firstMeasurementIndex = 0;
if(measurementStartedFlag){
pthread_mutex_lock(&Mutex);
measurementStartedFlag = false;
pthread_mutex_unlock(&Mutex);
}
carryOverFlag = false;
if (carryOverPacket)
delete carryOverPacket;
carryOverPacket = new char[generalData->packetSize];
if(RunningMask){
pthread_mutex_lock(&Mutex);
RunningMask = 0x0;
@ -165,84 +154,40 @@ void Listener::ResetParametersforNewMeasurement(){
}
void Listener::ThreadExecution() {
FILE_LOG (logDEBUG) << __AT__ << " called";
char* buffer;
fifo->GetNewAddress(buffer);
#ifdef FIFODEBUG
if (!index) cprintf(GREEN,"Listener %d, pop 0x%p buffer:%s\n", index,(void*)(buffer),buffer);
#endif
void Listener::RecordFirstIndices(uint64_t fnum) {
//listen to this fnum, later +1
currentFrameIndex = fnum;
int rc = 0;
measurementStartedFlag = true;
firstMeasurementIndex = fnum;
//udpsocket doesnt exist
if (!udpSocket) {
FILE_LOG(logERROR) << "Listening_Thread " << index << ": UDP Socket not created or shut down earlier";
(*((uint32_t*)buffer)) = 0;
StopListening(buffer);
return;
//start of entire acquisition
if (!acquisitionStartedFlag) {
acquisitionStartedFlag = true;
firstAcquisitionIndex = fnum;
}
//get data
if (*status != TRANSMITTING) {
rc = udpSocket->ReceiveDataOnly(buffer + generalData->fifoBufferHeaderSize,generalData->fifoBufferSize);
if (!index) cprintf(GREEN,"Listening %d: rc: %d\n",index,rc);
(*((uint32_t*)buffer)) = ((rc <= 0) ? 0 : rc);
}
//done acquiring
if (*status == TRANSMITTING) {
StopListening(buffer);
return;
}
uint64_t fnum; uint32_t pnum; uint32_t snum; uint64_t bcid=0;
generalData->GetHeaderInfo(index,buffer + generalData->fifoBufferHeaderSize,16,fnum,pnum,snum,bcid);
if (!index) cprintf(GREEN,"Listening %d: fnum:%lld, pnum:%d\n", index, (long long int)fnum, pnum);
//push into fifo
fifo->PushAddress(buffer);
if (!index) cprintf(GREEN,"%d First Acquisition Index:%lld\n"
"%d First Measurement Index:%lld\n",
index, (long long int)firstAcquisitionIndex,
index, (long long int)firstMeasurementIndex);
}
void Listener::StopListening(char* buf) {
FILE_LOG (logDEBUG) << __AT__ << " called";
int Listener::CreateUDPSockets() {
ShutDownUDPSocket();
cprintf(BLUE,"%d: End of Listening\n", index);
uint32_t numPackets = (uint32_t)(*((uint32_t*)buf));
//incomplete packets
if (numPackets > 0) {
fifo->PushAddress(buf);
fifo->GetNewAddress(buf);
#ifdef FIFODEBUG
if (!index) cprintf(GREEN,"Listener %d, Got incomplete, for dummy: pop 0x%p buffer:%s\n", index,(void*)(buf),buf);
#endif
//if eth is mistaken with ip address
if (strchr(eth,'.') != NULL){
strcpy(eth,"");
}
if(!strlen(eth)){
FILE_LOG(logWARNING) << "eth is empty. Listening to all";
}
//dummy
(*((uint32_t*)buf)) = DUMMY_PACKET_VALUE;
fifo->PushAddress(buf);
StopRunning();
//#ifdef DEBUG4
if (!index) FILE_LOG(logINFO) << "Listening Thread (" << *udpPortNumber << ")"
" Packets caught: " << numPacketsCaught;
//#endif
}
int Listener::CreateUDPSockets(const char* eth) {
FILE_LOG (logDEBUG) << __AT__ << " called";
udpSocket = new genericSocket(*udpPortNumber, genericSocket::UDP,
generalData->packetSize, eth, generalData->headerPacketSize);
generalData->packetSize, (strlen(eth)?eth:NULL), generalData->headerPacketSize);
int iret = udpSocket->getErrorStatus();
if(!iret){
cout << "UDP port opened at port " << *udpPortNumber << endl;
@ -256,8 +201,6 @@ int Listener::CreateUDPSockets(const char* eth) {
void Listener::ShutDownUDPSocket() {
FILE_LOG (logDEBUG) << __AT__ << " called";
if(udpSocket){
udpSocket->ShutDownSocket();
FILE_LOG(logINFO) << "Shut down UDP Socket " << index;
@ -266,3 +209,123 @@ void Listener::ShutDownUDPSocket() {
}
}
void Listener::ThreadExecution() {
char* buffer;
int rc = 0;
fifo->GetNewAddress(buffer);
#ifdef FIFODEBUG
if (!index) cprintf(GREEN,"Listener %d, pop 0x%p buffer:%s\n", index,(void*)(buffer),buffer);
#endif
//udpsocket doesnt exist
if (!udpSocket) {
FILE_LOG(logERROR) << "Listening_Thread " << index << ": UDP Socket not created or shut down earlier";
(*((uint32_t*)buffer)) = 0;
StopListening(buffer);
return;
}
//get data
if (*status != TRANSMITTING)
rc = ListenToAnImage(buffer + generalData->fifoBufferHeaderSize);
//done acquiring
if (*status == TRANSMITTING) {
StopListening(buffer);
return;
}
//error check
if (rc <= 0) cprintf(BG_RED,"Error:(Weird), UDP Sockets not shut down, but received nothing\n");
(*((uint32_t*)buffer)) = rc;
(*((uint64_t*)(buffer + FIFO_HEADER_NUMBYTES ))) = currentFrameIndex;
currentFrameIndex++;
//push into fifo
fifo->PushAddress(buffer);
}
void Listener::StopListening(char* buf) {
(*((uint32_t*)buf)) = DUMMY_PACKET_VALUE;
fifo->PushAddress(buf);
StopRunning();
#ifdef VERBOSE
cprintf(GREEN,"%d: Listening Packets (%d) : %d\n", index, *udpPortNumber, numPacketsCaught);
#endif
cprintf(GREEN,"%d: Listening Completed\n", index);
}
uint32_t Listener::ListenToAnImage(char* buf) {
uint32_t rc = 0;
uint64_t fnum = 0; uint32_t pnum = 0;
uint32_t offset = 0;
uint32_t currentpnum = 0;
int psize = generalData->packetSize;
//reset to -1
memset(buf,0xFF,psize);
//look for carry over
if (carryOverFlag) {
//check if its the current image packet
generalData->GetHeaderInfo(index,carryOverPacket,fnum,pnum);
if (fnum != currentFrameIndex) {
return generalData->imageSize;
}
carryOverFlag = false;
memcpy(buf,carryOverPacket, psize);
offset += psize;
}
while (offset < generalData->fifoBufferSize) {
//listen to new packet
rc += udpSocket->ReceiveDataOnly(buf + offset);
if (rc <= 0) return 0;
//update parameters
numPacketsCaught++; //record immediately to get more time before socket shutdown
numTotalPacketsCaught++;
generalData->GetHeaderInfo(index,buf + offset,fnum,pnum);
lastCaughtFrameIndex = fnum;
//#ifdef VERBOSE
if (!index && !pnum) cprintf(BLUE,"Listening %d: fnum:%lld, pnum:%d\n", index, (long long int)fnum, pnum);
//#endif
if (!measurementStartedFlag)
RecordFirstIndices(fnum);
//future packet
if(fnum != currentFrameIndex) {
carryOverFlag = true;
memcpy(carryOverPacket,buf + offset, psize);
return generalData->imageSize;
}
(*((uint64_t*)(buf - FILE_FRAME_HEADER_SIZE))) = fnum;
//future packet of same frame
if(pnum != currentpnum) {
memcpy(buf + (pnum * psize), buf + offset, psize);
memset(buf + offset, 0xFF, psize);
}
//update offset & current frame index
offset = (pnum + 1) * psize;
}
return generalData->imageSize;
}

16
slsReceiverSoftware/src/ThreadObject.cpp
View File

@ -16,20 +16,18 @@ ThreadObject::ThreadObject(int ind):
index(ind),
alive(false),
killThread(false),
thread(0) {
FILE_LOG (logDEBUG) << __AT__ << " called";
thread(0)
{
PrintMembers();
}
ThreadObject::~ThreadObject() {
FILE_LOG (logDEBUG) << __AT__ << " called";
DestroyThread();
}
void ThreadObject::PrintMembers() {
FILE_LOG (logDEBUG) << __AT__ << " called";
FILE_LOG (logDEBUG) << "Index : " << index
<< "\nalive: " << alive
<< "\nkillThread: " << killThread
@ -38,7 +36,6 @@ void ThreadObject::PrintMembers() {
void ThreadObject::DestroyThread() {
FILE_LOG (logDEBUG) << __AT__ << " called";
if(alive){
killThread = true;
sem_post(&semaphore);
@ -52,7 +49,6 @@ void ThreadObject::DestroyThread() {
int ThreadObject::CreateThread() {
FILE_LOG (logDEBUG) << __AT__ << " called";
if(alive){
FILE_LOG (logERROR) << "Cannot create thread " << index << ". Already alive";
return FAIL;
@ -72,16 +68,12 @@ int ThreadObject::CreateThread() {
void* ThreadObject::StartThread(void* thisPointer) {
FILE_LOG (logDEBUG) << __AT__ << " called";
((ThreadObject*)thisPointer)->RunningThread();
return thisPointer;
((ThreadObject*)thisPointer)->RunningThread();
return thisPointer;
}
void ThreadObject::RunningThread() {
FILE_LOG (logDEBUG) << __AT__ << " called";
while(true) {
while(IsRunning()) {

11
slsReceiverSoftware/src/UDPBaseImplementation.cpp
View File

@ -220,14 +220,17 @@ void UDPBaseImplementation::setFlippedData(int axis, int enable){
void UDPBaseImplementation::setFileFormat(const fileFormat f){
FILE_LOG(logDEBUG) << __AT__ << " starting";
if(f!=HDF5)
fileFormatType = f;
switch(f){
#ifdef HDF5C
else if((f==HDF5) && (myDetectorType == EIGER || myDetectorType == JUNGFRAU))
case HDF5:
fileFormatType = f;
#endif
default:
fileFormatType = f;
break;
}
FILE_LOG(logINFO) << "File Index:" << fileIndex;
FILE_LOG(logINFO) << "File Format:" << getFileFormatType(fileFormatType);
}

157
slsReceiverSoftware/src/UDPStandardImplementation.cpp
View File

@ -62,6 +62,7 @@ void UDPStandardImplementation::InitializeMembers() {
//*** receiver parameters ***
numThreads = 1;
numberofJobs = 1;
callbackAction = DO_EVERYTHING;
//*** mutex ***
pthread_mutex_init(&statusMutex,NULL);
@ -76,33 +77,69 @@ void UDPStandardImplementation::InitializeMembers() {
uint64_t UDPStandardImplementation::getTotalFramesCaught() const {
FILE_LOG (logDEBUG) << __AT__ << " starting";
uint64_t sum = 0;
uint32_t flagsum = 0;
vector<DataProcessor*>::const_iterator it;
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it)
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it) {
flagsum += ((*it)->GetMeasurementStartedFlag() ? 1 : 0);
sum += (*it)->GetNumTotalFramesCaught();
}
//no data processed
if (flagsum != dataProcessor.size()) return 0;
return (sum/dataProcessor.size());
}
uint64_t UDPStandardImplementation::getFramesCaught() const {
FILE_LOG (logDEBUG) << __AT__ << " starting";
uint64_t sum = 0;
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it)
uint32_t flagsum = 0;
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it) {
flagsum += ((*it)->GetAcquisitionStartedFlag() ? 1 : 0);
sum += (*it)->GetNumFramesCaught();
}
//no data processed
if (flagsum != dataProcessor.size()) return 0;
return (sum/dataProcessor.size());
}
int64_t UDPStandardImplementation::getAcquisitionIndex() const {
FILE_LOG (logDEBUG) << __AT__ << " starting";
//no data processed
if(!DataProcessor::GetAcquisitionStartedFlag())
return -1;
uint64_t sum = 0;
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it)
uint32_t flagsum = 0;
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it){
flagsum += ((*it)->GetAcquisitionStartedFlag() ? 1 : 0);
sum += (*it)->GetProcessedAcquisitionIndex();
}
//no data processed
if (flagsum != dataProcessor.size()) return -1;
return (sum/dataProcessor.size());
}
void UDPStandardImplementation::setFileFormat(const fileFormat f){
FILE_LOG(logDEBUG) << __AT__ << " starting";
//destroy file writer, set file format and create file writer
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it)
(*it)->SetFileFormat(f);
FILE_LOG(logINFO) << "File Format:" << getFileFormatType(fileFormatType);
}
void UDPStandardImplementation::setFileName(const char c[]) {
FILE_LOG (logDEBUG) << __AT__ << " starting";
@ -114,6 +151,8 @@ void UDPStandardImplementation::setFileName(const char c[]) {
if (uscore!=string::npos) {
if (sscanf(tempname.substr(uscore+1, tempname.size()-uscore-1).c_str(), "d%d", &detindex)) {
detID = detindex;
tempname=tempname.substr(0,uscore);
strcpy(fileName, tempname.c_str());
}
}
if (detindex == -1)
@ -146,6 +185,10 @@ int UDPStandardImplementation::setShortFrameEnable(const int i) {
Listener::SetGeneralData(generalData);
DataProcessor::SetGeneralData(generalData);
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it) {
(*it)->SetMaxFramesPerFile();
}
}
FILE_LOG (logINFO) << "Short Frame Enable: " << shortFrameEnable;
return OK;
@ -324,7 +367,7 @@ int UDPStandardImplementation::setDetectorType(const detectorType d) {
FILE_LOG (logDEBUG) << "Setting receiver type";
DeleteMembers();cout<<"size of fifo:"<<fifo.size()<<endl;
DeleteMembers();
InitializeMembers();
myDetectorType = d;
switch(myDetectorType) {
@ -357,6 +400,9 @@ int UDPStandardImplementation::setDetectorType(const detectorType d) {
numThreads = generalData->threadsPerReceiver;
fifoDepth = generalData->defaultFifoDepth;
//local network parameters
SetLocalNetworkParameters();
//create fifo structure
numberofJobs = -1;
if (SetupFifoStructure() == FAIL) {
@ -366,8 +412,9 @@ int UDPStandardImplementation::setDetectorType(const detectorType d) {
//create threads
for ( int i=0; i < numThreads; ++i ) {
listener.push_back(new Listener(fifo[i], &status, &udpPortNum[i]));
dataProcessor.push_back(new DataProcessor(fifo[i], &status, &statusMutex));
listener.push_back(new Listener(fifo[i], &status, &udpPortNum[i], eth));
dataProcessor.push_back(new DataProcessor(fifo[i], &status, &statusMutex, &fileFormatType, &fileWriteEnable,
&callbackAction, rawDataReadyCallBack,pRawDataReady));
if (Listener::GetErrorMask() || DataProcessor::GetErrorMask()) {
FILE_LOG (logERROR) << "Error: Could not creates listener/dataprocessor threads (index:" << i << ")";
for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it)
@ -380,8 +427,11 @@ int UDPStandardImplementation::setDetectorType(const detectorType d) {
}
}
//local network parameters
SetLocalNetworkParameters();
//set up writer and callbacks
for (vector<DataProcessor*>::const_iterator it = dataProcessor.begin(); it != dataProcessor.end(); ++it)
(*it)->SetupFileWriter(fileName, filePath, &fileIndex, &frameIndexEnable,
&overwriteEnable, &detID, &numThreads);
FILE_LOG (logDEBUG) << " Detector type set to " << getDetectorType(d);
return OK;
@ -413,6 +463,7 @@ int UDPStandardImplementation::startReceiver(char *c) {
FILE_LOG(logERROR) << c;
return FAIL;
}
cout << "Listener Ready ..." << endl;
if(fileWriteEnable){
if (SetupWriter() == FAIL) {
@ -421,6 +472,24 @@ int UDPStandardImplementation::startReceiver(char *c) {
return FAIL;
}
}
cout << "Processor Ready ..." << endl;
//callbacks
callbackAction = DO_EVERYTHING;
if (startAcquisitionCallBack) /** file path and file index not required?? or need to include detector index? do they need the datasize? its given for write data anyway */
callbackAction=startAcquisitionCallBack(filePath, fileName, fileIndex,
(generalData->fifoBufferSize) * numberofJobs + (generalData->fifoBufferHeaderSize), pStartAcquisition);
if (callbackAction < DO_EVERYTHING) {
FILE_LOG(logINFO) << "Call back activated. Data saving must be taken care of by user in call back.";
if (rawDataReadyCallBack) {
FILE_LOG(logINFO) << "Data Write has been defined externally";
}
} else if(!fileWriteEnable) {
FILE_LOG(logINFO) << "Data will not be saved";
}
//change status
pthread_mutex_lock(&statusMutex);
@ -432,6 +501,7 @@ int UDPStandardImplementation::startReceiver(char *c) {
FILE_LOG(logINFO) << "Receiver Started";
FILE_LOG(logINFO) << "Status: " << runStatusType(status);
return OK;
}
@ -443,10 +513,41 @@ void UDPStandardImplementation::stopReceiver(){
//set status to transmitting
startReadout();
//wait until status is run_finished
while(status == TRANSMITTING){
//wait for the processes to be done
while(Listener::GetRunningMask()){
usleep(5000);
}
while(DataProcessor::GetRunningMask()){
usleep(5000);
}
pthread_mutex_lock(&statusMutex);
status = RUN_FINISHED;
pthread_mutex_unlock(&(statusMutex));
FILE_LOG(logINFO) << "Status: " << runStatusType(status);
{ //statistics
int tot = 0;
for (int i = 0; i < numThreads; i++) {
tot += dataProcessor[i]->GetNumFramesCaught();
if (dataProcessor[i]->GetNumFramesCaught() < numberOfFrames) {
cprintf(RED, "\nPort %d\n",udpPortNum[i]);
cprintf(RED, "Missing Packets \t: %lld\n",(long long int)numberOfFrames*generalData->packetsPerFrame-listener[i]->GetTotalPacketsCaught());
cprintf(RED, "Frames Caught \t\t: %lld\n",(long long int)dataProcessor[i]->GetNumFramesCaught());
cprintf(RED, "Last Frame Number Caught :%lld\n",(long long int)listener[i]->GetLastFrameIndexCaught());
}else{
cprintf(GREEN, "\nPort %d\n",udpPortNum[i]);
cprintf(GREEN, "Missing Packets \t: %lld\n",(long long int)numberOfFrames*generalData->packetsPerFrame-listener[i]->GetTotalPacketsCaught());
cprintf(GREEN, "Frames Caught \t\t: %lld\n",(long long int)dataProcessor[i]->GetNumFramesCaught());
cprintf(GREEN, "Last Frame Number Caught :%lld\n",(long long int)listener[i]->GetLastFrameIndexCaught());
}
}
//callback
if (acquisitionFinishedCallBack)
acquisitionFinishedCallBack((int)(tot/numThreads), pAcquisitionFinished);
}
//change status
pthread_mutex_lock(&statusMutex);
@ -473,18 +574,11 @@ void UDPStandardImplementation::startReadout(){
for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it)
totalP += (*it)->GetTotalPacketsCaught();
//current udp buffer received
int currentReceivedInBuffer=0,prevReceivedInBuffer=-1;
for (vector<Listener*>::const_iterator it = listener.begin(); it != listener.end(); ++it)
currentReceivedInBuffer += (*it)->GetNumReceivedinUDPBuffer();
//wait for all packets
if((unsigned long long int)totalP!=numberOfFrames*generalData->packetsPerFrame*listener.size()){
//wait as long as there is change from prev totalP,
//and also change from received in buffer to previous value
//(as one listens to many at a time, shouldnt cut off in between)
while((prev != totalP) || (prevReceivedInBuffer!= currentReceivedInBuffer)){
while(prev != totalP){
#ifdef VERY_VERBOSE
cprintf(MAGENTA,"waiting for all packets prevP:%d totalP:%d PrevBuffer:%d currentBuffer:%d\n",prev,totalP,prevReceivedInBuffer,currentReceivedInBuffer);
@ -494,16 +588,11 @@ void UDPStandardImplementation::startReadout(){
prev = totalP;
totalP = 0;
prevReceivedInBuffer = currentReceivedInBuffer;
currentReceivedInBuffer = 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)
totalP += (*it)->GetTotalPacketsCaught();
currentReceivedInBuffer += (*it)->GetNumReceivedinUDPBuffer();
}
#ifdef VERY_VERBOSE
cprintf(MAGENTA,"\tupdated: totalP:%d currently in buffer:%d\n",totalP,currentReceivedInBuffer);
cprintf(MAGENTA,"\tupdated: totalP:%d\n",totalP);
#endif
}
}
@ -644,18 +733,9 @@ void UDPStandardImplementation::ResetParametersforNewMeasurement() {
int UDPStandardImplementation::CreateUDPSockets() {
FILE_LOG (logDEBUG) << __AT__ << " called";
shutDownUDPSockets();
//if eth is mistaken with ip address
if (strchr(eth,'.') != NULL){
strcpy(eth,"");
}
if(!strlen(eth)){
FILE_LOG(logWARNING) << "eth is empty. Listening to all";
}
bool error = false;
for (unsigned int i = 0; i < listener.size(); ++i)
if (listener[i]->CreateUDPSockets((strlen(eth)?eth:NULL)) == FAIL) {
if (listener[i]->CreateUDPSockets() == FAIL) {
error = true;
break;
}
@ -665,7 +745,6 @@ int UDPStandardImplementation::CreateUDPSockets() {
}
FILE_LOG(logDEBUG) << "UDP socket(s) created successfully.";
cout << "Listener Ready ..." << endl;
return OK;
}