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

eiger gui works and can read frames, other gui functionalities not implemented

Browse Source
This commit is contained in:
Maliakal Dhanya
2014-06-18 16:35:34 +02:00
parent 3a71049f1c
commit 5f82381b1e
4 changed files with 449 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files

130
slsReceiverSoftware/slsReceiver/slsReceiverTCPIPInterface.cpp
View File

@ -1012,6 +1012,8 @@ int slsReceiverTCPIPInterface::read_frame(){
switch(myDetectorType){ switch(myDetectorType){
case MOENCH: case MOENCH:
return moench_read_frame(); return moench_read_frame();
case EIGER:
return eiger_read_frame();
default: default:
return gotthard_read_frame(); return gotthard_read_frame();
} }
@ -1036,7 +1038,7 @@ int slsReceiverTCPIPInterface::moench_read_frame(){
char* raw = new char[bufferSize]; char* raw = new char[bufferSize];
uint32_t startIndex=0; uint32_t startIndex=0;
int index = 0,bindex = 0, offset=0; uint32_t index = 0,bindex = 0, offset=0;
strcpy(mess,"Could not read frame\n"); strcpy(mess,"Could not read frame\n");
@ -1053,7 +1055,7 @@ int slsReceiverTCPIPInterface::moench_read_frame(){
else{ else{
ret = OK; ret = OK;
startIndex=slsReceiverFunctions->getStartFrameIndex(); startIndex=slsReceiverFunctions->getStartFrameIndex();
slsReceiverFunctions->readFrame(fName,&raw); slsReceiverFunctions->readFrame(fName,&raw,index);
/**send garbage with -1 index to try again*/ /**send garbage with -1 index to try again*/
if (raw == NULL){ if (raw == NULL){
@ -1223,7 +1225,7 @@ int slsReceiverTCPIPInterface::gotthard_read_frame(){
}else{ }else{
ret = OK; ret = OK;
startIndex=slsReceiverFunctions->getStartFrameIndex(); startIndex=slsReceiverFunctions->getStartFrameIndex();
slsReceiverFunctions->readFrame(fName,&raw); slsReceiverFunctions->readFrame(fName,&raw,index);
/**send garbage with -1 index to try again*/ /**send garbage with -1 index to try again*/
if (raw == NULL){ if (raw == NULL){
@ -1334,6 +1336,128 @@ int slsReceiverTCPIPInterface::gotthard_read_frame(){
int slsReceiverTCPIPInterface::eiger_read_frame(){
ret=OK;
char fName[MAX_STR_LENGTH]="";
int arg = -1,i;
uint32_t index=0;
int bufferSize = EIGER_BUFFER_SIZE;
int onebuffersize = EIGER_BUFFER_SIZE/EIGER_PACKETS_PER_FRAME;
int onedatasize = EIGER_DATA_BYTES/EIGER_PACKETS_PER_FRAME;
int numpackets = EIGER_PACKETS_PER_FRAME;
char* raw = new char[bufferSize];
char* origVal = new char[bufferSize];
char* retval = new char[EIGER_DATA_BYTES];
/*
//retval is a full frame
int rnel = bufferSize/(sizeof(int));
int* retval = new int[rnel];
int* origVal = new int[rnel];
//all initialized to 0
for(i=0;i<rnel;i++) retval[i]=0;
for(i=0;i<rnel;i++) origVal[i]=0;
*/
strcpy(mess,"Could not read frame\n");
/*
typedef struct
{
unsigned char num1[4];
unsigned char num2[4];
} eiger_packet_header;
*/
// execute action if the arguments correctly arrived
#ifdef SLS_RECEIVER_UDP_FUNCTIONS
/**send garbage with -1 index to try again*/
if(!slsReceiverFunctions->getFramesCaught()){
arg=-1;
cout<<"haven't caught any frame yet"<<endl;
}
/** acq started */
else{
ret = OK;
/** read a frame */
slsReceiverFunctions->readFrame(fName,&raw, index);
#ifdef VERBOSE
cout << "index:" << dec << index << endl;
#endif
/**send garbage with -1 index to try again*/
if (raw == NULL){
arg = -1;
#ifdef VERBOSE
cout<<"data not ready for gui yet"<<endl;
#endif
}
/**proper frame*/
else{
memcpy(origVal,raw,bufferSize);
raw=NULL;
int copyindex=8;
int retindex=0;
for(int i=0;i<numpackets;++i){
//cout<<"p"<<i<<":"<<(htonl(*(uint32_t*)((eiger_packet_header *)((uint32_t*)(origVal[i] + HEADER_SIZE_NUM_TOT_PACKETS)))->num2)&0xff)<<"\t\t";
memcpy(retval+retindex ,origVal+copyindex ,onedatasize);
copyindex += 16+onedatasize;
retindex += onedatasize;
}
arg = index-1;
}
}
#ifdef VERBOSE
if(arg!=-1){
cout << "fName:" << fName << endl;
cout << "findex:" << arg << endl;
}
#endif
#endif
if(ret==OK && socket->differentClients){
cout << "Force update" << endl;
ret=FORCE_UPDATE;
}
// send answer
socket->SendDataOnly(&ret,sizeof(ret));
if(ret==FAIL){
cout << "mess:" << mess << endl;
socket->SendDataOnly(mess,sizeof(mess));
}
else{
socket->SendDataOnly(fName,MAX_STR_LENGTH);
socket->SendDataOnly(&arg,sizeof(arg));
socket->SendDataOnly(retval,EIGER_DATA_BYTES);
}
delete [] retval;
delete [] origVal;
delete [] raw;
return ret;
}
int slsReceiverTCPIPInterface::set_read_frequency(){ int slsReceiverTCPIPInterface::set_read_frequency(){
ret=OK; ret=OK;
int retval=-1; int retval=-1;

3
slsReceiverSoftware/slsReceiver/slsReceiverTCPIPInterface.h
View File

@ -162,6 +162,9 @@ private:
/** moench specific read frame */ /** moench specific read frame */
int moench_read_frame(); int moench_read_frame();
/** eiger specific read frame */
int eiger_read_frame();
/** Sets the receiver to send every nth frame to gui, or only upon gui request */ /** Sets the receiver to send every nth frame to gui, or only upon gui request */
int set_read_frequency(); int set_read_frequency();

489
slsReceiverSoftware/slsReceiver/slsReceiverUDPFunctions.cpp
View File

@ -1,4 +1,4 @@
/*#ifdef SLS_RECEIVER_UDP_FUNCTIONS*/ #ifdef SLS_RECEIVER_UDP_FUNCTIONS
/********************************************//** /********************************************//**
* @file slsReceiverUDPFunctions.cpp * @file slsReceiverUDPFunctions.cpp
* @short does all the functions for a receiver, set/get parameters, start/stop etc. * @short does all the functions for a receiver, set/get parameters, start/stop etc.
@ -34,7 +34,8 @@ slsReceiverUDPFunctions::slsReceiverUDPFunctions():
thread_started(0), thread_started(0),
eth(NULL), eth(NULL),
latestData(NULL), latestData(NULL),
guiFileName(NULL){ guiFileName(NULL),
guiFrameNumber(0){
for(int i=0;i<MAX_NUM_LISTENING_THREADS;i++){ for(int i=0;i<MAX_NUM_LISTENING_THREADS;i++){
udpSocket[i] = NULL; udpSocket[i] = NULL;
server_port[i] = DEFAULT_UDP_PORTNO+i; server_port[i] = DEFAULT_UDP_PORTNO+i;
@ -178,6 +179,7 @@ void slsReceiverUDPFunctions::initializeMembers(){
latestData = NULL; latestData = NULL;
guiFileName = NULL; guiFileName = NULL;
guiData = NULL; guiData = NULL;
guiFrameNumber = 0;
sfilefd = NULL; sfilefd = NULL;
cmSub = NULL; cmSub = NULL;
@ -788,13 +790,15 @@ void slsReceiverUDPFunctions::setupFifoStructure(){
/** acquisition functions */ /** acquisition functions */
void slsReceiverUDPFunctions::readFrame(char* c,char** raw){ void slsReceiverUDPFunctions::readFrame(char* c,char** raw, uint32_t &fnum){
//point to gui data //point to gui data
if (guiData == NULL) if (guiData == NULL)
guiData = latestData; guiData = latestData;
//copy data and filename //copy data and filename
strcpy(c,guiFileName); strcpy(c,guiFileName);
fnum = guiFrameNumber;
//could not get gui data //could not get gui data
if(!guiDataReady){ if(!guiDataReady){
@ -819,7 +823,8 @@ void slsReceiverUDPFunctions::readFrame(char* c,char** raw){
void slsReceiverUDPFunctions::copyFrameToGui(char* startbuf){ void slsReceiverUDPFunctions::copyFrameToGui(char* startbuf[], uint32_t fnum, char* buf){
//random read when gui not ready //random read when gui not ready
if((!nFrameToGui) && (!guiData)){ if((!nFrameToGui) && (!guiData)){
pthread_mutex_lock(&dataReadyMutex); pthread_mutex_lock(&dataReadyMutex);
@ -835,9 +840,17 @@ void slsReceiverUDPFunctions::copyFrameToGui(char* startbuf){
pthread_mutex_lock(&dataReadyMutex); pthread_mutex_lock(&dataReadyMutex);
guiDataReady=0; guiDataReady=0;
//send the first one //eiger
for(int i=0;i<numListeningThreads;i++) if(startbuf != NULL){
memcpy(latestData,(void*)(startbuf + HEADER_SIZE_NUM_TOT_PACKETS + i *(bufferSize+ HEADER_SIZE_NUM_TOT_PACKETS)),bufferSize); for(int i=numListeningThreads;i<packetsPerFrame+numListeningThreads;++i){
memcpy((((char*)latestData)+((i-numListeningThreads)*bufferSize)) ,startbuf[i] + HEADER_SIZE_NUM_TOT_PACKETS,bufferSize);
//cout<<"p"<<i<<":"<<(htonl(*(uint32_t*)((eiger_packet_header *)((uint32_t*)(startbuf[i] + HEADER_SIZE_NUM_TOT_PACKETS)))->num2)&0xff)<<"\t\t";
}
guiFrameNumber = fnum;
}else//other detectors
memcpy(latestData,buf,bufferSize);
strcpy(guiFileName,savefilename); strcpy(guiFileName,savefilename);
guiDataReady=1; guiDataReady=1;
pthread_mutex_unlock(&dataReadyMutex); pthread_mutex_unlock(&dataReadyMutex);
@ -1071,6 +1084,7 @@ int slsReceiverUDPFunctions::setupWriter(){
guiData = NULL; guiData = NULL;
guiDataReady=0; guiDataReady=0;
strcpy(guiFileName,""); strcpy(guiFileName,"");
guiFrameNumber = 0;
cbAction = DO_EVERYTHING; cbAction = DO_EVERYTHING;
pthread_mutex_lock(&status_mutex); pthread_mutex_lock(&status_mutex);
@ -1402,6 +1416,8 @@ void slsReceiverUDPFunctions::startReadout(){
//wait so that all packets which take time has arrived //wait so that all packets which take time has arrived
usleep(50000); usleep(50000);
/********************************************/
usleep(1000000);
pthread_mutex_lock(&status_mutex); pthread_mutex_lock(&status_mutex);
status = TRANSMITTING; status = TRANSMITTING;
pthread_mutex_unlock(&status_mutex); pthread_mutex_unlock(&status_mutex);
@ -1528,7 +1544,7 @@ int slsReceiverUDPFunctions::startListening(){
//start indices for each start of scan/acquisition - eiger does it before //start indices for each start of scan/acquisition - eiger does it before
if((!measurementStarted) && (rc > 0)) if((!measurementStarted) && (rc > 0) && (!ithread))
startFrameIndices(ithread); startFrameIndices(ithread);
//problem in receiving or end of acquisition //problem in receiving or end of acquisition
@ -1620,11 +1636,11 @@ int slsReceiverUDPFunctions::startListening(){
(*((uint16_t*)(buffer[ithread]))) = packetcount; (*((uint16_t*)(buffer[ithread]))) = packetcount;
totalListeningFrameCount[ithread] += packetcount; totalListeningFrameCount[ithread] += packetcount;
#ifdef VERYDEBUG #ifdef VERYDEBUG
if(!ithread) cout << "totalListeningFrameCount[" << ithread << "]:" << totalListeningFrameCount[ithread] << endl; cout<<dec<<ithread<<" listener going to push fifo:"<<(void*)(buffer[ithread])<<endl;
#endif #endif
while(!fifo[ithread]->push(buffer[ithread])); while(!fifo[ithread]->push(buffer[ithread]));
#ifdef VERYDEBUG #ifdef VERYDEBUG
cout << "*** pushed into listening fifo" << endl; if(!ithread) cout << ithread << " *** pushed into listening fifo" << endl;
#endif #endif
} }
@ -1652,8 +1668,6 @@ int slsReceiverUDPFunctions::startListening(){
int slsReceiverUDPFunctions::startWriting(){ int slsReceiverUDPFunctions::startWriting(){
int ithread = currentWriterThreadIndex; int ithread = currentWriterThreadIndex;
#ifdef VERYVERBOSE #ifdef VERYVERBOSE
@ -1664,16 +1678,17 @@ int slsReceiverUDPFunctions::startWriting(){
int numpackets, nf; int numpackets, nf;
uint32_t tempframenum; uint32_t tempframenum;
char* wbuf[MAX_NUM_LISTENING_THREADS]; char* wbuf[packetsPerFrame+numListeningThreads];//interleaved + 2 ports header
char *data=new char[bufferSize]; char *d=new char[bufferSize];
int iFrame = 0;
int xmax=0,ymax=0; int xmax=0,ymax=0;
int ret,i; int ret,i,j;
int w_index=0;
int packetsPerThread = packetsPerFrame/numListeningThreads;
while(1){ while(1){
nf = 0; nf = 0;
iFrame = 0; w_index = 0;
if(myDetectorType == MOENCH){ if(myDetectorType == MOENCH){
xmax = MOENCH_PIXELS_IN_ONE_ROW-1; xmax = MOENCH_PIXELS_IN_ONE_ROW-1;
ymax = MOENCH_PIXELS_IN_ONE_ROW-1; ymax = MOENCH_PIXELS_IN_ONE_ROW-1;
@ -1688,92 +1703,48 @@ int slsReceiverUDPFunctions::startWriting(){
} }
while((1<<ithread)&writerthreads_mask){ while((1<<ithread)&writerthreads_mask){
#ifdef VERYDEBUG #ifdef VERYDEBUG
cout << ithread << " ***waiting to pop out of writing fifo" << endl; cout << ithread << " ***waiting to pop out of writing fifo" << endl;
#endif #endif
//pop //pop
for(i=0;i<numListeningThreads;i++) for(i=0;i<numListeningThreads;++i){
fifo[i]->pop(wbuf[i]); fifo[i]->pop(wbuf[w_index+i]);
numpackets = (uint16_t)(*((uint16_t*)wbuf[0])); numpackets = (uint16_t)(*((uint16_t*)wbuf[w_index]));
#ifdef VERYDEBUG #ifdef VERYDEBUG
cout << ithread << " numpackets:" << dec << numpackets << endl; if((!w_index)||(!numpackets)) cout << ithread << " numpackets:" << dec << numpackets << endl;
cout << ithread << " *** popped from fifo " << numpackets << endl;
#endif #endif
}
#ifdef VERYDEBUG
cout << ithread << " numpackets:" << dec << numpackets << endl;
cout << ithread << " *** writer popped from fifo " << (void*) wbuf[w_index]<< endl;
cout << ithread << " *** writer popped from fifo " << (void*) wbuf[w_index+1]<< endl;
#endif
//last dummy packet //last dummy packet
if(numpackets == 0xFFFF){ if(numpackets == 0xFFFF){
#ifdef VERYDEBUG stopWriting(ithread,wbuf,w_index);
cout << ithread << " **********************popped last dummy frame:" << (void*)wbuf[0] << endl;
#endif
//free fifo
for(i=0;i<numListeningThreads;i++)
while(!fifoFree[i]->push(wbuf[i]));
#ifdef VERYDEBUG
cout << ithread << " fifo freed:" << (void*)wbuf[i] << endl;
#endif
//all threads need to close file, reset mask and exit loop
closeFile(ithread);
pthread_mutex_lock(&status_mutex);
writerthreads_mask^=(1<<ithread);
#ifdef VERYDEBUG
cout << ithread << " Resetting mask of current writing thread. New Mask: " << writerthreads_mask << endl;
#endif
pthread_mutex_unlock(&status_mutex);
//only thread 0 needs to do this
//check if all jobs are done and wait
//change status to run finished
if(ithread == 0){
if(dataCompression){
cout << "Waiting for jobs to be done.. current mask:" << hex << writerthreads_mask << endl;
while(writerthreads_mask){
/*cout << "." << flush;*/
usleep(50000);
}
cout<<" Jobs Done!"<<endl;
}
//to make sure listening threads are done before you update status, as that returns to client
while(listeningthreads_mask)
usleep(5000);
//update status
pthread_mutex_lock(&status_mutex);
status = RUN_FINISHED;
pthread_mutex_unlock(&(status_mutex));
//report
cout << "Status: Run Finished" << endl;
cout << "Total Packets Caught:" << dec << totalPacketsCaught << endl;
cout << "Total Frames Caught:"<< dec << (totalPacketsCaught/packetsPerFrame) << endl;
//acquisition end
if (acquisitionFinishedCallBack)
acquisitionFinishedCallBack((totalPacketsCaught/packetsPerFrame), pAcquisitionFinished);
}
continue; continue;
} }
//for progress //for progress
if(myDetectorType == EIGER){ if(myDetectorType == EIGER){
if(!numpackets) if(!numpackets)
tempframenum = htonl(*(unsigned int*)((eiger_image_header *)((char*)(wbuf[0] + HEADER_SIZE_NUM_TOT_PACKETS)))->fnum); tempframenum = htonl(*(unsigned int*)((eiger_image_header *)((char*)(wbuf[w_index] + HEADER_SIZE_NUM_TOT_PACKETS)))->fnum);
}else if ((myDetectorType == GOTTHARD) && (shortFrame == -1)) }else if ((myDetectorType == GOTTHARD) && (shortFrame == -1))
tempframenum = (((((uint32_t)(*((uint32_t*)(wbuf[0] + HEADER_SIZE_NUM_TOT_PACKETS))))+1)& (frameIndexMask)) >> frameIndexOffset); tempframenum = (((((uint32_t)(*((uint32_t*)(wbuf[w_index] + HEADER_SIZE_NUM_TOT_PACKETS))))+1)& (frameIndexMask)) >> frameIndexOffset);
else else
tempframenum = ((((uint32_t)(*((uint32_t*)(wbuf[0] + HEADER_SIZE_NUM_TOT_PACKETS))))& (frameIndexMask)) >> frameIndexOffset); tempframenum = ((((uint32_t)(*((uint32_t*)(wbuf[w_index] + HEADER_SIZE_NUM_TOT_PACKETS))))& (frameIndexMask)) >> frameIndexOffset);
if(numWriterThreads == 1) if(numWriterThreads == 1)
currframenum = tempframenum; currframenum = tempframenum;
@ -1784,136 +1755,58 @@ int slsReceiverUDPFunctions::startWriting(){
pthread_mutex_unlock(&progress_mutex); pthread_mutex_unlock(&progress_mutex);
} }
#ifdef VERYDEBUG #ifdef VERYDEBUG
cout << endl <<ithread << " tempframenum:" << dec << tempframenum << " curframenum:" << currframenum << endl; if(!numpackets) cout << endl <<ithread << " tempframenum:" << dec << tempframenum << " curframenum:" << currframenum << endl;
#endif #endif
/*if(numpackets)
cout<<"p:"<<(htonl(*(uint32_t*)((eiger_packet_header *)((uint32_t*)(wbuf[0] + HEADER_SIZE_NUM_TOT_PACKETS)))->num2)&0xff)<<"\t\t"<<endl;
*/
//without datacompression: write datacall back, or write data, free fifo //without datacompression: write datacall back, or write data, free fifo
if(!dataCompression){ if(!dataCompression){
if (cbAction < DO_EVERYTHING){
for(i=0;i<numListeningThreads;i++)
rawDataReadyCallBack(currframenum, wbuf[i], numpackets * onePacketSize, sfilefd, guiData,pRawDataReady);
}else if (numpackets > 0){
for(i=0;i<numListeningThreads;i++)
writeToFile_withoutCompression(wbuf[i], numpackets,currframenum);
}
//copy to gui
if (numpackets > 0){
cout<<"numpackets:"<<numpackets<<" wbuf:"<<(void*)wbuf[0]<<endl;
copyFrameToGui(wbuf[0]);
}
for(i=0;i<numListeningThreads;i++)
while(!fifoFree[i]->push(wbuf[i]));
#ifdef VERYVERBOSE
cout<<"buf freed:"<<(void*)wbuf[0]<<endl;
#endif
}
//data compression
else{
#if defined(MYROOT1) && defined(ALLFILE_DEBUG)
writeToFile_withoutCompression(wbuf, numpackets,currframenum);
#endif
eventType thisEvent = PEDESTAL;
int ndata;
char* buff = 0;
data = wbuf[0]+ HEADER_SIZE_NUM_TOT_PACKETS;
int remainingsize = numpackets * onePacketSize;
int np;
int once = 0;
double tot, tl, tr, bl, br;
int xmin = 1, ymin = 1, ix, iy;
while(buff = receiverdata[ithread]->findNextFrame(data,ndata,remainingsize)){
np = ndata/onePacketSize;
//cout<<"buff framnum:"<<ithread <<":"<< ((((uint32_t)(*((uint32_t*)buff)))& (frameIndexMask)) >> frameIndexOffset)<<endl;
if ((np == packetsPerFrame) && (buff!=NULL)){
if(nf == 1000) cout << "Thread " << ithread << ": pedestal done " << endl;
singlePhotonDet[ithread]->newFrame();
//only for moench
if(commonModeSubtractionEnable){
for(ix = xmin - 1; ix < xmax+1; ix++){
for(iy = ymin - 1; iy < ymax+1; iy++){
thisEvent = singlePhotonDet[ithread]->getEventType(buff, ix, iy, 0);
}
}
}
for(ix = xmin - 1; ix < xmax+1; ix++)
for(iy = ymin - 1; iy < ymax+1; iy++){
thisEvent=singlePhotonDet[ithread]->getEventType(buff, ix, iy, commonModeSubtractionEnable);
if (nf>1000) {
tot=0;
tl=0;
tr=0;
bl=0;
br=0;
if (thisEvent==PHOTON_MAX) {
iFrame=receiverdata[ithread]->getFrameNumber(buff);
#ifdef MYROOT1
myTree[ithread]->Fill();
//cout << "Fill in event: frmNr: " << iFrame << " ix " << ix << " iy " << iy << " type " << thisEvent << endl;
#else
pthread_mutex_lock(&write_mutex);
if((enableFileWrite) && (sfilefd))
singlePhotonDet[ithread]->writeCluster(sfilefd);
pthread_mutex_unlock(&write_mutex);
#endif
}
}
}
nf++;
#ifndef ALLFILE
pthread_mutex_lock(&progress_mutex);
packetsInFile += packetsPerFrame;
packetsCaught += packetsPerFrame;
totalPacketsCaught += packetsPerFrame;
if(packetsInFile >= maxPacketsPerFile)
createNewFile();
pthread_mutex_unlock(&progress_mutex);
#endif
if(!once){
copyFrameToGui(buff);
once = 1;
}
}
remainingsize -= ((buff + ndata) - data);
data = buff + ndata;
if(data > (wbuf[0] + HEADER_SIZE_NUM_TOT_PACKETS + numpackets * onePacketSize) )
cout <<" **************ERROR SHOULD NOT COME HERE, Error 142536!"<<endl;
if (cbAction < DO_EVERYTHING){
for(i=0;i<numListeningThreads;++i)
rawDataReadyCallBack(currframenum, wbuf[w_index+i], numpackets * onePacketSize, sfilefd, guiData,pRawDataReady);
}else if (numpackets > 0){
for(i=0;i<numListeningThreads;++i)
writeToFile_withoutCompression(wbuf[w_index+i], numpackets,currframenum);
} }
while(!fifoFree[0]->push(wbuf[0]));
if(myDetectorType == EIGER) {
//last packet
if((packetsPerThread-1)==(htonl(*(uint32_t*)((eiger_packet_header *)((uint32_t*)(wbuf[w_index] + HEADER_SIZE_NUM_TOT_PACKETS)))->num2)&0xff)){
copyFrameToGui(wbuf,currframenum);
for(j=0;j<w_index;j=j+numListeningThreads){
for(i=0;i<numListeningThreads;++i){
while(!fifoFree[i]->push(wbuf[j+i]));
}
}
w_index = 0;
}else
w_index+=numListeningThreads;
}
else{
//copy to gui
copyFrameToGui(NULL,-1,wbuf[0]+HEADER_SIZE_NUM_TOT_PACKETS);
#ifdef VERYVERBOSE
cout << ithread << " finished copying" << endl;
#endif
while(!fifoFree[0]->push(wbuf[0]));
#ifdef VERYVERBOSE #ifdef VERYVERBOSE
cout<<"buf freed:"<<(void*)wbuf[0]<<endl; cout<<"buf freed:"<<(void*)wbuf[0]<<endl;
#endif #endif
}
} }
//data compression
else
handleDataCompression(ithread,wbuf,numpackets,d, xmax, ymax, nf);
} }
#ifdef VERYVERBOSE #ifdef VERYVERBOSE
cout << ithread << " gonna wait for 1st sem" << endl; cout << ithread << " gonna wait for 1st sem" << endl;
@ -2016,7 +1909,7 @@ int i;
if(rc > 0){ if(rc > 0){
pc = (rc/onePacketSize); pc = (rc/onePacketSize);
#ifdef VERYDEBUG #ifdef VERYDEBUG
cout << ithread << " *** last packetcount:" << packetcount << endl; cout << ithread << " *** last packetcount:" << pc << endl;
#endif #endif
(*((uint16_t*)(buffer[ithread]))) = pc; (*((uint16_t*)(buffer[ithread]))) = pc;
totalListeningFrameCount[ithread] += pc; totalListeningFrameCount[ithread] += pc;
@ -2031,6 +1924,9 @@ int i;
for(i=0;i<numWriterThreads;++i){ for(i=0;i<numWriterThreads;++i){
fifoFree[ithread]->pop(buffer[ithread]); fifoFree[ithread]->pop(buffer[ithread]);
(*((uint16_t*)(buffer[ithread]))) = 0xFFFF; (*((uint16_t*)(buffer[ithread]))) = 0xFFFF;
#ifdef VERYDEBUG
cout << ithread << " going to push in dummy buffer:" << (void*)buffer[ithread] << " with num packets:"<< (*((uint16_t*)(buffer[ithread]))) << endl;
#endif
while(!fifo[ithread]->push(buffer[ithread])); while(!fifo[ithread]->push(buffer[ithread]));
#ifdef VERYDEBUG #ifdef VERYDEBUG
cout << ithread << " pushed in dummy buffer:" << (void*)buffer[ithread] << endl; cout << ithread << " pushed in dummy buffer:" << (void*)buffer[ithread] << endl;
@ -2046,7 +1942,7 @@ int i;
pthread_mutex_unlock(&(status_mutex)); pthread_mutex_unlock(&(status_mutex));
#ifdef VERYDEBUG #ifdef VERYDEBUG
cout << ithread << ": Frames listened to " << ((totalListeningFrameCount[ithread]/packetsPerFrame)/numListeningThreads) << endl; cout << ithread << ": Frames listened to " << dec << ((totalListeningFrameCount[ithread]*numListeningThreads)/packetsPerFrame) << endl;
#endif #endif
//waiting for all listening threads to be done, to print final count of frames listened to //waiting for all listening threads to be done, to print final count of frames listened to
@ -2061,7 +1957,7 @@ int i;
t = 0; t = 0;
for(i=0;i<numListeningThreads;++i) for(i=0;i<numListeningThreads;++i)
t += totalListeningFrameCount[i]; t += totalListeningFrameCount[i];
cout << "Total frames listened to " << (t/packetsPerFrame) << endl; cout << "Total frames listened to " << dec <<(t/packetsPerFrame) << endl;
#endif #endif
} }
@ -2070,6 +1966,79 @@ int i;
void slsReceiverUDPFunctions::stopWriting(int ithread, char* wbuffer[], int wIndex){
int i,j;
#ifdef VERYDEBUG
cout << ithread << " **********************popped last dummy frame:" << (void*)wbuffer[wIndex] << endl;
#endif
//free fifo
for(i=0;i<numListeningThreads;++i)
for(j=0;j<wIndex;++j){
while(!fifoFree[i]->push(wbuffer[wIndex+j]));
#ifdef VERYDEBUG
cout << ithread << " fifo freed:" << (void*)wbuffer[wIndex+j] << endl;
#endif
}
//all threads need to close file, reset mask and exit loop
closeFile(ithread);
pthread_mutex_lock(&status_mutex);
writerthreads_mask^=(1<<ithread);
#ifdef VERYDEBUG
cout << ithread << " Resetting mask of current writing thread. New Mask: " << writerthreads_mask << endl;
#endif
pthread_mutex_unlock(&status_mutex);
//only thread 0 needs to do this
//check if all jobs are done and wait
//change status to run finished
if(ithread == 0){
if(dataCompression){
cout << "Waiting for jobs to be done.. current mask:" << hex << writerthreads_mask << endl;
while(writerthreads_mask){
/*cout << "." << flush;*/
usleep(50000);
}
cout<<" Jobs Done!"<<endl;
}
//to make sure listening threads are done before you update status, as that returns to client
while(listeningthreads_mask)
usleep(5000);
//update status
pthread_mutex_lock(&status_mutex);
status = RUN_FINISHED;
pthread_mutex_unlock(&(status_mutex));
//report
cout << "Status: Run Finished" << endl;
cout << "Total Packets Caught:" << dec << totalPacketsCaught << endl;
cout << "Total Frames Caught:"<< dec << (totalPacketsCaught/packetsPerFrame) << endl;
//acquisition end
if (acquisitionFinishedCallBack)
acquisitionFinishedCallBack((totalPacketsCaught/packetsPerFrame), pAcquisitionFinished);
}
}
void slsReceiverUDPFunctions::writeToFile_withoutCompression(char* buf,int numpackets, uint32_t framenum){ void slsReceiverUDPFunctions::writeToFile_withoutCompression(char* buf,int numpackets, uint32_t framenum){
int packetsToSave, offset,lastpacket; int packetsToSave, offset,lastpacket;
uint32_t tempframenum = framenum; uint32_t tempframenum = framenum;
@ -2159,10 +2128,132 @@ void slsReceiverUDPFunctions::writeToFile_withoutCompression(char* buf,int numpa
if(numWriterThreads > 1) if(numWriterThreads > 1)
pthread_mutex_unlock(&write_mutex); pthread_mutex_unlock(&write_mutex);
} }
}
void slsReceiverUDPFunctions::handleDataCompression(int ithread, char* wbuffer[], int &npackets, char* data, int xmax, int ymax, int &nf){
#if defined(MYROOT1) && defined(ALLFILE_DEBUG)
writeToFile_withoutCompression(wbuf[0], numpackets,currframenum);
#endif
eventType thisEvent = PEDESTAL;
int ndata;
char* buff = 0;
data = wbuffer[0]+ HEADER_SIZE_NUM_TOT_PACKETS;
int remainingsize = npackets * onePacketSize;
int np;
int once = 0;
double tot, tl, tr, bl, br;
int xmin = 1, ymin = 1, ix, iy;
while(buff = receiverdata[ithread]->findNextFrame(data,ndata,remainingsize)){
np = ndata/onePacketSize;
//cout<<"buff framnum:"<<ithread <<":"<< ((((uint32_t)(*((uint32_t*)buff)))& (frameIndexMask)) >> frameIndexOffset)<<endl;
if ((np == packetsPerFrame) && (buff!=NULL)){
if(nf == 1000) cout << "Thread " << ithread << ": pedestal done " << endl;
singlePhotonDet[ithread]->newFrame();
//only for moench
if(commonModeSubtractionEnable){
for(ix = xmin - 1; ix < xmax+1; ix++){
for(iy = ymin - 1; iy < ymax+1; iy++){
thisEvent = singlePhotonDet[ithread]->getEventType(buff, ix, iy, 0);
}
}
}
for(ix = xmin - 1; ix < xmax+1; ix++)
for(iy = ymin - 1; iy < ymax+1; iy++){
thisEvent=singlePhotonDet[ithread]->getEventType(buff, ix, iy, commonModeSubtractionEnable);
if (nf>1000) {
tot=0;
tl=0;
tr=0;
bl=0;
br=0;
if (thisEvent==PHOTON_MAX) {
receiverdata[ithread]->getFrameNumber(buff);
//iFrame=receiverdata[ithread]->getFrameNumber(buff);
#ifdef MYROOT1
myTree[ithread]->Fill();
//cout << "Fill in event: frmNr: " << iFrame << " ix " << ix << " iy " << iy << " type " << thisEvent << endl;
#else
pthread_mutex_lock(&write_mutex);
if((enableFileWrite) && (sfilefd))
singlePhotonDet[ithread]->writeCluster(sfilefd);
pthread_mutex_unlock(&write_mutex);
#endif
}
}
}
nf++;
#ifndef ALLFILE
pthread_mutex_lock(&progress_mutex);
packetsInFile += packetsPerFrame;
packetsCaught += packetsPerFrame;
totalPacketsCaught += packetsPerFrame;
if(packetsInFile >= maxPacketsPerFile)
createNewFile();
pthread_mutex_unlock(&progress_mutex);
#endif
if(!once){
copyFrameToGui(NULL,-1,buff);
once = 1;
}
}
remainingsize -= ((buff + ndata) - data);
data = buff + ndata;
if(data > (wbuffer[0] + HEADER_SIZE_NUM_TOT_PACKETS + npackets * onePacketSize) )
cout <<" **************ERROR SHOULD NOT COME HERE, Error 142536!"<<endl;
}
while(!fifoFree[0]->push(wbuffer[0]));
#ifdef VERYVERBOSE
cout<<"buf freed:"<<(void*)wbuffer[0]<<endl;
#endif
} }
/*#endif*/
#endif

33
slsReceiverSoftware/slsReceiver/slsReceiverUDPFunctions.h
View File

@ -1,4 +1,4 @@
/*#ifdef SLS_RECEIVER_UDP_FUNCTIONS*/ #ifdef SLS_RECEIVER_UDP_FUNCTIONS
#ifndef SLS_RECEIVER_UDP_FUNCTIONS_H #ifndef SLS_RECEIVER_UDP_FUNCTIONS_H
#define SLS_RECEIVER_UDP_FUNCTIONS_H #define SLS_RECEIVER_UDP_FUNCTIONS_H
/********************************************//** /********************************************//**
@ -239,8 +239,9 @@ public:
* Returns the buffer-current frame read by receiver * Returns the buffer-current frame read by receiver
* @param c pointer to current file name * @param c pointer to current file name
* @param raw address of pointer, pointing to current frame to send to gui * @param raw address of pointer, pointing to current frame to send to gui
* @param fnum frame number for eiger as it is not in the packet
*/ */
void readFrame(char* c,char** raw); void readFrame(char* c,char** raw, uint32_t &fnum);
/** /**
* Closes all files * Closes all files
@ -292,7 +293,7 @@ private:
* Copy frames to gui * Copy frames to gui
* uses semaphore for nth frame mode * uses semaphore for nth frame mode
*/ */
void copyFrameToGui(char* startbuf); void copyFrameToGui(char* startbuf[], uint32_t fnum=-1, char* buf=NULL);
/** /**
* creates udp sockets * creates udp sockets
@ -392,6 +393,27 @@ private:
*/ */
void stopListening(int ithread, int rc, int &pc, int &t); void stopListening(int ithread, int rc, int &pc, int &t);
/**
* When acquisition is over, this is called
* @param ithread listening thread number
* @param wbuffer writer buffer
* @param wIndex writer Index
*/
void stopWriting(int ithread, char* wbuffer[], int wIndex);
/**
* data compression for each fifo output
* @param ithread listening thread number
* @param wbuffer writer buffer
* @param npackets number of packets from the fifo
* @param data pointer to the next packet start
* @param xmax max pixels in x direction
* @param ymax max pixels in y direction
* @param nf nf
*/
void handleDataCompression(int ithread, char* wbuffer[], int &npackets, char* data, int xmax, int ymax, int &nf);
/** structure of an eiger image header*/ /** structure of an eiger image header*/
typedef struct typedef struct
@ -541,6 +563,9 @@ private:
/** points to the filename to send to gui */ /** points to the filename to send to gui */
char* guiFileName; char* guiFileName;
/** temporary number for eiger frame number as its not included in the packet */
uint32_t guiFrameNumber;
/** send every nth frame to gui or only upon gui request*/ /** send every nth frame to gui or only upon gui request*/
int nFrameToGui; int nFrameToGui;
@ -731,4 +756,4 @@ public:
#endif #endif
/*#endif*/ #endif