Complete rewrite of receiver, with new changes

slsReceiverSoftware/src/BinaryFileWriter.cpp

@@ -0,0 +1,22 @@
+/************************************************
+ * @file BinaryFileWriter.h
+ * @short sets/gets properties for the binary file,
+ * creates/closes the file and writes data to it
+ ***********************************************/
+
+#include "BinaryFileWriter.h"
+
+#include <iostream>
+using namespace std;
+
+
+BinaryFileWriter::BinaryFileWriter(char* fname):
+		FileWriter(fname) {
+
+}
+
+
+BinaryFileWriter::~BinaryFileWriter() {
+
+}
+

slsReceiverSoftware/src/DataProcessor.cpp

@@ -0,0 +1,112 @@
+/************************************************
+ * @file DataProcessor.h
+ * @short creates data processor thread that
+ * pulls pointers to memory addresses from fifos
+ * and processes data stored in them & writes them to file
+ ***********************************************/
+
+
+#include "DataProcessor.h"
+#include "Fifo.h"
+
+#include <iostream>
+#include <cstring>
+using namespace std;
+
+const string DataProcessor::TypeName = "DataProcessor";
+
+int DataProcessor::NumberofDataProcessors(0);
+
+uint64_t DataProcessor::ErrorMask(0x0);
+
+uint64_t DataProcessor::RunningMask(0x0);
+
+pthread_mutex_t DataProcessor::Mutex = PTHREAD_MUTEX_INITIALIZER;
+
+
+DataProcessor::DataProcessor(Fifo*& f) : ThreadObject(NumberofDataProcessors), fifo(f) {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+
+	if(ThreadObject::CreateThread()){
+		pthread_mutex_lock(&Mutex);
+		ErrorMask ^= (1<<index);
+		pthread_mutex_unlock(&Mutex);
+	}
+	NumberofDataProcessors++;
+	FILE_LOG(logDEBUG) << "Number of DataProcessors: " << NumberofDataProcessors << endl;
+}
+
+
+DataProcessor::~DataProcessor() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	ThreadObject::DestroyThread();
+	NumberofDataProcessors--;
+}
+
+/** static functions */
+
+
+uint64_t DataProcessor::GetErrorMask() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	return ErrorMask;
+}
+
+
+void DataProcessor::ResetRunningMask() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	pthread_mutex_lock(&Mutex);
+	RunningMask = 0x0;
+	pthread_mutex_unlock(&Mutex);
+}
+
+
+/** non static functions */
+
+
+string DataProcessor::GetType(){
+	return TypeName;
+}
+
+
+bool DataProcessor::IsRunning() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	return ((1 << index) & RunningMask);
+}
+
+
+void DataProcessor::StartRunning() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	pthread_mutex_lock(&Mutex);
+	RunningMask |= (1<<index);
+	pthread_mutex_unlock(&Mutex);
+}
+
+
+void DataProcessor::StopRunning() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	pthread_mutex_lock(&Mutex);
+	RunningMask ^= (1<<index);
+	pthread_mutex_unlock(&Mutex);
+}
+
+
+void DataProcessor::SetFifo(Fifo*& f) {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	fifo = f;
+}
+
+
+void DataProcessor::ThreadExecution() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+
+	char* buffer=0;
+	fifo->PopAddress(buffer);
+#ifdef FIFODEBUG
+	cprintf(BLUE,"DataProcessor %d, pop 0x%p buffer:%s\n", index,(void*)(buffer),buffer);
+#endif
+
+	if(!strcmp(buffer,"done")){
+		StopRunning();
+	}
+	fifo->FreeAddress(buffer);
+}

slsReceiverSoftware/src/DataStreamer.cpp

