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

compression files created in separate threads and nt in tcp

Browse Source 
git-svn-id: file:///afs/psi.ch/project/sls_det_software/svn/slsDetectorSoftware@743 951219d9-93cf-4727-9268-0efd64621fa3
This commit is contained in:
l_maliakal_d 2014-02-26 15:32:02 +00:00
parent a892a8854c
commit 7371aea454
2 changed files with 116 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

172
slsDetectorSoftware/slsReceiver/slsReceiverFunctionList.cpp
View File

@ -166,15 +166,14 @@ slsReceiverFunctionList::slsReceiverFunctionList(detectorType det):
slsReceiverFunctionList::~slsReceiverFunctionList(){ slsReceiverFunctionList::~slsReceiverFunctionList(){
closeFile(-1); createListeningThreads(true);
createWriterThreads(true);
for(int i=0;i<numWriterThreads;i++){ for(int i=0;i<numWriterThreads;i++){
if(singlePhotonDet[i]) if(singlePhotonDet[i])
delete singlePhotonDet[i]; delete singlePhotonDet[i];
if(receiverdata[i]) if(receiverdata[i])
delete receiverdata[i]; delete receiverdata[i];
} }
createListeningThreads(true);
createWriterThreads(true);
if(udpSocket) delete udpSocket; if(udpSocket) delete udpSocket;
if(eth) delete [] eth; if(eth) delete [] eth;
@ -598,7 +597,7 @@ int slsReceiverFunctionList::createListeningThreads(bool destroy){
if(!destroy){ if(!destroy){
//listening thread //listening thread
cout << "Creating Listening Thread" << endl; cout << "Creating Listening Thread" << endl;
sem_init(&listensmp,0,1); sem_init(&listensmp,1,0);
if(pthread_create(&listening_thread, NULL,startListeningThread, (void*) this)){ if(pthread_create(&listening_thread, NULL,startListeningThread, (void*) this)){
cout << "Could not create listening thread" << endl; cout << "Could not create listening thread" << endl;
return FAIL; return FAIL;
@ -633,6 +632,7 @@ int slsReceiverFunctionList::createWriterThreads(bool destroy){
pthread_mutex_lock(&status_mutex); pthread_mutex_lock(&status_mutex);
writerthreads_mask = 0x0; writerthreads_mask = 0x0;
createfile_mask = 0x0;
pthread_mutex_unlock(&(status_mutex)); pthread_mutex_unlock(&(status_mutex));
@ -644,7 +644,7 @@ int slsReceiverFunctionList::createWriterThreads(bool destroy){
currentWriterThreadIndex = -1; currentWriterThreadIndex = -1;
for(i = 0; i < numWriterThreads; ++i){ for(i = 0; i < numWriterThreads; ++i){
sem_init(&writersmp[i],0,1); sem_init(&writersmp[i],1,0);
thread_started = 0; thread_started = 0;
currentWriterThreadIndex = i; currentWriterThreadIndex = i;
if(pthread_create(&writing_thread[i], NULL,startWritingThread, (void*) this)){ if(pthread_create(&writing_thread[i], NULL,startWritingThread, (void*) this)){
@ -727,10 +727,11 @@ int slsReceiverFunctionList::setupWriter(){
guiDataReady=0; guiDataReady=0;
strcpy(guiFileName,""); strcpy(guiFileName,"");
cbAction = DO_EVERYTHING; cbAction = DO_EVERYTHING;
int ret,ret1 = OK;
pthread_mutex_lock(&status_mutex); pthread_mutex_lock(&status_mutex);
writerthreads_mask = 0x0; writerthreads_mask = 0x0;
createfile_mask = 0x0;
ret_createfile = OK;
pthread_mutex_unlock(&status_mutex); pthread_mutex_unlock(&status_mutex);
//printouts //printouts
@ -756,19 +757,32 @@ int slsReceiverFunctionList::setupWriter(){
//creating first file //creating first file
if(!dataCompression) pthread_mutex_lock(&status_mutex);
ret = createNewFile(); for(int i=0;i<numWriterThreads;i++)
else{ createfile_mask|=(1<<i);
for(int i=0;i<numWriterThreads;i++){ pthread_mutex_unlock(&status_mutex);
pthread_mutex_lock(&write_mutex);
ret1 = createCompressionFile(i,0); for(int i=0;i<numWriterThreads;i++){
pthread_mutex_unlock(&write_mutex); #ifdef VERYDEBUG
if(ret1 == FAIL) cout << i << " gonna post 1st sem" << endl;
ret = FAIL; #endif
sem_post(&writersmp[i]);
//wait for each file to be created
while((1<<i)&createfile_mask){
cout<<"*"<<flush;
usleep(5000);
} }
} }
//wait till its created
/*while(createfile_mask){
cout<<"*"<<flush;
usleep(5000);
}*/
if (createfile_mask)
cout <<"*********************************************sooo weird:"<<createfile_mask<<endl;
return ret_createfile;
return ret;
} }
@ -778,15 +792,15 @@ int slsReceiverFunctionList::setupWriter(){
int slsReceiverFunctionList::createCompressionFile(int ithr, int iframe){ int slsReceiverFunctionList::createCompressionFile(int ithr, int iframe){
#ifdef MYROOT1 #ifdef MYROOT1
char temp[MAX_STR_LENGTH];
//create file name for gui purposes, and set up acquistion parameters //create file name for gui purposes, and set up acquistion parameters
sprintf(savefilename, "%s/%s_fxxx_%d_%d.root", filePath,fileName,fileIndex,ithr); sprintf(temp, "%s/%s_fxxx_%d_%d.root", filePath,fileName,fileIndex,ithr);
//file //file
myFile[ithr] = new TFile(savefilename,"RECREATE");/** later return error if it exists */ myFile[ithr] = new TFile(temp,"RECREATE");/** later return error if it exists */
cout<<"File created: "<<savefilename<<endl; cout << "Thread " << ithr << ": created File: "<< temp << endl;
//tree //tree
sprintf(savefilename, "%s_fxxx_%d_%d",fileName,fileIndex,ithr); sprintf(temp, "%s_fxxx_%d_%d",fileName,fileIndex,ithr);
myTree[ithr]=singlePhotonDet[ithr]->initEventTree(savefilename, &iframe); myTree[ithr]=singlePhotonDet[ithr]->initEventTree(temp, &iframe);
//resets the pedestalSubtraction array and the commonModeSubtraction //resets the pedestalSubtraction array and the commonModeSubtraction
singlePhotonDet[ithr]->newDataSet(); singlePhotonDet[ithr]->newDataSet();
if(myFile[ithr]==NULL){ if(myFile[ithr]==NULL){
@ -879,30 +893,33 @@ void slsReceiverFunctionList::closeFile(int ithr){
//datacompression //datacompression
else{ else{
#ifdef MYROOT1 #ifdef MYROOT1
if(ithr == -1){ pthread_mutex_lock(&write_mutex);
for(int i=0;i<numWriterThreads;i++) //write to file
closeFile(i); if(myTree[ithr] && myFile[ithr]){
} myFile[ithr] = myTree[ithr]->GetCurrentFile();
else{
//write to file if(myFile[ithr]->Write())
if(myTree[ithr] && myFile[ithr]){ //->Write(tall->GetName(),TObject::kOverwrite);
myFile[ithr] = myTree[ithr]->GetCurrentFile(); cout << "Thread " << ithr <<": wrote frames to file" << endl;
if(myFile[ithr]->Write()) else
cout << "Thread " << ithr <<" wrote frames to file" << endl; cout << "Thread " << ithr << ": could not write frames to file" << endl;
else
cout << "Thread " << ithr << " could not write frames to file" << endl; }else
}else cout << "Thread " << ithr << ": could not write frames to file: No file or No Tree" << endl;
cout << "Thread " << ithr << " could not write frames to file: No file or No Tree" << endl; //close file
//close file if(myTree[ithr] && myFile[ithr])
if(myTree[ithr] && myFile[ithr]) myFile[ithr] = myTree[ithr]->GetCurrentFile();
myFile[ithr] = myTree[ithr]->GetCurrentFile(); if(myFile[ithr] != NULL)
if(myFile[ithr] != NULL) myFile[ithr]->Close();
myFile[ithr]->Close(); myFile[ithr] = NULL;
myFile[ithr] = NULL; myTree[ithr] = NULL;
myTree[ithr] = NULL; pthread_mutex_unlock(&write_mutex);
}
#endif #endif
} }
#ifdef VERBOSE
cout << ithr << " out of close file" << endl;
#endif
} }
@ -937,13 +954,14 @@ int slsReceiverFunctionList::startReceiver(char message[]){
sprintf(message,"Could not create file %s.\n",savefilename); sprintf(message,"Could not create file %s.\n",savefilename);
return FAIL; return FAIL;
} }
cout << "Successfully created file(s)" << endl;
//done to give the gui some proper name instead of always the last file name //done to give the gui some proper name instead of always the last file name
if(dataCompression) if(dataCompression)
sprintf(savefilename, "%s/%s_fxxx_%d_xx.root", filePath,fileName,fileIndex); sprintf(savefilename, "%s/%s_fxxx_%d_xx.root", filePath,fileName,fileIndex);
//initialize semaphore //initialize semaphore
sem_init(&smp,0,1); sem_init(&smp,1,0);
//status //status
pthread_mutex_lock(&status_mutex); pthread_mutex_lock(&status_mutex);
@ -956,15 +974,10 @@ int slsReceiverFunctionList::startReceiver(char message[]){
//start listening /writing //start listening /writing
sem_post(&listensmp); sem_post(&listensmp);
/*int k;*/
for(int i=0; i < numWriterThreads; ++i){ for(int i=0; i < numWriterThreads; ++i){
/*sem_getvalue(&writersmp[i],&k);*/
sem_post(&writersmp[i]); sem_post(&writersmp[i]);
/*sem_getvalue(&writersmp[i],&k);*/
} }
cout << "Receiver Started.\nStatus:" << status << endl; cout << "Receiver Started.\nStatus:" << status << endl;
return OK; return OK;
@ -1268,6 +1281,7 @@ int slsReceiverFunctionList::startWriting(){
char *data=new char[bufferSize]; char *data=new char[bufferSize];
int iFrame = 0; int iFrame = 0;
int xmax=0,ymax=0; int xmax=0,ymax=0;
int ret;
while(1){ while(1){
@ -1320,9 +1334,7 @@ int slsReceiverFunctionList::startWriting(){
//all threads need to close file, reset mask and exit loop //all threads need to close file, reset mask and exit loop
pthread_mutex_lock(&write_mutex);
closeFile(ithread); closeFile(ithread);
pthread_mutex_unlock(&write_mutex);
pthread_mutex_lock(&status_mutex); pthread_mutex_lock(&status_mutex);
writerthreads_mask^=(1<<ithread); writerthreads_mask^=(1<<ithread);
#ifdef VERYDEBUG #ifdef VERYDEBUG
@ -1331,6 +1343,7 @@ int slsReceiverFunctionList::startWriting(){
pthread_mutex_unlock(&status_mutex); pthread_mutex_unlock(&status_mutex);
//only thread 0 needs to do this //only thread 0 needs to do this
//check if all jobs are done and wait //check if all jobs are done and wait
//change status to run finished //change status to run finished
@ -1429,7 +1442,7 @@ int slsReceiverFunctionList::startWriting(){
//cout<<"buff framnum:"<<ithread <<":"<< ((((uint32_t)(*((uint32_t*)buff)))& (frameIndexMask)) >> frameIndexOffset)<<endl; //cout<<"buff framnum:"<<ithread <<":"<< ((((uint32_t)(*((uint32_t*)buff)))& (frameIndexMask)) >> frameIndexOffset)<<endl;
if ((np == packetsPerFrame) && (buff!=NULL)){ if ((np == packetsPerFrame) && (buff!=NULL)){
if(nf == 1000) cout << " pedestal done " << endl; if(nf == 1000) cout << "Thread " << ithread << ": pedestal done " << endl;
singlePhotonDet[ithread]->newFrame(); singlePhotonDet[ithread]->newFrame();
@ -1456,9 +1469,7 @@ int slsReceiverFunctionList::startWriting(){
if (thisEvent==PHOTON_MAX) { if (thisEvent==PHOTON_MAX) {
iFrame=receiverdata[ithread]->getFrameNumber(buff); iFrame=receiverdata[ithread]->getFrameNumber(buff);
pthread_mutex_lock(&write_mutex);
myTree[ithread]->Fill(); myTree[ithread]->Fill();
pthread_mutex_unlock(&write_mutex);
//cout << "Fill in event: frmNr: " << iFrame << " ix " << ix << " iy " << iy << " type " << thisEvent << endl; //cout << "Fill in event: frmNr: " << iFrame << " ix " << ix << " iy " << iy << " type " << thisEvent << endl;
} }
} }
@ -1494,17 +1505,54 @@ int slsReceiverFunctionList::startWriting(){
#endif #endif
} }
} }
#ifdef VERYVERBOSE
/*int k; cout << ithread << " gonna wait for 1st sem" << endl;
sem_getvalue(&writersmp[ithread],&k); #endif
cout<<ithread<<" waiting for sem:"<<k <<endl;*/
//wait //wait
sem_wait(&writersmp[ithread]); sem_wait(&writersmp[ithread]);
if(killAllWritingThreads){
if(killAllWritingThreads) cout << ithread << " good bye thread" << endl;
closeFile(ithread);
pthread_exit(NULL); pthread_exit(NULL);
}
#ifdef VERYVERBOSE
cout << ithread << " got 1st post" << endl;
#endif
if((1<<ithread)&createfile_mask){
if(dataCompression){
pthread_mutex_lock(&write_mutex);
ret = createCompressionFile(ithread,0);
pthread_mutex_unlock(&write_mutex);
if(ret == FAIL)
ret_createfile = FAIL;
}else{
ret = createNewFile();
if(ret == FAIL)
ret_createfile = FAIL;
}
//let tcp know
pthread_mutex_lock(&status_mutex);
createfile_mask^=(1<<ithread);
pthread_mutex_unlock(&status_mutex);
}
#ifdef VERYVERBOSE
cout << ithread << " gonna wait for 2nd sem" << endl;
#endif
//wait
sem_wait(&writersmp[ithread]);
if(killAllWritingThreads){
cout << ithread << " Goodbye thread" << endl;
closeFile(ithread);
pthread_exit(NULL);
}
#ifdef VERYVERBOSE
cout << ithread << " got 2nd post" << endl;
#endif
} }

6
slsDetectorSoftware/slsReceiver/slsReceiverFunctionList.h
View File

@ -469,6 +469,12 @@ private:
/** mask showing which threads are running */ /** mask showing which threads are running */
volatile uint32_t writerthreads_mask; volatile uint32_t writerthreads_mask;
/** mask showing which threads have created files*/
volatile uint32_t createfile_mask;
/** OK if file created was successful */
int ret_createfile;
/** 0 if listening thread is idle, 1 otherwise */ /** 0 if listening thread is idle, 1 otherwise */
int listening_thread_running; int listening_thread_running;