Starting class templates

core-buffer/include/RamBuffer.hpp

@@ -31,8 +31,7 @@ public:
                      ModuleFrame &meta,
                      char *data) const;
     char* read_image(const uint64_t pulse_id) const;
-    void assemble_image(
-            const uint64_t pulse_id, ImageMetadata &image_meta) const;
+    void assemble_image(const uint64_t pulse_id, ImageMetadata &image_meta) const;
 };
 
 

core-buffer/include/image.hpp

@@ -0,0 +1,29 @@
+#ifndef IMAGE_HPP
+#define JUNGFRAU_H
+
+#include <cstdint>
+#include <vector>
+#include "formats.hpp"
+
+
+class array_t {
+    public:
+        // Constructor
+        array_t(size_t i_size): m_container(i_size) {};
+
+        // Access methods
+        ModuleFrame* meta(){ return &m_metadata; };
+        char* data(){ return m_container.data(); };
+        size_t size(){ return m_container.size(); };
+    protected:
+        std::vector<char> m_container;
+        ModuleFrame m_metadata;
+};
+
+
+
+
+
+
+
+#endif  //IMAGE_HPP

jfj-combined/CMakeLists.txt

@@ -0,0 +1,12 @@
+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 jf_udp_recv)
+target_link_libraries(jfj-udp-recv jfj-udp-recv-lib zmq rt)
+
+enable_testing()
+add_subdirectory(test/)

jfj-combined/README.md

@@ -0,0 +1,164 @@
+# 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-combined/include/JfjFrameCache.hpp

@@ -0,0 +1,99 @@
+#ifndef FRAME_CACHE_HPP
+#define FRAME_CACHE_HPP
+
+#include <cstddef>
+#include <stdexcept>
+#include <iostream>
+#include <mutex>
+#include <vector>
+#include <atomic>
+#include <functional>
+#include <thread>
+
+#include "jungfraujoch.hpp"
+
+
+/** Frame cache
+
+	Reimplemented RamBuffer for the better handling of image assembly and concurrency.
+	The class operates on in-memory arrays via pointer/reference access. It uses a
+	linearly increasing pulseID index to provide some headroom for collecting frames
+	from multiple detectors.
+	**/
+class FrameCache{
+public:
+    FrameCache(uint64_t capacity, uint64_t line_size, uint64_t block_size, std::function<void(ImageMetadata*, std::vector<char>*)> callback):
+            m_capacity(capacity), m_linesize(line_size), m_blocksize(block_size), f_send(callback),
+            m_vlock(capacity), m_valid(capacity), m_fill(capacity), m_meta(capacity), m_data(capacity) {
+        // Reserve the data buffer
+        for(auto& it: m_data){ it.resize(m_linesize*m_blocksize); }
+    };
+
+
+    /** Emplace a specific frame and module **/
+    void emplace(uint64_t pulseID, uint32_t moduleID, char* ptr_source, ModuleFrame* ptr_meta){
+        uint64_t idx = pulseID % m_capacity;
+
+        // Wait for unlocking block
+        while(m_vlock[idx]){ std::this_thread::yield(); }
+
+        // Invalid cache line: Just start a new line
+        if(m_valid[idx]){ start_line(idx, ptr_meta); }
+
+        // A new frame is starting
+        if(ptr_meta->frame_index != m_meta[idx].frame_index){
+            flush_line(idx);
+            start_line(idx, ptr_meta);
+        }
+
+        m_fill[idx]++;
+        char* ptr_dest = m_data[idx].data() + moduleID * m_blocksize;
+        memcpy(ptr_dest, (void*)ptr_source, m_blocksize);
+        memcpy(&m_meta[idx], (void*)ptr_meta, sizeof(ModuleFrame));
+    }
+
+
+    void flush_all(){
+        for(int64_t idx=0; idx< m_capacity; idx++){
+            flush_line(idx);
+        }
+    }
+
+    void flush_line(uint64_t idx){
+        if(m_valid[idx]){
+            m_vlock[idx] = 1;
+            std::cout << "Send action" << std::endl;
+            f_send(&m_meta[idx], &m_data[idx]);
+
+            m_valid[idx] = 0;
+            m_fill[idx] = 0;
+            m_vlock[idx] = 0;
+        }
+    }
+
+    void start_line(uint64_t idx, ModuleFrame* ptr_meta){
+        m_vlock[idx] = 1;
+        m_meta[idx].pulse_id = ptr_meta->pulse_id;
+        m_meta[idx].frame_index = ptr_meta->frame_index;
+        m_meta[idx].daq_rec = ptr_meta->daq_rec;
+        m_meta[idx].is_good_image = true;
+        m_valid[idx] = 1;
+        m_fill[idx] = 0;
+        m_vlock[idx] = 0;
+    }
+
+private:
+    const uint64_t m_capacity;
+    const uint64_t m_linesize;
+    const uint64_t m_blocksize;
+    std::function<void(ImageMetadata*, std::vector<char>*)> f_send;
+
+    /** Main container and mutex guard **/
+    std::vector<std::atomic<uint32_t>> m_vlock;
+    std::vector<std::atomic<uint32_t>> m_valid;
+    std::vector<std::atomic<uint32_t>> m_fill;
+    std::vector<ImageMetadata> m_meta;
+    std::vector<std::vector<char>> m_data;
+};
+
+#endif // FRAME_CACHE_HPP