@@ -0,0 +1,94 @@
+/************************************************
+ * @file DataStreamer.h
+ * @short streams data from receiver via ZMQ
+ ***********************************************/
+
+
+#include "DataStreamer.h"
+
+#include <iostream>
+using namespace std;
+
+const string DataStreamer::TypeName = "DataStreamer";
+
+int DataStreamer::NumberofDataStreamers(0);
+
+uint64_t DataStreamer::ErrorMask(0x0);
+
+uint64_t DataStreamer::RunningMask(0x0);
+
+pthread_mutex_t DataStreamer::Mutex = PTHREAD_MUTEX_INITIALIZER;
+
+
+DataStreamer::DataStreamer() : ThreadObject(NumberofDataStreamers) {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+
+	if(ThreadObject::CreateThread()){
+		pthread_mutex_lock(&Mutex);
+		ErrorMask ^= (1<<index);
+		pthread_mutex_unlock(&Mutex);
+	}
+
+	NumberofDataStreamers++;
+	FILE_LOG(logDEBUG) << "Number of DataStreamers: " << NumberofDataStreamers << endl;
+}
+
+
+DataStreamer::~DataStreamer() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	ThreadObject::DestroyThread();
+	NumberofDataStreamers--;
+}
+
+/** static functions */
+
+
+uint64_t DataStreamer::GetErrorMask() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	return ErrorMask;
+}
+
+
+void DataStreamer::ResetRunningMask() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	pthread_mutex_lock(&Mutex);
+	RunningMask = 0x0;
+	pthread_mutex_unlock(&Mutex);
+}
+
+
+/** non static functions */
+
+
+string DataStreamer::GetType(){
+	return TypeName;
+}
+
+
+bool DataStreamer::IsRunning() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	return ((1 << index) & RunningMask);
+}
+
+
+void DataStreamer::StartRunning() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	pthread_mutex_lock(&Mutex);
+	RunningMask |= (1<<index);
+	pthread_mutex_unlock(&Mutex);
+}
+
+
+void DataStreamer::StopRunning() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	pthread_mutex_lock(&Mutex);
+	RunningMask ^= (1<<index);
+	pthread_mutex_unlock(&Mutex);
+}
+
+void DataStreamer::ThreadExecution() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+
+}
+
+

slsReceiverSoftware/src/Fifo.cpp

@@ -0,0 +1,100 @@
+/************************************************
+ * @file Fifo.h
+ * @short constructs the fifo structure
+ * which is a circular buffer with pointers to
+ * parts of allocated memory
+ ***********************************************/
+
+#include "Fifo.h"
+
+#include <iostream>
+#include <cstdlib>
+using namespace std;
+
+int Fifo::NumberofFifoClassObjects(0);
+
+Fifo::Fifo(uint32_t fifoItemSize, uint32_t fifoDepth, bool &success):
+		memory(0),
+		fifoBound(0),
+		fifoFree(0) {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	index = NumberofFifoClassObjects++;
+	if(CreateFifos(fifoItemSize, fifoDepth) == FAIL)
+		success = false;
+}
+
+
+Fifo::~Fifo() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	DestroyFifos();
+	NumberofFifoClassObjects--;
+}
+
+
+
+int Fifo::CreateFifos(uint32_t fifoItemSize, uint32_t fifoDepth) {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+
+	//destroy if not already
+	DestroyFifos();
+	//create fifos
+	fifoBound = new CircularFifo<char>(fifoDepth);
+	fifoFree = new CircularFifo<char>(fifoDepth);
+	//allocate memory
+	memory = (char*) calloc (fifoItemSize * fifoDepth, sizeof(char));
+	if (memory == NULL){
+		FILE_LOG(logERROR) << "Could not allocate memory for fifos";
+		return FAIL;
+	}
+
+	{ //push free addresses into fifoFree fifo
+		char* buffer = memory;
+		while (buffer < (memory + fifoItemSize * (fifoDepth-1))) {
+			sprintf(buffer,"memory");
+#ifdef FIFODEBUG
+			cprintf(MAGENTA,"Fifofree %d: value:%d, pop 0x%p\n", index, fifoFree->getSemValue(), (void*)(buffer));
+#endif
+			FreeAddress(buffer);
+			buffer += fifoItemSize;
+		}
+	}
+	FILE_LOG(logINFO) << "Fifo Structure " << index << " reconstructed";
+	return OK;
+}
+
+
+void Fifo::DestroyFifos(){
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+
+	if (fifoBound) {
+		delete fifoBound;
+		fifoBound = 0;
+	}
+	if (fifoFree) {
+		delete fifoFree;
+		fifoFree = 0;
+	}
+	if(memory) {
+		free(memory);
+		memory = 0;
+	}
+}
+
+
+
+void Fifo::GetNewAddress(char*& address) {
+	fifoFree->pop(address);
+}
+
+void Fifo::FreeAddress(char*& address) {
+	while(!fifoFree->push(address));
+}
+
+void Fifo::PushAddress(char*& address) {
+	while(!fifoBound->push(address));
+}
+
+void Fifo::PopAddress(char*& address) {
+	fifoBound->pop(address);
+}
+

