almost there

slsDetectorSoftware/multiSlsDetector/multiSlsDetector.cpp

@@ -5070,10 +5070,14 @@ void multiSlsDetector::startReceivingDataThread(){
 	void *zmqsocket;
 	context = zmq_ctx_new();
 	zmqsocket = zmq_socket(context, ZMQ_PULL);
+	//int hwmval = 10;
+	//zmq_setsockopt(zmqsocket,ZMQ_RCVHWM,&hwmval,sizeof(hwmval)); //set receive HIGH WATER MARK (8-9ms slower)
 	zmq_connect(zmqsocket, hostname);
 	cout << "ZMQ Client of " << ithread << " at " << hostname << endl;
 	cprintf(BLUE,"%d Created socket\n",ithread);
-
+	/*
+	zmq_pollitem_t pollitem = {zmqsocket, 0 , ZMQ_POLLIN , 0};
+*/
 	//initializations
 	int numReadoutPerDetector = 1;
 	bool jungfrau = false;
@@ -5086,6 +5090,7 @@ void multiSlsDetector::startReceivingDataThread(){
 	int* image = new int[nel];
 	int len,idet = 0;
 	singleframe[ithread]=NULL;
+	int datavalue = 2;
 	threadStarted = true;					//let calling function know thread started and obtained current
 
 
@@ -5102,82 +5107,112 @@ void multiSlsDetector::startReceivingDataThread(){
 
 		//scan header-------------------------------------------------------------------
 		zmq_msg_init (&message);
-		len = zmq_msg_recv(&message, zmqsocket, 0);
+		while(1){
-		if (len == -1) {
+
-			zmq_msg_close(&message);
+			len = zmq_msg_recv(&message, zmqsocket, ZMQ_DONTWAIT);
-			cprintf(RED, "%d message null\n",ithread);
+			if(len>0)
-			continue;
+				break;//also comment out the next recv
+/*
+			zmq_poll(&pollitem, 1, 0);
+			//received something, get out
+			if(pollitem.revents & ZMQ_POLLIN){
+				pollitem.revents = 0;
+				break;
+			}
+*/
+			//received nothing
+			else if (receiverStoppedFlag){
+				//one more chance if receiver stopped
+				datavalue--;
+
+				if(!datavalue){
+					//#ifdef VERYVERBOSE
+					cprintf(RED,"End of socket for %d\n", ithread);
+					//#endif
+					singleframe[ithread] = NULL;
+					break;
+				}
+
+				//wait to check again only if receiver stopped
+				//usleep(4000);
+			}
+			usleep(4000);
 		}
 
-		// error if you print it
+		if(datavalue){
-		// cout << ithread << " header len:"<<len<<" value:"<< (char*)zmq_msg_data(&message)<<endl;
+			//len = zmq_msg_recv(&message, zmqsocket, 0);
-		//cprintf(BLUE,"%d header %d\n",ithread,len);
+			if (len == -1) {
-		rapidjson::Document d;
+				zmq_msg_close(&message);
-		d.Parse( (char*)zmq_msg_data(&message), zmq_msg_size(&message));
+				cprintf(RED, "%d message null\n",ithread);
-#ifdef VERYVERBOSE
+				continue;
-		// htype is an array of strings
+			}
-		rapidjson::Value::Array htype = d["htype"].GetArray();
-		for(int i=0; i< htype.Size(); i++)
-			std::cout << ithread << "htype: " << htype[i].GetString() << std::endl;
-		// shape is an array of ints
-		rapidjson::Value::Array shape = d["shape"].GetArray();
-		cout << ithread << "shape: ";
-		for(int i=0; i< shape.Size(); i++)
-			cout << ithread << shape[i].GetInt() << " ";
-		cout << endl;
 
-		cout << ithread << "type: " << d["type"].GetString() << endl;
+			// error if you print it
+			// cout << ithread << " header len:"<<len<<" value:"<< (char*)zmq_msg_data(&message)<<endl;
+			//cprintf(BLUE,"%d header %d\n",ithread,len);
+			rapidjson::Document d;
+			d.Parse( (char*)zmq_msg_data(&message), zmq_msg_size(&message));
+#ifdef VERYVERBOSE
+			// htype is an array of strings
+			rapidjson::Value::Array htype = d["htype"].GetArray();
+			for(int i=0; i< htype.Size(); i++)
+				std::cout << ithread << "htype: " << htype[i].GetString() << std::endl;
+			// shape is an array of ints
+			rapidjson::Value::Array shape = d["shape"].GetArray();
+			cout << ithread << "shape: ";
+			for(int i=0; i< shape.Size(); i++)
+				cout << ithread << shape[i].GetInt() << " ";
+			cout << endl;
+
+			cout << ithread << "type: " << d["type"].GetString() << endl;
 
 #endif
-		if(!ithread){
+			if(!ithread){
-			currentAcquisitionIndex 	= d["acqIndex"].GetInt();
+				currentAcquisitionIndex 	= d["acqIndex"].GetInt();
-			currentFrameIndex 			= d["fIndex"].GetInt();
+				currentFrameIndex 			= d["fIndex"].GetInt();
-			currentSubFrameIndex 		= d["subfnum"].GetInt();
+				currentSubFrameIndex 		= d["subfnum"].GetInt();
-			strcpy(currentFileName		 ,d["fname"].GetString());
+				strcpy(currentFileName		 ,d["fname"].GetString());
 #ifdef VERYVERBOSE
-			cout << "Acquisition index: " << currentAcquisitionIndex << endl;
+				cout << "Acquisition index: " << currentAcquisitionIndex << endl;
-			cout << "Frame index: " << currentFrameIndex << endl;
+				cout << "Frame index: " << currentFrameIndex << endl;
-			cout << "Subframe index: " << currentSubFrameIndex << endl;
+				cout << "Subframe index: " << currentSubFrameIndex << endl;
-			cout << "File name: " << currentFileName << endl;
+				cout << "File name: " << currentFileName << endl;
 #endif
-			if(currentFrameIndex ==-1) cprintf(RED,"multi frame index -1!!\n");
+				if(currentFrameIndex ==-1) cprintf(RED,"multi frame index -1!!\n");
-		}
+			}
-		singleframe[ithread]=image;
+			singleframe[ithread]=image;
 
-		// close the message
+			// close the message
-		zmq_msg_close(&message);
-
-
-		//scan data-------------------------------------------------------------------
-		zmq_msg_init (&message);
-		len = zmq_msg_recv(&message, zmqsocket, 0);
-
-		//cprintf(BLUE,"%d data %d\n",ithread,len);
-		//end of socket ("end")
-		if (len < 1024*256 ) {
-			if(!len) cprintf(RED,"Received no data in socket for %d\n", ithread);
-			//#ifdef VERYVERBOSE
-			cprintf(RED,"End of socket for %d\n", ithread);
-			//#endif
 			zmq_msg_close(&message);
-			singleframe[ithread] = NULL;
-			//break;
-		}
-		else{
-			//actual data
-			//cprintf(BLUE,"%d actual dataaa\n",ithread);
-			memcpy((char*)(singleframe[ithread]),(char*)zmq_msg_data(&message),singleDatabytes/numReadoutPerDetector);
 
 
-			//jungfrau masking adcval
+			//scan data-------------------------------------------------------------------
-			if(jungfrau){
+			zmq_msg_init (&message);
-				for(unsigned int i=0;i<nel;i++){
+			len = zmq_msg_recv(&message, zmqsocket, 0);
-					singleframe[ithread][i] = (singleframe[ithread][i] & 0x3FFF3FFF);
+
+			//cprintf(BLUE,"%d data %d\n",ithread,len);
+			//end of socket ("end")
+			if (len < 1024*256 ) {
+				cprintf(RED,"Received weird packet size %d in socket for %d\n", len, ithread);
+				memset((char*)(singleframe[ithread]),0xFF,singleDatabytes/numReadoutPerDetector);
+			}
+			else{
+				//actual data
+				//cprintf(BLUE,"%d actual dataaa\n",ithread);
+				memcpy((char*)(singleframe[ithread]),(char*)zmq_msg_data(&message),singleDatabytes/numReadoutPerDetector);
+
+
+				//jungfrau masking adcval
+				if(jungfrau){
+					for(unsigned int i=0;i<nel;i++){
+						singleframe[ithread][i] = (singleframe[ithread][i] & 0x3FFF3FFF);
+					}
 				}
 			}
 		}
 		sem_post(&sem_singledone[ithread]);//let multi know is ready
 		zmq_msg_close(&message); // close the message
+		datavalue = 2;
 	}
 
 	cprintf(RED,"%d Closing socket\n",ithread);

slsDetectorSoftware/slsDetector/slsDetectorUtils.cpp

@@ -58,6 +58,7 @@ int  slsDetectorUtils::acquire(int delflag){
   if(!receiver){
 	  setDetectorIndex(-1);
   }else{
+	receiverStoppedFlag = 0;
 	  //put receiver read frequency to random if no gui
 	  int ret = setReadReceiverFrequency(0);
 	  if(ret>0 && (dataReady == NULL)){
@@ -148,17 +149,21 @@ int  slsDetectorUtils::acquire(int delflag){
   if(receiver){
 	  if(getReceiverStatus()!=IDLE)
 		  stopReceiver();
-	  if(setReceiverOnline()==OFFLINE_FLAG)
+	  if(setReceiverOnline()==OFFLINE_FLAG){
 		  *stoppedFlag=1;
-
+		  receiverStoppedFlag = 1;
+	  }
 	  //multi detectors shouldnt have different receiver read frequencies enabled/disabled
     	if(setReadReceiverFrequency(0) < 0){
 	  		std::cout << "Error: The receiver read frequency is invalid:" << setReadReceiverFrequency(0) << std::endl;
-	  		 *stoppedFlag=1;
+	  		*stoppedFlag=1;
+	  		receiverStoppedFlag = 1;
 	  	}
 
-	  if(setReceiverOnline()==OFFLINE_FLAG)
+	  if(setReceiverOnline()==OFFLINE_FLAG){
 		  *stoppedFlag=1;
+		  receiverStoppedFlag = 1;
+	  }
   }
 
 
@@ -291,6 +296,7 @@ int  slsDetectorUtils::acquire(int delflag){
 	    	if(setReceiverOnline()==OFFLINE_FLAG){
 	    		stopReceiver();
 	    		*stoppedFlag=1;
+	    		receiverStoppedFlag = 1;
 	    		pthread_mutex_unlock(&mg);
 	    		break;
 	    	}
@@ -298,6 +304,7 @@ int  slsDetectorUtils::acquire(int delflag){
 	    	if(startReceiver() == FAIL) {
 	    		stopReceiver();
 	    		*stoppedFlag=1;
+	    		receiverStoppedFlag = 1;
 	    		pthread_mutex_unlock(&mg);
 	    		break;
 	    	}
@@ -344,6 +351,7 @@ int  slsDetectorUtils::acquire(int delflag){
 	  pthread_mutex_lock(&mg);
 	  //offline
 	  if(setReceiverOnline()==OFFLINE_FLAG){
+		  receiverStoppedFlag = 1;
 		  if ((getDetectorsType()==GOTTHARD) || (getDetectorsType()==MOENCH) || (getDetectorsType()==JUNGFRAU) ){
 			  if((*correctionMask)&(1<<WRITE_FILE))
 				  closeDataFile();
@@ -352,6 +360,7 @@ int  slsDetectorUtils::acquire(int delflag){
 	  //online
 	  else{
 		  stopReceiver();
+		  receiverStoppedFlag = 1;
 		 // cout<<"***********receiver stopped"<<endl;
 	  }
 	  pthread_mutex_unlock(&mg);

slsDetectorSoftware/slsDetector/slsDetectorUtils.h

@@ -837,6 +837,7 @@ virtual int setReceiverFifoDepth(int i = -1)=0;
  
   //protected:
   int *stoppedFlag;	 
+  int receiverStoppedFlag;
 
   int64_t *timerValue;
   detectorSettings *currentSettings;