jfj-combined/include/JfjFrameStats.hpp

@@ -0,0 +1,31 @@
+#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-combined/include/JfjFrameWorker.hpp

@@ -0,0 +1,37 @@
+#ifndef SF_DAQ_BUFFER_JFJ_UDPRECEIVER_HPP
+#define SF_DAQ_BUFFER_JFJ_UDPRECEIVER_HPP
+
+#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 JfjFrameWorker {
+    PacketUdpReceiver m_udp_receiver;
+    bool in_progress = false;
+    uint64_t m_frame_index = 0;
+    const uint64_t m_num_modules;
+    const uint64_t m_num_packets;
+    const uint64_t m_num_data_bytes;
+
+    // PacketBuffer<jfjoch_packet_t, buffer_config::BUFFER_UDP_N_RECV_MSG> m_buffer;
+    PacketBuffer<jfjoch_packet_t, 64> 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);
+
+    std::function<void(uint64_t index, uint32_t module, char* ptr_data, ModuleFrame* ptr_meta)> f_push_callback;
+public:
+    JfjFrameUdpReceiver(const uint16_t port, std::function<void(uint64_t index, uint32_t module, char* ptr_data, ModuleFrame* ptr_meta)> callback);
+    virtual ~JfjFrameUdpReceiver();
+    uint64_t get_frame_from_udp(ModuleFrame& metadata, char* frame_buffer);
+    void run();
+};
+
+#endif //SF_DAQ_BUFFER_JFJ_UDPRECEIVER_HPP

jfj-combined/include/PacketBuffer.hpp

@@ -0,0 +1,113 @@
+#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());
+        // std::cout << "Received " << this->idx_write << " frames" << std::endl;
+        // 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-combined/include/PacketUdpReceiver.hpp

@@ -0,0 +1,26 @@
+#ifndef UDPRECEIVER_H
+#define UDPRECEIVER_H
+
+#if defined(WIN32) || defined(_WIN32) || defined(MINGW32)
+    #include <winsock2.h>
+#else
+    #include <sys/socket.h>
+#endif // defined
+
+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-combined/include/Watchdog.hpp

@@ -0,0 +1,72 @@
+#ifndef FRAME_CACHE_HPP
+#define FRAME_CACHE_HPP
+
+#include <thread>
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <mutex>
+#include <iostream>
+
+
+class Watchdog{
+public:
+    Watchdog(uint32_t timeout, std::function<void()> callback): m_timeout(timeout), m_callback(callback) {
+        m_timeout = timeout;
+        m_callback = callback;
+        m_running = false;
+    };
+    ~Watchdog() { Stop(); };
+    void Start();
+    void Stop();
+    void Kick();
+
+private:
+    std::atomic<bool> m_running = false;
+    std::function<void()> m_callback;
+    uint32_t m_timeout;
+    std::chrono::time_point m_lastKick;
+
+
+    std::thread m_thread;
+    std::mutex m_mutex;
+    steady_clock::time_point m_lastPetTime;
+    std::condition_variable m_stopCondition;
+    void Loop();
+};
+
+
+void Watchdog::Start(){
+    std::unique_lock<std::mutex> lock(m_mutex);
+    if(m_running == false){
+        m_running = true;
+        m_lastKick = std::chrono::steady_clock::now();
+        m_thread = std::thread(&Watchdog::Loop, this);
+    }
+}
+
+void Watchdog::Stop(){
+    std::unique_lock<std::mutex> locker(m_mutex);
+    if(m_running == true){
+        m_running = false;
+        m_thread.join();
+    }
+}
+
+void Watchdog::Kick(){
+    std::unique_lock<std::mutex> locker(m_mutex);
+    m_lastKick = steady_clock::now();
+}
+
+void Watchdog::Loop(){
+    while(m_running){
+        if((std::chrono::now() - m_last_kick) < m_timeout){
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        } else {
+            std::cout << "Expired timer" << std::endl;
+            m_callback();
+        }
+    }
+}
+
+#endif // FRAME_CACHE_HPP