slsReceiverSoftware/src/FileReader.cpp

@@ -0,0 +1,19 @@
+/********************************************//**
+ * @file FileWriter.cpp
+ * @short sets/gets properties for the file, creates/closes the file and writes data to it
+ ***********************************************/
+
+
+
+#include "FileWriter.h"
+
+#include <iostream>
+using namespace std;
+
+
+
+FileWriter::FileWriter(){}
+
+
+FileWriter::~FileWriter() {}
+

slsReceiverSoftware/src/FileWriter.cpp

@@ -0,0 +1,24 @@
+/************************************************
+ * @file FileWriter.h
+ * @short sets/gets properties for the file,
+ * creates/closes the file and writes data to it
+ ***********************************************/
+
+#include "FileWriter.h"
+
+#include <iostream>
+using namespace std;
+
+
+FileWriter::FileWriter(char* fname):
+		fileName(fname) {
+		cout<<"fileName:"<<fileName<<endl;
+}
+
+FileWriter::~FileWriter() {
+
+}
+
+char* FileWriter::GetFileName() {
+	return fileName;
+}

slsReceiverSoftware/src/HDF5FileWriter.cpp

@@ -0,0 +1,23 @@
+/************************************************
+ * @file HDF5FileWriter.h
+ * @short sets/gets properties for the HDF5 file,
+ * creates/closes the file and writes data to it
+ ***********************************************/
+
+#include "HDF5FileWriter.h"
+
+#include <iostream>
+using namespace std;
+
+
+HDF5FileWriter::HDF5FileWriter(char* fname):
+				FileWriter(fname) {
+
+}
+
+
+HDF5FileWriter::~HDF5FileWriter() {
+
+}
+
+

slsReceiverSoftware/src/Listener.cpp

@@ -0,0 +1,118 @@
+/************************************************
+ * @file Listener.cpp
+ * @short creates the listener thread that
+ * listens to udp sockets, writes data to memory
+ * & puts pointers to their memory addresses into fifos
+ ***********************************************/
+
+
+#include "Listener.h"
+#include "Fifo.h"
+
+#include <iostream>
+#include <cstring>
+using namespace std;
+
+const string Listener::TypeName = "Listener";
+
+int Listener::NumberofListeners(0);
+
+uint64_t Listener::ErrorMask(0x0);
+
+uint64_t Listener::RunningMask(0x0);
+
+pthread_mutex_t Listener::Mutex = PTHREAD_MUTEX_INITIALIZER;
+
+Listener::Listener(Fifo*& f) : ThreadObject(NumberofListeners), fifo(f) {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+
+	if(ThreadObject::CreateThread()){
+		pthread_mutex_lock(&Mutex);
+		ErrorMask ^= (1<<index);
+		pthread_mutex_unlock(&Mutex);
+	}
+	count = 0;
+	NumberofListeners++;
+	FILE_LOG(logDEBUG) << "Number of Listeners: " << NumberofListeners << endl;
+}
+
+
+Listener::~Listener() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	ThreadObject::DestroyThread();
+	NumberofListeners--;
+}
+
+/** static functions */
+
+
+uint64_t Listener::GetErrorMask() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	return ErrorMask;
+}
+
+
+void Listener::ResetRunningMask() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	pthread_mutex_lock(&Mutex);
+	RunningMask = 0x0;
+	pthread_mutex_unlock(&Mutex);
+}
+
+
+/** non static functions */
+
+string Listener::GetType(){
+	return TypeName;
+}
+
+
+bool Listener::IsRunning() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	return ((1 << index) & RunningMask);
+}
+
+
+void Listener::StartRunning() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	pthread_mutex_lock(&Mutex);
+	RunningMask |= (1<<index);
+	pthread_mutex_unlock(&Mutex);
+}
+
+
+void Listener::StopRunning() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	pthread_mutex_lock(&Mutex);
+	RunningMask ^= (1<<index);
+	pthread_mutex_unlock(&Mutex);
+}
+
+
+void Listener::SetFifo(Fifo*& f) {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	fifo = f;
+}
+
+
+void Listener::ThreadExecution() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+
+	char* buffer;
+	fifo->GetNewAddress(buffer);
+#ifdef FIFODEBUG
+	cprintf(GREEN,"Listener %d, pop 0x%p buffer:%s\n", index,(void*)(buffer),buffer);
+#endif
+
+	strcpy(buffer,"changed");
+
+	if(count == 3){
+		strcpy(buffer,"done\0");
+		StopRunning();
+	}
+
+	fifo->PushAddress(buffer);
+	count++;
+}
+
+

