Rename buffer project

sf-buffer-recv/src/FrameUdpReceiver.cpp

@@ -0,0 +1,138 @@
+#include <cstring>
+#include <jungfrau.hpp>
+#include "FrameUdpReceiver.hpp"
+
+using namespace std;
+using namespace buffer_config;
+
+FrameUdpReceiver::FrameUdpReceiver(
+        const uint16_t port,
+        const int source_id) :
+            source_id_(source_id)
+{
+    udp_receiver_.bind(port);
+
+    for (int i = 0; i < BUFFER_UDP_N_RECV_MSG; i++) {
+        recv_buff_ptr_[i].iov_base = (void*) &(packet_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 = &sock_from_[i];
+        msgs_[i].msg_hdr.msg_namelen = sizeof(sockaddr_in);
+    }
+}
+
+FrameUdpReceiver::~FrameUdpReceiver() {
+    udp_receiver_.disconnect();
+}
+
+inline void FrameUdpReceiver::init_frame(
+        ModuleFrame& frame_metadata, const int i_packet)
+{
+    frame_metadata.pulse_id = packet_buffer_[i_packet].bunchid;
+    frame_metadata.frame_index = packet_buffer_[i_packet].framenum;
+    frame_metadata.daq_rec = (uint64_t) packet_buffer_[i_packet].debug;
+    frame_metadata.module_id = (int64_t) source_id_;
+}
+
+inline void FrameUdpReceiver::copy_packet_to_buffers(
+        ModuleFrame& metadata, char* frame_buffer, const int i_packet)
+{
+    size_t frame_buffer_offset =
+            JUNGFRAU_DATA_BYTES_PER_PACKET * packet_buffer_[i_packet].packetnum;
+    memcpy(
+            (void*) (frame_buffer + frame_buffer_offset),
+            packet_buffer_[i_packet].data,
+            JUNGFRAU_DATA_BYTES_PER_PACKET);
+
+    metadata.n_recv_packets++;
+}
+
+inline uint64_t FrameUdpReceiver::process_packets(
+        const int start_offset,
+        ModuleFrame& metadata,
+        char* frame_buffer)
+{
+    for (
+            int i_packet=start_offset;
+            i_packet < packet_buffer_n_packets_;
+            i_packet++) {
+
+        // First packet for this frame.
+        if (metadata.pulse_id == 0) {
+            init_frame(metadata, i_packet);
+
+        // Happens if the last packet from the previous frame gets lost.
+        // In the jungfrau_packet, pulse_id is called bunchid.
+        } else if (metadata.pulse_id != packet_buffer_[i_packet].bunchid) {
+            packet_buffer_loaded_ = true;
+            // Continue on this packet.
+            packet_buffer_offset_ = i_packet;
+
+            return metadata.pulse_id;
+        }
+
+        copy_packet_to_buffers(metadata, frame_buffer, i_packet);
+
+        // Last frame packet received. Frame finished.
+        if (packet_buffer_[i_packet].packetnum ==
+            JF_N_PACKETS_PER_FRAME - 1)
+        {
+            // Buffer is loaded only if this is not the last message.
+            if (i_packet+1 != packet_buffer_n_packets_) {
+                packet_buffer_loaded_ = true;
+                // Continue on next packet.
+                packet_buffer_offset_ = i_packet + 1;
+
+            // If i_packet is the last packet the buffer is empty.
+            } else {
+                packet_buffer_loaded_ = false;
+                packet_buffer_offset_ = 0;
+            }
+
+            return metadata.pulse_id;
+        }
+    }
+    // We emptied the buffer.
+    packet_buffer_loaded_ = false;
+    packet_buffer_offset_ = 0;
+
+    return 0;
+}
+
+uint64_t FrameUdpReceiver::get_frame_from_udp(
+        ModuleFrame& metadata, char* frame_buffer)
+{
+    // Reset the metadata and frame buffer for the next frame.
+    metadata.pulse_id = 0;
+    metadata.n_recv_packets = 0;
+    memset(frame_buffer, 0, JUNGFRAU_DATA_BYTES_PER_FRAME);
+
+    // Happens when last packet from previous frame was missed.
+    if (packet_buffer_loaded_) {
+
+        auto pulse_id = process_packets(
+                packet_buffer_offset_, metadata, frame_buffer);
+
+        if (pulse_id != 0) {
+            return pulse_id;
+        }
+    }
+
+    while (true) {
+
+        packet_buffer_n_packets_ = udp_receiver_.receive_many(
+                msgs_, BUFFER_UDP_N_RECV_MSG);
+
+        if (packet_buffer_n_packets_ == 0) {
+            continue;
+        }
+
+        auto pulse_id = process_packets(0, metadata, frame_buffer);
+
+        if (pulse_id != 0) {
+            return pulse_id;
+        }
+    }
+}

sf-buffer-recv/src/PacketUdpReceiver.cpp

@@ -0,0 +1,89 @@
+#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);
+
+    if (data_len < 0) {
+        return false;
+    }
+
+    if (data_len != buffer_n_bytes) {
+        return false;
+    }
+
+    return true;
+}
+
+void PacketUdpReceiver::disconnect()
+{
+    close(socket_fd_);
+    socket_fd_ = -1;
+}

