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

changed receiver to work with many writer threads for only receiver without compression

Browse Source 
git-svn-id: file:///afs/psi.ch/project/sls_det_software/svn/slsDetectorSoftware@707 951219d9-93cf-4727-9268-0efd64621fa3
This commit is contained in:
l_maliakal_d 2013-12-16 10:07:08 +00:00
parent d2958eb0a4
commit a485e33e82
7 changed files with 1237 additions and 1303 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
slsDetectorSoftware/MySocketTCP/genericSocket.h
View File

@ -541,7 +541,7 @@ enum communicationProtocol{
}; };
int ReceiveDataOnly(void* buf,int length){ int ReceiveDataOnly(void* buf,int length=0){
if (buf==NULL) return -1; if (buf==NULL) return -1;
@ -562,16 +562,26 @@ enum communicationProtocol{
break; break;
case UDP: case UDP:
if (socketDescriptor<0) return -1; if (socketDescriptor<0) return -1;
// while(length>0){ //if length given
for(int i=0;i<packets_per_frame;i++){ if(length){
while(length>0){
nsending=packet_size; nsending=packet_size;
//nsending = (length>packet_size) ? packet_size:length;
nsent = recvfrom(socketDescriptor,(char*)buf+total_sent,nsending, 0, (struct sockaddr *) &clientAddress, &clientAddress_length); nsent = recvfrom(socketDescriptor,(char*)buf+total_sent,nsending, 0, (struct sockaddr *) &clientAddress, &clientAddress_length);
if(!nsent) break; if(!nsent) break;
length-=nsent; length-=nsent;
total_sent+=nsent; total_sent+=nsent;
} }
}
//depends on packets per frame
else{
for(int i=0;i<packets_per_frame;i++){
nsending=packet_size;
nsent = recvfrom(socketDescriptor,(char*)buf+total_sent,nsending, 0, (struct sockaddr *) &clientAddress, &clientAddress_length);
if(!nsent) break;
length-=nsent;
total_sent+=nsent;
}
}
break; break;
default: default:
; ;

114
slsDetectorSoftware/slsDetectorAnalysis/postProcessing.cpp
View File

@ -482,7 +482,7 @@ void* postProcessing::processData(int delflag) {
pthread_mutex_lock(&mg); pthread_mutex_lock(&mg);
int* receiverData = readFrameFromReceiver(currentfName,currentfIndex); int* receiverData = readFrameFromReceiver(currentfName,currentfIndex);
pthread_mutex_unlock(&mg); pthread_mutex_unlock(&mg);
/*cout<<"index:"<<dec<<currentfIndex<<endl<<endl<<endl;;*/
//if detector returned null //if detector returned null
if(setReceiverOnline()==OFFLINE_FLAG) if(setReceiverOnline()==OFFLINE_FLAG)
receiverData = NULL; receiverData = NULL;
@ -521,118 +521,6 @@ void* postProcessing::processData(int delflag) {
} }
} }
} }
/*
int prevCaught=-1;
int caught = 0;
int prog = 0;
bool newData=false;
char currentfName[MAX_STR_LENGTH]="";
int currentfIndex=0;
int read_freq = setReadReceiverFrequency(0);
#ifdef VERBOSE
std::cout << "receiver read freq:" << read_freq << std::endl;
#endif
//always read nth data
if (read_freq != 0){
newData = true;
if (!dataReady){
std::cout << "Error: receiver read freq should be > 0 only when using gui." << std::endl;
std::cout << "Current receiver read frequency: " << read_freq << std::endl;
}
}
while(1){
cout.flush();
cout<<flush;
usleep(20000);
//get progress
pthread_mutex_lock(&mg);
if(setReceiverOnline()==ONLINE_FLAG)
prog=getReceiverCurrentFrameIndex();//getFramesCaughtByReceiver();//caught=getReceiverCurrentFrameIndex();
pthread_mutex_unlock(&mg);
if(setReceiverOnline()==OFFLINE_FLAG)
prog=prevCaught;
if(prevCaught == -1)
setCurrentProgress(0);
else
setCurrentProgress(prog);
if (checkJoinThread()) break;
if (dataReady){
// new Data? for random read
if (!read_freq){
caught = prog;
if (caught > prevCaught)
newData=true;
else
newData=false;
#ifdef VERBOSE
std::cout << "caught:" << caught << " prevcaught:" << prevCaught << " newData:" << newData << std::endl;
#endif
prevCaught=caught;
}
//read frame if new data or nth frame reading
if (newData){
if(setReceiverOnline()==ONLINE_FLAG){
//get data
strcpy(currentfName,"");
pthread_mutex_lock(&mg);
int* receiverData = readFrameFromReceiver(currentfName,currentfIndex);//if(currentfIndex!=-1)cout<<"--currentfIndex:"<<currentfIndex<<endl;
pthread_mutex_unlock(&mg);
//if detector returned null
if(setReceiverOnline()==OFFLINE_FLAG)
receiverData = NULL;
if(receiverData == NULL){
currentfIndex = -1;
cout<<"****Detector Data returned is NULL***"<<endl;
}
// determine if new Data for nth frame read
if (read_freq){
caught = currentfIndex;
#ifdef VERBOSE
std::cout << "caught:" << caught << " prevcaught:" << prevCaught << std::endl;
#endif
//delete if not new data
if((caught == prevCaught) || (caught == -1))
currentfIndex = -1;
else
prevCaught=caught;
}
//not garbage frame
if (currentfIndex >= 0) {
fdata = decodeData(receiverData);
delete [] receiverData;
if ((fdata) && (dataReady)){
thisData = new detectorData(fdata,NULL,NULL,getCurrentProgress(),currentfName,getTotalNumberOfChannels());
dataReady(thisData, currentfIndex, pCallbackArg);
delete thisData;
fdata = NULL;
}
}
else{
;//cout<<"****Detector returned mismatched indices/garbage or acquisition is over. Trying again.***"<<endl;
}
}
}
}
}*/
} }
return 0; return 0;