jfj-combined/src/JfjFrameStats.cpp

@@ -0,0 +1,71 @@
+#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-combined/src/JfjFrameWorker.cpp

@@ -0,0 +1,151 @@
+#include <cstring>
+#include "JfjFrameWorker.hpp"
+
+using namespace std;
+using namespace buffer_config;
+
+
+
+JfjFrameWorker::JfjFrameWorker(const uint16_t port, std::function<void(uint64_t index, uint32_t module, char* ptr_data, ModuleFrame* ptr_meta)> callback):
+        m_num_modules(n_modules), m_num_packets(n_modules*JFJOCH_N_PACKETS_PER_MODULE),
+        m_num_data_bytes(n_modules*JFJOCH_DATA_BYTES_PER_MODULE), f_push_callback(callback) {
+    m_udp_receiver.bind(port);
+}
+
+JfjFrameWorker::~JfjFrameWorker() {
+    m_udp_receiver.disconnect();
+}
+
+inline void JfjFrameWorker::init_frame(ModuleFrame& metadata, const jfjoch_packet_t& 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 JfjFrameWorker::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;
+            if(this->in_progress){
+                this->in_progress = false;
+                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;
+        this->in_progress = true;
+
+        // 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 == m_num_packets - 1){
+            this->in_progress = false;
+            return metadata.pulse_id;
+        }
+    }
+
+    // We emptied the buffer.
+   // m_buffer.reset();
+    return 0;
+}
+
+uint64_t JfjFrameWorker::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, m_num_data_bytes);
+
+
+    // 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.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; }
+    }
+}
+
+
+
+
+
+std::generator<uint64_t> JfjFrameWorker::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, m_num_data_bytes);
+
+
+    // Process leftover packages in the buffer
+    if (!m_buffer.is_empty()) {
+        auto pulse_id = process_packets(metadata, frame_buffer);
+        if (pulse_id != 0) { co_yield pulse_id; }
+    }
+
+
+    while (true) {
+        // Receive new packages (pass if none)...
+        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) { co_yield pulse_id; }
+    }
+}
+
+void JfjFrameWorker::run(){
+
+
+    // Might be better creating a structure for double buffering
+    ModuleFrame frameMeta;
+    char* dataBuffer = new char[JFJOCH_DATA_BYTES_PER_MODULE];
+
+    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 = co_await get_frame_from_udp(&frameMeta, dataBuffer);
+
+        if(pulse_id>1000){
+            f_push_callback(pulse_id, m_moduleID, dataBuffer, frameMeta);
+        }
+    }
+
+    delete[] dataBuffer;
+}
+
+
+
+
+
+
+
+
+
+
+
+
+

jfj-combined/src/PacketUdpReceiver.cpp

@@ -0,0 +1,66 @@
+#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-combined/src/dummyMain.cpp

@@ -0,0 +1,27 @@
+#include <iostream>
+#include <stdexcept>
+#include <zmq.h>
+
+#include "formats.hpp"
+#include "../include/JfjFrameCache.hpp"
+#include "../include/JfjFrameWorker.hpp"
+
+
+void dummy_sender(ImageMetadata* meta, std::vector<char>* data){
+    std::cout << "Sending " << meta->frame_index << std::endl;
+}
+
+
+
+int main (int argc, char *argv[]) {
+
+
+    FrameCache cache(32, 3, JFJOCH_DATA_BYTES_PER_MODULE, &dummy_sender);
+
+
+
+    JfjFrameWorker W0(5340, 0, cache.emplace);
+    JfjFrameWorker W1(5341, 1, cache.emplace);
+    JfjFrameWorker W2(5342, 2, cache.emplace);
+
+}

jfj-combined/src/main.cpp

@@ -0,0 +1,74 @@
+#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, 8);
+    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[8 * JFJOCH_DATA_BYTES_PER_MODULE];
+
+    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-combined/test/CMakeLists.txt

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

jfj-combined/test/main.cpp

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

jfj-combined/test/mock/udp.hpp