sf-buffer-recv/src/main.cpp

@@ -0,0 +1,108 @@
+#include <iostream>
+#include <stdexcept>
+#include <zmq.h>
+#include <chrono>
+#include <sstream>
+#include <zconf.h>
+#include <RamBuffer.hpp>
+
+#include "formats.hpp"
+#include "buffer_config.hpp"
+#include "jungfrau.hpp"
+#include "FrameUdpReceiver.hpp"
+
+using namespace std;
+using namespace chrono;
+using namespace buffer_config;
+
+void* get_live_stream_socket(const string& detector_name, const int source_id)
+{
+    stringstream ipc_stream;
+    string LIVE_IPC_URL = BUFFER_LIVE_IPC_URL + detector_name + "-";
+    ipc_stream << LIVE_IPC_URL << source_id;
+    const auto ipc_address = ipc_stream.str();
+
+    void* ctx = zmq_ctx_new();
+    void* socket = zmq_socket(ctx, ZMQ_PUB);
+
+    const int sndhwm = BUFFER_ZMQ_SNDHWM;
+    if (zmq_setsockopt(socket, ZMQ_SNDHWM, &sndhwm, sizeof(sndhwm)) != 0) {
+        throw runtime_error(zmq_strerror(errno));
+    }
+
+    const int linger = 0;
+    if (zmq_setsockopt(socket, ZMQ_LINGER, &linger, sizeof(linger)) != 0) {
+        throw runtime_error(zmq_strerror(errno));
+    }
+
+    if (zmq_bind(socket, ipc_address.c_str()) != 0) {
+        throw runtime_error(zmq_strerror(errno));
+    }
+
+    return socket;
+}
+
+int main (int argc, char *argv[]) {
+
+    if (argc != 6) {
+        cout << endl;
+        cout << "Usage: sf_buffer [detector_name] [n_modules] [device_name]";
+        cout << " [udp_port] [root_folder] [source_id]";
+        cout << endl;
+        cout << "\tdetector_name: Detector name, example JF07T32V01" << endl;
+        cout << "\tn_modules: Number of modules in the detector." << endl;
+        cout << "\tdevice_name: Name to write to disk." << endl;
+        cout << "\tudp_port: UDP port to connect to." << endl;
+        cout << "\troot_folder: FS root folder." << endl;
+        cout << "\tsource_id: ID of the source for live stream." << endl;
+        cout << endl;
+
+        exit(-1);
+    }
+
+    string detector_name = string(argv[1]);
+    int n_modules = atoi(argv[2]);
+    string device_name = string(argv[3]);
+    int udp_port = atoi(argv[4]);
+    string root_folder = string(argv[5]);
+    int source_id = atoi(argv[6]);
+
+    uint64_t stats_counter(0);
+    uint64_t n_missed_packets = 0;
+    uint64_t n_corrupted_frames = 0;
+
+    FrameUdpReceiver receiver(udp_port, source_id);
+    RamBuffer buffer(detector_name, n_modules);
+
+    auto binary_buffer = new BufferBinaryFormat();
+    auto socket = get_live_stream_socket(detector_name, source_id);
+
+    while (true) {
+
+        auto pulse_id = receiver.get_frame_from_udp(
+                binary_buffer->metadata, binary_buffer->data);
+
+        buffer.write_frame(&(binary_buffer->metadata),
+                           &(binary_buffer->data[0]));
+
+        zmq_send(socket, &pulse_id, sizeof(pulse_id), 0);
+
+        if (binary_buffer->metadata.n_recv_packets < JF_N_PACKETS_PER_FRAME) {
+            n_missed_packets += JF_N_PACKETS_PER_FRAME -
+                    binary_buffer->metadata.n_recv_packets;
+            n_corrupted_frames++;
+        }
+
+        stats_counter++;
+        if (stats_counter == STATS_MODULO) {
+            cout << "sf_buffer:device_name " << device_name;
+            cout << " sf_buffer:n_missed_packets " << n_missed_packets;
+            cout << " sf_buffer:n_corrupted_frames " << n_corrupted_frames;
+            cout << endl;
+
+            stats_counter = 0;
+            n_missed_packets = 0;
+            n_corrupted_frames = 0;
+        }
+    }
+}