910/frame synchronizer (#1086)

* wip: draft of frame synchronizer, semaphores not done yet * signal handler not affecting semaphore inside lambda function * finally works with sem * install targets cmake error fix * reducing the number of data call backs. incoming from developer * WIP: of synchronisation (#969) * WIP of synchronisation * working so far if everything goes right * added all information into json headers * valid json * allow frame synchronizer to have access to static libzmq when compiling on conda (libzeromq-devel not installed by default * upto date with multirecieverapp for invalid arguments and help * formatting * remove warnings * changes to print * removed prints * no need for print frames to be called * minor * commnet * adding json header in start callback, imagesize in data callback and formatted * fixed sanitizer issues. 1 left for ctrl+C - zmq_msg_t should be deleted, not freed. Same with the char arrays and semaphores. * fixed sanitizer issues and made it more readable * moving clearing old frames to new startacq just in case it has to process soem frames before the callback * created a callback mutex to register the callbacks when using threads for different Receiver objects instead of child processes, remove the clean up of the frames (deleting from a signal is thread unsafe) from the siginterrupt handler, reading or setting terminate should also be inside the mutex, pass receiver object index so that only the first one cleans up the shared structure --------- Co-authored-by: Felix Engelmann <[email protected]>
slsdetectorgroup · Feb 18, 2025 · 7494e41 · 7494e41
1 parent 7607fdc
commit 7494e41
Show file tree

Hide file tree

Showing 7 changed files with 700 additions and 17 deletions.
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -249,8 +249,8 @@ endif()
 if(SLS_USE_SANITIZER)
     target_compile_options(slsProjectOptions INTERFACE -fsanitize=address,undefined -fno-omit-frame-pointer)
     target_link_libraries(slsProjectOptions INTERFACE -fsanitize=address,undefined)
-    # target_compile_options(slsProjectOptions INTERFACE -fsanitize=thread)
-    # target_link_libraries(slsProjectOptions INTERFACE -fsanitize=thread)
+    #target_compile_options(slsProjectOptions INTERFACE -fsanitize=thread -fno-omit-frame-pointer)
+    #target_link_libraries(slsProjectOptions INTERFACE -fsanitize=thread)
 endif()
 
 

diff --git a/python/scripts/frameSynchronizerPullSocket.py b/python/scripts/frameSynchronizerPullSocket.py
@@ -0,0 +1,19 @@
+# SPDX-License-Identifier: LGPL-3.0-or-other
+# Copyright (C) 2021 Contributors to the SLS Detector Package
+# Script to get combined zmq packets from frame synchronizer using pull zmq sockets
+import json
+import zmq
+
+c = zmq.Context()
+s = c.socket(zmq.PULL)
+s.connect("tcp://127.0.0.1:5555")
+
+while True:
+     m = s.recv_multipart()
+     for p in m:
+         if p.startswith(b"{"):
+             print(p.decode().strip())
+         else:
+             print("binary")
+     print("--------")
+
diff --git a/slsReceiverSoftware/CMakeLists.txt b/slsReceiverSoftware/CMakeLists.txt
@@ -138,13 +138,34 @@ if (SLS_USE_RECEIVER_BINARIES)
         slsProjectWarnings
     )
 
-    install(TARGETS slsReceiver slsMultiReceiver
+    add_executable(slsFrameSynchronizer
+        src/FrameSynchronizerApp.cpp
+    )
+
+    set_target_properties(slsFrameSynchronizer PROPERTIES
+        RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/bin
+    )
+    if((CMAKE_BUILD_TYPE STREQUAL "Release") AND SLS_LTO_AVAILABLE)
+        set_property(TARGET slsFrameSynchronizer PROPERTY INTERPROCEDURAL_OPTIMIZATION True)
+    endif()
+
+    target_link_libraries(slsFrameSynchronizer 
+    PUBLIC
+        slsReceiverStatic
+        pthread
+        rt
+    PRIVATE
+        slsProjectWarnings
+        "$<BUILD_INTERFACE:libzmq-static>"
+    )
+
+    install(TARGETS slsReceiver slsMultiReceiver slsFrameSynchronizer
         EXPORT "${TARGETS_EXPORT_NAME}"
         RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR}
         LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR}
         ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR}
         PUBLIC_HEADER DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}/sls
