almost done

slsReceiverSoftware/Makefile

@@ -12,10 +12,13 @@ PROGS		= 	$(DESTDIR)/slsReceiver
 CFLAGS= -g -DC_ONLY -fPIC
 #FLAGS+=  #-DVERBOSE -DVERYVERBOSE
 
-DFLAGS= -g -DDACS_INT  -DSLS_RECEIVER_UDP_FUNCTIONS 
+DFLAGS= -g -DDACS_INT  -DSLS_RECEIVER_UDP_FUNCTIONS #-DVERBOSE
 
 INCLUDES?= $(INCLUDESRXR) -I include/ -I ../slsDetectorCalibration
 
+LIBZMQDIR = include
+LIBZMQ =  -L$(LIBZMQDIR) -Wl,-rpath=$(LIBZMQDIR) -lzmq
+
 #-Iinclude -I../slsDetectorCalibration -I$(ASM) 
 
 SRC_CLNT = MySocketTCP.cpp UDPInterface.cpp  UDPBaseImplementation.cpp UDPStandardImplementation.cpp slsReceiverTCPIPInterface.cpp slsReceiver.cpp slsReceiverUsers.cpp utilities.cpp 
@@ -49,9 +52,9 @@ intdoc: $(SRC_H) $(SRC_CLNT)
 
 $(BUILDDIR)/%.o : $(SRCDIR)/%.cpp Makefile
 ifeq ($(ROOTSLS),yes)
-	$(CXX) -DROOTSLS -o $@ -c $< $(INCLUDES) $(DFLAGS)  $(ROOTFLAGS) -fPIC  $(EPICSFLAGS) $(LDFLAGRXR) -L/usr/lib64/ $(FLAGS)
+	$(CXX) -DROOTSLS -o $@ -c $< $(INCLUDES) $(DFLAGS)  $(ROOTFLAGS) -fPIC  $(EPICSFLAGS) $(LDFLAGRXR) -L/usr/lib64/ $(FLAGS) 
 else	
-	$(CXX) -o $@ -c $< $(INCLUDES) $(DFLAGS) -fPIC  $(EPICSFLAGS) $(LDFLAGRXR) -lpthread $(FLAGS)
+	$(CXX) -o $@ -c $< $(INCLUDES) $(DFLAGS) -fPIC  $(EPICSFLAGS) $(LDFLAGRXR) -lpthread $(FLAGS) $(LIBZMQ)  -lrt 
 endif
 
 lib:  $(OBJS) $(DESTDIR)/libSlsReceiver.so $(DESTDIR)/libSlsReceiver.a 
@@ -60,7 +63,7 @@ receiver: $(DESTDIR)/slsReceiver
 
 
 $(DESTDIR)/libSlsReceiver.so: $(OBJS) 
-	$(CXX)  -shared -Wl,-soname,libSlsReceiver.so -o libSlsReceiver.so  $(OBJS) -lc $(INCLUDES) $(DFLAGS) $(FLAGS) $(EPICSFLAGS)  -L/usr/lib64 -lpthread
+	$(CXX)  -shared -Wl,-soname,libSlsReceiver.so -o libSlsReceiver.so  $(OBJS) -lc $(INCLUDES) $(DFLAGS) $(FLAGS) $(EPICSFLAGS)  -L/usr/lib64 -lpthread $(LIBZMQ) -lrt 
 	$(shell test -d $(DESTDIR) || mkdir -p $(DESTDIR))
 	mv libSlsReceiver.so $(DESTDIR)
 
@@ -70,12 +73,12 @@ $(DESTDIR)/libSlsReceiver.a: $(OBJS)
 
 
 $(DESTDIR)/slsReceiver: 	lib
-	$(CXX)   -o $@  $(SRCDIR)/$(MAIN_SRC)  $(FLAGS)  $(INCLUDES) $(CLAGS) $(LIBS) $(LDFLAGRXR) -fPIC 
+	$(CXX)   -o $@  $(SRCDIR)/$(MAIN_SRC)  $(FLAGS)  $(INCLUDES) $(CLAGS) $(LIBS) $(LDFLAGRXR) -fPIC  $(LIBZMQ) -lrt 
 #$(EIGERFLAGS)
 
 
 $(DESTDIR)/dummyReceiver: 	lib
-	$(CXX)   -o $@  $(SRCDIR)/$(DUMMY_MAIN_SRC)  $(FLAGS)  $(INCLUDES) $(CLAGS) $(LIBS) $(LDFLAGRXR) -fPIC 
+	$(CXX)   -o $@  $(SRCDIR)/$(DUMMY_MAIN_SRC)  $(FLAGS)  $(INCLUDES) $(CLAGS) $(LIBS) $(LDFLAGRXR) -fPIC $(LIBZMQ)  -lrt 
 #$(EIGERFLAGS)
 
 

slsReceiverSoftware/include/UDPStandardImplementation.h

@@ -70,6 +70,13 @@ class UDPStandardImplementation: private virtual slsReceiverDefs, public UDPBase
 	void configure(map<string, string> config_map);
 
 	//*** file parameters***
+	/**
+	 * Set File Name Prefix (without frame index, file index and extension (_d0_f000000000000_8.raw))
+	 * Does not check for file existence since it is created only at startReceiver
+	 * @param c file name (max of 1000 characters)
+	 */
+	void setFileName(const char c[]);
+
 	/**
 	 * Overridden method
 	 * Set data compression, by saving only hits (so far implemented only for Moench and Gotthard)
@@ -500,10 +507,11 @@ private:
 	 * Get Frame Number
 	 * @param ithread writer thread index
 	 * @param wbuffer writer buffer
-	 * @param tempframenumber reference to the frame number
+	 * @param framenumber reference to the frame number
+	 * @param packetnumber reference to the packet number
 	 * @return OK or FAIL
 	 */
-	int getFrameNumber(int ithread, char* wbuffer, uint64_t &tempframenumber);
+	int getFrameandPacketNumber(int ithread, char* wbuffer, uint64_t &framenumber, uint32_t &packetnumber);
 
 	/**
 	 * Find offset upto this frame number and write it to file
@@ -530,6 +538,9 @@ private:
 #endif
 
 	//**detector parameters***
+	/*Detector Readout ID*/
+	int detID;
+
 	/** Size of 1 buffer processed at a time */
 	int bufferSize;
 
@@ -655,17 +666,17 @@ private:
 	/** Current Frame copied for GUI */
 	char* latestData[MAX_NUMBER_OF_WRITER_THREADS];
 
-	/** If Data to be sent to GUI is ready */
-	bool guiDataReady[MAX_NUMBER_OF_WRITER_THREADS];
-
-	/** Pointer to data to be sent to GUI */
-	char* guiData[MAX_NUMBER_OF_WRITER_THREADS];
-
 	/** Pointer to file name to be sent to GUI */
 	char guiFileName[MAX_NUMBER_OF_WRITER_THREADS][MAX_STR_LENGTH];
 
+	/** Number of packets copied to be sent to gui (others padded) */
+	int guiNumPackets[MAX_NUMBER_OF_WRITER_THREADS];
+
 	/** Semaphore to synchronize Writer and GuiReader threads*/
-	sem_t writerGuiSemaphore[MAX_NUMBER_OF_WRITER_THREADS];
+	sem_t writerGuiSemaphore[MAX_NUMBER_OF_WRITER_THREADS]; //datacompression, only first thread sends to gui
+
+	/** Semaphore to synchronize Writer and GuiReader threads*/
+	sem_t dataCallbackWriterSemaphore[MAX_NUMBER_OF_WRITER_THREADS]; //datacompression, only first thread sends to gui
 
 	/** counter for nth frame to gui */
 	int frametoGuiCounter[MAX_NUMBER_OF_WRITER_THREADS];
@@ -673,6 +684,9 @@ private:
 
 
 	//***data call back thread parameters***
+	/** Ensures if zmq threads created successfully */
+	bool zmqThreadStarted;
+
 	/** Number of data callback Threads */
 	int numberofDataCallbackThreads;
 

slsReceiverSoftware/include/genericSocket.h

@@ -81,6 +81,7 @@ using namespace std;
 #define DEFAULT_BACKLOG 5
 #define DEFAULT_UDP_PORTNO 50001
 #define DEFAULT_GUI_PORTNO 65000