slsReceiverSoftware/src/ThreadObject.cpp

@@ -0,0 +1,113 @@
+/************************************************
+ * @file ThreadObject.h
+ * @short creates/destroys a thread
+ ***********************************************/
+
+
+
+#include "ThreadObject.h"
+
+#include <iostream>
+using namespace std;
+
+
+
+ThreadObject::ThreadObject(int ind):
+		index(ind),
+		alive(false),
+		killThread(false),
+		thread(0) {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	PrintMembers();
+}
+
+
+ThreadObject::~ThreadObject() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	DestroyThread();
+}
+
+
+void ThreadObject::PrintMembers() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	FILE_LOG(logDEBUG) 	<< "Index : " << 	index
+						<< "\nalive: " <<	alive
+						<< "\nkillThread: " << killThread
+						<< "\npthread: " << thread;
+}
+
+
+void  ThreadObject::DestroyThread() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	if(alive){
+		killThread = true;
+		sem_post(&semaphore);
+		pthread_join(thread,NULL);
+		sem_destroy(&semaphore);
+		killThread = false;
+		alive = false;
+		FILE_LOG(logDEBUG) << GetType() << " thread with index " << index << " destroyed successfully.";
+	}
+}
+
+
+int ThreadObject::CreateThread() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+	if(alive){
+		FILE_LOG(logERROR) << "Cannot create thread " << index << ". Already alive";
+		return FAIL;
+	}
+	sem_init(&semaphore,1,0);
+	killThread = false;
+
+	if(pthread_create(&thread, NULL,StartThread, (void*) this)){
+		FILE_LOG(logERROR) << "Could not create "  << GetType() << " thread with index " << index;
+		return FAIL;
+	}
+	alive = true;
+	FILE_LOG(logINFO) << GetType() << " thread " << index << " created successfully.";
+
+	return OK;
+}
+
+
+void* ThreadObject::StartThread(void* thisPointer) {
+		FILE_LOG(logDEBUG) << __AT__ << " called";
+		((ThreadObject*)thisPointer)->RunningThread();
+		return thisPointer;
+}
+
+
+void ThreadObject::RunningThread() {
+	FILE_LOG(logDEBUG) << __AT__ << " called";
+
+
+	while(true)	{
+
+		while(IsRunning()) {
+
+			ThreadExecution();
+
+		}/*--end of inner loop */
+
+
+		//wait till the next acquisition
+		sem_wait(&semaphore);
+
+		if(killThread)	{
+			cprintf(BLUE,"%s Thread %d: Goodbye\n",GetType().c_str(),index);
+			pthread_exit(NULL);
+		}
+
+	}/*--end of loop for each acquisition (outer loop) */
+}
+
+
+
+
+void ThreadObject::Continue() {
+	sem_post(&semaphore);
+}
+
+
+

slsReceiverSoftware/src/UDPBaseImplementation.cpp

