Cleanup

jf-zmqstreamer/CMakeLists.txt

@@ -0,0 +1,12 @@
+file(GLOB SOURCES src/*.cpp)
+
+add_library(jf-zmqstreamer-lib STATIC ${SOURCES})
+target_include_directories(jf-zmqstreamer-lib PUBLIC include/)
+target_link_libraries(jf-zmqstreamer-lib external core-buffer-lib)
+
+add_executable(jf-zmqstreamer src/main.cpp)
+set_target_properties(jf-zmqstreamer PROPERTIES OUTPUT_NAME jfj_combined)
+target_link_libraries(jf-zmqstreamer jf-zmqstreamer-lib zmq rt pthread)
+
+enable_testing()
+add_subdirectory(test/)

jf-zmqstreamer/include/FrameCache.hpp

@@ -91,7 +91,6 @@ public:
         }
     }
 
-
 protected:
     /** Flush and start a new line
 

jf-zmqstreamer/include/FrameWorker.hpp

@@ -10,13 +10,13 @@
 #include "PacketBuffer.hpp"
 #include "FrameStats.hpp"
 
-/** JungfrauJoch UDP receiver
+/** Jungfrau UDP receiver
 
     Capture UDP data stream from Jungfrau(Joch) FPGA card.
     NOTE: This design will not scale well for higher frame rates...
     TODO: Direct copy into FrameCache buffer (saves a memcopy)
 **/
-class JfjFrameWorker {
+class FrameWorker {
     const std::string m_moduleName;
     const uint64_t m_moduleID;
 
@@ -35,14 +35,13 @@ class JfjFrameWorker {
 
     std::function<void(uint64_t, uint64_t, BufferBinaryFormat&)> f_push_callback;
 public:
-    JfjFrameWorker(const uint16_t port, std::string moduleName, const uint32_t moduleID,
-                   std::function<void(uint64_t, uint64_t, BufferBinaryFormat&)> callback);
-    virtual ~JfjFrameWorker();
+    FrameWorker(const uint16_t port, std::string moduleName, const uint32_t moduleID, std::function<void(uint64_t, uint64_t, BufferBinaryFormat&)> callback);
+    virtual ~FrameWorker();
     void run();
     
     // Copy semantics : OFF
-    JfjFrameWorker(JfjFrameWorker const &) = delete;
-    JfjFrameWorker& operator=(JfjFrameWorker const &) = delete;
+    FrameWorker(FrameWorker const &) = delete;
+    FrameWorker& operator=(FrameWorker const &) = delete;
 };
 
 #endif //SF_DAQ_BUFFER_JFJ_FRAMEWORKER_HPP

jf-zmqstreamer/src/FrameWorker.cpp

@@ -3,13 +3,13 @@
 #include "FrameWorker.hpp"
 
 
-JfjFrameWorker::JfjFrameWorker(const uint16_t port, std::string moduleName, const uint32_t moduleID, std::function<void(uint64_t, uint64_t, BufferBinaryFormat&)> callback):
+FrameWorker::FrameWorker(const uint16_t port, std::string moduleName, const uint32_t moduleID, std::function<void(uint64_t, uint64_t, BufferBinaryFormat&)> callback):
         m_moduleName(moduleName), m_moduleID(moduleID), m_moduleStats(moduleName, moduleID, 10.0), f_push_callback(callback) {
     m_udp_receiver.bind(port);
 }
 
 
-JfjFrameWorker::~JfjFrameWorker() {
+FrameWorker::~FrameWorker() {
     m_udp_receiver.disconnect();
 }
 
@@ -18,7 +18,7 @@ JfjFrameWorker::~JfjFrameWorker() {
     Drains the buffer either until it's empty or the current frame is finished.
     Has some optimizations and safety checks before segfaulting right away...
     TODO: Direct memcopy into FrameCache for more speed!                    **/
-inline uint64_t JfjFrameWorker::process_packets(BufferBinaryFormat& buffer){
+inline uint64_t FrameWorker::process_packets(BufferBinaryFormat& buffer){
 
     while(!m_buffer.is_empty()){
         // Happens if the last packet from the previous frame gets lost.
@@ -68,7 +68,7 @@ inline uint64_t JfjFrameWorker::process_packets(BufferBinaryFormat& buffer){
     return 0;
 }
 
-uint64_t JfjFrameWorker::get_frame(BufferBinaryFormat& buffer){
+uint64_t FrameWorker::get_frame(BufferBinaryFormat& buffer){
     // Reset the metadata and frame buffer for the next frame
     std::memset(&buffer, 0, sizeof(buffer));
     uint64_t pulse_id = 0;
@@ -85,7 +85,7 @@ uint64_t JfjFrameWorker::get_frame(BufferBinaryFormat& buffer){
 }
 
 
-void JfjFrameWorker::run(){
+void FrameWorker::run(){
     std::cout << "Running worker loop" << std::endl;
     // Might be better creating a structure for double buffering
     BufferBinaryFormat buffer;

jf-zmqstreamer/src/main.cpp

@@ -35,20 +35,20 @@ int main (int argc, char *argv[]) {
 
 
     std::cout << "Creating frame workers..." << std::endl;
-    std::vector<std::shared_ptr<JfjFrameWorker>> vWorkers;
+    std::vector<std::shared_ptr<FrameWorker>> vWorkers;
     for(int mm=0; mm<config.n_modules; mm++){
         // Module name (not really used...)
         char m_name[128];
         snprintf(m_name, 128, "M%02d", mm);
         std::string moduleName(m_name);
-        vWorkers.emplace_back( std::make_shared<JfjFrameWorker>(config.start_udp_port+mm, moduleName, mm, push_cb) );
+        vWorkers.emplace_back( std::make_shared<FrameWorker>(config.start_udp_port+mm, moduleName, mm, push_cb) );
     }
 
 
     std::cout << "Starting frame worker threads..." << std::endl;
     std::vector<std::thread> vThreads;
     for(int mm=0; mm<config.n_modules; mm++){
-        vThreads.push_back( std::thread(&JfjFrameWorker::run, vWorkers[mm].get()) );
+        vThreads.push_back( std::thread(&FrameWorker::run, vWorkers[mm].get()) );
     }
 
     for(auto& it: vThreads){

jfj-combined/CMakeLists.txt

@@ -1,12 +0,0 @@
-file(GLOB SOURCES src/*.cpp)
-
-add_library(jfj-combined-lib STATIC ${SOURCES})
-target_include_directories(jfj-combined-lib PUBLIC include/)
-target_link_libraries(jfj-combined-lib external core-buffer-lib)
-
-add_executable(jfj-combined src/main.cpp)
-set_target_properties(jfj-combined PROPERTIES OUTPUT_NAME jfj_combined)
-target_link_libraries(jfj-combined jfj-combined-lib zmq rt pthread)
-
-enable_testing()
-add_subdirectory(test/)

jfj-udp-recv/CMakeLists.txt

@@ -1,12 +0,0 @@
-file(GLOB SOURCES src/*.cpp)
-
-add_library(jfj-udp-recv-lib STATIC ${SOURCES})
-target_include_directories(jfj-udp-recv-lib PUBLIC include/)
-target_link_libraries(jfj-udp-recv-lib external core-buffer-lib)
-
-add_executable(jfj-udp-recv src/main.cpp)
-set_target_properties(jfj-udp-recv PROPERTIES OUTPUT_NAME jfj_udp_recv)
-target_link_libraries(jfj-udp-recv jfj-udp-recv-lib zmq rt)
-
-enable_testing()
-add_subdirectory(test/)

jfj-udp-recv/README.md

@@ -1,164 +0,0 @@
-# sf-buffer
-sf-buffer is the component that receives the detector data in form of UDP 
-packages and writes them down to disk to a binary format. In addition, it 
-sends a copy of the module frame to sf-stream via ZMQ.
-
-Each sf-buffer process is taking care of a single detector module. The 
-processes are all independent and do not rely on any external data input 
-to maximize isolation and possible interactions in our system.
-
-The main design principle is simplicity and decoupling:
-
-- No interprocess dependencies/communication.
-- No dependencies on external libraries (as much as possible).
-- Using POSIX as much as possible.
-
-We are optimizing for maintainability and long term stability. Performance is 
-of concern only if the performance criteria are not met.
-
-## Overview
-
-![image_buffer_overview](../docs/sf_daq_buffer-overview-buffer.jpg)
-
-sf-buffer is a single threaded application (without counting the ZMQ IO threads)
-that does both receiving, assembling, writing and sending in the same thread.
-
-### UDP receiving
-
-Each process listens to one udp port. Packets coming to this udp port are 
-assembled into frames. Frames (either complete or with missing packets) are 
-passed forward. The number of received packets is saved so we can later 
-(at image assembly time) determine if the frame is valid or not. At this point 
-we do no validation.
-
-We are currently using **recvmmsg** to minimize the number of switches to 
-kernel mode.
-
-We expect all packets to come in order or not come at all. Once we see the 
-package for the next pulse_id we can assume no more packages are coming for 
-the previous one, and send the assembled frame down the program.
-
-### File writing
-
-Files are written to disk in frames - one write to disk per frame. This gives 
-us a relaxed 10ms interval of 1 MB writes.
-
-#### File format
-
-The binary file on disk is just a serialization of multiple 
-**BufferBinaryFormat** structs:
-```c++
-#pragma pack(push)
-#pragma pack(1)
-struct ModuleFrame {
-    uint64_t pulse_id;
-    uint64_t frame_index;
-    uint64_t daq_rec;
-    uint64_t n_recv_packets;
-    uint64_t module_id;
-};
-#pragma pack(pop)
-
-#pragma pack(push)
-#pragma pack(1)
-struct BufferBinaryFormat {
-    const char FORMAT_MARKER = 0xBE;
-    ModuleFrame meta;
-    char data[buffer_config::MODULE_N_BYTES];
-};
-#pragma pack(pop)
-```
-
-![file_layout_image](../docs/sf_daq_buffer-FileLayout.jpg)
-
-Each frame is composed by:
-
-- **FORMAT\_MARKER** (0xBE) - a control byte to determine the validity of the frame.
-- **ModuleFrame** - frame meta used in image assembly phase.
-- **Data** - assembled frame from a single module.
-
-Frames are written one after another to a specific offset in the file. The 
-offset is calculated based on the pulse_id, so each frame has a specific place 
-in the file and there is no need to have an index for frame retrieval.
-
-The offset where a specific pulse_id is written in a file is calculated:
-
-```c++
-// We save 1000 pulses in each file.
-const uint64_t FILE_MOD = 1000
-
-// Relative index of pulse_id inside file.
-size_t file_base = pulse_id % FILE_MOD;
-// Offset in bytes of relative index in file.
-size_t file_offset = file_base * sizeof(BufferBinaryFormat);
-```
-
-We now know where to look for data inside the file, but we still don't know 
-inside which file to look. For this we need to discuss the folder structure.
-
-#### Folder structure
-
-The folder (as well as file) structure is deterministic in the sense that given 
-a specific pulse_id, we can directly calculate the folder, file, and file 
-offset where the data is stored. This allows us to have independent writing 
-and reading from the buffer without building any indexes.
-
-The binary files written by sf_buffer are saved to:
-
-[detector_folder]/[module_folder]/[data_folder]/[data_file].bin
-
-- **detector\_folder** should always be passed as an absolute path. This is the 
-container that holds all data related to a specific detector.
-- **module\_folder** is usually composed like "M00", "M01". It separates data 
-from different modules of one detector.
-- **data\_folder** and **data\_file** are automatically calculated based on the 
-current pulse_id, FOLDER_MOD and FILE_MOD attributes. This folders act as our 
-index for accessing data.
-
-![folder_layout_image](../docs/sf_daq_buffer-FolderLayout.jpg)
-
-```c++
-// FOLDER_MOD = 100000
-int data_folder = (pulse_id % FOLDER_MOD) * FOLDER_MOD; 
-// FILE_MOD = 1000
-int data_file = (pulse_id % FILE_MOD) * FILE_MOD; 
-```
-
-The data_folder and data_file folders are named as the first pulse_id that 
-should be stored inside them.
-
-FOLDER_MOD == 100000 means that each data_folder will contain data for 100000
-pulses, while FILE_MOD == 1000 means that each file inside the data_folder 
-will contain 1000 pulses. The total number of data_files in each data_folder 
-will therefore be **FILE\_MOD / FOLDER\_MOD = 100**.
-
-#### Analyzing the buffer on disk
-In **sf-utils** there is a Python module that allows you to read directly the 
-buffer in order to debug it or to verify the consistency between the HDF5 file 
-and the received data.
-
-- VerifyH5DataConsistency.py checks the consistency between the H5 file and 
-buffer.
-- BinaryBufferReader.py reads the buffer and prints meta. The class inside 
-can also be used in external scripts.
-
-### ZMQ sending
-
-A copy of the data written to disk is also send via ZMQ to the sf-stream. This 
-is used to provide live viewing / processing capabilities. Each module data is 
-sent separately, and this is later assembled in the sf-stream.
-
-We use the PUB/SUB mechanism for distributing this data - we cannot control the 
-rate of the producer, and we would like to avoid distributed image assembly 
-if possible, so PUSH/PULL does not make sense in this case.
-
-We provide no guarantees on live data delivery, but in practice the number of 
-dropped or incomplete frames in currently negligible.
-
-The protocol is a serialization of the same data structures we use to 
-write on disk (no need for additional memory operations before sending out 
-data). It uses a 2 part multipart ZMQ message:
-
-- The first part is a serialization of the ModuleFrame struct (see above).
-- The second part is the data field in the BufferBinaryFormat struct (the frame
-data).

jfj-udp-recv/include/JfjFrameStats.hpp

@@ -1,31 +0,0 @@
-#include <cstddef>
-#include <formats.hpp>
-#include <chrono>
-
-#ifndef SF_DAQ_BUFFER_FRAMESTATS_HPP
-#define SF_DAQ_BUFFER_FRAMESTATS_HPP
-
-
-class FrameStats {
-    const std::string detector_name_;
-    const int module_id_;
-    size_t stats_time_;
-
-    int frames_counter_;
-    int n_missed_packets_;
-    int n_corrupted_frames_;
-    int n_corrupted_pulse_id_;
-    std::chrono::time_point<std::chrono::steady_clock> stats_interval_start_;
-
-    void reset_counters();
-    void print_stats();
-
-public:
-    FrameStats(const std::string &detector_name,
-               const int module_id,
-               const size_t stats_time);
-    void record_stats(const ModuleFrame &meta, const bool bad_pulse_id);
-};
-
-
-#endif //SF_DAQ_BUFFER_FRAMESTATS_HPP

jfj-udp-recv/include/JfjFrameUdpReceiver.hpp

@@ -1,32 +0,0 @@
-#ifndef SF_DAQ_BUFFER_JOCHUDPRECEIVER_HPP
-#define SF_DAQ_BUFFER_JOCHUDPRECEIVER_HPP
-
-#include <netinet/in.h>
-#include "PacketUdpReceiver.hpp"
-#include "formats.hpp"
-#include "buffer_config.hpp"
-#include "PacketBuffer.hpp"
-#include "jungfraujoch.hpp"
-
-/** JungfrauJoch UDP receiver
-
-    Wrapper class to capture frames from the UDP stream of the JungfrauJoch FPGA card.
-    NOTE: This design will not scale well for higher frame rates...
-**/
-class JfjFrameUdpReceiver {
-    PacketUdpReceiver m_udp_receiver;
-    uint64_t m_frame_index;
-
-    PacketBuffer<jfjoch_packet_t, buffer_config::BUFFER_UDP_N_RECV_MSG> m_buffer;
-
-    inline void init_frame(ModuleFrame& frame_metadata, const jfjoch_packet_t& c_packet);
-    inline uint64_t process_packets(ModuleFrame& metadata, char* frame_buffer);
-
-public:
-    JfjFrameUdpReceiver(const uint16_t port);
-    virtual ~JfjFrameUdpReceiver();
-    uint64_t get_frame_from_udp(ModuleFrame& metadata, char* frame_buffer);
-};
-
-
-#endif //SF_DAQ_BUFFER_JOCHUDPRECEIVER_HPP

jfj-udp-recv/include/PacketBuffer.hpp

@@ -1,112 +0,0 @@
-#ifndef CIRCULAR_BUFFER_TEMPLATE_HPP
-#define CIRCULAR_BUFFER_TEMPLATE_HPP
-
-#include <cstddef>
-#include <stdexcept>
-#include <iostream>
-#include <mutex>
-#include <sys/socket.h>
-#include <netinet/in.h>
-
-
-/** Linear data buffer (NOT FIFO)
-
-	Simplified data buffer that provides pop and push operations and
-	bundles the actual container with metadata required by <sockets.h>.
-	It stores the actual data in an accessible C-style array. **/
-template <typename T, size_t CAPACITY>
-class PacketBuffer{
-public:
-    PacketBuffer() {
-        for (int i = 0; i < CAPACITY; i++) {
-            m_recv_buff_ptr[i].iov_base = (void*) &(m_container[i]);
-            m_recv_buff_ptr[i].iov_len = sizeof(T);
-
-            // C-structure as expected by <sockets.h>
-            m_msgs[i].msg_hdr.msg_iov = &m_recv_buff_ptr[i];
-            m_msgs[i].msg_hdr.msg_iovlen = 1;
-            m_msgs[i].msg_hdr.msg_name = &m_sock_from[i];
-            m_msgs[i].msg_hdr.msg_namelen = sizeof(sockaddr_in);
-        }
-     };
-    // ~PacketBuffer() {};
-
-    /**Diagnostics**/
-    int size() const { return ( idx_write-idx_read ); }
-    int capacity() const { return m_capacity; }
-    bool is_full() const { return bool(idx_write >= m_capacity); }
-    bool is_empty() const { return bool(idx_write <= idx_read); }
-
-    /**Operators**/
-    void reset(){ idx_write = 0; idx_read = 0; };		// Reset the buffer
-    T& container(){ return m_container; };              // Direct container reference
-    mmsghdr& msgs(){ return m_msgs; };
-
-    /**Element access**/
-    T& pop_front();             //Destructive read
-    const T& peek_front();      //Non-destructive read
-    void push_back(T item);     //Write new element to buffer
-
-    /**Fill from UDP receiver**/
-    template <typename TY>
-    void fill_from(TY& recv){
-        std::lock_guard<std::mutex> g_guard(m_mutex);
-        this->idx_write = recv.receive_many(m_msgs, this->capacity());
-        // Returns -1 with errno=11 if no data received
-        if(idx_write==-1){ idx_write = 0; }
-        this->idx_read = 0;
-    }
-
-private:
-    // Main container
-    T m_container[CAPACITY];
-    const size_t m_capacity = CAPACITY;
-    /**Guards**/
-    std::mutex m_mutex;
-    /**Read and write index**/
-    int idx_write = 0;
-    int idx_read = 0;
-
-    // C-structures as expected by <sockets.h>
-    mmsghdr m_msgs[CAPACITY];
-    iovec m_recv_buff_ptr[CAPACITY];
-    sockaddr_in m_sock_from[CAPACITY];
-};
-
-
-/*********************************************************************/
-/*********************************************************************/
-/*********************************************************************/
-
-/** Destructive read
-	Standard read access to queues (i.e. progress the read pointer).
-	Throws 'std::length_error' if container is empty. **/
-template <typename T, size_t CAPACITY>
-T& PacketBuffer<T, CAPACITY>::pop_front(){
-    std::lock_guard<std::mutex> g_guard(m_mutex);
-    if(this->is_empty()){ throw std::out_of_range("Attempted to read empty queue!"); }
-    idx_read++;
-    return m_container[idx_read-1];
-}
-
-/** Non-destructive read
-	Standard, non-destructive read access (does not progress the read pointer).
-	Throws 'std::length_error' if container is empty. **/
-template <typename T, size_t CAPACITY>
-const T& PacketBuffer<T, CAPACITY>::peek_front(){
-    std::lock_guard<std::mutex> g_guard(m_mutex);
-	if(this->is_empty()){ throw std::out_of_range("Attempted to read empty queue!"); }
-    return m_container[idx_read];
-}
-
-
-/** Push an element into the end of the buffer**/
-template <typename T, size_t CAPACITY>
-void PacketBuffer<T, CAPACITY>::push_back(T item){
-    std::lock_guard<std::mutex> g_guard(m_mutex);
-    if(this->is_full()){ throw std::out_of_range("Attempted to write a full buffer!"); }
-    m_container[idx_write] = item;
-    idx_write++;
-}
-
-#endif // CIRCULAR_BUFFER_TEMPLATE_HPP

jfj-udp-recv/include/PacketUdpReceiver.hpp

@@ -1,22 +0,0 @@
-#ifndef UDPRECEIVER_H
-#define UDPRECEIVER_H
-
-#include <sys/socket.h>
-
-class PacketUdpReceiver {
-
-    int socket_fd_;
-
-public:
-    PacketUdpReceiver();
-    virtual ~PacketUdpReceiver();
-
-    bool receive(void* buffer, const size_t buffer_n_bytes);
-    int receive_many(mmsghdr* msgs, const size_t n_msgs);
-
-    void bind(const uint16_t port);
-    void disconnect();
-};
-
-
-#endif //LIB_CPP_H5_WRITER_UDPRECEIVER_H

jfj-udp-recv/src/JfjFrameStats.cpp

@@ -1,71 +0,0 @@
-#include <iostream>
-#include "JfjFrameStats.hpp"
-
-using namespace std;
-using namespace chrono;
-
-FrameStats::FrameStats(
-        const std::string &detector_name,
-        const int module_id,
-        const size_t stats_time) :
-            detector_name_(detector_name),
-            module_id_(module_id),
-            stats_time_(stats_time)
-{
-   reset_counters();
-}
-
-void FrameStats::reset_counters()
-{
-    frames_counter_ = 0;
-    n_missed_packets_ = 0;
-    n_corrupted_frames_ = 0;
-    n_corrupted_pulse_id_ = 0;
-    stats_interval_start_ = steady_clock::now();
-}
-
-void FrameStats::record_stats(const ModuleFrame &meta, const bool bad_pulse_id)
-{
-
-    if (bad_pulse_id) {
-        n_corrupted_pulse_id_++;
-    } 
-
-    if (meta.n_recv_packets < JF_N_PACKETS_PER_FRAME) {
-        n_missed_packets_ += JF_N_PACKETS_PER_FRAME - meta.n_recv_packets;
-        n_corrupted_frames_++;
-    }
-
-    frames_counter_++;
-
-    auto time_passed = duration_cast<milliseconds>(
-             steady_clock::now()-stats_interval_start_).count(); 
-
-    if (time_passed >= stats_time_*1000) {
-        print_stats();
-        reset_counters();
-    }
-}
-
-void FrameStats::print_stats()
-{
-    auto interval_ms_duration = duration_cast<milliseconds>(
-            steady_clock::now()-stats_interval_start_).count();
-    // * 1000 because milliseconds, + 250 because of truncation.
-    int rep_rate = ((frames_counter_ * 1000) + 250) / interval_ms_duration;
-    uint64_t timestamp = time_point_cast<nanoseconds>(
-            system_clock::now()).time_since_epoch().count();
-
-    // Output in InfluxDB line protocol
-    cout << "jf_udp_recv";
-    cout << ",detector_name=" << detector_name_;
-    cout << ",module_name=M" << module_id_;
-    cout << " ";
-    cout << "n_missed_packets=" << n_missed_packets_ << "i";
-    cout << ",n_corrupted_frames=" << n_corrupted_frames_ << "i";
-    cout << ",repetition_rate=" << rep_rate << "i";
-    cout << ",n_corrupted_pulse_ids=" << n_corrupted_pulse_id_ << "i";
-    cout << " ";
-    cout << timestamp;
-    cout << endl;
-}

jfj-udp-recv/src/JfjFrameUdpReceiver.cpp

@@ -1,86 +0,0 @@
-#include <cstring>
-#include "JfjFrameUdpReceiver.hpp"
-
-using namespace std;
-using namespace buffer_config;
-
-std::ostream &operator<<(std::ostream &os, jfjoch_packet_t const &packet) { 
-    os << "Frame number: " << packet.framenum << std::endl;
-    os << "Packet number: " << packet.packetnum << std::endl;
-    os << "Bunch id: " << packet.bunchid << std::endl;
-    os << std::endl;
-    return os;
-}
-
-JfjFrameUdpReceiver::JfjFrameUdpReceiver(const uint16_t port) {
-    m_udp_receiver.bind(port);
-}
-
-JfjFrameUdpReceiver::~JfjFrameUdpReceiver() {
-    m_udp_receiver.disconnect();
-}
-
-inline void JfjFrameUdpReceiver::init_frame(ModuleFrame& metadata, const jfjoch_packet_t& c_packet) {
-    // std::cout << c_packet;
-      
-    metadata.pulse_id = c_packet.bunchid;
-    metadata.frame_index = c_packet.framenum;
-    metadata.daq_rec = (uint64_t) c_packet.debug;
-    metadata.module_id = (int64_t) 0;
-}
-
-inline uint64_t JfjFrameUdpReceiver::process_packets(ModuleFrame& metadata, char* frame_buffer){
-
-    while(!m_buffer.is_empty()){
-        // Happens if the last packet from the previous frame gets lost.
-        if (m_frame_index != m_buffer.peek_front().framenum) {
-            m_frame_index = m_buffer.peek_front().framenum;
-            std::cout << "Peeked pulse: " << metadata.pulse_id << std::endl;
-            return metadata.pulse_id;
-        }
-
-        // Otherwise pop the queue (and set current frame index)
-        jfjoch_packet_t& c_packet = m_buffer.pop_front();
-        m_frame_index = c_packet.framenum;
-
-        // Always copy metadata (otherwise problem when 0th packet gets lost)
-        this->init_frame(metadata, c_packet);
-
-        // Copy data to frame buffer
-        size_t offset = JFJOCH_DATA_BYTES_PER_PACKET * c_packet.packetnum;
-        memcpy( (void*) (frame_buffer + offset), c_packet.data, JFJOCH_DATA_BYTES_PER_PACKET);
-        metadata.n_recv_packets++;
-
-        // Last frame packet received. Frame finished.
-        if (c_packet.packetnum == JFJOCH_N_PACKETS_PER_FRAME - 1){
-            return metadata.pulse_id;
-        }
-    }
-
-    // We emptied the buffer.
-   // m_buffer.reset();
-    return 0;
-}
-
-uint64_t JfjFrameUdpReceiver::get_frame_from_udp(ModuleFrame& metadata, char* frame_buffer){
-    // Reset the metadata and frame buffer for the next frame. (really needed?)
-    metadata.pulse_id = 0;
-    metadata.n_recv_packets = 0;
-    memset(frame_buffer, 0, JFJOCH_DATA_BYTES_PER_FRAME);
-    // Process leftover packages in the buffer
-    if (!m_buffer.is_empty()) {
-        auto pulse_id = process_packets(metadata, frame_buffer);
-        if (pulse_id != 0) { return pulse_id; }
-    }
-
-    while (true) {
-        // Receive new packages (pass if none)...
-        m_buffer.reset();
-        m_buffer.fill_from(m_udp_receiver);      
-        if (m_buffer.is_empty()) { continue; }
-
-        // ... and process them
-        auto pulse_id = process_packets(metadata, frame_buffer);
-        if (pulse_id != 0) { return pulse_id; }
-    }
-}

jfj-udp-recv/src/PacketUdpReceiver.cpp

@@ -1,66 +0,0 @@
-#include <netinet/in.h>
-#include <iostream>
-#include "PacketUdpReceiver.hpp"
-#include "jungfrau.hpp"
-#include <unistd.h>
-#include <cstring>
-#include "buffer_config.hpp"
-
-using namespace std;
-using namespace buffer_config;
-
-PacketUdpReceiver::PacketUdpReceiver() : socket_fd_(-1) { }
-
-PacketUdpReceiver::~PacketUdpReceiver() {
-    disconnect();
-}
-
-void PacketUdpReceiver::bind(const uint16_t port){
-    if (socket_fd_ > -1) {
-        throw runtime_error("Socket already bound.");
-    }
-
-    socket_fd_ = socket(AF_INET, SOCK_DGRAM, 0);
-    if (socket_fd_ < 0) {
-        throw runtime_error("Cannot open socket.");
-    }
-
-    sockaddr_in server_address = {0};
-    server_address.sin_family = AF_INET;
-    server_address.sin_addr.s_addr = INADDR_ANY;
-    server_address.sin_port = htons(port);
-
-    timeval udp_socket_timeout;
-    udp_socket_timeout.tv_sec = 0;
-    udp_socket_timeout.tv_usec = BUFFER_UDP_US_TIMEOUT;
-
-    if (setsockopt(socket_fd_, SOL_SOCKET, SO_RCVTIMEO, &udp_socket_timeout, sizeof(timeval)) == -1) {
-        throw runtime_error("Cannot set SO_RCVTIMEO. " + string(strerror(errno)));
-    }
-
-    if (setsockopt(socket_fd_, SOL_SOCKET, SO_RCVBUF, &BUFFER_UDP_RCVBUF_BYTES, sizeof(int)) == -1) {
-        throw runtime_error("Cannot set SO_RCVBUF. " + string(strerror(errno)));
-    };
-    //TODO: try to set SO_RCVLOWAT
-
-    auto bind_result = ::bind(socket_fd_, reinterpret_cast<const sockaddr *>(&server_address), sizeof(server_address));
-
-    if (bind_result < 0) {
-        throw runtime_error("Cannot bind socket.");
-    }
-}
-
-int PacketUdpReceiver::receive_many(mmsghdr* msgs, const size_t n_msgs){
-    return recvmmsg(socket_fd_, msgs, n_msgs, 0, 0);
-}
-
-bool PacketUdpReceiver::receive(void* buffer, const size_t buffer_n_bytes){
-    auto data_len = recv(socket_fd_, buffer, buffer_n_bytes, 0);
-
-    return (data_len == buffer_n_bytes) ? true : false;
-}
-
-void PacketUdpReceiver::disconnect(){
-    close(socket_fd_);
-    socket_fd_ = -1;
-}

jfj-udp-recv/src/main.cpp

@@ -1,74 +0,0 @@
-#include <iostream>
-#include <stdexcept>
-#include <zmq.h>
-#include <RamBuffer.hpp>
-
-#include "formats.hpp"
-#include "buffer_config.hpp"
-#include "JfjFrameUdpReceiver.hpp"
-#include "BufferUtils.hpp"
-#include "JfjFrameStats.hpp"
-
-using namespace std;
-using namespace chrono;
-using namespace buffer_config;
-using namespace BufferUtils;
-
-int main (int argc, char *argv[]) {
-
-    if (argc != 3) {
-        cout << endl;
-        cout << "Usage: jfj_udp_recv [detector_json_filename]" << endl;
-        cout << "\tdetector_json_filename: detector config file path." << endl;
-        cout << endl;
-
-        exit(-1);
-    }
-
-    const auto config = read_json_config(string(argv[1]));
-
-    const auto udp_port = config.start_udp_port;
-    JfjFrameUdpReceiver receiver(udp_port);
-    RamBuffer buffer(config.detector_name, config.n_modules);
-    FrameStats stats(config.detector_name, 0, STATS_TIME);
-
-    auto ctx = zmq_ctx_new();
-    zmq_ctx_set(ctx, ZMQ_IO_THREADS, ZMQ_IO_THREADS);
-    auto sender = BufferUtils::bind_socket(ctx, config.detector_name, "jungfraujoch");
-
-    // Might be better creating a structure for double buffering
-    ModuleFrame frameMeta;
-    ImageMetadata imageMeta;
-    char* dataBuffer = new char[JFJOCH_DATA_BYTES_PER_FRAME];
-
-    uint64_t pulse_id_previous = 0;
-    uint64_t frame_index_previous = 0;
-
-
-    while (true) {
-        // NOTE: Needs to be pipelined for really high frame rates
-        auto pulse_id = receiver.get_frame_from_udp(frameMeta, dataBuffer);
-
-        bool bad_pulse_id = false;
-
-        if ( ( frameMeta.frame_index != (frame_index_previous+1) ) || ( (pulse_id-pulse_id_previous) < 0 ) || ( (pulse_id-pulse_id_previous) > 1000 ) ) {
-            bad_pulse_id = true;
-        } else {
-            imageMeta.pulse_id = frameMeta.pulse_id;
-            imageMeta.frame_index = frameMeta.frame_index;
-            imageMeta.daq_rec = frameMeta.daq_rec;
-            imageMeta.is_good_image = true;            
-            
-            buffer.write_frame(frameMeta, dataBuffer);            
-            zmq_send(sender, &imageMeta, sizeof(imageMeta), 0);
-        }
-
-        stats.record_stats(frameMeta, bad_pulse_id);
-
-        pulse_id_previous = pulse_id;
-        frame_index_previous = frameMeta.frame_index;
-
-    }
-
-    delete[] dataBuffer;
-}

jfj-udp-recv/test/CMakeLists.txt

@@ -1,8 +0,0 @@
-add_executable(jfj-udp-recv-tests main.cpp)
-
-target_link_libraries(jfj-udp-recv-tests
-        core-buffer-lib
-        jfj-udp-recv-lib
-        gtest
-        )
-

jfj-udp-recv/test/main.cpp

@@ -1,10 +0,0 @@
-#include "gtest/gtest.h"
-#include "test_PacketUdpReceiver.cpp"
-#include "test_FrameUdpReceiver.cpp"
-
-using namespace std;
-
-int main(int argc, char **argv) {
-  ::testing::InitGoogleTest(&argc, argv);
-  return RUN_ALL_TESTS();
-}

jfj-udp-recv/test/mock/udp.hpp

@@ -1,16 +0,0 @@
-#ifndef MOCK_UDP_H
-#define MOCK_UDP_H
-
-const int MOCK_UDP_PORT(13000);
-
-sockaddr_in get_server_address(uint16_t udp_port)
-{
-    sockaddr_in server_address = {0};
-    server_address.sin_family = AF_INET;
-    server_address.sin_addr.s_addr = INADDR_ANY;
-    server_address.sin_port = htons(udp_port);
-
-    return server_address;
-}
-
-#endif

jfj-udp-recv/test/test_FrameUdpReceiver.cpp

@@ -1,230 +0,0 @@
-#include <netinet/in.h>
-#include <jungfraujoch.hpp>
-#include "gtest/gtest.h"
-#include "JfjFrameUdpReceiver.hpp"
-#include "mock/udp.hpp"
-
-#include <thread>
-#include <chrono>
-#include <future>
-
-using namespace std;
-
-TEST(BufferUdpReceiver, simple_recv)
-{
-    int n_packets = JFJOCH_N_PACKETS_PER_FRAME;
-    int n_frames = 5;
-
-    uint16_t udp_port = MOCK_UDP_PORT;
-    auto server_address = get_server_address(udp_port);
-    auto send_socket_fd = socket(AF_INET, SOCK_DGRAM, 0);
-    ASSERT_TRUE(send_socket_fd >= 0);
-
-    JfjFrameUdpReceiver udp_receiver(udp_port);
-
-    auto handle = async(launch::async, [&](){
-        for (int64_t i_frame=0; i_frame < n_frames; i_frame++){
-            for (size_t i_packet=0; i_packet<n_packets; i_packet++) {
-                jfjoch_packet_t send_udp_buffer;
-                send_udp_buffer.packetnum = i_packet;
-                send_udp_buffer.bunchid = i_frame + 1;
-                send_udp_buffer.framenum = i_frame + 1000;
-                send_udp_buffer.debug = i_frame + 10000;
-
-                ::sendto(
-                        send_socket_fd,
-                        &send_udp_buffer,
-                        JFJOCH_BYTES_PER_PACKET,
-                        0,
-                        (sockaddr*) &server_address,
-                        sizeof(server_address));
-            }
-        }
-    });
-
-    handle.wait();
-
-    ModuleFrame metadata;
-    auto frame_buffer = make_unique<char[]>(JFJOCH_DATA_BYTES_PER_FRAME);
-
-    for (int i_frame=0; i_frame < n_frames; i_frame++) {
-        auto pulse_id = udp_receiver.get_frame_from_udp(
-                metadata, frame_buffer.get());
-
-        ASSERT_EQ(i_frame + 1, pulse_id);
-        ASSERT_EQ(metadata.frame_index, i_frame + 1000);
-        ASSERT_EQ(metadata.daq_rec, i_frame + 10000);
-        // -1 because we skipped a packet.
-        ASSERT_EQ(metadata.n_recv_packets, n_packets);
-    }
-
-    ::close(send_socket_fd);
-}
-
-TEST(BufferUdpReceiver, missing_middle_packet)
-{
-    int n_packets = JFJOCH_N_PACKETS_PER_FRAME;
-    int n_frames = 3;
-
-    uint16_t udp_port = MOCK_UDP_PORT;
-    auto server_address = get_server_address(udp_port);
-    auto send_socket_fd = socket(AF_INET, SOCK_DGRAM, 0);
-    ASSERT_TRUE(send_socket_fd >= 0);
-
-    JfjFrameUdpReceiver udp_receiver(udp_port);
-
-    auto handle = async(launch::async, [&](){
-        for (int64_t i_frame=0; i_frame < n_frames; i_frame++){
-            for (size_t i_packet=0; i_packet<n_packets; i_packet++) {
-                // Skip some random middle packet.
-                if (i_packet == 10) {
-                    continue;
-                }
-
-                jfjoch_packet_t send_udp_buffer;
-                send_udp_buffer.packetnum = i_packet;
-                send_udp_buffer.bunchid = i_frame + 1;
-                send_udp_buffer.framenum = i_frame + 1000;
-                send_udp_buffer.debug = i_frame + 10000;
-
-                ::sendto(
-                        send_socket_fd,
-                        &send_udp_buffer,
-                        JFJOCH_BYTES_PER_PACKET,
-                        0,
-                        (sockaddr*) &server_address,
-                        sizeof(server_address));
-            }
-        }
-    });
-
-    handle.wait();
-
-    ModuleFrame metadata;
-    auto frame_buffer = make_unique<char[]>(JFJOCH_DATA_BYTES_PER_FRAME);
-
-    for (int i_frame=0; i_frame < n_frames; i_frame++) {
-        auto pulse_id = udp_receiver.get_frame_from_udp(
-                metadata, frame_buffer.get());
-
-        ASSERT_EQ(i_frame + 1, pulse_id);
-        ASSERT_EQ(metadata.frame_index, i_frame + 1000);
-        ASSERT_EQ(metadata.daq_rec, i_frame + 10000);
-        // -1 because we skipped a packet.
-        ASSERT_EQ(metadata.n_recv_packets, n_packets - 1);
-    }
-
-    ::close(send_socket_fd);
-}
-
-TEST(BufferUdpReceiver, missing_first_packet)
-{
-    auto n_packets = JFJOCH_N_PACKETS_PER_FRAME;
-    int n_frames = 3;
-
-    uint16_t udp_port = MOCK_UDP_PORT;
-    auto server_address = get_server_address(udp_port);
-    auto send_socket_fd = socket(AF_INET, SOCK_DGRAM, 0);
-    ASSERT_TRUE(send_socket_fd >= 0);
-
-    JfjFrameUdpReceiver udp_receiver(udp_port);
-
-    auto handle = async(launch::async, [&](){
-        for (int64_t i_frame=0; i_frame < n_frames; i_frame++){
-            for (size_t i_packet=0; i_packet<n_packets; i_packet++) {
-                // Skip first packet.
-                if (i_packet == 0) {
-                    continue;
-                }
-
-                jfjoch_packet_t send_udp_buffer;
-                send_udp_buffer.packetnum = i_packet;
-                send_udp_buffer.bunchid = i_frame + 1;
-                send_udp_buffer.framenum = i_frame + 1000;
-                send_udp_buffer.debug = i_frame + 10000;
-
-                ::sendto(
-                        send_socket_fd,
-                        &send_udp_buffer,
-                        JUNGFRAU_BYTES_PER_PACKET,
-                        0,
-                        (sockaddr*) &server_address,
-                        sizeof(server_address));
-            }
-        }
-    });
-
-    handle.wait();
-
-    ModuleFrame metadata;
-    auto frame_buffer = make_unique<char[]>(JFJOCH_DATA_BYTES_PER_FRAME);
-
-    for (int i_frame=0; i_frame < n_frames; i_frame++) {
-        auto pulse_id = udp_receiver.get_frame_from_udp(
-                metadata, frame_buffer.get());
-
-        ASSERT_EQ(i_frame + 1, pulse_id);
-        ASSERT_EQ(metadata.frame_index, i_frame + 1000);
-        ASSERT_EQ(metadata.daq_rec, i_frame + 10000);
-        // -1 because we skipped a packet.
-        ASSERT_EQ(metadata.n_recv_packets, n_packets - 1);
-    }
-
-    ::close(send_socket_fd);
-}
-
-TEST(BufferUdpReceiver, missing_last_packet)
-{
-    int n_packets = JFJOCH_N_PACKETS_PER_FRAME;
-    int n_frames = 3;
-
-    uint16_t udp_port = MOCK_UDP_PORT;
-    auto server_address = get_server_address(udp_port);
-    auto send_socket_fd = socket(AF_INET, SOCK_DGRAM, 0);
-    ASSERT_TRUE(send_socket_fd >= 0);
-
-    JfjFrameUdpReceiver udp_receiver(udp_port);
-
-    auto handle = async(launch::async, [&](){
-        for (int64_t i_frame=0; i_frame < n_frames; i_frame++){
-            for (size_t i_packet=0; i_packet<n_packets; i_packet++) {
-                // Skip the last packet.
-                if (i_packet == n_packets-1) {
-                    continue;
-                }
-
-                jfjoch_packet_t send_udp_buffer;
-                send_udp_buffer.packetnum = i_packet;
-                send_udp_buffer.bunchid = i_frame + 1;
-                send_udp_buffer.framenum = i_frame + 1000;
-                send_udp_buffer.debug = i_frame + 10000;
-
-                ::sendto(
-                        send_socket_fd,
-                        &send_udp_buffer,
-                        JUNGFRAU_BYTES_PER_PACKET,
-                        0,
-                        (sockaddr*) &server_address,
-                        sizeof(server_address));
-            }
-        }
-    });
-
-    handle.wait();
-
-    ModuleFrame metadata;
-    auto frame_buffer = make_unique<char[]>(JFJOCH_DATA_BYTES_PER_FRAME);
-
-    // n_frames -1 because the last frame is not complete.
-    for (int i_frame=0; i_frame < n_frames - 1; i_frame++) {
-        auto pulse_id = udp_receiver.get_frame_from_udp(metadata, frame_buffer.get());
-
-        ASSERT_EQ(i_frame + 1, pulse_id);
-        ASSERT_EQ(metadata.frame_index, i_frame + 1000);
-        ASSERT_EQ(metadata.daq_rec, i_frame + 10000);
-        // -1 because we skipped a packet.
-        ASSERT_EQ(metadata.n_recv_packets, n_packets - 1);
-    }
-
-    ::close(send_socket_fd);
-}

jfj-udp-recv/test/test_PacketUdpReceiver.cpp

@@ -1,170 +0,0 @@
-#include <netinet/in.h>
-#include <jungfrau.hpp>
-#include "gtest/gtest.h"
-#include "mock/udp.hpp"
-#include "PacketUdpReceiver.hpp"
-
-#include <thread>
-#include <chrono>
-
-using namespace std;
-
-TEST(PacketUdpReceiver, simple_recv)
-{
-    uint16_t udp_port = MOCK_UDP_PORT;
-
-    auto send_socket_fd = socket(AF_INET,SOCK_DGRAM,0);
-    ASSERT_TRUE(send_socket_fd >= 0);
-
-    PacketUdpReceiver udp_receiver;
-    udp_receiver.bind(udp_port);
-
-    jungfrau_packet send_udp_buffer;
-    send_udp_buffer.packetnum = 91;
-    send_udp_buffer.framenum = 92;
-    send_udp_buffer.bunchid = 93;
-    send_udp_buffer.debug = 94;
-
-    auto server_address = get_server_address(udp_port);
-    ::sendto(
-            send_socket_fd,
-            &send_udp_buffer,
-            JUNGFRAU_BYTES_PER_PACKET,
-            0,
-            (sockaddr*) &server_address,
-            sizeof(server_address));
-
-    this_thread::sleep_for(chrono::milliseconds(100));
-
-    jungfrau_packet recv_udp_buffer;
-    ASSERT_TRUE(udp_receiver.receive(
-            &recv_udp_buffer, JUNGFRAU_BYTES_PER_PACKET));
-
-    EXPECT_EQ(send_udp_buffer.packetnum, recv_udp_buffer.packetnum);
-    EXPECT_EQ(send_udp_buffer.framenum, recv_udp_buffer.framenum);
-    EXPECT_EQ(send_udp_buffer.bunchid, recv_udp_buffer.bunchid);
-    EXPECT_EQ(send_udp_buffer.debug, recv_udp_buffer.debug);
-
-    ASSERT_FALSE(udp_receiver.receive(
-            &recv_udp_buffer, JUNGFRAU_BYTES_PER_PACKET));
-
-    udp_receiver.disconnect();
-    ::close(send_socket_fd);
-}
-
-TEST(PacketUdpReceiver, false_recv)
-{
-    uint16_t udp_port = MOCK_UDP_PORT;
-
-    auto send_socket_fd = socket(AF_INET,SOCK_DGRAM,0);
-    ASSERT_TRUE(send_socket_fd >= 0);
-
-    PacketUdpReceiver udp_receiver;
-    udp_receiver.bind(udp_port);
-
-    jungfrau_packet send_udp_buffer;
-    jungfrau_packet recv_udp_buffer;
-
-    auto server_address = get_server_address(udp_port);
-
-    ::sendto(
-            send_socket_fd,
-            &send_udp_buffer,
-            JUNGFRAU_BYTES_PER_PACKET-1,
-            0,
-            (sockaddr*) &server_address,
-            sizeof(server_address));
-
-    ASSERT_FALSE(udp_receiver.receive(
-            &recv_udp_buffer, JUNGFRAU_BYTES_PER_PACKET));
-
-    ::sendto(
-            send_socket_fd,
-            &send_udp_buffer,
-            JUNGFRAU_BYTES_PER_PACKET,
-            0,
-            (sockaddr*) &server_address,
-            sizeof(server_address));
-
-    ASSERT_TRUE(udp_receiver.receive(
-            &recv_udp_buffer, JUNGFRAU_BYTES_PER_PACKET));
-
-    ::sendto(
-            send_socket_fd,
-            &send_udp_buffer,
-            JUNGFRAU_BYTES_PER_PACKET-1,
-            0,
-            (sockaddr*) &server_address,
-            sizeof(server_address));
-
-    ASSERT_TRUE(udp_receiver.receive(
-            &recv_udp_buffer, JUNGFRAU_BYTES_PER_PACKET-1));
-
-    udp_receiver.disconnect();
-    ::close(send_socket_fd);
-}
-
-TEST(PacketUdpReceiver, receive_many)
-{
-    auto n_msg_buffer = JF_N_PACKETS_PER_FRAME;
-    jungfrau_packet recv_buffer[n_msg_buffer];
-    iovec recv_buff_ptr[n_msg_buffer];
-    struct mmsghdr msgs[n_msg_buffer];
-    struct sockaddr_in sockFrom[n_msg_buffer];
-
-    for (int i = 0; i < n_msg_buffer; i++) {
-        recv_buff_ptr[i].iov_base = (void*) &(recv_buffer[i]);
-        recv_buff_ptr[i].iov_len = sizeof(jungfrau_packet);
-
-        msgs[i].msg_hdr.msg_iov = &recv_buff_ptr[i];
-        msgs[i].msg_hdr.msg_iovlen = 1;
-        msgs[i].msg_hdr.msg_name = &sockFrom[i];
-        msgs[i].msg_hdr.msg_namelen = sizeof(sockaddr_in);
-    }
-
-    uint16_t udp_port = MOCK_UDP_PORT;
-
-    auto send_socket_fd = socket(AF_INET,SOCK_DGRAM,0);
-    ASSERT_TRUE(send_socket_fd >= 0);
-
-    PacketUdpReceiver udp_receiver;
-    udp_receiver.bind(udp_port);
-
-    jungfrau_packet send_udp_buffer;
-
-    auto server_address = get_server_address(udp_port);
-
-    send_udp_buffer.bunchid = 0;
-    ::sendto(
-            send_socket_fd,
-            &send_udp_buffer,
-            JUNGFRAU_BYTES_PER_PACKET,
-            0,
-            (sockaddr*) &server_address,
-            sizeof(server_address));
-
-    send_udp_buffer.bunchid = 1;
-    ::sendto(
-            send_socket_fd,
-            &send_udp_buffer,
-            JUNGFRAU_BYTES_PER_PACKET,
-            0,
-            (sockaddr*) &server_address,
-            sizeof(server_address));
-
-    this_thread::sleep_for(chrono::milliseconds(10));
-
-    auto n_msgs = udp_receiver.receive_many(msgs, JF_N_PACKETS_PER_FRAME);
-    ASSERT_EQ(n_msgs, 2);
-
-    for (size_t i=0;i<n_msgs;i++) {
-        ASSERT_EQ(msgs[i].msg_len, JUNGFRAU_BYTES_PER_PACKET);
-        ASSERT_EQ(recv_buffer[i].bunchid, i);
-    }
-
-    n_msgs = udp_receiver.receive_many(msgs, JF_N_PACKETS_PER_FRAME);
-    ASSERT_EQ(n_msgs, -1);
-
-    udp_receiver.disconnect();
-    ::close(send_socket_fd);
-}