+#define DEFAULT_ZMQ_PORTNO 70001
 
 class genericSocket{
 

slsReceiverSoftware/include/zmq.h

@@ -0,0 +1,416 @@
+/*
+    Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file
+
+    This file is part of 0MQ.
+
+    0MQ is free software; you can redistribute it and/or modify it under
+    the terms of the GNU Lesser General Public License as published by
+    the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+
+    0MQ is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Lesser General Public License for more details.
+
+    You should have received a copy of the GNU Lesser General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+    *************************************************************************
+    NOTE to contributors. This file comprises the principal public contract
+    for ZeroMQ API users (along with zmq_utils.h). Any change to this file
+    supplied in a stable release SHOULD not break existing applications.
+    In practice this means that the value of constants must not change, and
+    that old values may not be reused for new constants.
+    *************************************************************************
+*/
+
+#ifndef __ZMQ_H_INCLUDED__
+#define __ZMQ_H_INCLUDED__
+
+/*  Version macros for compile-time API version detection                     */
+#define ZMQ_VERSION_MAJOR 4
+#define ZMQ_VERSION_MINOR 0
+#define ZMQ_VERSION_PATCH 4
+
+#define ZMQ_MAKE_VERSION(major, minor, patch) \
+    ((major) * 10000 + (minor) * 100 + (patch))
+#define ZMQ_VERSION \
+    ZMQ_MAKE_VERSION(ZMQ_VERSION_MAJOR, ZMQ_VERSION_MINOR, ZMQ_VERSION_PATCH)
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#if !defined _WIN32_WCE
+#include <errno.h>
+#endif
+#include <stddef.h>
+#include <stdio.h>
+#if defined _WIN32
+#include <winsock2.h>
+#endif
+
+/*  Handle DSO symbol visibility                                             */
+#if defined _WIN32
+#   if defined ZMQ_STATIC
+#       define ZMQ_EXPORT
+#   elif defined DLL_EXPORT
+#       define ZMQ_EXPORT __declspec(dllexport)
+#   else
+#       define ZMQ_EXPORT __declspec(dllimport)
+#   endif
+#else
+#   if defined __SUNPRO_C  || defined __SUNPRO_CC
+#       define ZMQ_EXPORT __global
+#   elif (defined __GNUC__ && __GNUC__ >= 4) || defined __INTEL_COMPILER
+#       define ZMQ_EXPORT __attribute__ ((visibility("default")))
+#   else
+#       define ZMQ_EXPORT
+#   endif
+#endif
+
+/*  Define integer types needed for event interface                          */
+#if defined ZMQ_HAVE_SOLARIS || defined ZMQ_HAVE_OPENVMS
+#   include <inttypes.h>
+#elif defined _MSC_VER && _MSC_VER < 1600
+#   ifndef int32_t
+typedef __int32 int32_t;
+#   endif
+#   ifndef uint16_t
+typedef unsigned __int16 uint16_t;
+#   endif
+#   ifndef uint8_t
+typedef unsigned __int8 uint8_t;
+#   endif
+#else
+#   include <stdint.h>
+#endif
+
+
+/******************************************************************************/
+/*  0MQ errors.                                                               */
+/******************************************************************************/
+
+/*  A number random enough not to collide with different errno ranges on      */
+/*  different OSes. The assumption is that error_t is at least 32-bit type.   */
+#define ZMQ_HAUSNUMERO 156384712
+
+/*  On Windows platform some of the standard POSIX errnos are not defined.    */
+#ifndef ENOTSUP
+#define ENOTSUP (ZMQ_HAUSNUMERO + 1)
+#endif
+#ifndef EPROTONOSUPPORT
+#define EPROTONOSUPPORT (ZMQ_HAUSNUMERO + 2)
+#endif
+#ifndef ENOBUFS
+#define ENOBUFS (ZMQ_HAUSNUMERO + 3)
+#endif
+#ifndef ENETDOWN
+#define ENETDOWN (ZMQ_HAUSNUMERO + 4)
+#endif
+#ifndef EADDRINUSE
+#define EADDRINUSE (ZMQ_HAUSNUMERO + 5)
+#endif
+#ifndef EADDRNOTAVAIL
+#define EADDRNOTAVAIL (ZMQ_HAUSNUMERO + 6)
+#endif
+#ifndef ECONNREFUSED
+#define ECONNREFUSED (ZMQ_HAUSNUMERO + 7)
+#endif
+#ifndef EINPROGRESS
+#define EINPROGRESS (ZMQ_HAUSNUMERO + 8)
+#endif
+#ifndef ENOTSOCK
+#define ENOTSOCK (ZMQ_HAUSNUMERO + 9)
+#endif
+#ifndef EMSGSIZE
+#define EMSGSIZE (ZMQ_HAUSNUMERO + 10)
+#endif
+#ifndef EAFNOSUPPORT
+#define EAFNOSUPPORT (ZMQ_HAUSNUMERO + 11)
+#endif
+#ifndef ENETUNREACH
+#define ENETUNREACH (ZMQ_HAUSNUMERO + 12)
+#endif
+#ifndef ECONNABORTED
+#define ECONNABORTED (ZMQ_HAUSNUMERO + 13)
+#endif
+#ifndef ECONNRESET
+#define ECONNRESET (ZMQ_HAUSNUMERO + 14)
+#endif
+#ifndef ENOTCONN
+#define ENOTCONN (ZMQ_HAUSNUMERO + 15)
+#endif
+#ifndef ETIMEDOUT
+#define ETIMEDOUT (ZMQ_HAUSNUMERO + 16)
+#endif
+#ifndef EHOSTUNREACH
+#define EHOSTUNREACH (ZMQ_HAUSNUMERO + 17)
+#endif
+#ifndef ENETRESET
+#define ENETRESET (ZMQ_HAUSNUMERO + 18)
+#endif
+
+/*  Native 0MQ error codes.                                                   */
+#define EFSM (ZMQ_HAUSNUMERO + 51)
+#define ENOCOMPATPROTO (ZMQ_HAUSNUMERO + 52)
+#define ETERM (ZMQ_HAUSNUMERO + 53)
+#define EMTHREAD (ZMQ_HAUSNUMERO + 54)
+
+/*  Run-time API version detection                                            */
+ZMQ_EXPORT void zmq_version (int *major, int *minor, int *patch);
+
+/*  This function retrieves the errno as it is known to 0MQ library. The goal */
+/*  of this function is to make the code 100% portable, including where 0MQ   */
+/*  compiled with certain CRT library (on Windows) is linked to an            */
+/*  application that uses different CRT library.                              */
+ZMQ_EXPORT int zmq_errno (void);
+
+/*  Resolves system errors and 0MQ errors to human-readable string.           */
+ZMQ_EXPORT const char *zmq_strerror (int errnum);
+
+/******************************************************************************/
+/*  0MQ infrastructure (a.k.a. context) initialisation & termination.         */
+/******************************************************************************/
+
+/*  New API                                                                   */
+/*  Context options                                                           */
+#define ZMQ_IO_THREADS  1
+#define ZMQ_MAX_SOCKETS 2
+
+/*  Default for new contexts                                                  */
+#define ZMQ_IO_THREADS_DFLT  1
+#define ZMQ_MAX_SOCKETS_DFLT 1023
+
+ZMQ_EXPORT void *zmq_ctx_new (void);
+ZMQ_EXPORT int zmq_ctx_term (void *context);
+ZMQ_EXPORT int zmq_ctx_shutdown (void *ctx_);
+ZMQ_EXPORT int zmq_ctx_set (void *context, int option, int optval);
+ZMQ_EXPORT int zmq_ctx_get (void *context, int option);
+
+/*  Old (legacy) API                                                          */
+ZMQ_EXPORT void *zmq_init (int io_threads);
+ZMQ_EXPORT int zmq_term (void *context);
+ZMQ_EXPORT int zmq_ctx_destroy (void *context);
+
+
+/******************************************************************************/
+/*  0MQ message definition.                                                   */
+/******************************************************************************/
+
+typedef struct zmq_msg_t {unsigned char _ [32];} zmq_msg_t;
+
+typedef void (zmq_free_fn) (void *data, void *hint);
+
+ZMQ_EXPORT int zmq_msg_init (zmq_msg_t *msg);
+ZMQ_EXPORT int zmq_msg_init_size (zmq_msg_t *msg, size_t size);
+ZMQ_EXPORT int zmq_msg_init_data (zmq_msg_t *msg, void *data,
+    size_t size, zmq_free_fn *ffn, void *hint);
+ZMQ_EXPORT int zmq_msg_send (zmq_msg_t *msg, void *s, int flags);
+ZMQ_EXPORT int zmq_msg_recv (zmq_msg_t *msg, void *s, int flags);
+ZMQ_EXPORT int zmq_msg_close (zmq_msg_t *msg);
+ZMQ_EXPORT int zmq_msg_move (zmq_msg_t *dest, zmq_msg_t *src);
+ZMQ_EXPORT int zmq_msg_copy (zmq_msg_t *dest, zmq_msg_t *src);
+ZMQ_EXPORT void *zmq_msg_data (zmq_msg_t *msg);
+ZMQ_EXPORT size_t zmq_msg_size (zmq_msg_t *msg);
+ZMQ_EXPORT int zmq_msg_more (zmq_msg_t *msg);
+ZMQ_EXPORT int zmq_msg_get (zmq_msg_t *msg, int option);
+ZMQ_EXPORT int zmq_msg_set (zmq_msg_t *msg, int option, int optval);
+
+
+/******************************************************************************/
+/*  0MQ socket definition.                                                    */
+/******************************************************************************/
+
+/*  Socket types.                                                             */
+#define ZMQ_PAIR 0
+#define ZMQ_PUB 1
+#define ZMQ_SUB 2
+#define ZMQ_REQ 3
+#define ZMQ_REP 4
+#define ZMQ_DEALER 5
+#define ZMQ_ROUTER 6
+#define ZMQ_PULL 7
+#define ZMQ_PUSH 8
+#define ZMQ_XPUB 9
+#define ZMQ_XSUB 10
+#define ZMQ_STREAM 11
+
+/*  Deprecated aliases                                                        */
+#define ZMQ_XREQ ZMQ_DEALER
+#define ZMQ_XREP ZMQ_ROUTER
+
+/*  Socket options.                                                           */
+#define ZMQ_AFFINITY 4
+#define ZMQ_IDENTITY 5
+#define ZMQ_SUBSCRIBE 6
+#define ZMQ_UNSUBSCRIBE 7
+#define ZMQ_RATE 8
+#define ZMQ_RECOVERY_IVL 9
+#define ZMQ_SNDBUF 11
+#define ZMQ_RCVBUF 12
+#define ZMQ_RCVMORE 13
+#define ZMQ_FD 14
+#define ZMQ_EVENTS 15
+#define ZMQ_TYPE 16
+#define ZMQ_LINGER 17
+#define ZMQ_RECONNECT_IVL 18
+#define ZMQ_BACKLOG 19
+#define ZMQ_RECONNECT_IVL_MAX 21
+#define ZMQ_MAXMSGSIZE 22
+#define ZMQ_SNDHWM 23
+#define ZMQ_RCVHWM 24
+#define ZMQ_MULTICAST_HOPS 25
+#define ZMQ_RCVTIMEO 27
+#define ZMQ_SNDTIMEO 28
+#define ZMQ_LAST_ENDPOINT 32
+#define ZMQ_ROUTER_MANDATORY 33
+#define ZMQ_TCP_KEEPALIVE 34
+#define ZMQ_TCP_KEEPALIVE_CNT 35
+#define ZMQ_TCP_KEEPALIVE_IDLE 36
+#define ZMQ_TCP_KEEPALIVE_INTVL 37
+#define ZMQ_TCP_ACCEPT_FILTER 38
+#define ZMQ_IMMEDIATE 39
+#define ZMQ_XPUB_VERBOSE 40
+#define ZMQ_ROUTER_RAW 41
+#define ZMQ_IPV6 42
+#define ZMQ_MECHANISM 43
+#define ZMQ_PLAIN_SERVER 44
+#define ZMQ_PLAIN_USERNAME 45
+#define ZMQ_PLAIN_PASSWORD 46
+#define ZMQ_CURVE_SERVER 47
+#define ZMQ_CURVE_PUBLICKEY 48
+#define ZMQ_CURVE_SECRETKEY 49
+#define ZMQ_CURVE_SERVERKEY 50
+#define ZMQ_PROBE_ROUTER 51
+#define ZMQ_REQ_CORRELATE 52
+#define ZMQ_REQ_RELAXED 53
+#define ZMQ_CONFLATE 54
+#define ZMQ_ZAP_DOMAIN 55
+
+/*  Message options                                                           */
+#define ZMQ_MORE 1
+
+/*  Send/recv options.                                                        */
+#define ZMQ_DONTWAIT 1
+#define ZMQ_SNDMORE 2
+
+/*  Security mechanisms                                                       */
+#define ZMQ_NULL 0
+#define ZMQ_PLAIN 1
+#define ZMQ_CURVE 2
+
+/*  Deprecated options and aliases                                            */
+#define ZMQ_IPV4ONLY                31
+#define ZMQ_DELAY_ATTACH_ON_CONNECT ZMQ_IMMEDIATE
+#define ZMQ_NOBLOCK                 ZMQ_DONTWAIT
+#define ZMQ_FAIL_UNROUTABLE         ZMQ_ROUTER_MANDATORY
+#define ZMQ_ROUTER_BEHAVIOR         ZMQ_ROUTER_MANDATORY
+
+/******************************************************************************/
+/*  0MQ socket events and monitoring                                          */
+/******************************************************************************/
+
+/*  Socket transport events (tcp and ipc only)                                */
+#define ZMQ_EVENT_CONNECTED 1
+#define ZMQ_EVENT_CONNECT_DELAYED 2
+#define ZMQ_EVENT_CONNECT_RETRIED 4
+
+#define ZMQ_EVENT_LISTENING 8
+#define ZMQ_EVENT_BIND_FAILED 16
+
+#define ZMQ_EVENT_ACCEPTED 32
+#define ZMQ_EVENT_ACCEPT_FAILED 64
+
+#define ZMQ_EVENT_CLOSED 128
+#define ZMQ_EVENT_CLOSE_FAILED 256
+#define ZMQ_EVENT_DISCONNECTED 512
+#define ZMQ_EVENT_MONITOR_STOPPED 1024
+
+#define ZMQ_EVENT_ALL ( ZMQ_EVENT_CONNECTED | ZMQ_EVENT_CONNECT_DELAYED | \
+                        ZMQ_EVENT_CONNECT_RETRIED | ZMQ_EVENT_LISTENING | \
+                        ZMQ_EVENT_BIND_FAILED | ZMQ_EVENT_ACCEPTED | \
+                        ZMQ_EVENT_ACCEPT_FAILED | ZMQ_EVENT_CLOSED | \
+                        ZMQ_EVENT_CLOSE_FAILED | ZMQ_EVENT_DISCONNECTED | \
+                        ZMQ_EVENT_MONITOR_STOPPED)
+
+/*  Socket event data  */
+typedef struct {
+    uint16_t event;  // id of the event as bitfield
+    int32_t  value ; // value is either error code, fd or reconnect interval
+} zmq_event_t;
+
+ZMQ_EXPORT void *zmq_socket (void *, int type);
+ZMQ_EXPORT int zmq_close (void *s);
+ZMQ_EXPORT int zmq_setsockopt (void *s, int option, const void *optval,
+    size_t optvallen);
+ZMQ_EXPORT int zmq_getsockopt (void *s, int option, void *optval,
+    size_t *optvallen);
+ZMQ_EXPORT int zmq_bind (void *s, const char *addr);
+ZMQ_EXPORT int zmq_connect (void *s, const char *addr);
+ZMQ_EXPORT int zmq_unbind (void *s, const char *addr);
+ZMQ_EXPORT int zmq_disconnect (void *s, const char *addr);
+ZMQ_EXPORT int zmq_send (void *s, const void *buf, size_t len, int flags);
+ZMQ_EXPORT int zmq_send_const (void *s, const void *buf, size_t len, int flags);
+ZMQ_EXPORT int zmq_recv (void *s, void *buf, size_t len, int flags);
+ZMQ_EXPORT int zmq_socket_monitor (void *s, const char *addr, int events);
+
+ZMQ_EXPORT int zmq_sendmsg (void *s, zmq_msg_t *msg, int flags);
+ZMQ_EXPORT int zmq_recvmsg (void *s, zmq_msg_t *msg, int flags);
+
+/*  Experimental                                                              */
+struct iovec;
+
+ZMQ_EXPORT int zmq_sendiov (void *s, struct iovec *iov, size_t count, int flags);
+ZMQ_EXPORT int zmq_recviov (void *s, struct iovec *iov, size_t *count, int flags);
+
+/******************************************************************************/
+/*  I/O multiplexing.                                                         */
+/******************************************************************************/
+
+#define ZMQ_POLLIN 1
+#define ZMQ_POLLOUT 2
+#define ZMQ_POLLERR 4
+
+typedef struct
+{
+    void *socket;
+#if defined _WIN32
+    SOCKET fd;
+#else
+    int fd;
+#endif
+    short events;
+    short revents;
+} zmq_pollitem_t;
+
+#define ZMQ_POLLITEMS_DFLT 16
+
+ZMQ_EXPORT int zmq_poll (zmq_pollitem_t *items, int nitems, long timeout);
+
+/*  Built-in message proxy (3-way) */
+
+ZMQ_EXPORT int zmq_proxy (void *frontend, void *backend, void *capture);
+
+/*  Encode a binary key as printable text using ZMQ RFC 32  */
+ZMQ_EXPORT char *zmq_z85_encode (char *dest, uint8_t *data, size_t size);
+
+/*  Encode a binary key from printable text per ZMQ RFC 32  */
+ZMQ_EXPORT uint8_t *zmq_z85_decode (uint8_t *dest, char *string);
+
+/*  Deprecated aliases */
+#define ZMQ_STREAMER 1
+#define ZMQ_FORWARDER 2
+#define ZMQ_QUEUE 3
+/*  Deprecated method */
+ZMQ_EXPORT int zmq_device (int type, void *frontend, void *backend);
+
+#undef ZMQ_EXPORT
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

