almost done

slsDetectorSoftware/Makefile

@@ -18,6 +18,9 @@ INCLUDES?= -IcommonFiles -IslsDetector -I../slsReceiverSoftware/MySocketTCP -Ius
 SRC_CLNT=slsDetectorAnalysis/fileIO.cpp  usersFunctions/usersFunctions.cpp slsDetector/slsDetectorUtils.cpp slsDetector/slsDetectorCommand.cpp slsDetectorAnalysis/angularConversion.cpp slsDetectorAnalysis/angularConversionStatic.cpp slsDetectorAnalysis/energyConversion.cpp slsDetector/slsDetectorActions.cpp slsDetectorAnalysis/postProcessing.cpp slsDetector/slsDetector.cpp multiSlsDetector/multiSlsDetector.cpp slsDetectorAnalysis/postProcessingFuncs.cpp slsReceiverInterface/receiverInterface.cpp slsDetector/slsDetectorUsers.cpp threadFiles/CondVar.cpp  threadFiles/Mutex.cpp threadFiles/ThreadPool.cpp   #../slsReceiverSoftware/MySocketTCP/MySocketTCP.cpp 
 
 
+LIBZMQDIR = ../slsReceiverSoftware/include
+LIBZMQ =  -L$(LIBZMQDIR) -Wl,-rpath=$(LIBZMQDIR) -lzmq
+
 $(info )
 $(info #######################################)
 $(info #    Compiling slsDetectorSoftware    #)
@@ -66,14 +69,14 @@ gotthardVirtualServer:  $(SRC_MYTHEN_SVC)
 
 
 %.o : %.cpp %.h Makefile
-	$(CXX) -o $@ -c $< $(INCLUDES) $(DFLAGS) -fPIC  $(EPICSFLAGS) -lpthread #$(FLAGS)
+	$(CXX) -o $@ -c $< $(INCLUDES) $(DFLAGS) -fPIC  $(EPICSFLAGS) -lpthread  -lrt $(LIBZMQ) #$(FLAGS)
 
 
 package:  $(OBJS) $(DESTDIR)/libSlsDetector.so $(DESTDIR)/libSlsDetector.a 
 
 
 $(DESTDIR)/libSlsDetector.so: $(OBJS) 
-	$(CXX)  -shared -Wl,-soname,libSlsDetector.so -o libSlsDetector.so  $(OBJS)  -lc $(INCLUDES) $(DFLAGS) $(FLAGS) $(EPICSFLAGS)  -L/usr/lib64 -lpthread
+	$(CXX)  -shared -Wl,-soname,libSlsDetector.so -o libSlsDetector.so  $(OBJS)  -lc $(INCLUDES) $(DFLAGS) $(FLAGS) $(EPICSFLAGS)  -L/usr/lib64 -lpthread -lrt $(LIBZMQ)
 	$(shell test -d $(DESTDIR) || mkdir -p $(DESTDIR))
 	mv libSlsDetector.so $(DESTDIR)
 

slsDetectorSoftware/multiSlsDetector/multiSlsDetector.cpp

@@ -267,28 +267,30 @@ multiSlsDetector::multiSlsDetector(int id) :  slsDetectorUtils(), shmId(-1)
   getNMods();
   getMaxMods();
   threadpool = 0;
-	if(createThreadPool() == FAIL)
+  zmqthreadpool = 0;
+	if(createThreadPool(&threadpool) == FAIL)
 		exit(-1);
 
 }
 
 multiSlsDetector::~multiSlsDetector() {
   //removeSlsDetector();
-	destroyThreadPool();
+	destroyThreadPool(&threadpool);
+	destroyThreadPool(&zmqthreadpool);
 }
 
 
-int multiSlsDetector::createThreadPool(){
+int multiSlsDetector::createThreadPool(ThreadPool** t){
-	if(threadpool){
+	if(*t){
-		threadpool->destroy_threadpool();
+		(ThreadPool*)(*t)->destroy_threadpool();
-		threadpool=0;
+		*t=0;
 	}
 	if(thisMultiDetector->numberOfDetectors < 1){
 		cout << "No detectors attached to create threadpool" << endl;
 		return OK;
 	}
-	threadpool = new ThreadPool(thisMultiDetector->numberOfDetectors);
+	*t = new ThreadPool(thisMultiDetector->numberOfDetectors);
-	switch(threadpool->initialize_threadpool()){
+	switch(((ThreadPool*)(*t))->initialize_threadpool()){
 	case 0:
 		cerr << "Failed to initialize thread pool!" << endl;
 		return FAIL;
@@ -299,19 +301,19 @@ int multiSlsDetector::createThreadPool(){
 		break;
 	default:
 #ifdef VERBOSE
-		cout << "Initialized Threadpool" << endl;
+		cout << "Initialized Threadpool " << *t << endl;
 #endif
 		break;
 	}
 	return OK;
 }
 
-void multiSlsDetector::destroyThreadPool(){
+void multiSlsDetector::destroyThreadPool(ThreadPool** t){
-	if(threadpool){
+	if(*t){
-		threadpool->destroy_threadpool();
+		(ThreadPool*)(*t)->destroy_threadpool();
-		threadpool=0;
+		*t=0;
 #ifdef VERBOSE
-		cout<<"Destroyed Threadpool"<<endl;
+		cout<<"Destroyed Threadpool "<< *t << endl;
 #endif
 	}
 }
@@ -392,8 +394,7 @@ int multiSlsDetector::addSlsDetector(int id, int pos) {
 
   //set offsets
   updateOffsets();
-  destroyThreadPool();
+  if(createThreadPool(&threadpool) == FAIL)
-  if(createThreadPool() == FAIL)
 	exit(-1);
 
 
@@ -864,8 +865,7 @@ int multiSlsDetector::removeSlsDetector(int pos) {
   }
 
   updateOffsets();
-  destroyThreadPool();
+  if(createThreadPool(&threadpool) == FAIL)
-  if(createThreadPool() == FAIL)
 	exit(-1);
 
   return thisMultiDetector->numberOfDetectors;
@@ -1196,6 +1196,7 @@ slsDetectorDefs::detectorSettings multiSlsDetector::getSettings(int pos) {
 				threadpool->add_task(task);
 			}
 		}
+		threadpool->startExecuting();
 		threadpool->wait_for_tasks_to_complete();
 		for(int idet=posmin; idet<posmax; idet++){
 			if(detectors[idet]){
@@ -1243,6 +1244,7 @@ slsDetectorDefs::detectorSettings multiSlsDetector::setSettings(detectorSettings
 				threadpool->add_task(task);
 			}
 		}
+		threadpool->startExecuting();
 		threadpool->wait_for_tasks_to_complete();
 		for(int idet=posmin; idet<posmax; idet++){
 			if(detectors[idet]){
@@ -1626,6 +1628,7 @@ int multiSlsDetector::startAndReadAllNoWait(){
 				threadpool->add_task(task);
 			}
 		}
+		threadpool->startExecuting();
 		threadpool->wait_for_tasks_to_complete();
 		for(int idet=posmin; idet<posmax; idet++){
 			if((idet!=thisMultiDetector->masterPosition) && (detectors[idet])){
@@ -3318,6 +3321,7 @@ char* multiSlsDetector::setNetworkParameter(networkParameter p, string s){
 					threadpool->add_task(task);
 				}
 			}
+			threadpool->startExecuting();
 			threadpool->wait_for_tasks_to_complete();
 			for(int idet=0; idet<thisMultiDetector->numberOfDetectors; idet++){
 				if(detectors[idet]){
@@ -3926,6 +3930,7 @@ int multiSlsDetector::executeTrimming(trimMode mode, int par1, int par2, int imo
 					threadpool->add_task(task);
 				}
 			}
+			threadpool->startExecuting();
 			threadpool->wait_for_tasks_to_complete();
 			for(int idet=0; idet<thisMultiDetector->numberOfDetectors; idet++){
 				if(detectors[idet]){
@@ -3977,6 +3982,7 @@ int multiSlsDetector::loadSettingsFile(string fname, int imod) {
 				  threadpool->add_task(task);
 			  }
 		  }
+			threadpool->startExecuting();
 		  threadpool->wait_for_tasks_to_complete();
 		  for(int idet=0; idet<thisMultiDetector->numberOfDetectors; idet++){
 			  if(detectors[idet]){
@@ -4051,6 +4057,7 @@ int multiSlsDetector::setAllTrimbits(int val, int imod){
 					threadpool->add_task(task);
 				}
 			}
+			threadpool->startExecuting();
 			threadpool->wait_for_tasks_to_complete();
 			for(int idet=0; idet<thisMultiDetector->numberOfDetectors; idet++){
 				if(detectors[idet]){
@@ -4100,6 +4107,7 @@ int multiSlsDetector::loadCalibrationFile(string fname, int imod) {
 					threadpool->add_task(task);
 				}
 			}
+			threadpool->startExecuting();
 			threadpool->wait_for_tasks_to_complete();
 			for(int idet=0; idet<thisMultiDetector->numberOfDetectors; idet++){
 				if(detectors[idet]){
@@ -4752,6 +4760,7 @@ int multiSlsDetector::startReceiver(){
 				threadpool->add_task(task);
 			}
 		}
+		threadpool->startExecuting();
 		threadpool->wait_for_tasks_to_complete();
 		for(int idet=posmin; idet<posmax; idet++){
 			if((idet!=thisMultiDetector->masterPosition) && (detectors[idet])){
@@ -4813,6 +4822,7 @@ int multiSlsDetector::stopReceiver(){
 				threadpool->add_task(task);
 			}
 		}
+		threadpool->startExecuting();
 		threadpool->wait_for_tasks_to_complete();
 		for(int idet=posmin; idet<posmax; idet++){
 			if((idet!=thisMultiDetector->masterPosition) && (detectors[idet])){
@@ -4948,7 +4958,173 @@ int multiSlsDetector::resetFramesCaught() {
 
 
 
-int* multiSlsDetector::readFrameFromReceiver(char* fName,  int &acquisitionIndex, int &frameIndex, int &subFrameIndex){
+void multiSlsDetector::readFrameFromReceiver(){
+	int value;
+	if(createThreadPool(&zmqthreadpool) == FAIL){
+		cprintf(BG_RED,"Error: Could not create the zmq threads\n");
+		return;
+	}
+	zmqthreadpool->setzeromqThread();
+
+	//determine number of half readouts and maxX and maxY
+	int maxX=0,maxY=0;
+	int numReadout = 1;
+	if(getDetectorsType() == EIGER){
+		numReadout = 2;
+		maxX = thisMultiDetector->numberOfChannel[X];
+		maxY = thisMultiDetector->numberOfChannel[Y];
+	}
+
+	//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;
+	int slsdatabytes = 0, slsmaxchannels = 0, bytesperchannel = 0, slsmaxX = 0, slsmaxY=0;
+	if(!zmqthreadpool){
+		cout << "Error in creating threadpool. Exiting" << endl;
+		return;
+	}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);
+				}
+				Task* task = new Task(new func00_t<void, slsDetector>(&slsDetector::readFrameFromReceiver,detectors[idet]));
+				zmqthreadpool->add_task(task);
+				if(!slsdatabytes){
+					slsdatabytes = detectors[idet]->getDataBytes();
+					slsmaxchannels = detectors[idet]->getMaxNumberOfChannels();
+					bytesperchannel = slsdatabytes/slsmaxchannels;
+					slsmaxX = detectors[idet]->getTotalNumberOfChannels(X);
+					slsmaxY = detectors[idet]->getTotalNumberOfChannels(Y);
+				}
+				//set mask
+				runningMask|=(1<<(idet*numReadout));
+				if(numReadout>1)
+					runningMask|=(1<<(idet*numReadout+1));
+
+			}
+		}
+	}
+
+	zmqthreadpool->startExecuting();//tell them to start
+
+
+	int nel=(thisMultiDetector->dataBytes)/sizeof(int);
+
+	if(nel <= 0){
+		cout << "Multislsdetector databytes not valid :" << thisMultiDetector->dataBytes << endl;
+		return;
+	}
+	int* multiframe=new int[nel];
+	int* p = multiframe;
+	int idet,offsetY,offsetX;
+	int halfreadoutoffset =  (slsmaxX/numReadout);
+	//after reconstruction
+	int framecount=0;
+	int nx =getTotalNumberOfChannels(slsDetectorDefs::X);
+	int ny =getTotalNumberOfChannels(slsDetectorDefs::Y);
+
+
+
+	while(true){
+		memset(((char*)multiframe),0x0,slsdatabytes*thisMultiDetector->numberOfDetectors);
+
+		for(int ireadout=0; ireadout<thisMultiDetector->numberOfDetectors*numReadout; ++ireadout){
+			idet = ireadout/numReadout;
+
+			if(detectors[idet]){
+				if((1 << ireadout) & runningMask){
+
+
+					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){
+							break;
+						}
+						continue;
+					}
+
+
+					//assemble data
+					//eiger, so interleaving
+					if(maxX){
+
+						if(ireadout == 3){
+							offsetY = (maxY - (thisMultiDetector->offsetY[idet] + slsmaxY)) * maxX * bytesperchannel;
+							if(!(ireadout%numReadout)) 	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
+							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);
+
+							}
+
+
+						}//end of ireadout
+
+					}
+					//no interleaving, just add to the end
+					//numReadout always 1 here
+					else{
+						memcpy(p,multiframe,slsdatabytes);
+						p+=slsdatabytes/sizeof(int);
+					}
+
+
+
+				}
+			}
+		}//end of for loop
+
+		if(!runningMask){
+			break;
+		}
+
+		//send data to callback
+
+		fdata = decodeData(multiframe);
+		if ((fdata) && (dataReady)){
+			thisData = new detectorData(fdata,NULL,NULL,getCurrentProgress(),"noname.raw",nx,ny);
+			dataReady(thisData, framecount, framecount, pCallbackArg);//should be fnum and subfnum from json header
+			delete thisData;
+			fdata = NULL;
+			cout<<"Send frame #"<< framecount << " to gui"<<endl;
+		}
+
+		framecount++;
+		setCurrentProgress(framecount);
+
+	}
+
+	zmqthreadpool->wait_for_tasks_to_complete();
+	destroyThreadPool(&zmqthreadpool);
+	delete[] multiframe;
+
+
+
+
+
+
+
+
+/*
 	int nel=(thisMultiDetector->dataBytes)/sizeof(int);
 	if(nel <= 0){
 		cout << "Multislsdetector databytes not valid :" << thisMultiDetector->dataBytes << endl;
@@ -5039,6 +5215,7 @@ int* multiSlsDetector::readFrameFromReceiver(char* fName,  int &acquisitionIndex
 	if((getDetectorsType() == EIGER) &&(complete ==FAIL))
 		acquisitionIndex = -1;
 	return retval;
+	*/
 };
 
 
@@ -5468,6 +5645,7 @@ int multiSlsDetector::pulsePixel(int n,int x,int y) {
 				threadpool->add_task(task);
 			}
 		}
+		threadpool->startExecuting();
 		threadpool->wait_for_tasks_to_complete();
 		for(int idet=0; idet<thisMultiDetector->numberOfDetectors; idet++){
 			if(detectors[idet]){
@@ -5504,6 +5682,7 @@ int multiSlsDetector::pulsePixelNMove(int n,int x,int y) {
 				threadpool->add_task(task);
 			}
 		}
+		threadpool->startExecuting();
 		threadpool->wait_for_tasks_to_complete();
 		for(int idet=0; idet<thisMultiDetector->numberOfDetectors; idet++){
 			if(detectors[idet]){
@@ -5540,6 +5719,7 @@ int multiSlsDetector::pulseChip(int n) {
 				threadpool->add_task(task);
 			}
 		}
+		threadpool->startExecuting();
 		threadpool->wait_for_tasks_to_complete();
 		for(int idet=0; idet<thisMultiDetector->numberOfDetectors; idet++){
 			if(detectors[idet]){

slsDetectorSoftware/multiSlsDetector/multiSlsDetector.h

@@ -245,10 +245,10 @@ class multiSlsDetector  : public slsDetectorUtils {
    * Creates all the threads in the threadpool
     \returns OK or FAIL
   */
-  int createThreadPool();
+  int createThreadPool(ThreadPool** t);
 
   /** destroys all the threads in the threadpool */
-  void destroyThreadPool();
+  void destroyThreadPool(ThreadPool** t);
   
   /** frees the shared memory occpied by the sharedMultiSlsDetector structure */
   int freeSharedMemory() ;
@@ -1180,14 +1180,9 @@ class multiSlsDetector  : public slsDetectorUtils {
  int resetFramesCaught();
 
  /**
-  * Reads a frame from receiver
+  * Reads frames from receiver through a constant socket
-  * @param fName file name of current frame()
-  * @param acquisitionIndex current acquisition index
-  * @param frameIndex current frame index (for each scan)
-  * @param subFrameIndex current sub frame index (for 32 bit mode for eiger)
-  /returns a frame read from recever
  */
- int* readFrameFromReceiver(char* fName, int &acquisitionIndex, int &frameIndex, int &subFrameIndex);
+ void readFrameFromReceiver();
 
   /** Locks/Unlocks the connection to the receiver
       /param lock sets (1), usets (0), gets (-1) the lock
@@ -1378,7 +1373,7 @@ class multiSlsDetector  : public slsDetectorUtils {
 
  private:
   ThreadPool* threadpool;
-
+  ThreadPool* zmqthreadpool;
 
 
 };

slsDetectorSoftware/slsDetector/slsDetector.cpp

@@ -10,7 +10,7 @@
 #include <cstdlib>
 #include <math.h>
 #include "gitInfoLib.h"
-
+#include <zmq.h>
 
 int slsDetector::initSharedMemory(detectorType type, int id) {
 
@@ -7145,58 +7145,125 @@ int slsDetector::resetFramesCaught(){
 
 
 
-int* slsDetector::readFrameFromReceiver(char* fName,  int &acquisitionIndex, int &frameIndex, int &subFrameIndex){
+void slsDetector::readFrameFromReceiver(){
-	int fnum=F_READ_RECEIVER_FRAME;
-	int nel=thisDetector->dataBytes/sizeof(int);
-	int* retval=new int[nel];
-	int ret=FAIL;
-	int n;
-	char mess[MAX_STR_LENGTH]="Nothing";
 
-	if (setReceiverOnline(ONLINE_FLAG)==ONLINE_FLAG) {
+	//determine number of half readouts
-#ifdef VERBOSE
+	int numReadout = 1;
-		std::cout<< "slsDetector: Reading frame from receiver "<< thisDetector->dataBytes << " " <<nel <<std::endl;
+	if(thisDetector->myDetectorType == EIGER)	numReadout = 2;
-#endif
+	int readoutId = detId*numReadout;
-		if (connectData() == OK){
+	volatile uint64_t runningMask = 0x0;
-			dataSocket->SendDataOnly(&fnum,sizeof(fnum));
-			dataSocket->ReceiveDataOnly(&ret,sizeof(ret));
 
-			if (ret==FAIL) {
+	//server details
-				n= dataSocket->ReceiveDataOnly(mess,sizeof(mess));
+	char hostname[numReadout][100];
-				std::cout<< "Detector returned: " << mess << " " << n << std::endl;
+	int portno[numReadout];
-				delete [] retval;
+	int nel=(thisDetector->dataBytes/numReadout)/sizeof(int);
-				disconnectData();
+	for(int i=0;i<numReadout;++i){
-				return NULL;
+		portno[i] = DEFAULT_ZMQ_PORTNO + (readoutId+i);
-			} else {
+		sprintf(hostname[i], "%s%d", "tcp://127.0.0.1:",portno[i]);
-				n=dataSocket->ReceiveDataOnly(fName,MAX_STR_LENGTH);
+		//cout << "ZMQ Client of " << readoutId+i << " at " << hostname[i] << endl;
-				n=dataSocket->ReceiveDataOnly(&acquisitionIndex,sizeof(acquisitionIndex));
-				n=dataSocket->ReceiveDataOnly(&frameIndex,sizeof(frameIndex));
-				if(thisDetector->myDetectorType == EIGER)
-					n=dataSocket->ReceiveDataOnly(&subFrameIndex,sizeof(subFrameIndex));
-				n=dataSocket->ReceiveDataOnly(retval,thisDetector->dataBytes);
 
-#ifdef VERBOSE
+		parentDet->slsframe[readoutId+i]=new int[nel];
-				std::cout<< "Received "<< n << " data bytes" << std::endl;
-#endif
-				if (n!=thisDetector->dataBytes) {
-					std::cout<<endl<< "wrong data size received: received " << n << " but expected from receiver " << thisDetector->dataBytes << std::endl;
-					ret=FAIL;
-					delete [] retval;
-					disconnectData();
-					return NULL;
 	}
 
+
+	//loop though the half readouts to start sockets
+	void *context[numReadout];
+	void *zmqsocket[numReadout];
+	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]);
+
+		runningMask|=(1<<(i));
+
+	}
+
+
+	//receive msgs and let multi know
+	zmq_msg_t message;
+	int len,idet = 0;
+	int framecount=0;
+
+
+	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
+
+				// 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);
+				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
+
+					runningMask^=(1<<idet);
+					//all done, get out
+					if(!runningMask){
+						break;
+					}
+					continue;
+				}
+
+				cout<<"Received on " << readoutId+idet << " for frame " << framecount << endl;
+
+				if(zmq_msg_data(&message)==NULL)
+						cprintf(RED,"GOT NULL FROM ZMQ\n");
+				//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]);
+
 					//jungfrau masking adcval
 					if(thisDetector->myDetectorType == JUNGFRAU){
 						for(unsigned int i=0;i<nel;i++){
-						retval[i] = (retval[i] & 0x3FFF3FFF);
+							parentDet->slsframe[readoutId+idet][i] = (parentDet->slsframe[readoutId+idet][i] & 0x3FFF3FFF);
 						}
 					}
+				//}
+				sem_post(&parentDet->sem_slsdone[readoutId+idet]);//let multi know is ready
+
 			}
-			disconnectData();
+		}//end of for loop
+
+		if(!runningMask){
+			break;
 		}
+
 	}
-	return retval;
+	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];
+	}
+
 };
 
 

slsDetectorSoftware/slsDetector/slsDetector.h

@@ -1569,14 +1569,9 @@ class slsDetector : public slsDetectorUtils, public energyConversion {
  int resetFramesCaught();
 
  /**
-  * Reads a frame from receiver
+  * Reads frames from receiver through a constant socket
-  * @param fName file name of current frame()
-  * @param acquisitionIndex current acquisition index
-  * @param frameIndex current frame index (for each scan)
-  * @param subFrameIndex current sub frame index (for 32 bit mode for eiger)
-  /returns a frame read from recever
  */
- int* readFrameFromReceiver(char* fName, int &acquisitionIndex, int &frameIndex, int &subFrameIndex);
+ void readFrameFromReceiver();
 
   /** Locks/Unlocks the connection to the receiver
       /param lock sets (1), usets (0), gets (-1) the lock

slsDetectorSoftware/slsDetector/slsDetectorBase.h

@@ -507,14 +507,9 @@ class slsDetectorBase :  public virtual slsDetectorDefs, public virtual errorDef
 
 
   /**
-   * Reads a frame from receiver
+   * Reads frames from receiver through a constant socket
-   * @param fName file name of current frame()
-   * @param acquisitionIndex current acquisition index
-   * @param frameIndex current frame index (for each scan)
-   * @param subFrameIndex current sub frame index (for 32 bit mode for eiger)
-   /returns a frame read from recever
   */
-  virtual int* readFrameFromReceiver(char* fName, int &acquisitionIndex, int &frameIndex, int &subFrameIndex)=0;
+  virtual void readFrameFromReceiver()=0;
 
 
   /** Sets the read receiver frequency

slsDetectorSoftware/slsDetector/slsDetectorUtils.cpp

@@ -72,10 +72,6 @@ int  slsDetectorUtils::acquire(int delflag){
 
   int multiframe = nc*nf;
 
-  pthread_mutex_lock(&mg);
-  acquiringDone = 0;
-  pthread_mutex_unlock(&mg);
-
   // setTotalProgress();
   //moved these 2 here for measurement change
   progressIndex=0;
@@ -162,7 +158,6 @@ int  slsDetectorUtils::acquire(int delflag){
 
 
   if (*threadedProcessing) {
-	sem_init(&sem_queue,0,0);
     startThread(delflag);
   }
 #ifdef VERBOSE
@@ -341,48 +336,13 @@ int  slsDetectorUtils::acquire(int delflag){
 
 	  //offline
 	  if(setReceiverOnline()==OFFLINE_FLAG){
-	  // wait until data processing thread has finished the data
-		  pthread_mutex_lock(&mg);
-		  acquiringDone = 1;
-		  pthread_mutex_unlock(&mg);
-		  if (*threadedProcessing) {
-			  sem_wait(&sem_queue);
-			  pthread_mutex_lock(&mg);
-			  acquiringDone = 0;
-			  pthread_mutex_unlock(&mg);
-		  }
-
-#ifdef VERBOSE
-	  cout << "check data queue size " << endl;
-#endif
-	  /*while (dataQueueSize()){
-	    //#ifdef VERBOSE
-	    cout << "AAAAAAAAA check data queue size " << endl;
-	    //#endif
-	    usleep(100000);
-	  }*/
-
 		  if ((getDetectorsType()==GOTTHARD) || (getDetectorsType()==MOENCH) || (getDetectorsType()==JUNGFRAU) ){
 			  if((*correctionMask)&(1<<WRITE_FILE))
 				  closeDataFile();
 		  }
-
 	  }
-
-
 	  //online
 	  else{
-		  pthread_mutex_lock(&mg);
-		  acquiringDone = 1;
-		  pthread_mutex_unlock(&mg);
-
-		  // wait until data processing thread has taken the last frame
-		  if (*threadedProcessing) {
-			  sem_wait(&sem_queue);
-			  pthread_mutex_lock(&mg);
-			  acquiringDone = 0;
-			  pthread_mutex_unlock(&mg);
-		  }
 		  pthread_mutex_lock(&mg);
 		  stopReceiver();
 		  pthread_mutex_unlock(&mg);
@@ -391,7 +351,6 @@ int  slsDetectorUtils::acquire(int delflag){
 
 
 
-
 	  pthread_mutex_lock(&mp);
 	  if (*stoppedFlag==0) {
 	    executeAction(headerAfter);
@@ -506,7 +465,6 @@ int  slsDetectorUtils::acquire(int delflag){
 #endif
     setJoinThread(1);
     pthread_join(dataProcessingThread, &status);
-    sem_destroy(&sem_queue);
 #ifdef VERBOSE
     cout << "data processing thread joined" << endl;
 #endif

slsDetectorSoftware/slsDetector/slsDetectorUtils.h

@@ -29,7 +29,7 @@ extern "C" {
 #include <sstream>
 #include <queue>
 #include <math.h>
-
+#include <semaphore.h>
 using namespace std;
 
 
@@ -641,14 +641,9 @@ virtual int getReceiverCurrentFrameIndex()=0;
 virtual int resetFramesCaught()=0;
 
 /**
- * Reads a frame from receiver
+ * Reads frames from receiver through a constant socket
- * @param fName file name of current frame()
- * @param acquisitionIndex current acquisition index
- * @param frameIndex current frame index (for each scan)
- * @param subFrameIndex current sub frame index (for 32 bit mode for eiger)
- /returns a frame read from recever
 */
-virtual int* readFrameFromReceiver(char* fName, int &acquisitionIndex, int &frameIndex, int &subFrameIndex)=0;
+virtual void readFrameFromReceiver()=0;
 
 
 /**
@@ -850,6 +845,14 @@ virtual int setReceiverFifoDepth(int i = -1)=0;
   int (*progress_call)(double,void*);
   void *pProgressCallArg;
   
+  public:
+  //data call back
+  //individual sls and multi
+  sem_t sem_slsdone[MAXDET];
+  sem_t sem_slswait[MAXDET];
+  int* slsframe[MAXDET];
+
+
 };
 
 

slsDetectorSoftware/slsDetectorAnalysis/postProcessing.cpp

@@ -7,6 +7,7 @@
 #include "usersFunctions.h"
 #endif
 
+
 //#define VERBOSE
 
 static void* startProcessData(void *n){
@@ -461,12 +462,7 @@ void* postProcessing::processData(int delflag) {
 			
 			
 			}
-				/** IF detector acquisition is done, let the acquire() thread know to finish up and force join thread */
+
-			if(acquiringDone){
-			  sem_post(&sem_queue);
-			  //	cout << "Sem posted" << endl;
-			} //else
-				// cout << "Sem not posted" << endl;
 			/* IF THERE ARE NO DATA look if acquisition is finished */
 			if (checkJoinThread()) {
 			  if (dataQueueSize()==0) {
@@ -488,6 +484,14 @@ void* postProcessing::processData(int delflag) {
 	else{
 
 
+
+		readFrameFromReceiver();
+
+
+
+
+
+/*
 		int progress = 0;
 		char currentfName[MAX_STR_LENGTH]="";
 		int caught = -1;
@@ -663,6 +667,8 @@ void* postProcessing::processData(int delflag) {
 			}
 
 		}
+*/
+
 	}
 
 	return 0;

slsDetectorSoftware/slsDetectorAnalysis/postProcessing.h

@@ -17,7 +17,7 @@
 #include <sstream>
 #include <queue>
 #include <math.h>
-#include <semaphore.h>
+
 
 class postProcessingFuncs;
 
@@ -300,11 +300,7 @@ s
  /** data queue size */
   int queuesize;
 
-  /** queue mutex */
-  sem_t sem_queue;
 
-  /** set when detector finishes acquiring */
-  int acquiringDone;
 
 
   /**
@@ -333,18 +329,23 @@ s
 
 
 
- private:
   double *fdata;
-  
   int (*dataReady)(detectorData*,int, int,void*);
   void *pCallbackArg; 
+  detectorData *thisData;
+  
+ private:
+ // double *fdata;
+
+ // int (*dataReady)(detectorData*,int, int,void*);
+ // void *pCallbackArg;
 
   int (*rawDataReady)(double*,int,void*); 
   void *pRawDataArg; 
   
   
   postProcessingFuncs *ppFun;
-  detectorData *thisData;
+  //detectorData *thisData;
   
 
   double *ang;
@@ -374,4 +375,5 @@ s
 
 
 
+
 #endif

slsDetectorSoftware/threadFiles/Task.h

@@ -16,6 +16,18 @@ using namespace std;
 
 class slsDetector;
 
+template<typename _Ret, typename _Class>
+class func00_t{
+public:
+	func00_t(_Ret (_Class::*fn)(),_Class* ptr):
+		m_fn(fn),m_ptr(ptr){}
+	~func00_t() {}
+	void operator()() const {((m_ptr->*m_fn)());}
+private:
+	_Class* m_ptr;
+	_Ret (_Class::*m_fn)();
+};
+
 template<typename _Ret, typename _Class, typename _Store>
 class func0_t{
 public:
@@ -94,27 +106,32 @@ private:
 class Task:  public virtual slsDetectorDefs{
 public:
 	/* Return: int, Param: int */
-	Task(func1_t <int,slsDetector,int,int>* t):									m1(t),m2(0),m3(0),m4(0),m5(0),m6(0),m7(0),m8(0),m9(0),m10(0),m11(0){};
+	Task(func1_t <int,slsDetector,int,int>* t):									m1(t),m2(0),m3(0),m4(0),m5(0),m6(0),m7(0),m8(0),m9(0),m10(0),m11(0),m12(0){};
 	/* Return: int, Param: string,int */
-	Task(func2_t <int,slsDetector,string,int,int>* t):							m1(0),m2(t),m3(0),m4(0),m5(0),m6(0),m7(0),m8(0),m9(0),m10(0),m11(0){};
+	Task(func2_t <int,slsDetector,string,int,int>* t):							m1(0),m2(t),m3(0),m4(0),m5(0),m6(0),m7(0),m8(0),m9(0),m10(0),m11(0),m12(0){};
 	/* Return: string, Param: string */
-	Task(func1_t <string,slsDetector,string,string>* t):						m1(0),m2(0),m3(t),m4(0),m5(0),m6(0),m7(0),m8(0),m9(0),m10(0),m11(0){};
+	Task(func1_t <string,slsDetector,string,string>* t):						m1(0),m2(0),m3(t),m4(0),m5(0),m6(0),m7(0),m8(0),m9(0),m10(0),m11(0),m12(0){};
 	/* Return: char*, Param: char* */
-	Task(func1_t <char*,slsDetector,char*,string>* t):							m1(0),m2(0),m3(0),m4(t),m5(0),m6(0),m7(0),m8(0),m9(0),m10(0),m11(0){};
+	Task(func1_t <char*,slsDetector,char*,string>* t):							m1(0),m2(0),m3(0),m4(t),m5(0),m6(0),m7(0),m8(0),m9(0),m10(0),m11(0),m12(0){};
 	/* Return: detectorSettings, Param: int */
-	Task(func1_t <detectorSettings,slsDetector,int,int>* t):					m1(0),m2(0),m3(0),m4(0),m5(t),m6(0),m7(0),m8(0),m9(0),m10(0),m11(0){};
+	Task(func1_t <detectorSettings,slsDetector,int,int>* t):					m1(0),m2(0),m3(0),m4(0),m5(t),m6(0),m7(0),m8(0),m9(0),m10(0),m11(0),m12(0){};
 	/* Return: detectorSettings, Param: detectorSettings,int */
-	Task(func2_t <detectorSettings,slsDetector,detectorSettings,int,int>* t):	m1(0),m2(0),m3(0),m4(0),m5(0),m6(t),m7(0),m8(0),m9(0),m10(0),m11(0){};
+	Task(func2_t <detectorSettings,slsDetector,detectorSettings,int,int>* t):	m1(0),m2(0),m3(0),m4(0),m5(0),m6(t),m7(0),m8(0),m9(0),m10(0),m11(0),m12(0){};
 	/* Return: int, Param: int,int */
-	Task(func2_t <int,slsDetector,int,int,int>* t):	 							m1(0),m2(0),m3(0),m4(0),m5(0),m6(0),m7(t),m8(0),m9(0),m10(0),m11(0){};
+	Task(func2_t <int,slsDetector,int,int,int>* t):	 							m1(0),m2(0),m3(0),m4(0),m5(0),m6(0),m7(t),m8(0),m9(0),m10(0),m11(0),m12(0){};
 	/* Return: int, Param: int,int */
-	Task(func3_t <int,slsDetector,int,int,int,int>* t): 						m1(0),m2(0),m3(0),m4(0),m5(0),m6(0),m7(0),m8(t),m9(0),m10(0),m11(0){};
+	Task(func3_t <int,slsDetector,int,int,int,int>* t): 						m1(0),m2(0),m3(0),m4(0),m5(0),m6(0),m7(0),m8(t),m9(0),m10(0),m11(0),m12(0){};
 	/* Return: int, Param: trimMode,int,int,int */
-	Task(func4_t <int,slsDetector,trimMode,int,int,int,int>* t):				m1(0),m2(0),m3(0),m4(0),m5(0),m6(0),m7(0),m8(0),m9(t),m10(0),m11(0){};
+	Task(func4_t <int,slsDetector,trimMode,int,int,int,int>* t):				m1(0),m2(0),m3(0),m4(0),m5(0),m6(0),m7(0),m8(0),m9(t),m10(0),m11(0),m12(0){};
-	/* Return: int, Param: int */
+	/* Return: int, Param: none */
-	Task(func0_t <int,slsDetector,int>* t):										m1(0),m2(0),m3(0),m4(0),m5(0),m6(0),m7(0),m8(0),m9(0),m10(t),m11(0){};
+	Task(func0_t <int,slsDetector,int>* t):										m1(0),m2(0),m3(0),m4(0),m5(0),m6(0),m7(0),m8(0),m9(0),m10(t),m11(0),m12(0){};
 	/* Return: char*, Param: networkParameter,string,string */
-	Task(func2_t <char*,slsDetector,networkParameter,string,string>* t):		m1(0),m2(0),m3(0),m4(0),m5(0),m6(0),m7(0),m8(0),m9(0),m10(0),m11(t){};
+	Task(func2_t <char*,slsDetector,networkParameter,string,string>* t):		m1(0),m2(0),m3(0),m4(0),m5(0),m6(0),m7(0),m8(0),m9(0),m10(0),m11(t),m12(0){};
+	/* Return: void, Param: none */
+	Task(func00_t <void,slsDetector>* t):										m1(0),m2(0),m3(0),m4(0),m5(0),m6(0),m7(0),m8(0),m9(0),m10(0),m11(0),m12(t){};
+
+
+
 	~Task(){}
 
 	void operator()(){
@@ -129,6 +146,7 @@ public:
 		else if(m9)		(*m9)();
 		else if(m10)	(*m10)();
 		else if(m11)	(*m11)();
+		else if(m12)	(*m12)();
 	}
 
 private:
@@ -154,6 +172,8 @@ private:
 	func0_t <int,slsDetector,int>* m10;
 	/* Return: char*, Param: networkParameter,string,string */
 	func2_t <char*,slsDetector,networkParameter,string,string>* m11;
+	/* Return: void, Param: none */
+	func00_t <void,slsDetector>* m12;
 };
 
 

slsDetectorSoftware/threadFiles/ThreadPool.cpp

@@ -7,8 +7,10 @@ ThreadPool::ThreadPool(int pool_size) : m_pool_size(pool_size){
 #endif
 	m_tasks_loaded = false;
 	thread_started = false;
+	zmqthreadpool = false;
 	current_thread_number = -1;
 	number_of_ongoing_tasks = 0;
+	number_of_total_tasks = 0;
 }
 
 ThreadPool::~ThreadPool(){
@@ -34,6 +36,7 @@ int ThreadPool::initialize_threadpool(){
 	m_pool_state = STARTED;
 	int ret = -1;
 	sem_init(&semStart,1,0);
+	sem_init(&semDone,1,0);
 	for (int i = 0; i < m_pool_size; i++) {
 		pthread_t tid;
 		thread_started = false;
@@ -68,12 +71,15 @@ int ThreadPool::destroy_threadpool(){
 	for (int i = 0; i < m_pool_size; i++) {
 		void* result;
 		sem_post(&semStart);
+		sem_post(&semDone);
 		ret = pthread_join(m_threads[i], &result);
 		/*cout << "pthread_join() returned " << ret << ": " << strerror(errno) << endl;*/
 		m_task_cond_var.broadcast(); // try waking up a bunch of threads that are still waiting
 	}
 	sem_destroy(&semStart);
+	sem_destroy(&semDone);
 	number_of_ongoing_tasks = 0;
+	number_of_total_tasks = 0;
 	/* cout << m_pool_size << " threads exited from the thread pool" << endl;*/
 	return 0;
 }
@@ -109,25 +115,28 @@ void* ThreadPool::execute_thread(){
 		/*cout << "Unlocking: " << pthread_self() << endl;*/
 		m_task_mutex.unlock();
 
+		//if(zmqthreadpool) cout<<"***"<<ithread<<" semaphore start address wait:"<<&semStart<<endl;
 		sem_wait(&semStart);
 
+		//cout<<"***"<<ithread<<" checking out semaphore done address:"<<&semDone<<endl;
+
 		/*cout << ithread <<" Executing thread " << pthread_self() << endl;*/
 		// execute the task
 		(*task)(); // could also do task->run(arg);
 		/*cout << ithread <<" Done executing thread " << pthread_self() << endl;*/
 
-		m_all_tasks_mutex.lock();
+		m_task_mutex.lock();
 		number_of_ongoing_tasks--;
-		m_all_tasks_mutex.unlock();
+		m_task_mutex.unlock();
 
+		//if(zmqthreadpool) cout<<ithread <<" task  done: "<<number_of_ongoing_tasks<<endl;
 		//last task and check m_tasks_loaded to ensure done only once
 		if((!number_of_ongoing_tasks) && m_tasks_loaded){
-			/*cout << ithread << " all tasks done."<<endl;*/
+			//if(zmqthreadpool) cout << ithread << " all tasks done."<<endl;
-			m_all_tasks_mutex.lock();
 			m_tasks_loaded = false;
-			m_all_tasks_cond_var.signal();// wake up thread that is waiting for all tasks to be complete
-			m_all_tasks_mutex.unlock();
 		}
+		//if(zmqthreadpool) cout<<"***"<<ithread<<" semaphore done address post:"<<&semDone<<endl;
+		sem_post(&semDone);
 		delete task;
 		/*cout << ithread << " task deleted" << endl;*/
 	}
@@ -143,28 +152,37 @@ int ThreadPool::add_task(Task* task){
 	// TODO: put a limit on how many tasks can be added at most
 	m_tasks.push_back(task);
 	number_of_ongoing_tasks++;
+	number_of_total_tasks++;
 	m_task_cond_var.signal(); // wake up one thread that is waiting for a task to be available
 	m_task_mutex.unlock();
 	return 0;
 }
 
-void ThreadPool::wait_for_tasks_to_complete(){
+void ThreadPool::startExecuting(){
 	if(m_pool_size == 1)
 			return;
-	m_all_tasks_mutex.lock();
+
 		/*cout << "waiting for tasks: locked. gonna wait" << endl;*/
 		m_tasks_loaded = true;
 
 		//giving all threads permission to start as all tasks have been added
-	//using a different variable as number_of_ongoing_tasks is likely to change during the loop
+		//if(zmqthreadpool) cout<<"*** semaphore start address post:"<<&semStart<<endl;
-	int totalnumtasks = number_of_ongoing_tasks;
+		for(int i=0;i<number_of_total_tasks;i++)
-	for(int i=0;i<totalnumtasks;i++)
 			sem_post(&semStart);
 
-	while ((m_pool_state != STOPPED) && m_tasks_loaded) {
-	m_all_tasks_cond_var.wait(m_all_tasks_mutex.get_mutex_ptr());
-	}
-	/*cout << "waiting for tasks:totall out, must be locked again" << endl;*/
-	m_all_tasks_mutex.unlock();
 }
 
+void ThreadPool::wait_for_tasks_to_complete(){
+	if(m_pool_size == 1)
+		return;
+	//if(zmqthreadpool) cout<<"waiting for all tasks to be done "<<endl;
+	//if(zmqthreadpool) cout<<"*** semaphore done address wait:"<<&semDone<<endl;
+	for(int i=0;i<number_of_total_tasks;i++)
+		sem_wait(&semDone);
+	number_of_total_tasks = 0;
+	//if(zmqthreadpool) cout<<"complete"<<endl<<endl;
+}
+
+void ThreadPool::setzeromqThread(){
+	zmqthreadpool = true;
+}

slsDetectorSoftware/threadFiles/ThreadPool.h

@@ -26,7 +26,9 @@ public:
   int destroy_threadpool();
   void* execute_thread();
   int add_task(Task* task);
+  void startExecuting();
   void wait_for_tasks_to_complete();
+  void setzeromqThread();
 
 private:
   int m_pool_size;
@@ -36,16 +38,16 @@ private:
   std::deque<Task*> m_tasks;
   volatile int m_pool_state;
 
-  Mutex m_all_tasks_mutex;
-  CondVar m_all_tasks_cond_var;
   bool m_tasks_loaded;
   bool thread_started;
   int current_thread_number;
 
   //volatile uint64_t tasks_done_mask;
   volatile int number_of_ongoing_tasks;
+  volatile int number_of_total_tasks;
 
  sem_t semStart;
-
+ sem_t semDone;
+ bool zmqthreadpool;
 };