-)
+    )
 
 endif(SLS_USE_RECEIVER_BINARIES)
 

diff --git a/slsReceiverSoftware/src/ClientInterface.cpp b/slsReceiverSoftware/src/ClientInterface.cpp
@@ -33,6 +33,8 @@ using Interface = ServerInterface;
 #define gettid() syscall(SYS_gettid)
 #endif
 
+std::mutex ClientInterface::callbackMutex;
+
 ClientInterface::~ClientInterface() {
     killTcpThread = true;
     LOG(logINFO) << "Shutting down TCP Socket on port " << portNumber;
@@ -55,12 +57,14 @@ std::string ClientInterface::getReceiverVersion() { return APIRECEIVER; }
 /***callback functions***/
 void ClientInterface::registerCallBackStartAcquisition(
     int (*func)(const startCallbackHeader, void *), void *arg) {
+    std::lock_guard<std::mutex> lock(callbackMutex);
     startAcquisitionCallBack = func;
     pStartAcquisition = arg;
 }
 
 void ClientInterface::registerCallBackAcquisitionFinished(
     void (*func)(const endCallbackHeader, void *), void *arg) {
+    std::lock_guard<std::mutex> lock(callbackMutex);
     acquisitionFinishedCallBack = func;
     pAcquisitionFinished = arg;
 }
@@ -69,6 +73,7 @@ void ClientInterface::registerCallBackRawDataReady(
     void (*func)(sls_receiver_header &, dataCallbackHeader, char *, size_t &,
                  void *),
     void *arg) {
+    std::lock_guard<std::mutex> lock(callbackMutex);
     rawDataReadyCallBack = func;
     pRawDataReady = arg;
 }
@@ -461,15 +466,18 @@ void ClientInterface::setDetectorType(detectorType arg) {
                            std::string(e.what()) + ']');
     }
     // callbacks after (in setdetectortype, the object is reinitialized)
-    if (startAcquisitionCallBack != nullptr)
-        impl()->registerCallBackStartAcquisition(startAcquisitionCallBack,
-                                                 pStartAcquisition);
-    if (acquisitionFinishedCallBack != nullptr)
-        impl()->registerCallBackAcquisitionFinished(acquisitionFinishedCallBack,
-                                                    pAcquisitionFinished);
-    if (rawDataReadyCallBack != nullptr)
-        impl()->registerCallBackRawDataReady(rawDataReadyCallBack,
-                                             pRawDataReady);
+    {
+        std::lock_guard<std::mutex> lock(callbackMutex);
+        if (startAcquisitionCallBack != nullptr)
+            impl()->registerCallBackStartAcquisition(startAcquisitionCallBack,
+                                                     pStartAcquisition);
+        if (acquisitionFinishedCallBack != nullptr)
+            impl()->registerCallBackAcquisitionFinished(
+                acquisitionFinishedCallBack, pAcquisitionFinished);
+        if (rawDataReadyCallBack != nullptr)
+            impl()->registerCallBackRawDataReady(rawDataReadyCallBack,
+                                                 pRawDataReady);
+    }
 
     impl()->setThreadIds(parentThreadId, tcpThreadId);
 }

diff --git a/slsReceiverSoftware/src/ClientInterface.h b/slsReceiverSoftware/src/ClientInterface.h
@@ -194,6 +194,8 @@ class ClientInterface : private virtual slsDetectorDefs {
     pid_t tcpThreadId{0};
     std::vector<std::string> udpips =
         std::vector<std::string>(MAX_NUMBER_OF_LISTENING_THREADS);
+    // necessary if Receiver objects using threads with callbacks
+    static std::mutex callbackMutex;
 };
 
 } // namespace sls