Make core-buffer folder with all common stuff used in the buffer

core-buffer/CMakeLists.txt

@@ -0,0 +1,13 @@
+file(GLOB SOURCES
+        src/*.cpp)
+
+add_library(core-buffer STATIC ${SOURCES})
+target_include_directories(core-buffer PUBLIC include/)
+target_link_libraries(core-buffer core-writer)
+
+if(CMAKE_BUILD_TYPE STREQUAL "Debug")
+    target_compile_definitions(core-buffer PRIVATE DEBUG_OUTPUT)
+endif()
+
+enable_testing()
+add_subdirectory(test/)

core-buffer/include/buffer_utils.hpp

@@ -0,0 +1,35 @@
+#ifndef BUFFER_UTILS_HPP
+#define BUFFER_UTILS_HPP
+
+const size_t FILE_MOD = 1000;
+const size_t FOLDER_MOD = 100000;
+
+std::string get_filename(
+        std::string root_folder,
+        std::string device_name,
+        uint64_t pulse_id)
+{
+    uint64_t folder_base = pulse_id / FOLDER_MOD;
+    folder_base *= FOLDER_MOD;
+
+    uint64_t file_base = pulse_id / FILE_MOD;
+    file_base *= FILE_MOD;
+
+    std::stringstream folder;
+    folder << root_folder << "/";
+    folder << device_name << "/";
+    folder << folder_base << "/";
+    folder << file_base << ".h5";
+
+    return folder.str();
+}
+
+std::size_t get_file_frame_index(uint64_t pulse_id)
+{
+    uint64_t file_base = pulse_id / FILE_MOD;
+    file_base *= FILE_MOD;
+
+    return pulse_id - file_base;
+}
+
+#endif //BUFFER_UTILS_HPP

core-buffer/src/UdpReceiver.cpp

@@ -0,0 +1,91 @@
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <iostream>
+#include "UdpReceiver.hpp"
+
+using namespace std;
+
+UdpReceiver::UdpReceiver() :
+    socket_fd_(-1)
+{
+}
+
+UdpReceiver::~UdpReceiver()
+{
+    close();
+}
+
+void UdpReceiver::bind(const uint16_t port, const size_t usec_timeout)
+{
+    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.");
+    }
+
+    #ifdef DEBUG_OUTPUT
+        using namespace date;
+        using namespace chrono;
+        cout << "[" << system_clock::now() << "]";
+        cout << "[UdpReceiver::bind]";
+        cout << " port " << port;
+        cout << " usec_timeout " << usec_timeout << endl;
+    #endif
+
+    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);
+
+    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.");
+    }
+
+    struct timeval udp_socket_timeout;
+    udp_socket_timeout.tv_sec = 0;
+    udp_socket_timeout.tv_usec = usec_timeout;
+
+    setsockopt(
+            socket_fd_,
+            SOL_SOCKET,
+            SO_RCVTIMEO,
+            (const char*)&udp_socket_timeout,
+            sizeof(struct timeval));
+}
+
+bool UdpReceiver::receive(void* buffer, size_t buffer_n_bytes)
+{
+    auto data_len = recv(socket_fd_, buffer, buffer_n_bytes, 0);
+
+    if (data_len < 0) {
+        return false;
+    }
+
+    if (data_len != buffer_n_bytes) {
+        #ifdef DEBUG_OUTPUT
+            using namespace date;
+            using namespace chrono;
+            cout << "[" << system_clock::now() << "]";
+            cout << "[UdpReceiver::receive]";
+            cout << " Expected buffer_n_bytes " << buffer_n_bytes;
+            cout << " but got data_len " << data_len << endl;
+        #endif
+        return false;
+    }
+
+    return true;
+}
+
+void UdpReceiver::close()
+{
+    ::close(socket_fd_);
+    socket_fd_ = -1;
+}

core-buffer/src/UdpRecvModule.cpp

@@ -0,0 +1,178 @@
+#include "UdpRecvModule.hpp"
+#include "jungfrau.hpp"
+#include <iostream>
+#include <UdpReceiver.hpp>
+
+using namespace std;
+
+UdpRecvModule::UdpRecvModule(RingBuffer<UdpFrameMetadata>& ring_buffer) :
+            ring_buffer_(ring_buffer),
+            is_receiving_(false)
+{
+
+}
+
+UdpRecvModule::~UdpRecvModule()
+{
+    stop_recv();
+}
+
+void UdpRecvModule::start_recv(
+        const uint16_t udp_port,
+        const size_t frame_n_bytes)
+{
+    if (is_receiving_ == true) {
+        std::stringstream err_msg;
+
+        using namespace date;
+        using namespace chrono;
+        err_msg << "[" << system_clock::now() << "]";
+        err_msg << "[UdpRecvModule::start_recv]";
+        err_msg << " Receivers already running." << endl;
+
+        throw runtime_error(err_msg.str());
+    }
+
+    #ifdef DEBUG_OUTPUT
+        using namespace date;
+        using namespace chrono;
+        cout << "[" << system_clock::now() << "]";
+        cout << "[UdpRecvModule::start_recv]";
+        cout << " Starting with ";
+        cout << "udp_port " << udp_port << endl;
+    #endif
+
+    is_receiving_ = true;
+
+    if (receiving_thread_.joinable()) {
+        receiving_thread_.join();
+    }
+
+    receiving_thread_ = thread(
+            &UdpRecvModule::receive_thread, this,
+            udp_port,
+            frame_n_bytes);
+}
+
+void UdpRecvModule::stop_recv()
+{
+#ifdef DEBUG_OUTPUT
+    using namespace date;
+    using namespace chrono;
+    cout << "[" << system_clock::now() << "]";
+    cout << "UdpRecvModule::stop_recv";
+    cout << " Stop receiving." << endl;
+#endif
+
+    is_receiving_ = false;
+
+    if (receiving_thread_.joinable()) {
+        receiving_thread_.join();
+    }
+}
+
+void UdpRecvModule::receive_thread(
+        const uint16_t udp_port,
+        const size_t frame_size)
+{
+    try {
+        ring_buffer_.initialize(frame_size);
+
+        UdpReceiver udp_receiver;
+        udp_receiver.bind(udp_port);
+
+        auto metadata = make_shared<UdpFrameMetadata>();
+        metadata->frame_bytes_size = JUNGFRAU_DATA_BYTES_PER_FRAME;
+        metadata->pulse_id = 0;
+        metadata->n_recv_packets = 0;
+
+        char* frame_buffer = ring_buffer_.reserve(metadata);
+
+        jungfrau_packet packet_buffer;
+
+        while (is_receiving_.load(memory_order_relaxed)) {
+
+            if (!udp_receiver.receive(
+                    &packet_buffer,
+                    JUNGFRAU_BYTES_PER_PACKET)) {
+                continue;
+            }
+
+            auto* frame_metadata = metadata.get();
+
+            // TODO: Horrible. Breake it down into methods.
+
+            // First packet for this frame.
+            if (frame_metadata->pulse_id == 0) {
+                frame_metadata->frame_index = packet_buffer.framenum;
+                frame_metadata->pulse_id = packet_buffer.bunchid;
+                frame_metadata->recv_packets_1 = ~(uint64_t)0;
+                frame_metadata->recv_packets_2 = ~(uint64_t)0;
+                frame_metadata->daq_rec = packet_buffer.debug;
+            // Packet from new frame, while we lost the last packet of
+            // previous frame.
+            } else if (frame_metadata->pulse_id != packet_buffer.bunchid) {
+                ring_buffer_.commit(metadata);
+
+                metadata = make_shared<UdpFrameMetadata>();
+                metadata->frame_bytes_size = JUNGFRAU_DATA_BYTES_PER_FRAME;
+                metadata->pulse_id = 0;
+                metadata->n_recv_packets = 0;
+
+                frame_buffer = ring_buffer_.reserve(metadata);
+                memset(frame_buffer, 0, JUNGFRAU_DATA_BYTES_PER_FRAME);
+
+                frame_metadata->frame_index = packet_buffer.framenum;
+                frame_metadata->pulse_id = packet_buffer.bunchid;
+                frame_metadata->recv_packets_1 = ~(uint64_t)0;
+                frame_metadata->recv_packets_2 = ~(uint64_t)0;
+                frame_metadata->daq_rec = packet_buffer.debug;
+            }
+
+            size_t frame_buffer_offset =
+                    JUNGFRAU_DATA_BYTES_PER_PACKET * packet_buffer.packetnum;
+
+            memcpy(
+                    (void*) (frame_buffer + frame_buffer_offset),
+                    packet_buffer.data,
+                    JUNGFRAU_DATA_BYTES_PER_PACKET);
+
+            frame_metadata->n_recv_packets++;
+
+            if (packet_buffer.packetnum < 64) {
+                frame_metadata->recv_packets_1 ^=
+                        (uint64_t)1 << packet_buffer.packetnum;
+            } else {
+                frame_metadata->recv_packets_2 ^=
+                        (uint64_t)1 << (packet_buffer.packetnum - 64);
+            }
+
+            // Frame finished with last packet.
+            if (packet_buffer.packetnum == JUNGFRAU_N_PACKETS_PER_FRAME-1)
+            {
+                ring_buffer_.commit(metadata);
+
+                metadata = make_shared<UdpFrameMetadata>();
+                metadata->frame_bytes_size = JUNGFRAU_DATA_BYTES_PER_FRAME;
+                metadata->pulse_id = 0;
+                metadata->n_recv_packets = 0;
+
+                frame_buffer = ring_buffer_.reserve(metadata);
+                memset(frame_buffer, 0, JUNGFRAU_DATA_BYTES_PER_FRAME);
+            }
+        }
+
+    } catch (const std::exception& e) {
+        is_receiving_ = false;
+
+        using namespace date;
+        using namespace chrono;
+
+        cout << "[" << system_clock::now() << "]";
+        cout << "[UdpRecvModule::receive_thread]";
+        cout << " Stopped because of exception: " << endl;
+        cout << e.what() << endl;
+
+        throw;
+    }
+}

core-buffer/test/CMakeLists.txt

@@ -0,0 +1,5 @@
+add_executable(core-buffer_tests test_main.cpp)
+
+target_link_libraries(core-buffer_tests
+        core-buffer
+        gtest)

core-buffer/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

core-buffer/test/test_UdpReceiver.cpp

@@ -0,0 +1,99 @@
+#include <netinet/in.h>
+#include <jungfrau.hpp>
+#include "gtest/gtest.h"
+#include "UdpReceiver.hpp"
+#include "mock/udp.hpp"
+
+
+TEST(UdpReceiver, 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);
+
+    UdpReceiver 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));
+
+    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.close();
+    ::close(send_socket_fd);
+}
+
+TEST(UdpReceiver, 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);
+
+    UdpReceiver 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.close();
+    ::close(send_socket_fd);
+}

core-buffer/test/test_UdpRecvModule.cpp

@@ -0,0 +1,91 @@
+#include "gtest/gtest.h"
+#include "UdpRecvModule.hpp"
+#include "jungfrau.hpp"
+#include "mock/udp.hpp"
+
+using namespace std;
+
+TEST(UdpRecvModule, basic_interaction)
+{
+    uint16_t udp_port(12000);
+
+    RingBuffer<UdpFrameMetadata> ring_buffer(10);
+    UdpRecvModule udp_recv_module(ring_buffer);
+
+    udp_recv_module.start_recv(udp_port, JUNGFRAU_DATA_BYTES_PER_FRAME);
+
+    EXPECT_THROW(
+            udp_recv_module.start_recv(udp_port, JUNGFRAU_BYTES_PER_PACKET),
+            runtime_error);
+
+    // Stop should never throw an exception.
+    udp_recv_module.stop_recv();
+    EXPECT_NO_THROW(udp_recv_module.stop_recv());
+}
+
+TEST(UdpRecvModule, simple_recv)
+{
+    uint16_t udp_port(MOCK_UDP_PORT);
+    size_t n_msg(128);
+
+    RingBuffer<UdpFrameMetadata> ring_buffer(10);
+    UdpRecvModule udp_recv_module(ring_buffer);
+
+    udp_recv_module.start_recv(udp_port, JUNGFRAU_DATA_BYTES_PER_FRAME);
+
+    this_thread::sleep_for(chrono::milliseconds(100));
+
+    // The first slot should be already reserved in the ring buffer.
+    ASSERT_FALSE(ring_buffer.is_empty());
+
+    auto send_socket_fd = socket(AF_INET,SOCK_DGRAM,0);
+    ASSERT_TRUE(send_socket_fd >= 0);
+
+    auto server_address = get_server_address(udp_port);
+
+    jungfrau_packet send_udp_buffer;
+    send_udp_buffer.bunchid = 100;
+    send_udp_buffer.debug = 1000;
+
+    send_udp_buffer.framenum = 1;
+    for (size_t i=0; i<n_msg; i++) {
+        send_udp_buffer.packetnum = i;
+
+        ::sendto(
+                send_socket_fd,
+                &send_udp_buffer,
+                JUNGFRAU_BYTES_PER_PACKET,
+                0,
+                (sockaddr*) &server_address,
+                sizeof(server_address));
+    }
+
+    ASSERT_FALSE(ring_buffer.is_empty());
+    auto result = ring_buffer.read();
+    // The slot should be reserved, but not yet committed.
+    // Only with next frame packet commit.
+    ASSERT_TRUE(result.first == nullptr);
+
+    // When packet from new frame is received, the previous frame should be
+    // committed to the ring buffer.
+    send_udp_buffer.framenum = 2;
+    for (size_t i=0; i<128; i++){
+        send_udp_buffer.packetnum = i;
+
+        ::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));
+
+    ASSERT_FALSE(ring_buffer.is_empty());
+    auto result2 = ring_buffer.read();
+    ASSERT_TRUE(result2.first != nullptr);
+
+    ::close(send_socket_fd);
+}

core-buffer/test/test_buffer_utils.cpp

@@ -0,0 +1,47 @@
+#include "gtest/gtest.h"
+#include "buffer_utils.hpp"
+
+using namespace std;
+
+TEST(BufferUtils, get_filename)
+{
+    auto expected_file = "/root/device-1/12300000/12345000.h5";
+
+    auto root_folder = "/root";
+    auto device_name = "device-1";
+
+    auto result = get_filename(
+            root_folder,
+            device_name,
+            12345000);
+
+    ASSERT_EQ(result, expected_file);
+
+    auto result2 = get_filename(
+            root_folder,
+            device_name,
+            12345999);
+
+    ASSERT_EQ(result2, expected_file);
+
+    auto result3 = get_filename(
+            root_folder,
+            device_name,
+            12346000);
+
+    ASSERT_NE(result3, expected_file);
+
+    auto result4 = get_filename(
+            root_folder,
+            device_name,
+            12344999);
+
+    ASSERT_NE(result4, expected_file);
+}
+
+TEST(BufferUtils, get_file_frame_index)
+{
+    ASSERT_EQ(get_file_frame_index(12345000), 0);
+    ASSERT_EQ(get_file_frame_index(12345543), 543);
+    ASSERT_EQ(get_file_frame_index(12345999), 999);
+}

core-buffer/test/test_main.cpp

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