sf-daq/sf_daq_buffer
0
0
Fork 0
mirror of https://github.com/paulscherrerinstitute/sf_daq_buffer.git synced 2026-05-01 08:52:21 +02:00
Code Issues Packages Projects Releases Wiki Activity

Isolate detector differences to FrameUdpRecv

Browse Source
This commit is contained in:
babic_a
2021-07-05 15:13:46 +02:00
parent f9e8b2750a
commit a64ea62c88
3 changed files with 53 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
std-udp-recv/include/FrameUdpReceiver.hpp
+13 -11
View File
@@ -6,18 +6,20 @@
#include "formats.hpp"
#include "buffer_config.hpp"
#ifdef USE_EIGER
#include "eiger.hpp"
#else
#include "jungfrau.hpp"
#endif
class FrameUdpReceiver {
const int module_id_;
const int bit_depth_;
const size_t n_packets_per_frame_;
const size_t data_bytes_per_frame_;
PacketUdpReceiver udp_receiver_;
const int n_packets_per_frame_;
det_packet packet_buffer_[buffer_config::BUFFER_UDP_N_RECV_MSG];
iovec recv_buff_ptr_[buffer_config::BUFFER_UDP_N_RECV_MSG];
mmsghdr msgs_[buffer_config::BUFFER_UDP_N_RECV_MSG];
sockaddr_in sock_from_[buffer_config::BUFFER_UDP_N_RECV_MSG];
det_packet* const packet_buffer_;
iovec* const recv_buff_ptr_;
mmsghdr* const msgs_;
sockaddr_in* const sock_from_;
bool packet_buffer_loaded_ = false;
int packet_buffer_n_packets_ = 0;
@@ -30,9 +32,9 @@ class FrameUdpReceiver {
const int n_packets, ModuleFrame& metadata, char* frame_buffer);
public:
FrameUdpReceiver(const int module_id, const uint16_t port, const int n_modules, const int n_submodules, const int bit_depth);
FrameUdpReceiver(const uint16_t port, const int n_packets_per_frame);
virtual ~FrameUdpReceiver();
uint64_t get_frame_from_udp(ModuleFrame& metadata, char* frame_buffer);
uint64_t get_frame_from_udp(ModuleFrame& meta, char* frame_buffer);
};
std-udp-recv/src/FrameUdpReceiver.cpp
+38 -48
View File
@@ -3,37 +3,32 @@
#include <ostream>
#include <iostream>
#include <chrono>
#include <unistd.h>
#include "date.h"
using namespace std;
using namespace buffer_config;
FrameUdpReceiver::FrameUdpReceiver(
const int module_id,
const uint16_t port,
const int n_modules,
const int n_submodules,
const int bit_depth):
module_id_(module_id),
bit_depth_(bit_depth),
n_packets_per_frame_(bit_depth_ * MODULE_N_PIXELS / 8 / DATA_BYTES_PER_PACKET / n_modules * n_submodules),
data_bytes_per_frame_(n_packets_per_frame_ * DATA_BYTES_PER_PACKET)
const uint16_t port, const int n_packets_per_frame) :
n_packets_per_frame_(n_packets_per_frame),
packet_buffer_(new det_packet[n_packets_per_frame_]),
recv_buff_ptr_(new iovec[n_packets_per_frame_]),
msgs_(new mmsghdr[n_packets_per_frame_]),
sock_from_(new sockaddr_in[n_packets_per_frame_])
{
#ifdef DEBUG_OUTPUT
using namespace date;
cout << " [" << std::chrono::system_clock::now();
cout << "] [FrameUdpReceiver::FrameUdpReceiver] :";
cout << " Details of FrameUdpReceiver:";
cout << "module_id: " << module_id_;
cout << " || port: " << port;
cout << " || bit_depth : " << bit_depth_;
cout << " || n_packets_per_frame_ : " << n_packets_per_frame_;
cout << " || data_bytes_per_frame_: " << data_bytes_per_frame_ << " !!";
cout << " || n_packets_per_frame: " << n_packets_per_frame_;
cout << endl;
#endif
udp_receiver_.bind(port);
for (int i = 0; i < BUFFER_UDP_N_RECV_MSG; i++) {
for (int i = 0; i < n_packets_per_frame_; i++) {
recv_buff_ptr_[i].iov_base = (void*) &(packet_buffer_[i]);
recv_buff_ptr_[i].iov_len = sizeof(det_packet);
@@ -46,28 +41,28 @@ FrameUdpReceiver::FrameUdpReceiver(
FrameUdpReceiver::~FrameUdpReceiver() {
udp_receiver_.disconnect();
delete[] packet_buffer_;
delete[] recv_buff_ptr_;
delete[] msgs_;
delete[] sock_from_;
}
inline void FrameUdpReceiver::init_frame(
ModuleFrame& frame_metadata, const int i_packet)
{
// Eiger has no pulse_id, frame number instead
frame_metadata.pulse_id = packet_buffer_[i_packet].framenum;
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) module_id_;
frame_metadata.bit_depth = (int16_t) bit_depth_;
frame_metadata.row = (int16_t) packet_buffer_[i_packet].row;
frame_metadata.column = (int16_t) packet_buffer_[i_packet].column;
frame_metadata.pos_y = (int16_t) packet_buffer_[i_packet].row;
frame_metadata.pos_x = (int16_t) packet_buffer_[i_packet].column;
#ifdef DEBUG_OUTPUT
using namespace date;
cout << " [" << std::chrono::system_clock::now();
cout << "] [FrameUdpReceiver::init_frame] :";
cout << "module_id: " << module_id_;
cout << " || row: " << frame_metadata.row;
cout << " || column: " << frame_metadata.column;
cout << " || pos_y: " << frame_metadata.pos_x;
cout << " || pos_x: " << frame_metadata.pos_x;
cout << " || pulse_id: " << frame_metadata.pulse_id;
cout << " || frame_index: " << frame_metadata.frame_index;
cout << endl;
@@ -86,7 +81,6 @@ inline void FrameUdpReceiver::copy_packet_to_buffers(
DATA_BYTES_PER_PACKET);
metadata.n_recv_packets++;
// cout << "[ frame" << metadata.frame_index << "] NUMBER OF RECV PACKETS : " << metadata.n_recv_packets ;
}
inline uint64_t FrameUdpReceiver::process_packets(
@@ -99,11 +93,12 @@ inline uint64_t FrameUdpReceiver::process_packets(
i_packet++) {
// First packet for this frame.
if (metadata.pulse_id == 0) {
if (metadata.frame_index == 0) {
init_frame(metadata, i_packet);
// Happens if the last packet from the previous frame gets lost.
// In the jungfrau_packet, framenum is the trigger number (how many triggers from detector power-on) happened
// In the jungfrau_packet, framenum is the trigger number
// (how many triggers from detector power-on) happened
} else if (metadata.frame_index != packet_buffer_[i_packet].framenum) {
packet_buffer_loaded_ = true;
// Continue on this packet.
@@ -121,19 +116,19 @@ inline uint64_t FrameUdpReceiver::process_packets(
#ifdef DEBUG_OUTPUT
using namespace date;
cout << " [" << std::chrono::system_clock::now();
cout << "] [FrameUdpReceiver::process_packets] :";
cout << " Frame " << metadata.frame_index << " || ";
cout << "] [frameudpreceiver::process_packets] :";
cout << " frame " << metadata.frame_index << " || ";
cout << packet_buffer_[i_packet].packetnum << " packets received.";
cout << " PULSE ID "<< metadata.pulse_id;
cout << " pulse id "<< metadata.pulse_id;
cout << endl;
#endif
// Buffer is loaded only if this is not the last message.
// 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.
// continue on next packet.
packet_buffer_offset_ = i_packet + 1;
// If i_packet is the last packet the buffer is empty.
// if i_packet is the last packet the buffer is empty.
} else {
packet_buffer_loaded_ = false;
packet_buffer_offset_ = 0;
@@ -150,36 +145,31 @@ inline uint64_t FrameUdpReceiver::process_packets(
}
uint64_t FrameUdpReceiver::get_frame_from_udp(
ModuleFrame& metadata, char* frame_buffer)
ModuleFrame& meta, 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, 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;
auto frame_index = process_packets(
packet_buffer_offset_, meta, frame_buffer);
if (frame_index != 0) {
return frame_index;
}
}
while (true) {
packet_buffer_n_packets_ = udp_receiver_.receive_many(
msgs_, BUFFER_UDP_N_RECV_MSG);
msgs_, n_packets_per_frame_);
if (packet_buffer_n_packets_ == 0) {
continue;
}
auto pulse_id = process_packets(0, metadata, frame_buffer);
auto frame_index = process_packets(0, meta, frame_buffer);
if (pulse_id != 0) {
return pulse_id;
if (frame_index != 0) {
return frame_index;
}
}
}
std-udp-recv/src/main.cpp
+2 -7
View File
@@ -10,12 +10,6 @@
#include "FrameStats.hpp"
#include "UdpRecvConfig.hpp"
#ifdef USE_EIGER
#include "eiger.hpp"
#else
# include "jungfrau.hpp"
#endif
using namespace std;
using namespace chrono;
using namespace buffer_config;
@@ -27,7 +21,8 @@ int main (int argc, char *argv[]) {
if (argc != 4) {
cout << endl;
cout << "Usage: std_udp_recv [udp_recv_config_filename] [module_id] [bit_depth]";
cout << "Usage: std_udp_recv [udp_recv_config_filename] [module_id] "
"[bit_depth]";
cout << endl;
cout << "\tudp_recv_config_filename: detector config file path." << endl;
cout << "\tmodule_id: id of the module for this process." << endl;