@@ -191,7 +191,7 @@ uint32_t UDPBaseImplementation::getFrameToGuiFrequency() const{	FILE_LOG(logDEBU
 
 uint32_t UDPBaseImplementation::getFrameToGuiTimer() const{	FILE_LOG(logDEBUG) << __AT__ << " starting";	return frameToGuiTimerinMS;}
 
-uint32_t UDPBaseImplementation::getDataStreamEnable() const{	FILE_LOG(logDEBUG) << __AT__ << " starting";	return dataStreamEnable;}
+bool UDPBaseImplementation::getDataStreamEnable() const{	FILE_LOG(logDEBUG) << __AT__ << " starting";	return dataStreamEnable;}
 
 uint64_t UDPBaseImplementation::getAcquisitionPeriod() const{	FILE_LOG(logDEBUG) << __AT__ << " starting";	return acquisitionPeriod;}
 
@@ -342,11 +342,13 @@ void UDPBaseImplementation::setEthernetInterface(const char* c){
 
 
 /***acquisition parameters***/
-void UDPBaseImplementation::setShortFrameEnable(const int i){
+int UDPBaseImplementation::setShortFrameEnable(const int i){
 	FILE_LOG(logDEBUG) << __AT__ << " starting";
 
 	shortFrameEnable = i;
 	FILE_LOG(logINFO) << "Short Frame Enable: " << stringEnable(shortFrameEnable);
+	//overrridden child classes might return FAIL
+	return OK;
 }
 
 int UDPBaseImplementation::setFrameToGuiFrequency(const uint32_t freq){
@@ -367,7 +369,7 @@ void UDPBaseImplementation::setFrameToGuiTimer(const uint32_t time_in_ms){
 }
 
 
-uint32_t UDPBaseImplementation::setDataStreamEnable(const uint32_t enable){
+int UDPBaseImplementation::setDataStreamEnable(const bool enable){
 	FILE_LOG(logDEBUG) << __AT__ << " starting";
 
 	dataStreamEnable = enable;

slsReceiverSoftware/src/main.cpp

@@ -101,7 +101,7 @@ int main(int argc, char *argv[]) {
 	//receiver->registerCallBackRawDataReady(rawDataReadyCallBack,NULL);
 
 
-
+/*
 	//start tcp server thread
 	if(receiver->start() == slsReceiverDefs::OK){
 		FILE_LOG(logDEBUG1) << "DONE!" << endl;
@@ -113,7 +113,7 @@ int main(int argc, char *argv[]) {
 		//stop tcp server thread, stop udp socket
 		receiver->stop();
 	}
-
+*/
 	deleteReceiver(receiver);
 	cout << "Goodbye!" << endl;
 	return 0;

slsReceiverSoftware/src/slsReceiver.cpp

@@ -127,6 +127,7 @@ slsReceiver::slsReceiver(int argc, char *argv[], int &success){
 		udp_interface = UDPInterface::create(udp_interface_type);
 		udp_interface->configure(configuration_map);
 #endif
+		udp_interface = UDPInterface::create("standard");
 		tcpipInterface = new slsReceiverTCPIPInterface(success, udp_interface, tcpip_port_no);
 	}
 }

slsReceiverSoftware/src/slsReceiverTCPIPInterface.cpp

@@ -23,8 +23,8 @@ using namespace std;
 
 
 slsReceiverTCPIPInterface::~slsReceiverTCPIPInterface() {
-	stop();
-	if(mySock) {delete mySock; mySock=NULL;}
+	/*stop();
+	if(mySock) {delete mySock; mySock=NULL;}*/
 }
 
 slsReceiverTCPIPInterface::slsReceiverTCPIPInterface(int &success, UDPInterface* rbase, int pn):
@@ -58,7 +58,7 @@ slsReceiverTCPIPInterface::slsReceiverTCPIPInterface(int &success, UDPInterface*
 
 	success=OK;
 
-	//create socket
+	/*//create socket
 	if(success == OK){
 		mySock = new MySocketTCP(port_no);
 		if (mySock->getErrorStatus()) {
@@ -76,7 +76,7 @@ slsReceiverTCPIPInterface::slsReceiverTCPIPInterface(int &success, UDPInterface*
 			cout << "Function table assigned." << endl;
 #endif
 		}
-	}
+	}*/
 
 }
 
@@ -2927,8 +2927,8 @@ int slsReceiverTCPIPInterface::set_fifo_depth() {
 			ret=FAIL;
 		}else{
 			retval = receiverBase->getFifoDepth();
-			if(value >= 0 && retval != value)
-				ret = FAIL;
+			/*if(value >= 0 && retval != value)
+				ret = FAIL;*/
 		}
 
 	}