slsReceiverSoftware/src/UDPStandardImplementation.cpp

@@ -17,6 +17,9 @@
 #include <string.h>
 #include <stdint.h>
 #include <ctime>
+
+#include <zmq.h>	//zmq
+
 using namespace std;
 
 #define WRITE_HEADERS
@@ -44,6 +47,7 @@ UDPStandardImplementation::UDPStandardImplementation(){
 	}else if(system("echo 250000 > /proc/sys/net/core/netdev_max_backlog")){
 		FILE_LOG(logDEBUG) << "Warning: No root permission to change max length of input queue in file /proc/sys/net/core/netdev_max_backlog";
 	}
+
 	/** permanent setting by heiner
 	    net.core.rmem_max = 104857600 # 100MiB
 	    net.core.netdev_max_backlog = 250000
@@ -85,12 +89,12 @@ void UDPStandardImplementation::deleteMembers(){
 	}
 	for(int i=0; i<MAX_NUMBER_OF_WRITER_THREADS; i++){
 		if(latestData[i]) 	{delete[] latestData[i]; 	latestData[i] = NULL;}
-		guiData[i] = NULL;
 	}
 	//kill threads
 	if(threadStarted){
 		createListeningThreads(true);
 		createWriterThreads(true);
+		threadStarted = false;
 	}
 }
 
@@ -124,6 +128,7 @@ void UDPStandardImplementation::initializeMembers(){
 	FILE_LOG(logDEBUG) << "Info: Initializing members";
 
 	//***detector parameters***
+	detID = 0;
 	bufferSize = 0;
 	onePacketSize = 0;
 	oneDataSize = 0;
@@ -180,13 +185,13 @@ void UDPStandardImplementation::initializeMembers(){
 	//***receiver to GUI parameters***
 	for(int i=0; i<MAX_NUMBER_OF_WRITER_THREADS; i++){
 		latestData[i] = NULL;
-		guiDataReady[i] = false;
+		guiNumPackets[i] = 0;
-		guiData[i] = NULL;
 		strcpy(guiFileName[i],"");
 		frametoGuiCounter[i] = 0;
 	}
 
 	//***data callback thread parameters***
+	zmqThreadStarted = false;
 	numberofDataCallbackThreads = 1;
 	dataCallbackThreadsMask = 0x0;
 	killAllDataCallbackThreads = false;
@@ -321,6 +326,7 @@ int UDPStandardImplementation::setupFifoStructure(){
 	if(threadStarted){
 		createListeningThreads(true);
 		createWriterThreads(true);
+		threadStarted = false;
 	}
 
 
@@ -416,6 +422,44 @@ void UDPStandardImplementation::configure(map<string, string> config_map){
 }
 
 
+void UDPStandardImplementation::setFileName(const char c[]){
+	FILE_LOG(logDEBUG) << __AT__ << " starting";
+
+	char oldfilename[MAX_STR_LENGTH];
+	strcpy(oldfilename,fileName);
+
+	if(strlen(c))
+		strcpy(fileName, c);
+
+	if(strlen(fileName)){
+		int detindex = -1;
+		string tempname(fileName);
+		size_t uscore=tempname.rfind("_");
+		if (uscore!=string::npos){
+			if (sscanf(tempname.substr(uscore+1,tempname.size()-uscore-1).c_str(),"d%d",&detindex)) {
+				detID = detindex;
+			}
+		}
+		if(detindex == -1)
+			detID = 0;
+	}
+
+
+	if(dataCallbackEnabled && (strcmp(oldfilename,fileName))){cout<<"***Going to destroy data callback threads and create!!!"<<endl;
+		if(zmqThreadStarted){
+			createDataCallbackThreads(true);
+			zmqThreadStarted = false;
+		}
+		numberofDataCallbackThreads = MAX_NUMBER_OF_LISTENING_THREADS;
+		if(createDataCallbackThreads() == FAIL){
+			cprintf(BG_RED,"Error: Could not create data callback threads\n");
+		}
+	}
+
+
+	FILE_LOG(logINFO) << "File name:" << fileName;
+}
+
 int UDPStandardImplementation::setDataCompressionEnable(const bool b){
 	FILE_LOG(logDEBUG) << __AT__ << " starting";
 
@@ -442,9 +486,9 @@ int UDPStandardImplementation::setDataCompressionEnable(const bool b){
 	else
 		numberofWriterThreads = 1;
 	if(createWriterThreads() == FAIL){
-			cprintf(BG_RED,"Error: Could not create writer threads\n");
+		cprintf(BG_RED,"Error: Could not create writer threads\n");
-			return FAIL;
+		return FAIL;
-		}
+	}
 	//-- end of create writer threads
 	setThreadPriorities();
 
@@ -779,22 +823,26 @@ int UDPStandardImplementation::setDetectorType(const detectorType d){
 		pthread_mutex_unlock(&(statusMutex));
 		if(threadStarted){
 			createListeningThreads(true);
-			createDataCallbackThreads(true);
 			createWriterThreads(true);
 		}
 		numberofListeningThreads = MAX_NUMBER_OF_LISTENING_THREADS;
-		numberofDataCallbackThreads = MAX_NUMBER_OF_LISTENING_THREADS;
 		numberofWriterThreads = MAX_NUMBER_OF_WRITER_THREADS;
 	}
+	if(zmqThreadStarted)
+		createDataCallbackThreads(true);
+
 
 	//set up fifo structure -1 for numberofJobsPerBuffer ensure it is done
 	numberofJobsPerBuffer = -1;
 	setupFifoStructure();
 
+	numberofDataCallbackThreads = MAX_NUMBER_OF_LISTENING_THREADS;
+	if(dataCallbackEnabled)
+		createDataCallbackThreads();
+
 	//allocate for latest data (frame copy for gui), free variables
 	for(int i=0; i<MAX_NUMBER_OF_WRITER_THREADS; i++){
 		if(latestData[i]) 	{delete[] latestData[i]; 	latestData[i] = NULL;}
-		guiData[i] = NULL;
 		latestData[i] = new char[bufferSize];
 	}
 
@@ -803,7 +851,6 @@ int UDPStandardImplementation::setDetectorType(const detectorType d){
 	if(myDetectorType == EIGER){
 		for(int i=0; i<MAX_NUMBER_OF_WRITER_THREADS; i++)
 			updateFileHeader(i);
-		createDataCallbackThreads();
 	}
 
 	FILE_LOG(logDEBUG) << " Detector type set to " << getDetectorType(d);
@@ -867,8 +914,7 @@ int UDPStandardImplementation::startReceiver(char *c){
 		}
 		//reset gui variables
 		frametoGuiCounter[i] = 0;
-		guiData[i] = NULL;
+		guiNumPackets[i] = 0;
-		guiDataReady[i]=0;
 		strcpy(guiFileName[i],"");
 
 	}
@@ -920,8 +966,10 @@ int UDPStandardImplementation::startReceiver(char *c){
 		sprintf(completeFileName[0], "%s/%s_fxxx_%lld_xx.root", filePath,fileNamePerThread[0],(long long int)fileIndex);
 
 	//initialize semaphore to synchronize between writer and gui reader threads
-	for(int i=0;i<numberofWriterThreads;i++)
+	for(int i=0;i<numberofWriterThreads;i++){
 		sem_init(&writerGuiSemaphore[i],1,0);
+		sem_init(&dataCallbackWriterSemaphore[i],1,0);
+	}
 
 	//status and thread masks
 	pthread_mutex_lock(&statusMutex);
@@ -968,14 +1016,14 @@ void UDPStandardImplementation::stopReceiver(){
 
 	//wait until status is run_finished
 	while(status == TRANSMITTING){
-		for(int i=0; i < numberofWriterThreads; i++)
-			sem_post(&writerGuiSemaphore[i]);
 		usleep(5000);
 	}
 
 	//semaphore destroy
-	for(int i=0; i < numberofWriterThreads; i++)
+	for(int i=0; i < numberofWriterThreads; i++){
 		sem_destroy(&writerGuiSemaphore[i]);
+		sem_destroy(&dataCallbackWriterSemaphore[i]);
+	}
 
 	//change status
 	pthread_mutex_lock(&statusMutex);
@@ -1077,71 +1125,9 @@ void UDPStandardImplementation::startReadout(){
 void UDPStandardImplementation::readFrame(int ithread, char* c,char** raw, int64_t &startAcq, int64_t &startFrame){
 	FILE_LOG(logDEBUG) << __AT__ << " called";
 
-	//point to gui data, to let writer thread know that gui is back for data
-	if (guiData[ithread] == NULL){
-		guiData[ithread] = latestData[ithread];
-#ifdef DEBUG4
-		cprintf(CYAN,"Info: gui data not null anymore - ready to get data\n");
-#endif
-	}
-
-	//copy data and filename
-	strcpy(c,guiFileName[ithread]);
-	startAcq = startAcquisitionIndex;
-	startFrame = startFrameIndex;
-
-
-	//gui data not copied yet
-	if(!guiDataReady[ithread]){
-#ifdef DEBUG4
-		cprintf(CYAN,"Info: gui data not ready\n");
-#endif
-		*raw = NULL;
-	}
-
-	//gui data ready, pass address to gui to copy the data
-	else{
-#ifdef DEBUG4
-		cprintf(CYAN,"Info: gui data ready\n");
-#endif
-		*raw = guiData[ithread];
-
-		guiData[ithread] = NULL;
-
-		//for nth frame to gui, post semaphore so writer stops waiting
-		if((FrameToGuiFrequency) && (writerThreadsMask)){
-#ifdef DEBUG4
-			cprintf(CYAN,"Info: gonna post\n");
-#endif
-			//release after getting data
-			sem_post(&writerGuiSemaphore[ithread]);
-		}
-#ifdef DEBUG4
-		cprintf(CYAN,"Info: done post\n");
-#endif
-
-	}
 }
 
-/*
-void UDPStandardImplementation::resetGuiPointer(int ithread){
-	FILE_LOG(logDEBUG) << __AT__ << " called";
 
-	guiData[ithread] = NULL;
-
-	//for nth frame to gui, post semaphore so writer stops waiting
-	if((FrameToGuiFrequency) && (writerThreadsMask)){
-#ifdef DEBUG4
-		cprintf(CYAN,"Info: gonna post\n");
-#endif
-		//release after getting data
-		sem_post(&writerGuiSemaphore[ithread]);
-	}
-#ifdef DEBUG4
-	cprintf(CYAN,"Info: done post\n");
-#endif
-}
-*/
 
 void UDPStandardImplementation::closeFile(int ithread){
 	FILE_LOG(logDEBUG) << __AT__ << " called for " << ithread ;
@@ -1220,10 +1206,11 @@ int UDPStandardImplementation::createDataCallbackThreads(bool destroy){
 		for(int i = 0; i < numberofDataCallbackThreads; ++i){
 			sem_post(&dataCallbackSemaphore[i]);
 			pthread_join(dataCallbackThreads[i],NULL);
+			sem_destroy(&dataCallbackSemaphore[i]);
 			FILE_LOG(logDEBUG) << "." << flush;
 		}
 		killAllDataCallbackThreads = false;
-		threadStarted = false;
+		zmqThreadStarted = false;
 		FILE_LOG(logDEBUG) << "Info: Data Callback thread(s) destroyed";
 	}
 
@@ -1236,13 +1223,13 @@ int UDPStandardImplementation::createDataCallbackThreads(bool destroy){
 
 		for(int i = 0; i < numberofDataCallbackThreads; ++i){
 			sem_init(&dataCallbackSemaphore[i],1,0);
-			threadStarted = false;
+			zmqThreadStarted = false;
 			currentThreadIndex = i;
 			if(pthread_create(&dataCallbackThreads[i], NULL,startDataCallbackThread, (void*) this)){
 				FILE_LOG(logERROR) << "Could not create listening thread with index " << i;
 				return FAIL;
 			}
-			while(!threadStarted);
+			while(!zmqThreadStarted);
 			FILE_LOG(logDEBUG) << "." << flush;
 		}
 		FILE_LOG(logDEBUG) << "Info: Data Callback thread(s) created successfully.";
@@ -1270,6 +1257,7 @@ int UDPStandardImplementation::createListeningThreads(bool destroy){
 		for(int i = 0; i < numberofListeningThreads; ++i){
 			sem_post(&listenSemaphore[i]);
 			pthread_join(listeningThreads[i],NULL);
+			sem_destroy(&listenSemaphore[i]);
 			FILE_LOG(logDEBUG) << "." << flush;
 		}
 		killAllListeningThreads = false;
@@ -1321,6 +1309,7 @@ int UDPStandardImplementation::createWriterThreads(bool destroy){
 		for(int i = 0; i < numberofWriterThreads; ++i){
 			sem_post(&writerSemaphore[i]);
 			pthread_join(writingThreads[i],NULL);
+			sem_destroy(&writerSemaphore[i]);
 			FILE_LOG(logDEBUG) <<"."<<flush;
 		}
 		killAllWritingThreads = false;
@@ -1643,54 +1632,128 @@ void* UDPStandardImplementation::startWritingThread(void* this_pointer){
 
 
 
-void UDPStandardImplementation::startDataCallback(){
+void UDPStandardImplementation::startDataCallback(){cprintf(MAGENTA,"start data call back thread started %d\n",currentThreadIndex);
 	FILE_LOG(logDEBUG) << __AT__ << " called";
 
 	//set current thread value  index
 	int ithread = currentThreadIndex;
 	//let calling function know thread started and obtained current
-	threadStarted = 1;
+	zmqThreadStarted = 1;
-	char* buffer;
+
     // server address to bind
-    const char *hostName = "tcp://127.0.0.1:70001";/**increment this by ithread and detid*/
+    char hostName[100] = "tcp://127.0.0.1:";
+    int portno = DEFAULT_ZMQ_PORTNO + (detID*2+ithread);
+    sprintf(hostName,"%s%d",hostName,portno);
+    FILE_LOG(logINFO) << "Thread" << ithread << ": ZMQ Server at " << hostName;
 
 	/* outer loop - loops once for each acquisition */
 	//infinite loop, exited only to change dynamic range, 10G parameters etc (then recreated again)
 	while(true){
 
+		int oneframesize = oneDataSize * packetsPerFrame;
+		char* buffer = new char[packetsPerFrame*oneDataSize];
+		memset(buffer,0xFF,oneframesize);
+		int bufferoffset = 0;
+		int size = 0;
+		int offset=0;
+		int currentfnum = 0;
+		uint64_t fnum = 0;
+		uint32_t pnum = 0;
+
 		void *context = zmq_ctx_new();
 		// create a publisher
-		socket = zmq_socket(context, ZMQ_PUB);
+		void *zmqsocket = zmq_socket(context, ZMQ_PUB);
 		// bind
-		zmq_bind(socket,hostName);/**increment this by 1*/
+		zmq_bind(zmqsocket,hostName);
 
+
+		currentfnum = -1;
 		/* inner loop - loop for each buffer */
-		//until mask reset (udp sockets shut down by client)
+		//until mask reset (dummy pcaket got by writer)
 		while((1 << ithread) & dataCallbackThreadsMask){
 
-			//wait for data
+			//let the writer thread continue, while we process carry over if any
-			sem_wait(&dataCallbackSemaphore[ithread]);
+			sem_post(&writerGuiSemaphore[ithread]);
-			if(status == TRANSMITTING)
+			//wait for receiver to send more data
+			sem_wait(&dataCallbackWriterSemaphore[ithread]);
+
+			//everything is done
+			if(guiNumPackets[ithread] == dummyPacketValue){
+
+
+				 /**suing this in clientzmq_msg_more,
+				  * in serve use zmq_msg_send (&message, sender, ZMQ_SNDMORE); and 0 for last packet, but better to check lengt*/
+				zmq_send (zmqsocket, "end", 3, 0);
+
+
+				pthread_mutex_lock(&statusMutex);
+				dataCallbackThreadsMask^=(1<<ithread);
+				pthread_mutex_unlock(&statusMutex);
 				continue;
-			int numpackets = (uint32_t)(*( (uint32_t*) latestData)); /*latestdata should be size of one buffer*/
+			}
-			memcpy(buffer, latestData, numpackets*onePacketSize);/**read first bytes to get numpackets*/
+
+
+			size = guiNumPackets[ithread]*onePacketSize;
+			offset=0;
+
+			while(offset < size){
+
+				//until getting frame number is not error
+				while((size>0) && (getFrameandPacketNumber(ithread, latestData[ithread]+offset, fnum, pnum)==FAIL)){
+					offset+= onePacketSize;
+				}
+
+				//end of buffer
+				if(offset >= size)
+					break;
+
+				//new frame
+				if(currentfnum==-1){
+					currentfnum = fnum;
+				}
+
+				//last packet
+				if(pnum == packetsPerFrame){
+					memcpy(buffer+((pnum-1)*oneDataSize), latestData[ithread]+offset+8,oneDataSize);
+					offset+= onePacketSize;
+					zmq_send(zmqsocket, buffer, oneframesize, 0);
+					memset(buffer,0xFF,oneframesize);
+					currentfnum = -1;
+				}
+				//same frame (not last) or next frame
+				else {
+					//next frame
+					if(fnum > currentfnum){
+						zmq_send(zmqsocket, buffer, oneframesize, 0);
+						memset(buffer,0xFF,oneframesize);
+						currentfnum = fnum;
+					}
+
+					memcpy(buffer+((pnum-1)*oneDataSize), latestData[ithread]+offset+8,oneDataSize);
+					offset+= onePacketSize;
+				}
+
+			}
 
-			/*check if it should be added to previous (processlistening buffer for datacompression)*/
 
-			 zmq_send(socket, buffer.data(), buffer.size(), 0);
 
 		}/*--end of loop for each buffer (inner loop)*/
 
+		//free resources
+		delete[] buffer;
+	    zmq_unbind(zmqsocket, hostName);
+	    zmq_close(zmqsocket);
+	    zmq_ctx_destroy(context);
+
+
+
 		//end of acquisition, wait for next acquisition/change of parameters
 		sem_wait(&dataCallbackSemaphore[ithread]);
 
+
 		//check to exit thread (for change of parameters) - only EXIT possibility
 		if(killAllDataCallbackThreads){
-			cprintf(BLUE,"DataCallback_Thread %d:Goodbye!\n",ithread);
+			cprintf(MAGENTA,"DataCallback_Thread %d:Goodbye!\n",ithread);
-			//free resources at exit
-		    zmq_unbind(socket, hostName);
-		    zmq_close(socket);
-		    zmq_ctx_destroy(context);
 			pthread_exit(NULL);
 		}
 
@@ -1701,7 +1764,7 @@ void UDPStandardImplementation::startDataCallback(){
 
 
 
-void UDPStandardImplementation::startListening(){
+void UDPStandardImplementation::startListening(){cprintf(BLUE,"startlistening thread started %d\n",currentThreadIndex);
 	FILE_LOG(logDEBUG) << __AT__ << " called";
 
 	//set current thread value  index
@@ -2136,7 +2199,7 @@ uint32_t UDPStandardImplementation::processListeningBuffer(int ithread, int &cSi
 
 
 
-void UDPStandardImplementation::startWriting(){
+void UDPStandardImplementation::startWriting(){cprintf(GREEN,"start writing thread started %d\n",currentThreadIndex);
 	FILE_LOG(logDEBUG) << __AT__ << " called";
 
 	//set current thread value  index
@@ -2157,7 +2220,6 @@ void UDPStandardImplementation::startWriting(){
 
 		//--reset parameters before acquisition
 		nf = 0;
-		guiData[ithread] = latestData[ithread];  		//so that the first frame is always copied
 		if(dataCompressionEnable)
 			listenfifoIndex = 0;						//compression has only one listening thread
 
@@ -2255,7 +2317,7 @@ void UDPStandardImplementation::waitWritingBufferForNextAcquisition(int ithread)
 		if(myDetectorType == EIGER){
 			int detindex = -1;
 			string tempname(fileName);
-
+			//detid (more than 1 half module)
 			size_t uscore=tempname.rfind("_");
 			if (uscore!=string::npos){
 				if (sscanf(tempname.substr(uscore+1,tempname.size()-uscore-1).c_str(),"d%d",&detindex)) {
@@ -2263,11 +2325,12 @@ void UDPStandardImplementation::waitWritingBufferForNextAcquisition(int ithread)
 					sprintf(fileNamePerThread[ithread],"%s_d%d",tempname.c_str(),detindex*2+ithread);
 				}
 			}
-
+			//only one half module, so no detid
 			if(detindex == -1)
 				sprintf(fileNamePerThread[ithread],"%s_d%d",fileName,ithread);
 
-		}
+		}else
+			strcpy(fileNamePerThread[0],fileName);
 
 		if(dataCompressionEnable){
 #ifdef MYROOT1
@@ -2335,6 +2398,14 @@ void UDPStandardImplementation::stopWriting(int ithread, char* wbuffer){
 		cprintf(YELLOW,"Writing_Thread %d: Freeing dummy-end buffer. Pushed into fifofree %p for listener %d\n", ithread,(void*)(wbuffer),ithread);
 #endif
 
+	if(dataCallbackEnabled){
+		//ensure previous frame was processed
+		sem_wait(&writerGuiSemaphore[ithread]);
+		guiNumPackets[ithread] = dummyPacketValue;
+		//let it know its got data
+		sem_post(&dataCallbackWriterSemaphore[ithread]);
+	}
+
 
 	//all threads need to close file, reset mask and exit loop
 	closeFile(ithread);
@@ -2349,18 +2420,12 @@ void UDPStandardImplementation::stopWriting(int ithread, char* wbuffer){
 	//thread 0 waits for all threads to finish & print statistics
 	if(ithread == 0){
 		//wait for all other threads
-		if(dataCompressionEnable){
+		while(writerThreadsMask)
-			cprintf(GREEN,"Writing_Thread %d: Waiting for jobs to be done.. current mask:0x%x\n",ithread, writerThreadsMask);
+			usleep(5000);
-			while(writerThreadsMask){
-				/*cout << "." << flush;*/
-				usleep(50000);
-			}
-			cprintf(GREEN,"Writing_Thread %d: Jobs Done!\n",ithread);
-		}
-
 		//ensure listening threads done before updating status as it returns to client (from stopReceiver)
 		while(listeningThreadsMask)
 			usleep(5000);
+		//ensure datacallbacks threads are done
 		while(dataCallbackThreadsMask)
 			usleep(5000);
 		//update status
@@ -2412,7 +2477,8 @@ void UDPStandardImplementation::handleWithoutDataCompression(int ithread, char*
 
 	//get current frame number
 	uint64_t tempframenumber;
-	if(getFrameNumber(ithread, wbuffer + HEADER_SIZE_NUM_TOT_PACKETS,tempframenumber) == FAIL){
+	uint32_t pnum;
+	if(getFrameandPacketNumber(ithread, wbuffer + HEADER_SIZE_NUM_TOT_PACKETS,tempframenumber,pnum) == FAIL){
 		//error in frame number sent by fpga
 		while(!fifoFree[ithread]->push(wbuffer));
 
@@ -2432,7 +2498,7 @@ void UDPStandardImplementation::handleWithoutDataCompression(int ithread, char*
 	//callback to write data
 	if (cbAction < DO_EVERYTHING)
 		rawDataReadyCallBack((int)currentFrameNumber[ithread], wbuffer + HEADER_SIZE_NUM_TOT_PACKETS, npackets * onePacketSize,
-				sfilefd[ithread], guiData[ithread],pRawDataReady);//know which thread from sfilefd
+				sfilefd[ithread], latestData[ithread],pRawDataReady);//know which thread from sfilefd
 
 
 
@@ -2446,7 +2512,7 @@ void UDPStandardImplementation::handleWithoutDataCompression(int ithread, char*
 
 	//copy frame for gui
 	//if(npackets >= (packetsPerFrame/numberofListeningThreads))
-	if(npackets)
+	if(dataCallbackEnabled && npackets)
 		copyFrameToGui(ithread, wbuffer,npackets);
 #ifdef DEBUG4
 	cprintf(GREEN,"Writing_Thread: Copied frame\n");
@@ -2487,8 +2553,9 @@ void UDPStandardImplementation::writeFileWithoutCompression(int ithread, char* w
 			if(numpackets &&(lastFrameNumberInFile[ithread]>=0)){
 				//get start frame (required to create new file at the right juncture)
 				uint64_t startframe =-1;
+				uint32_t pnum;
 				//if(ithread) cout<<"getting start frame number"<<endl;
-				if(getFrameNumber(ithread, wbuffer + offset, startframe) == FAIL){
+				if(getFrameandPacketNumber(ithread, wbuffer + offset, startframe,pnum) == FAIL){
 					//error in frame number sent by fpga
 					while(!fifoFree[ithread]->push(wbuffer));
 					return;
@@ -2545,7 +2612,8 @@ void UDPStandardImplementation::writeFileWithoutCompression(int ithread, char* w
 		if(numpackets){
 			//get last frame number
 			uint64_t finalLastFrameNumberToSave = 0;
-			if(getFrameNumber(ithread, wbuffer + HEADER_SIZE_NUM_TOT_PACKETS + ((numpackets - 1) * onePacketSize), finalLastFrameNumberToSave) == FAIL){
+			uint32_t pnum;
+			if(getFrameandPacketNumber(ithread, wbuffer + HEADER_SIZE_NUM_TOT_PACKETS + ((numpackets - 1) * onePacketSize), finalLastFrameNumberToSave,pnum) == FAIL){
 				//error in frame number sent by fpga
 				while(!fifoFree[ithread]->push(wbuffer));
 				return;
@@ -2621,50 +2689,32 @@ void UDPStandardImplementation::updateFileHeader(int ithread){
 
 }
 
-
+//called only if datacallback enabled
 void UDPStandardImplementation::copyFrameToGui(int ithread, char* buffer, uint32_t numpackets){
 	FILE_LOG(logDEBUG) << __AT__ << " called";
 
-
-
 	//if nthe frame, wait for your turn (1st frame always shown as its zero)
 	if(FrameToGuiFrequency && ((frametoGuiCounter[ithread])%FrameToGuiFrequency));
 
 	//random read (gui ready) or nth frame read: gui needs data now or it is the first frame
 	else{
 
-		//tell datacallback to pick up data
-		sem_post(&dataCallbackSemaphore[ithread]);
-
-
-		memcpy(latestData[ithread],buffer , numpackets*onePacketSize);
-
 #ifdef DEBUG4
-		cprintf(GREEN,"Writing_Thread: CopyingFrame: Gui needs data now OR 1st frame\n");
+		cprintf(GREEN,"Writing_Thread: CopyingFrame: Going to copy data\n");
 #endif
-		pthread_mutex_lock(&dataReadyMutex);
+		//ensure previous frame was processed
-		guiDataReady[ithread]=0;
+		sem_wait(&writerGuiSemaphore[ithread]);
-#ifdef DEBUG4
+
-		cprintf(GREEN,"Writing_Thread: CopyingFrame: guidataready is  0, Copying data\n");
+		//copy date
-#endif
+		guiNumPackets[ithread] = numpackets;
-		memcpy(latestData[ithread],buffer , numpackets*onePacketSize);
 		strcpy(guiFileName[ithread],completeFileName[ithread]);
-		guiDataReady[ithread]=1;
+		memcpy(latestData[ithread],buffer+ HEADER_SIZE_NUM_TOT_PACKETS , numpackets*onePacketSize);
-		pthread_mutex_unlock(&dataReadyMutex);
+		//let it know its got data
-#ifdef DEBUG4
+		sem_post(&dataCallbackWriterSemaphore[ithread]);
-		cprintf(GREEN,"Writing_Thread: CopyingFrame: Copied Data, guidataready is 1\n");
-#endif
 
-		//nth frame read, block current process if the guireader hasnt read it yet
-		if(FrameToGuiFrequency){
 #ifdef DEBUG4
-			cprintf(GREEN,"Writing_Thread: CopyingFrame: Waiting after copying\n");
+		cprintf(GREEN,"Writing_Thread: CopyingFrame: Copied Data\n");
 #endif
-			sem_wait(&writerGuiSemaphore[ithread]);
-#ifdef DEBUG4
-			cprintf(GREEN,"Writing_Thread: CopyingFrame: Done waiting\n");
-#endif
-		}
 
 	}
 
@@ -2684,7 +2734,8 @@ void UDPStandardImplementation::handleDataCompression(int ithread, char* wbuffer
 
 	//get frame number
 	uint64_t tempframenumber=-1;
-	if(getFrameNumber(ithread, wbuffer + HEADER_SIZE_NUM_TOT_PACKETS, tempframenumber) == FAIL){
+	uint32_t pnum;
+	if(getFrameandPacketNumber(ithread, wbuffer + HEADER_SIZE_NUM_TOT_PACKETS, tempframenumber,pnum) == FAIL){
 		//error in frame number sent by fpga
 		while(!fifoFree[ithread]->push(wbuffer));
 		return;
@@ -2795,7 +2846,8 @@ void UDPStandardImplementation::handleDataCompression(int ithread, char* wbuffer
 
 #endif
 			if(!once){
-				copyFrameToGui(ithread, buff[0],(uint32_t)packetsPerFrame);
+				if(dataCallbackEnabled)
+					copyFrameToGui(ithread, buff[0],(uint32_t)packetsPerFrame);
 				once = 1;
 			}
 		}
@@ -2819,59 +2871,60 @@ void UDPStandardImplementation::handleDataCompression(int ithread, char* wbuffer
 
 
 
-int UDPStandardImplementation::getFrameNumber(int ithread, char* wbuffer, uint64_t &tempframenumber){
+int UDPStandardImplementation::getFrameandPacketNumber(int ithread, char* wbuffer, uint64_t &framenumber, uint32_t &packetnumber){
 	FILE_LOG(logDEBUG) << __AT__ << " called";
 
 	eiger_packet_footer_t* footer=0;
 	jfrau_packet_header_t* header=0;
-	int pnum=-1;
 
 	switch(myDetectorType){
 
 	case EIGER:
 		footer = (eiger_packet_footer_t*)(wbuffer + footerOffset);
-		tempframenumber = (uint32_t)(*( (uint64_t*) footer));
+		framenumber = (uint32_t)(*( (uint64_t*) footer));
 		//error in frame number sent by fpga
 		if(!((uint32_t)(*( (uint64_t*) footer)))){
-			tempframenumber = -1;
+			framenumber = -1;
 			FILE_LOG(logERROR) << "Fifo "<< ithread << ": Frame Number is zero from firmware.";
 			return FAIL;
 		}
+		packetnumber = (*( (uint16_t*) footer->packetNumber));
 #ifdef DEBUG4
 		if(!ithread) cprintf(GREEN,"Writing_Thread %d: fnum:%lld pnum:%d FPGA_fnum:%d footeroffset:%d\n",
 				ithread,
-				(long long int)tempframenumber,
+				(long long int)framenumber,
-				(*( (uint16_t*) footer->packetNumber)),
+				packetnumber,
-				(uint32_t)(*( (uint64_t*) footer)),
+				framenumber,
 				footerOffset);
 #endif
-		tempframenumber += (startFrameIndex - 1);
+		framenumber += (startFrameIndex - 1);
 		break;
 
 	case JUNGFRAU:
 		header = (jfrau_packet_header_t*)(wbuffer);
-		tempframenumber = (*( (uint32_t*) header->frameNumber))&frameIndexMask;
+		framenumber = (*( (uint32_t*) header->frameNumber))&frameIndexMask;
+		packetnumber = (uint32_t)(*( (uint8_t*) header->packetNumber));
 #ifdef DEBUG4
 		cprintf(GREEN, "Writing_Thread %d: fnum:%lld\t pnum:%d\n",
-				(long long int)tempframenumber,
+				(long long int)framenumber,
-				(*( (uint8_t*) header->packetNumber)));
+				packetnumber);
 #endif
-		tempframenumber += startFrameIndex;
+		framenumber += startFrameIndex;
 		break;
 
 	default:
-		tempframenumber = ((uint32_t)(*((uint32_t*)(wbuffer))));
+		framenumber = ((uint32_t)(*((uint32_t*)(wbuffer))));
 		//for gotthard and normal frame, increment frame number to separate fnum and pnum
 		if (myDetectorType == PROPIX ||(myDetectorType == GOTTHARD && shortFrameEnable == -1))
-			tempframenumber++;
+			framenumber++;
-		pnum = tempframenumber&packetIndexMask;
+		packetnumber = framenumber&packetIndexMask;
-		tempframenumber = (tempframenumber & frameIndexMask) >> frameIndexOffset;
+		framenumber = (framenumber & frameIndexMask) >> frameIndexOffset;
 #ifdef DEBUG4
 		cprintf(GREEN, "Writing_Thread %d: fnum:%lld\t pnum:%d\n",
-				(long long int)tempframenumber,
+				(long long int)framenumber,
-				pnum);
+				packetnumber);
 #endif
-		tempframenumber += startFrameIndex;
+		framenumber += startFrameIndex;
 		break;
 	}
 	return OK;
@@ -2890,9 +2943,10 @@ int UDPStandardImplementation::writeUptoFrameNumber(int ithread, char* wbuffer,
 	int endoffset = startoffset + numpackets * onePacketSize;
 	uint64_t tempframenumber=-1;
 	offset = endoffset;
+	uint32_t pnum;
 
 	//get last frame number
-	if(getFrameNumber(ithread, wbuffer + (endoffset-onePacketSize), tempframenumber) == FAIL){
+	if(getFrameandPacketNumber(ithread, wbuffer + (endoffset-onePacketSize), tempframenumber,pnum) == FAIL){
 		//error in frame number sent by fpga
 		while(!fifoFree[ithread]->push(wbuffer));
 		return FAIL;
@@ -2918,7 +2972,7 @@ int UDPStandardImplementation::writeUptoFrameNumber(int ithread, char* wbuffer,
 		offset -= bigIncrements;
 		if(offset<startoffset)
 			break;//if(ithread) cout<<"frame number at going backwards fast f#:"<<tempframenumber<< " offset:"<<offset<<endl;
-		if(getFrameNumber(ithread, wbuffer + offset, tempframenumber) == FAIL){
+		if(getFrameandPacketNumber(ithread, wbuffer + offset, tempframenumber,pnum) == FAIL){
 			//error in frame number sent by fpga
 			while(!fifoFree[ithread]->push(wbuffer));
 			return FAIL;
@@ -2926,7 +2980,7 @@ int UDPStandardImplementation::writeUptoFrameNumber(int ithread, char* wbuffer,
 	}
 	if(offset<startoffset){
 		offset = startoffset;//if(ithread) cout<<"offset < start offset f#:"<<tempframenumber<< " offset:"<<offset<<endl;
-		if(getFrameNumber(ithread, wbuffer + offset, tempframenumber) == FAIL){
+		if(getFrameandPacketNumber(ithread, wbuffer + offset, tempframenumber,pnum) == FAIL){
 			//error in frame number sent by fpga
 			while(!fifoFree[ithread]->push(wbuffer));
 			return FAIL;
@@ -2934,7 +2988,7 @@ int UDPStandardImplementation::writeUptoFrameNumber(int ithread, char* wbuffer,
 	}
 	while(tempframenumber<nextFrameNumber){
 		offset += onePacketSize;//if(ithread) cout<<"frame number at going forwards slow f#:"<<tempframenumber<< " offset:"<<offset<<endl;
-		if(getFrameNumber(ithread, wbuffer + offset, tempframenumber) == FAIL){
+		if(getFrameandPacketNumber(ithread, wbuffer + offset, tempframenumber,pnum) == FAIL){
 			//error in frame number sent by fpga
 			while(!fifoFree[ithread]->push(wbuffer));
 			return FAIL;

slsReceiverSoftware/src/slsReceiverTCPIPInterface.cpp

@@ -353,6 +353,10 @@ int slsReceiverTCPIPInterface::set_detector_type(){
 			sprintf(mess,"Receiver locked by %s\n", socket->lastClientIP);
 			ret=FAIL;
 		}
+		else if((receiverBase)&&(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING)){
+			strcpy(mess,"Can not set detector type while receiver not idle\n");
+			ret = FAIL;
+		}
 		else{
 
 			switch(dr){
@@ -443,6 +447,10 @@ int slsReceiverTCPIPInterface::set_file_name() {
 			strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 			ret=FAIL;
 		}
+		else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
+			strcpy(mess,"Can not set file name while receiver not idle\n");
+			ret = FAIL;
+		}
 		else{
 			receiverBase->setFileName(fName);
 			retval = receiverBase->getFileName();
@@ -506,15 +514,15 @@ int slsReceiverTCPIPInterface::set_file_dir() {
 		if (lockStatus==1 && socket->differentClients==1){
 			sprintf(mess,"Receiver locked by %s\n", socket->lastClientIP);
 			ret=FAIL;
-		}/*
+		}
-		else if((strlen(fPath))&&(receiverBase->getStatus()==RUNNING)){
-			strcpy(mess,"Can not set file path while receiver running\n");
-			ret = FAIL;
-		}*/
 		else if (receiverBase == NULL){
 			strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 			ret=FAIL;
 		}
+		else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
+			strcpy(mess,"Can not set file path while receiver not idle\n");
+			ret = FAIL;
+		}
 		else{
 			receiverBase->setFilePath(fPath);
 			retval = receiverBase->getFilePath();
@@ -584,6 +592,10 @@ int slsReceiverTCPIPInterface::set_file_index() {
 			strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 			ret=FAIL;
 		}
+		else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
+			strcpy(mess,"Can not set file index while receiver not idle\n");
+			ret = FAIL;
+		}
 		else{
 			if(index >= 0)
 				receiverBase->setFileIndex(index);
@@ -648,6 +660,10 @@ int slsReceiverTCPIPInterface::set_frame_index() {
 			strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 			ret=FAIL;
 		}
+		else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
+			strcpy(mess,"Can not set frame index while receiver not idle\n");
+			ret = FAIL;
+		}
 		else{
 			//client sets to 0, but for receiver it is just an enable
 			//client uses this value for other detectors not using receiver,
@@ -725,9 +741,9 @@ int slsReceiverTCPIPInterface::setup_udp(){
 			strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 			ret=FAIL;
 		}
-		else if(receiverBase->getStatus()==RUNNING){
+		else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
+			strcpy(mess,"Can not set up udp while receiver not idle\n");
 			ret = FAIL;
-			strcpy(mess,"cannot set up udp when receiver is running\n");
 		}
 		else{
 			//set up udp port
@@ -859,9 +875,12 @@ int slsReceiverTCPIPInterface::stop_receiver(){
 		ret=FAIL;
 	}
 	else{
-		if(receiverBase->getStatus()!=IDLE)
+		if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
 			receiverBase->stopReceiver();
+			cout<<"receiver stopped"<<endl;
+		}
 		s = receiverBase->getStatus();
+		cout<<"to stop, receiver status:"<<s<<endl;
 		if(s==IDLE)
 			ret = OK;
 		else{
@@ -1060,7 +1079,7 @@ int slsReceiverTCPIPInterface::set_short_frame() {
 			strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 			ret=FAIL;
 		}
-		else if(receiverBase->getStatus()==RUNNING){
+		else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
 			strcpy(mess,"Cannot set short frame while status is running\n");
 			ret=FAIL;
 		}
@@ -2077,6 +2096,10 @@ int slsReceiverTCPIPInterface::set_read_frequency(){
 			strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 			ret=FAIL;
 		}
+		else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
+			strcpy(mess,"Can not set receiver frequency mode while receiver not idle\n");
+			ret = FAIL;
+		}
 		/*
 		else if((receiverBase->getStatus()==RUNNING) && (index >= 0)){
 			ret = FAIL;
@@ -2142,6 +2165,10 @@ int slsReceiverTCPIPInterface::enable_file_write(){
 			strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 			ret=FAIL;
 		}
+		else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
+			strcpy(mess,"Can not set file write mode while receiver not idle\n");
+			ret = FAIL;
+		}
 		else{
 			if(enable >= 0)
 				receiverBase->setFileWriteEnable(enable);
@@ -2213,7 +2240,12 @@ int	slsReceiverTCPIPInterface::start_readout(){
 	if (receiverBase == NULL){
 		strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 		ret=FAIL;
-	}else{
+	}
+	else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
+		strcpy(mess,"Can not start receiver readout while receiver not idle\n");
+		ret = FAIL;
+	}
+	else{
 		receiverBase->startReadout();
 		retval = receiverBase->getStatus();
 		if((retval == TRANSMITTING) || (retval == RUN_FINISHED) || (retval == IDLE))
@@ -2269,6 +2301,10 @@ int slsReceiverTCPIPInterface::set_timer() {
 			strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 			ret=FAIL;
 		}
+		else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
+			strcpy(mess,"Can not set timer while receiver not idle\n");
+			ret = FAIL;
+		}
 		else{
 			if(index[0] == FRAME_PERIOD){
 				if(index[1]>=0){
@@ -2344,7 +2380,7 @@ int slsReceiverTCPIPInterface::enable_compression() {
 				strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 				ret=FAIL;
 			}
-			else if(receiverBase->getStatus()==RUNNING){
+			else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
 				strcpy(mess,"Cannot enable/disable compression while status is running\n");
 				ret=FAIL;
 			}
@@ -2413,6 +2449,10 @@ int slsReceiverTCPIPInterface::set_detector_hostname() {
 			strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 			ret=FAIL;
 		}
+		else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
+			strcpy(mess,"Can not set detector hostname while receiver not idle\n");
+			ret = FAIL;
+		}
 		else{
 			receiverBase->initialize(hostname);
 			retval = receiverBase->getDetectorHostname();
@@ -2494,7 +2534,12 @@ int slsReceiverTCPIPInterface::set_dynamic_range() {
 			if (receiverBase == NULL){
 				strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 				ret=FAIL;
-			}else{
+			}
+			else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
+				strcpy(mess,"Can not set dynamic range while receiver not idle\n");
+				ret = FAIL;
+			}
+			else{
 				if(dr > 0){
 					ret = receiverBase->setDynamicRange(dr);
 					if(ret == FAIL)
@@ -2571,6 +2616,10 @@ int slsReceiverTCPIPInterface::enable_overwrite() {
 			strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 			ret=FAIL;
 		}
+		else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
+			strcpy(mess,"Can not set overwrite mode while receiver not idle\n");
+			ret = FAIL;
+		}
 		else{
 			if(index >= 0)
 				receiverBase->setOverwriteEnable(index);
@@ -2634,6 +2683,10 @@ int slsReceiverTCPIPInterface::enable_tengiga() {
 			strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 			ret=FAIL;
 		}
+		else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
+			strcpy(mess,"Can not set up 1Giga/10Giga mode while receiver not idle\n");
+			ret = FAIL;
+		}
 		else{
 			if(val >= 0)
 				ret = receiverBase->setTenGigaEnable(val);
@@ -2699,7 +2752,7 @@ int slsReceiverTCPIPInterface::set_fifo_depth() {
 				strcpy(mess,SET_RECEIVER_ERR_MESSAGE);
 				ret=FAIL;
 			}
-			else if(receiverBase->getStatus()==RUNNING){
+			else if(receiverBase->getStatus()==RUNNING || receiverBase->getStatus()==TRANSMITTING){
 				strcpy(mess,"Cannot set/get fifo depth while status is running\n");
 				ret=FAIL;
 			}