@@ -0,0 +1,16 @@
+#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-combined/test/test_FrameUdpReceiver.cpp

@@ -0,0 +1,199 @@
+#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;
+#define NUM_TEST_MODULES 3
+
+
+TEST(BufferUdpReceiver, simple_recv){
+    int n_packets = NUM_TEST_MODULES * JFJOCH_N_PACKETS_PER_MODULE;
+    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, NUM_TEST_MODULES);
+
+    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[]>(NUM_TEST_MODULES*JFJOCH_DATA_BYTES_PER_MODULE);
+
+    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);
+        ASSERT_EQ(metadata.n_recv_packets, n_packets);
+    }
+
+    ::close(send_socket_fd);
+}
+
+TEST(BufferUdpReceiver, missing_middle_packet){
+    int n_packets = NUM_TEST_MODULES * JFJOCH_N_PACKETS_PER_MODULE;
+    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, NUM_TEST_MODULES);
+
+    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[]>(NUM_TEST_MODULES * JFJOCH_DATA_BYTES_PER_MODULE);
+
+    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 = NUM_TEST_MODULES * JFJOCH_N_PACKETS_PER_MODULE;
+    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, NUM_TEST_MODULES);
+
+    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[]>(NUM_TEST_MODULES * JFJOCH_DATA_BYTES_PER_MODULE);
+
+    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);
+        // -2 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 = NUM_TEST_MODULES * JFJOCH_N_PACKETS_PER_MODULE;
+    int n_frames = 4;
+
+    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, NUM_TEST_MODULES);
+
+    auto handle = async(launch::async, [&](){
+        for (int64_t i_frame=0; i_frame < n_frames+1; 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[]>(NUM_TEST_MODULES * JFJOCH_DATA_BYTES_PER_MODULE);
+
+    // 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-combined/test/test_PacketBuffer.cpp

@@ -0,0 +1,76 @@
+#include <netinet/in.h>
+#include <jungfraujoch.hpp>
+#include "gtest/gtest.h"
+#include "PacketBuffer.hpp"
+
+#include <thread>
+#include <chrono>
+#include <future>
+
+using namespace std;
+
+
+
+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;
+}
+
+
+
+
+class MockReceiver{
+    public:
+        uint64_t idx_packet = 42000;
+        uint64_t packet_per_frame = 512;
+        uint64_t num_bunches = 100;
+        uint64_t num_packets =50;
+        jfjoch_packet_t tmp;
+
+        uint64_t receive_many(mmsghdr* msgs, const size_t n_msgs){
+            // Receive 'num_packets numner of packets'
+            for(int ii=0; ii<num_packets; ii++){                
+                tmp.framenum = idx_packet / packet_per_frame;
+                tmp.bunchid = 1000 + idx_packet / packet_per_frame;
+                tmp.packetnum = idx_packet % packet_per_frame;
+                memcpy( msgs[ii].msg_hdr.msg_iov->iov_base, &tmp, sizeof(tmp));
+                idx_packet++;
+            }
+            return num_packets;
+        };
+};
+
+
+
+TEST(BufferUdpReceiver, packetbuffer_simple){
+
+    PacketBuffer<jfjoch_packet_t, 128> p_buffer;
+    MockReceiver mockery;
+    uint64_t prev_bunch, prev_packet;
+    jfjoch_packet_t p_pop;
+    
+    mockery.idx_packet = 7*512 + 13;
+    mockery.num_packets = 25;
+    
+    p_buffer.fill_from(mockery);
+    
+    // First packet
+    ASSERT_FALSE(p_buffer.is_empty());
+    ASSERT_EQ(p_buffer.size(), 25);  
+    
+    ASSERT_EQ(p_buffer.peek_front().bunchid, 1007);
+    ASSERT_EQ(p_buffer.peek_front().packetnum, 13);
+    prev_bunch = p_buffer.peek_front().bunchid;
+    prev_packet = p_buffer.peek_front().packetnum;
+    
+    p_pop = p_buffer.pop_front();
+    ASSERT_EQ(p_buffer.size(), 24);
+
+    ASSERT_EQ(p_pop.bunchid, prev_bunch);
+    ASSERT_EQ(p_pop.packetnum, prev_packet);
+    ASSERT_EQ(p_buffer.peek_front().bunchid, prev_bunch);
+    ASSERT_EQ(p_buffer.peek_front().packetnum, prev_packet+1);
+};

jfj-combined/test/test_PacketUdpReceiver.cpp

@@ -0,0 +1,170 @@
+#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);
+}