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

almosst

Browse Source
This commit is contained in:
Dhanya Maliakal
2016-09-15 12:16:01 +02:00
parent 0f9a841c4a
commit a821442b1a
8 changed files with 98 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files

202
slsDetectorSoftware/multiSlsDetector/multiSlsDetector.cpp
View File

@ -4959,27 +4959,27 @@ int multiSlsDetector::resetFramesCaught() {
void multiSlsDetector::readFrameFromReceiver(){
int value;
//Note:num threads = (num slsDets = num tasks)
//so, half slsdet readouts read serially in each task (eiger udp ports)
//create zmq threads
if(createThreadPool(&zmqthreadpool) == FAIL){
cprintf(BG_RED,"Error: Could not create the zmq threads\n");
return;
}
zmqthreadpool->setzeromqThread();
zmqthreadpool->setzeromqThread(); //for debugging
//determine number of half readouts and maxX and maxY
int maxX=0,maxY=0;
int numReadout = 1;
bool checkbottom = false;
int numReadoutPerDetector = 1;
if(getDetectorsType() == EIGER){
numReadout = 2;
numReadoutPerDetector = 2;
maxX = thisMultiDetector->numberOfChannel[X];
maxY = thisMultiDetector->numberOfChannel[Y];
checkbottom = true;
}
int numReadouts = numReadoutPerDetector * thisMultiDetector->numberOfDetectors;
//Note:num threads correspond to num detectors as task calls each slsdet
//(eiger udp ports/half readouts will have to do it serially)
//start all socket tasks
volatile uint64_t runningMask = 0x0;
@ -4990,11 +4990,13 @@ void multiSlsDetector::readFrameFromReceiver(){
}else{
for(int idet=0; idet<thisMultiDetector->numberOfDetectors; idet++){
if(detectors[idet]){
sem_init(&sem_slswait[idet*numReadout],1,0);
sem_init(&sem_slsdone[idet*numReadout],1,0);
if(numReadout>1){
sem_init(&sem_slswait[idet*numReadout+1],1,0);
sem_init(&sem_slsdone[idet*numReadout+1],1,0);
sem_init(&sem_slswait[idet*numReadoutPerDetector],1,0);
sem_init(&sem_slsdone[idet*numReadoutPerDetector],1,0);
sem_init(&sem_multiwait[idet*numReadoutPerDetector],1,0);
if(numReadoutPerDetector>1){
sem_init(&sem_slswait[idet*numReadoutPerDetector+1],1,0);
sem_init(&sem_slsdone[idet*numReadoutPerDetector+1],1,0);
sem_init(&sem_multiwait[idet*numReadoutPerDetector+1],1,0);
}
Task* task = new Task(new func00_t<void, slsDetector>(&slsDetector::readFrameFromReceiver,detectors[idet]));
zmqthreadpool->add_task(task);
@ -5006,19 +5008,21 @@ void multiSlsDetector::readFrameFromReceiver(){
slsmaxY = detectors[idet]->getTotalNumberOfChannels(Y);
}
//set mask
runningMask|=(1<<(idet*numReadout));
if(numReadout>1)
runningMask|=(1<<(idet*numReadout+1));
runningMask|=(1<<(idet*numReadoutPerDetector));
if(numReadoutPerDetector>1)
runningMask|=(1<<(idet*numReadoutPerDetector+1));
}
}
}
zmqthreadpool->startExecuting(); //tell them to start
for(int i=0;i<numReadouts;++i) //wait for all of them to have started
sem_wait(&sem_multiwait[i]);
sem_post(&dataThreadStartedSemaphore); //let utils:acquire continue to start measurement/acquisition
zmqthreadpool->startExecuting();//tell them to start
int nel=(thisMultiDetector->dataBytes)/sizeof(int);
if(nel <= 0){
cout << "Multislsdetector databytes not valid :" << thisMultiDetector->dataBytes << endl;
return;
@ -5026,51 +5030,43 @@ void multiSlsDetector::readFrameFromReceiver(){
int* multiframe=new int[nel];
int* p = multiframe;
int idet,offsetY,offsetX;
int halfreadoutoffset = (slsmaxX/numReadout);
int halfreadoutoffset = (slsmaxX/numReadoutPerDetector);
//after reconstruction
int framecount=0;
int nx =getTotalNumberOfChannels(slsDetectorDefs::X);
int ny =getTotalNumberOfChannels(slsDetectorDefs::Y);
//construct complete image and send to callback
while(true){
memset(((char*)multiframe),0x0,slsdatabytes*thisMultiDetector->numberOfDetectors);
for(int ireadout=0; ireadout<thisMultiDetector->numberOfDetectors*numReadout; ++ireadout){
idet = ireadout/numReadout;
memset(((char*)multiframe),0x0,slsdatabytes*thisMultiDetector->numberOfDetectors); //reset frame memory
for(int ireadout=0; ireadout<numReadouts; ++ireadout){
idet = ireadout/numReadoutPerDetector;
if(detectors[idet]){
if((1 << ireadout) & runningMask){
if((1 << ireadout) & runningMask){ //if running
sem_post(&sem_slswait[ireadout]); //sls to continue
sem_wait(&sem_slsdone[ireadout]); //wait for sls to copy
//this socket closed
if(slsframe[ireadout] == NULL){
runningMask^=(1<<ireadout);
//all done, get out
if(!runningMask){
if(!runningMask){ //all done, get out
break;
}
continue;
}
//assemble data
//eiger, so interleaving between ports in one readout itself
if(maxX){
//if(ireadout == 3){
if(maxX){ //eiger, so interleaving between ports in one readout itself
offsetY = (maxY - (thisMultiDetector->offsetY[idet] + slsmaxY)) * maxX * bytesperchannel;
if(!(ireadout%numReadout)) offsetX = thisMultiDetector->offsetX[idet];
else offsetX = thisMultiDetector->offsetX[idet] + halfreadoutoffset;
//the left half or right half
if(!(ireadout%numReadoutPerDetector))
offsetX = thisMultiDetector->offsetX[idet];
else
offsetX = thisMultiDetector->offsetX[idet] + halfreadoutoffset;
offsetX *= bytesperchannel;
//cprintf(BLUE,"offsetx:%d offsety:%d maxx:%d slsmaxX:%d slsmaxY:%d bytesperchannel:%d\n",
// offsetX,offsetY,maxX,slsmaxX,slsmaxY,bytesperchannel);
//cprintf(BLUE,"copying bytes:%d\n", (slsmaxX/numReadout)*bytesperchannel);
//itnerleaving with other detectors
@ -5078,21 +5074,16 @@ void multiSlsDetector::readFrameFromReceiver(){
if(((idet+1)%2) == 0){
for(int i=0;i<slsmaxY;++i)
memcpy(((char*)multiframe) + offsetY + offsetX + ((slsmaxY-i)*maxX*bytesperchannel),
(char*)slsframe[ireadout]+ i*(slsmaxX/numReadout)*bytesperchannel,
(slsmaxX/numReadout)*bytesperchannel);
(char*)slsframe[ireadout]+ i*(slsmaxX/numReadoutPerDetector)*bytesperchannel,
(slsmaxX/numReadoutPerDetector)*bytesperchannel);
}
//top
else{
for(int i=0;i<slsmaxY;++i)
memcpy(((char*)multiframe) + offsetY + offsetX + (i*maxX*bytesperchannel),
(char*)slsframe[ireadout]+ i*(slsmaxX/numReadout)*bytesperchannel,
(slsmaxX/numReadout)*bytesperchannel);
(char*)slsframe[ireadout]+ i*(slsmaxX/numReadoutPerDetector)*bytesperchannel,
(slsmaxX/numReadoutPerDetector)*bytesperchannel);
}
//}//end of ireadout
}
//no interleaving, just add to the end
//numReadout always 1 here
@ -5100,9 +5091,6 @@ void multiSlsDetector::readFrameFromReceiver(){
memcpy(p,multiframe,slsdatabytes);
p+=slsdatabytes/sizeof(int);
}
}
}
}//end of for loop
@ -5112,7 +5100,6 @@ void multiSlsDetector::readFrameFromReceiver(){
}
//send data to callback
fdata = decodeData(multiframe);
if ((fdata) && (dataReady)){
thisData = new detectorData(fdata,NULL,NULL,getCurrentProgress(),"noname.raw",nx,ny);
@ -5121,116 +5108,19 @@ void multiSlsDetector::readFrameFromReceiver(){
fdata = NULL;
//cout<<"Send frame #"<< framecount << " to gui"<<endl;
}
framecount++;
setCurrentProgress(framecount);
}
zmqthreadpool->wait_for_tasks_to_complete();
for(int i=0;i<numReadouts;++i){
sem_destroy(&sem_slsdone[i]);
sem_destroy(&sem_slswait[i]);
sem_destroy(&sem_multiwait[i]);
}
destroyThreadPool(&zmqthreadpool);
delete[] multiframe;
/*
int nel=(thisMultiDetector->dataBytes)/sizeof(int);
if(nel <= 0){
cout << "Multislsdetector databytes not valid :" << thisMultiDetector->dataBytes << endl;
acquisitionIndex = -1;
return NULL;
}
int n,complete=OK;
int i,k,offsetX, offsetY, maxX, maxY; double dr;
int* retval=new int[nel];
int *retdet = NULL, *p=retval;
string fullFName="";
string ext="";
int index=-1,f_index=-1,p_index=-1,det_index=-1;
double sv0=-1,sv1=-1;
if(getDetectorsType() == EIGER){
maxX = thisMultiDetector->numberOfChannel[X];
maxY = thisMultiDetector->numberOfChannel[Y];
}
for (int id=0; id<thisMultiDetector->numberOfDetectors; id++) {
if (detectors[id]) {
n=detectors[id]->getDataBytes();
retdet=detectors[id]->readFrameFromReceiver(fName, acquisitionIndex, frameIndex, subFrameIndex);
if(detectors[id]->getErrorMask())
setErrorMask(getErrorMask()|(1<<id));
if (retdet){
if (acquisitionIndex==-1){
complete = FAIL;
delete [] retdet;
}else{
n=detectors[id]->getDataBytes();
if(getDetectorsType() == EIGER){
//cout << "fname:"<<fName<<" findex:"<<fIndex<<endl;//cout<<"n:"<<n<<endl;//cout<<"maxchan:"<<detectors[id]->getMaxNumberOfChannels()<<" n:"<<n<<endl;
dr = (double)n/detectors[id]->getMaxNumberOfChannels();
k=(int)(detectors[id]->getMaxNumberOfChannels(X)*dr);//bit mode
//cout << "dr:"<<dr<<endl;//cout << "k:"<<k<<endl;
offsetY = (int)(((maxY - (thisMultiDetector->offsetY[id] + detectors[id]->getMaxNumberOfChannels(Y))) * maxX)*dr);//bit mode
offsetX = (int)(thisMultiDetector->offsetX[id]*dr);
//cout << "offsetY"<<offsetY<< " offsetX:"<<offsetX<<endl;
for(i=0; i< 256;i++){
memcpy((((char*)p) + offsetY + offsetX + ((int)(i*maxX*dr))) ,(((char*)retdet) + (i*k)),k);//bit mode
}
}
else{
memcpy(p,retdet,n);
p+=n/sizeof(int);
}
delete [] retdet;
//concatenate filenames
if(!fullFName.length()){
//assign file prefix
fullFName.assign(fileIO::getFileName());
if (strrchr(fName,'.')!=NULL){
ext.assign(fName);
size_t dot = ext.rfind(".");
if(dot != string::npos)
ext = ext.erase(0,dot);
else
ext = "";
//get variables
fileIOStatic::getVariablesFromFileName(fName,index, f_index, p_index, sv0, sv1, det_index);
//append scan and det variables
fullFName.append(fileIOStatic::getReceiverFileNameToConcatenate(fName));
}
}
//append only if scan variables are different
if(!fileIOStatic::verifySameFrame(fName,index,f_index, p_index, sv0, sv1, det_index)){
fullFName.append(fileIOStatic::getReceiverFileNameToConcatenate(fName));
}
}
}else {
#ifdef VERBOSE
cout << "Receiver for detector " << id << " does not have data left " << endl;
#endif
delete [] retval;
return NULL;
}
}
}
//append extension
fullFName.append(ext);
strcpy(fName,fullFName.c_str());
//if some of the receivers did not give data, dont count it
if((getDetectorsType() == EIGER) &&(complete ==FAIL))
acquisitionIndex = -1;
return retval;
*/
};
}

3
slsDetectorSoftware/multiSlsDetector/multiSlsDetector.h
View File

@ -1179,8 +1179,7 @@ class multiSlsDetector : public slsDetectorUtils {
*/
int resetFramesCaught();
/**
* Reads frames from receiver through a constant socket
/** Reads frames from receiver through a constant socket
*/
void readFrameFromReceiver();

53
slsDetectorSoftware/slsDetector/slsDetector.cpp
View File

@ -7161,7 +7161,6 @@ void slsDetector::readFrameFromReceiver(){
portno[i] = DEFAULT_ZMQ_PORTNO + (readoutId+i);
sprintf(hostname[i], "%s%d", "tcp://127.0.0.1:",portno[i]);
//cout << "ZMQ Client of " << readoutId+i << " at " << hostname[i] << endl;
parentDet->slsframe[readoutId+i]=new int[nel];
}
@ -7172,69 +7171,51 @@ void slsDetector::readFrameFromReceiver(){
for(int i=0;i<numReadout;++i){
context[i] = zmq_ctx_new();
zmqsocket[i] = zmq_socket(context[i], ZMQ_SUB);
// an empty string implies receiving any messages
zmq_setsockopt(zmqsocket[i], ZMQ_SUBSCRIBE, "", 0);
// connect to publisher
// the publisher server does not have to be started
zmq_connect(zmqsocket[i], hostname[i]);
zmq_setsockopt(zmqsocket[i], ZMQ_SUBSCRIBE, "", 0); // an empty string implies receiving any messages
zmq_connect(zmqsocket[i], hostname[i]);// connect to publisher,the publisher server does not have to be started
runningMask|=(1<<(i));
sem_post(&parentDet->sem_multiwait[readoutId+i]); //let multi know socket created
}
cout<<detId<<" started"<<endl;
//receive msgs and let multi know
zmq_msg_t message;
int len,idet = 0;
int framecount=0;
//read frame
while(true){
for(int idet=0; idet<numReadout; ++idet){
if((1 << idet) & runningMask){
sem_wait(&parentDet->sem_slswait[readoutId+idet]);//wait for it to be copied
//update indices
if(!idet) framecount++; //count only once
if(!idet) framecount++; //update indices, count only once
// receive a message, this is a blocking function
len = zmq_msg_init (&message); /* is this required? Xiaoqiang didnt have it*/
if(len) {cprintf(RED,"Failed to initialize message %d for %d\n",len,readoutId+idet); continue; }//error
len = zmq_msg_recv(&message, zmqsocket[idet], 0);
//int size = zmq_msg_size (&message);
//end of socket
if (len <= 3 ) {
if(!len) cprintf(RED,"Received no data in socket for %d\n", readoutId+idet);
//cout<<readoutId+idet <<" sls Received end data"<<endl;
parentDet->slsframe[readoutId+idet] = NULL;
sem_post(&parentDet->sem_slsdone[readoutId+idet]);//let multi know is ready
sem_post(&parentDet->sem_slsdone[readoutId+idet]); //let multi know is ready
runningMask^=(1<<idet);
cout<<detId<<" " << idet << " finished"<<endl;
//all done, get out
if(!runningMask){
cout<<detId<<" all done"<<endl;
//cout<<detId<<" " << idet << " finished"<<endl;
if(!runningMask){ //all done, get out
//cout<<detId<<" all done"<<endl;
break;
}
continue;
}
cout<<"Received on " << readoutId+idet << " for frame " << framecount << endl;
if(zmq_msg_data(&message)==NULL) cprintf(RED,"GOT NULL FROM ZMQ\n"); /*not needed most likely*/
if(zmq_msg_data(&message)==NULL)
cprintf(RED,"GOT NULL FROM ZMQ\n");
//cout<<"Received on " << readoutId+idet << " for frame " << framecount << endl;
//if(len == thisDetector->dataBytes/numReadout){//hoow to solve this
memcpy((char*)(parentDet->slsframe[readoutId+idet]),(char*)zmq_msg_data(&message),thisDetector->dataBytes/numReadout);
//memcpy((char*)(parentDet->slsframe[readoutId+idet]),zmq_msg_data(&message[idet]),thisDetector->dataBytes);
//check header, if incorrect frame, copy somewhere and assign a blank subframe
//parentDet->slsframe[readoutId+idet] = (int*)zmq_msg_data(&message[idet]);
//check header, if incorrect frame, copy somewhere and assign a blank subframe and also check size
//jungfrau masking adcval
if(thisDetector->myDetectorType == JUNGFRAU){
@ -7244,29 +7225,23 @@ void slsDetector::readFrameFromReceiver(){
}
//}
sem_post(&parentDet->sem_slsdone[readoutId+idet]);//let multi know is ready
}
}//end of for loop
if(!runningMask){
break;
}
}
zmq_msg_close(&message);
//close socket
for(int i=0;i<numReadout;i++){
zmq_disconnect(zmqsocket[i], hostname[i]);
zmq_close(zmqsocket[i]);
zmq_ctx_destroy(context[i]);
delete [] parentDet->slsframe[readoutId+i];
}
};
}

3
slsDetectorSoftware/slsDetector/slsDetector.h
View File

@ -1568,8 +1568,7 @@ class slsDetector : public slsDetectorUtils, public energyConversion {
*/
int resetFramesCaught();
/**
* Reads frames from receiver through a constant socket
/** Reads frames from receiver through a constant socket
*/
void readFrameFromReceiver();

3
slsDetectorSoftware/slsDetector/slsDetectorBase.h
View File

@ -506,8 +506,7 @@ class slsDetectorBase : public virtual slsDetectorDefs, public virtual errorDef
virtual int lockReceiver(int lock=-1)=0;
/**
* Reads frames from receiver through a constant socket
/** Reads frames from receiver through a constant socket
*/
virtual void readFrameFromReceiver()=0;

3
slsDetectorSoftware/slsDetector/slsDetectorUtils.cpp
View File

@ -157,6 +157,7 @@ int slsDetectorUtils::acquire(int delflag){
if (*threadedProcessing) {
if(dataReady)
sem_init(&dataThreadStartedSemaphore,1,0);
startThread(delflag);
}
@ -171,6 +172,7 @@ int slsDetectorUtils::acquire(int delflag){
if(*threadedProcessing && dataReady)
sem_wait(&dataThreadStartedSemaphore);
for(int im=0;im<nm;im++) {
#ifdef VERBOSE
@ -348,7 +350,6 @@ int slsDetectorUtils::acquire(int delflag){
pthread_mutex_lock(&mg);
stopReceiver();
pthread_mutex_unlock(&mg);
cout<<"receiver stopped"<<endl;
}

6
slsDetectorSoftware/slsDetector/slsDetectorUtils.h
View File

@ -640,8 +640,8 @@ virtual int getReceiverCurrentFrameIndex()=0;
*/
virtual int resetFramesCaught()=0;
/**
* Reads frames from receiver through a constant socket
/** Reads frames from receiver through a constant socket
*/
virtual void readFrameFromReceiver()=0;
@ -850,9 +850,11 @@ virtual int setReceiverFifoDepth(int i = -1)=0;
//individual sls and multi
sem_t sem_slsdone[MAXDET];
sem_t sem_slswait[MAXDET];
sem_t sem_multiwait[MAXDET];
int* slsframe[MAXDET];
};

32
slsDetectorSoftware/slsDetectorAnalysis/postProcessing.cpp
View File

@ -483,12 +483,36 @@ void* postProcessing::processData(int delflag) {
//receiver
else{
if(dataReady){
startdatathreads();
sem_post(&dataThreadStartedSemaphore);
}
readFrameFromReceiver();
}
//only update progress
else{
int caught = -1;
while(true){
cout.flush();
cout<<flush;
usleep(20000); //20ms need this else connecting error to receiver (too fast)
//get progress
if(setReceiverOnline() == ONLINE_FLAG){
pthread_mutex_lock(&mg);
caught = getFramesCaughtByReceiver();
pthread_mutex_unlock(&mg);
}
//updating progress
if(caught!= -1){
setCurrentProgress(caught);
#ifdef VERY_VERY_DEBUG
cout << "caught:" << caught << endl;
#endif
}
if (checkJoinThread()){
break;
}
}
}
cout<<"exiting from proccessing thread"<<endl;