12
slsDetectorSoftware/slsDetectorAnalysis/singlePhotonFilter.cpp
View File

@ -331,9 +331,9 @@ int singlePhotonFilter::verifyFrame(char *inData){
//for moench, put first packet last //for moench, put first packet last
if (pIndex == 0) if (pIndex == 0)
pIndex = packets_per_frame; pIndex = packets_per_frame;
#ifdef VERYVERBOSE //#ifdef VERYVERBOSE
cout<<"fi:"<<fIndex<< " pi:"<< pIndex << endl; cout<<"fi:"<<fIndex<< " pi:"<< pIndex << endl;
#endif //#endif
//firsttime //firsttime
if (firstTime){ if (firstTime){
firstTime = false; firstTime = false;
@ -345,8 +345,8 @@ int singlePhotonFilter::verifyFrame(char *inData){
//if it is not matching withthe frame number //if it is not matching withthe frame number
if (fIndex != fnum){ if (fIndex != fnum){
/*cout << "**Frame number doesnt match:Missing Packet! " << fnum << " " cout << "**Frame number doesnt match:Missing Packet! " << fnum << " "
"Expected f " << fnum << " p " << pnum + 1 << " received f " << fIndex << " p " << pIndex << endl;*/ "Expected f " << fnum << " p " << pnum + 1 << " received f " << fIndex << " p " << pIndex << endl;
if (ptot == 0) { if (ptot == 0) {
if (pIndex == 1)//so that its not moved to next line. if (pIndex == 1)//so that its not moved to next line.
@ -368,8 +368,8 @@ int singlePhotonFilter::verifyFrame(char *inData){
//if missing a packet, discard //if missing a packet, discard
else if (pIndex != pnum + 1){/**else */ else if (pIndex != pnum + 1){/**else */
/*cout << "**packet number doesnt match:Missing Packet! " << fnum << " " cout << "**packet number doesnt match:Missing Packet! " << fnum << " "
"Expected f" << fnum << " p " << pnum + 1 << " received f " << fnum << " p " << pIndex << endl;*/ "Expected f" << fnum << " p " << pnum + 1 << " received f " << fnum << " p " << pIndex << endl;
pnum = pIndex; pnum = pIndex;
ptot++; ptot++;
} }

1
slsDetectorSoftware/slsReceiver/receiver_defs.h
View File

@ -14,6 +14,7 @@
#define BUF_SIZE (16*1024*1024) //16mb #define BUF_SIZE (16*1024*1024) //16mb
#define SAMPLE_TIME_IN_NS 100000000//100ms #define SAMPLE_TIME_IN_NS 100000000//100ms
#define MAX_JOBS_PER_THREAD 1000 #define MAX_JOBS_PER_THREAD 1000
#define HEADER_SIZE_NUM_TOT_PACKETS 2
#define HEADER_SIZE_NUM_FRAMES 2 #define HEADER_SIZE_NUM_FRAMES 2
#define HEADER_SIZE_NUM_PACKETS 1 #define HEADER_SIZE_NUM_PACKETS 1

1971
slsDetectorSoftware/slsReceiver/slsReceiverFunctionList.cpp
View File

File diff suppressed because it is too large Load Diff

350
slsDetectorSoftware/slsReceiver/slsReceiverFunctionList.h
View File

@ -71,12 +71,12 @@ public:
/** /**
* Returns Frames Caught for each real time acquisition (eg. for each scan) * Returns Frames Caught for each real time acquisition (eg. for each scan)
*/ */
int getFramesCaught(){return framesCaught;}; int getFramesCaught(){return (packetsCaught/packetsPerFrame);};
/** /**
* Returns Total Frames Caught for an entire acquisition (including all scans) * Returns Total Frames Caught for an entire acquisition (including all scans)
*/ */
int getTotalFramesCaught(){return totalFramesCaught;}; int getTotalFramesCaught(){return (totalPacketsCaught/packetsPerFrame);};
/** /**
* Returns the frame index at start of each real time acquisition (eg. for each scan) * Returns the frame index at start of each real time acquisition (eg. for each scan)
@ -143,6 +143,52 @@ public:
*/ */
void resetTotalFramesCaught(); void resetTotalFramesCaught();
/**
* Set short frame
* @param i if shortframe i=1
*/
int setShortFrame(int i);
/**
* Set the variable to send every nth frame to gui
* or if 0,send frame only upon gui request
*/
int setNFrameToGui(int i);
/** set acquisition period if a positive number
*/
int64_t setAcquisitionPeriod(int64_t index);
/** enabl data compression, by saving only hits
*/
void enableDataCompression(bool enable);
/** get data compression, by saving only hits
*/
bool getDataCompression(){ return dataCompression;};
/** set status to transmitting and
* when fifo is empty later, sets status to run_finished
*/
void startReadout();
/**
* Returns the buffer-current frame read by receiver
* @param c pointer to current file name
* @param raw address of pointer, pointing to current frame to send to gui
*/
void readFrame(char* c,char** raw);
/** free fifo buffer, called back from single photon filter
*/
static void freeFifoBufferCallBack (char* fbuffer, void *this_pointer);
/**
* Call back from single photon filter to free writingfifo
* called from freeFifoBufferCallBack
*/
void freeFifoBuffer(char* fbuffer);
/** /**
* Starts Receiver - starts to listen for packets * Starts Receiver - starts to listen for packets
@ -157,50 +203,48 @@ public:
*/ */
int stopReceiver(); int stopReceiver();
/**
* Returns the buffer-current frame read by receiver
* @param c pointer to current file name
* @param raw address of pointer, pointing to current frame to send to gui
*/
void readFrame(char* c,char** raw);
/**
* Set short frame
* @param i if shortframe i=1
*/
int setShortFrame(int i);
/** set status to transmitting and
* when fifo is empty later, sets status to run_finished
*/
void startReadout();
/** enabl data compression, by saving only hits
*/
void enableDataCompression(bool enable){dataCompression = enable;if(filter)filter->enableCompression(enable);};
/** get data compression, by saving only hits
*/
bool getDataCompression(){ return dataCompression;};
/**
* Set the variable to send every nth frame to gui
* or if 0,send frame only upon gui request
*/
int setNFrameToGui(int i);
/** set acquisition period if a positive number
*/
int64_t setAcquisitionPeriod(int64_t index);
/** free fifo buffer, called back from single photon filter
*/
static void freeFifoBufferCallBack (char* fbuffer, void *this_pointer){((slsReceiverFunctionList*)this_pointer)->freeFifoBuffer(fbuffer);};
void freeFifoBuffer(char* fbuffer){fifofree->push(fbuffer);};
private: private:
/**
* Constructs the singlePhotonFilter object
*/
void setupFilter();
/**
* Copy frames to gui
* uses semaphore for nth frame mode
*/
void copyFrameToGui(char* startbuf);
/**
* set up fifo according to the new numjobsperthread
*/
void setupFifoStructure ();
/**
* creates udp socket
* \returns if success or fail
*/
int createUDPSocket();
/**
* create listening thread and many writer threads at class construction
* @param destroy is true to kill all threads and start again
*/
int createThreads(bool destroy = false);
/**
* initializes variables and creates the first file
* also does the startAcquisitionCallBack
* \returns FAIL or OK
*/
int setupWriter();
/**
* Creates new file
*\returns OK for succces or FAIL for failure
*/
int createNewFile();
/** /**
* Static function - Thread started which listens to packets. * Static function - Thread started which listens to packets.
@ -209,13 +253,6 @@ private:
*/ */
static void* startListeningThread(void *this_pointer); static void* startListeningThread(void *this_pointer);
/**
* Thread started which listens to packets.
* Called by startReceiver()
*
*/
int startListening();
/** /**
* Static function - Thread started which writes packets to file. * Static function - Thread started which writes packets to file.
* Called by startReceiver() * Called by startReceiver()
@ -223,6 +260,13 @@ private:
*/ */
static void* startWritingThread(void *this_pointer); static void* startWritingThread(void *this_pointer);
/**
* Thread started which listens to packets.
* Called by startReceiver()
*
*/
int startListening();
/** /**
* Thread started which writes packets to file. * Thread started which writes packets to file.
* Called by startReceiver() * Called by startReceiver()
@ -230,34 +274,38 @@ private:
*/ */
int startWriting(); int startWriting();
/**
* Creates new file
*\returns OK for succces or FAIL for failure
*/
int createNewFile();
/** /**
* Copy frames to gui * Writing to file without compression
* uses semaphore for nth frame mode * @param buf is the address of buffer popped out of fifo
* @param num
*/ */
void copyFrameToGui(char* startbuf); void writeToFile_withoutCompression(char* buf,int numpackets);
/** set up fifo according to the new numjobsperthread
*/
void setupFifoStructure ();
/**
* increment counters, pop and push fifos
*/
void processFrameForFifo();
/** detector type */ /** detector type */
detectorType myDetectorType; detectorType myDetectorType;
/** max frames per file **/ /** status of receiver */
int maxFramesPerFile; runStatus status;
/** UDP Socket between Receiver and Detector */
genericSocket* udpSocket;
/** Server UDP Port*/
int server_port;
/** ethernet interface or IP to listen to */
char *eth;
/** max packets per file **/
int maxPacketsPerFile;
/** File write enable */ /** File write enable */
int enableFileWrite; int enableFileWrite;
@ -277,9 +325,6 @@ private:
/** if frame index required in file name */ /** if frame index required in file name */
int frameIndexNeeded; int frameIndexNeeded;
/** Frames Caught for each real time acquisition (eg. for each scan) */
int framesCaught;
/* Acquisition started */ /* Acquisition started */
bool acqStarted; bool acqStarted;
@ -292,14 +337,14 @@ private:
/** Actual current frame index of each time acquisition (eg. for each scan) */ /** Actual current frame index of each time acquisition (eg. for each scan) */
uint32_t frameIndex; uint32_t frameIndex;
/** Total Frames Caught for an entire acquisition (including all scans) */ /** Frames Caught for each real time acquisition (eg. for each scan) */
int totalFramesCaught; int packetsCaught;
/** Total packets caught for an entire acquisition (including all scans) */ /** Total packets caught for an entire acquisition (including all scans) */
int totalPacketsCaught; int totalPacketsCaught;
/** Frames currently in current file, starts new file when it reaches max */ /** Pckets currently in current file, starts new file when it reaches max */
int framesInFile; int packetsInFile;
/** Frame index at start of an entire acquisition (including all scans) */ /** Frame index at start of an entire acquisition (including all scans) */
uint32_t startAcquisitionIndex; uint32_t startAcquisitionIndex;
@ -307,68 +352,38 @@ private:
/** Actual current frame index of an entire acquisition (including all scans) */ /** Actual current frame index of an entire acquisition (including all scans) */
uint32_t acquisitionIndex; uint32_t acquisitionIndex;
/** Previous Frame number from buffer */ /** number of packets per frame*/
uint32_t prevframenum; int packetsPerFrame;
/** thread listening to packets */ /** frame index mask */
pthread_t listening_thread; uint32_t frameIndexMask;
/** thread writing packets */ /** packet index mask */
pthread_t writing_thread; uint32_t packetIndexMask;
/** mutex for locking variable used by different threads */ /** frame index offset */
pthread_mutex_t status_mutex; int frameIndexOffset;
/** listening thread running */ /** acquisition period */
int listening_thread_running; int64_t acquisitionPeriod;
/** writing thread running */
int writing_thread_running;
/** status of receiver */
runStatus status;
/** Receiver buffer */
char* buffer;
/** Receiver buffer */
char *mem0, *memfull;
/** latest data */
char* latestData;
/** UDP Socket between Receiver and Detector */
genericSocket* udpSocket;
/** Server UDP Port*/
int server_port;
/** ethernet interface or IP to listen to */
char *eth;
/** Element structure to put inside a fifo */
struct dataStruct {
char* buffer;
int rc;
};
/** circular fifo to read and write data*/
CircularFifo<char>* fifo;
/** circular fifo to read and write data*/
CircularFifo<char>* fifofree;
/** fifo size */
unsigned int fifosize;
/** short frames */ /** short frames */
int shortFrame; int shortFrame;
/** current frame number */
uint32_t currframenum;
/** Previous Frame number from buffer */
uint32_t prevframenum;
/** buffer size can be 1286*2 or 518 or 1286*40 */ /** buffer size can be 1286*2 or 518 or 1286*40 */
int bufferSize; int bufferSize;
/** number of packets per frame*/ /** oen buffer size */
int packetsPerFrame; int onePacketSize;
/** latest data */
char* latestData;
/** gui data ready */ /** gui data ready */
int guiDataReady; int guiDataReady;
@ -379,17 +394,17 @@ private:
/** points to the filename to send to gui */ /** points to the filename to send to gui */
char* guiFileName; char* guiFileName;
/** current frame number */
uint32_t currframenum;
/** send every nth frame to gui or only upon gui request*/ /** send every nth frame to gui or only upon gui request*/
int nFrameToGui; int nFrameToGui;
/** frame index mask */ /** fifo size */
int frameIndexMask; unsigned int fifosize;
/** frame index offset */ /** number of jobs per thread for data compression */
int frameIndexOffset; int numJobsPerThread;
/** memory allocated for the buffer */
char *mem0;
/** datacompression - save only hits */ /** datacompression - save only hits */
bool dataCompression; bool dataCompression;
@ -397,35 +412,66 @@ private:
/** single photon filter */ /** single photon filter */
singlePhotonFilter *filter; singlePhotonFilter *filter;
/** oen buffer size */ /** circular fifo to store addresses of data read */
int oneBufferSize; CircularFifo<char>* fifo;
/** semaphore to synchronize writer and guireader threads */ /** circular fifo to store addresses of data already written and ready to be resued*/
sem_t smp; CircularFifo<char>* fifoFree;
/** guiDataReady mutex */ /** Receiver buffer */
pthread_mutex_t dataReadyMutex; char *buffer;
/** number of jobs per thread for data compression */ /** max number of writer threads */
int numJobsPerThread; const static int MAX_NUM_WRITER_THREADS = 15;
/** offset of current frame */ /** number of writer threads */
int currentFrameOffset; int numWriterThreads;
/** offset of current packet */ /** to know if listening and writer threads created properly */
int currentPacketOffset; int thread_started;
/** current packet count for current frame */ /** mask showing which threads are running */
int currentPacketCount; volatile int32_t writerthreads_mask;
/** current frame count for current buffer */ /** current writer thread index*/
int currentFrameCount; int currentWriterThreadIndex;
/** thread listening to packets */
pthread_t listening_thread;
/** thread writing packets */
pthread_t writing_thread[MAX_NUM_WRITER_THREADS];
/** total frame count the listening thread has listened to */ /** total frame count the listening thread has listened to */
int totalListeningFrameCount; int totalListeningFrameCount;
/** acquisition period */ /** 0 if receiver is idle, 1 otherwise */
int64_t acquisitionPeriod; int running;
//semaphores
/** semaphore to synchronize writer and guireader threads */
sem_t smp;
/** semaphore to synchronize listener thread */
sem_t listensmp;
/** semaphore to synchronize writer threads */
sem_t writersmp[MAX_NUM_WRITER_THREADS];
//mutex
/** guiDataReady mutex */
pthread_mutex_t dataReadyMutex;
/** mutex for status */
pthread_mutex_t status_mutex;
/** mutex for progress variable currframenum */
pthread_mutex_t progress_mutex;
/** mutex for writing data to file */
pthread_mutex_t write_mutex;
/** /**
callback arguments are callback arguments are
@ -490,14 +536,12 @@ public:
0 callback takes care of open,close,wrie file 0 callback takes care of open,close,wrie file
1 callback writes file, we have to open, close it 1 callback writes file, we have to open, close it
2 we open, close, write file, callback does not do anything 2 we open, close, write file, callback does not do anything
*/ */
void registerCallBackStartAcquisition(int (*func)(char*, char*,int, int, void*),void *arg){startAcquisitionCallBack=func; pStartAcquisition=arg;}; void registerCallBackStartAcquisition(int (*func)(char*, char*,int, int, void*),void *arg){startAcquisitionCallBack=func; pStartAcquisition=arg;};
/** /**
callback argument is callback argument is
toatal frames caught toatal frames caught
*/ */
void registerCallBackAcquisitionFinished(void (*func)(int, void*),void *arg){acquisitionFinishedCallBack=func; pAcquisitionFinished=arg;}; void registerCallBackAcquisitionFinished(void (*func)(int, void*),void *arg){acquisitionFinishedCallBack=func; pAcquisitionFinished=arg;};

6
slsDetectorSoftware/slsReceiver/slsReceiver_funcs.cpp
View File

@ -733,7 +733,7 @@ int slsReceiverFuncs::start_receiver(){
ret = FAIL; ret = FAIL;
} }
*/ */
else if(slsReceiverList->getStatus()!=RUNNING) else if(slsReceiverList->getStatus()==IDLE)
ret=slsReceiverList->startReceiver(mess); ret=slsReceiverList->startReceiver(mess);
#endif #endif
@ -1183,10 +1183,10 @@ int slsReceiverFuncs::gotthard_read_frame(){
cout << "index:" << hex << index << endl; cout << "index:" << hex << index << endl;
#endif #endif
}else{ }else{
bindex = (uint32_t)(*((uint32_t*)raw)); bindex = ((uint32_t)(*((uint32_t*)raw)))+1;
pindex = (bindex & GOTTHARD_PACKET_INDEX_MASK); pindex = (bindex & GOTTHARD_PACKET_INDEX_MASK);
index = ((bindex & GOTTHARD_FRAME_INDEX_MASK) >> GOTTHARD_FRAME_INDEX_OFFSET); index = ((bindex & GOTTHARD_FRAME_INDEX_MASK) >> GOTTHARD_FRAME_INDEX_OFFSET);
bindex2 = (uint32_t)(*((uint32_t*)((char*)(raw+onebuffersize)))); bindex2 = ((uint32_t)(*((uint32_t*)((char*)(raw+onebuffersize)))))+1;
pindex2 =(bindex2 & GOTTHARD_PACKET_INDEX_MASK); pindex2 =(bindex2 & GOTTHARD_PACKET_INDEX_MASK);
index2 =((bindex2 & GOTTHARD_FRAME_INDEX_MASK) >> GOTTHARD_FRAME_INDEX_OFFSET); index2 =((bindex2 & GOTTHARD_FRAME_INDEX_MASK) >> GOTTHARD_FRAME_INDEX_OFFSET);
#ifdef VERBOSE #ifdef VERBOSE