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

Refactored to linear buffer

Browse Source
This commit is contained in:
mohacsi_i
2021-06-01 10:22:10 +02:00
parent 0ae9eb1e88
commit 20e7376615
12 changed files with 494 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core-buffer/include/jungfraujoch.hpp
+1 -1
View File
@@ -17,7 +17,7 @@ struct jfjoch_packet_t {
uint32_t exptime;
uint32_t packetnum;
double bunchid;
uint64_t bunchid;
uint64_t timestamp;
uint16_t moduleID;
core-buffer/test/test_PacketBuffer.cpp
+254
View File
@@ -0,0 +1,254 @@
#include <netinet/in.h>
#include <jungfraujoch.hpp>
#include "gtest/gtest.h"
#include "PacketBuffer.hpp"
#include <thread>
#include <chrono>
#include <future>
using namespace std;
template <typename TY>
class MockReceiver<TY>{
public:
int idx_packet = 42000;
int packet_per_frame = 512;
int num_bunches = 100;
int num_packets =50;
int receive_many(mmsghdr* msgs, const size_t n_msgs){
// Receive 'num_packets numner of packets'
for(int ii=0; ii<std::min(num_packets, n_msgs); ii++){
jfjoch_packet_t& refer = std::reinterpret_cast<jfjoch_packet_t&>(mmsghdr[ii].msg_hdr.msg_iov->iov_base);
refer.bunchid = idx_packet / packet_per_frame;
refer.packetnum = idx_packet % packet_per_frame;
idx_packet++;
}
return std::min(num_packets, n_msgs);
};
};
//
//
//
//
//TEST(BufferUdpReceiver, simple_recv)
//{
// auto n_packets = JF_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 (int i_frame=0; i_frame < n_frames; i_frame++){
// for (size_t i_packet=0; i_packet<n_packets; i_packet++) {
// jungfrau_packet 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[]>(JUNGFRAU_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)
//{
// auto n_packets = JF_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, source_id);
//
// auto handle = async(launch::async, [&](){
// for (int 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;
// }
//
// jungfrau_packet 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[]>(JUNGFRAU_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 = JF_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 (int 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;
// }
//
// jungfrau_packet 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[]>(JUNGFRAU_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)
//{
// auto n_packets = JF_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 (int 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;
// }
//
// jungfrau_packet 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[]>(JUNGFRAU_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/include/DataBuffer.hpp
-92
View File
@@ -1,92 +0,0 @@
#ifndef CIRCULAR_BUFFER_TEMPLATE_HPP
#define CIRCULAR_BUFFER_TEMPLATE_HPP
#include <algorithm>
#include <cstddef>
#include <cassert>
#include <stdexcept>
#include <iostream>
/**Linear data buffer
A simplified version of FIFO.
**/
template <typename T, size_t CAP>
class DataBuffer{
public:
DataBuffer() {};
~DataBuffer() {};
/**Diagnostics**/
size_t size() const { return ( _write-_read ); }
size_t capacity() const { return _capacity; }
bool is_full(){ return ( (_write - _read)<_capacity ); }
bool is_empty(){ return (_write ==_read); }
/**Operators**/
void zero(){ memset(m_cont, 0, sizeof(m_cont)); }
T& operator[](size_t index); // Array subscript operator
T& container(){ return (_cont; } // Direct container reference
/**Element access**/
const T& pop_front(); //Destructive read
const T& get_front(); //Non-destructive read
void push_back(T item); //Write new element to buffer
/**Guards**/
std::mutex g_mutex;
private:
T m_cont[CAP];
const size_t m_capacity = CAP;
size_t ptr_write = 0;
size_t ptr_read = 0;
};
/** Array subscript operator
Throws 'std::length_error' if out of range.
**/
template<typename T>
T& DataBuffer<T>::operator[](size_t idx){
if(idx > m_capacity){
std::string msg = "Buffer index '" + std::to_string(idx) + "' is out of range with capacity '" + std::to_sting(m_capacity) + "'" + std::endl;
throw std::out_of_range(msg);
}
return m_buffer[idx];
}
template<typename T>
T& DataBuffer<T>::container(){
return m_buffer;
}
/*********************************************************************/
/** Destructive read (i.e. progress the read pointer) **/
template<typename T>
const T& DataBuffer<T>::pop_front(){
std::lock_guard<std::mutex> g_guard;
ptr_read++;
return _buffer[ptr_read-1];
}
/**Push a new element to the ringbuffer (do not progress read pointer)**/
template<typename T>
const T& DataBuffer<T>::peek_front(){
return m_buffer[ptr_read];
}
/**Push a new element to the ringbuffer**/
template<typename T>
void DataBuffer<T>::push_back(T item){
std::lock_guard<std::mutex> g_guard;
if(ptr_write==m_capacity-1){
std::string msg = "Buffer with '" + std::to_sting(m_capacity) + "' capacity is full" + std::endl;
throw std::out_of_range(msg);
}
m_buffer[ptr_write] = item;
ptr_write++;
}
#endif // CIRCULAR_BUFFER_TEMPLATE_HPP
jfj-udp-recv/include/JFJochUdpReceiver.hpp
-33
View File
@@ -1,33 +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"
class JFJochUdpReceiver {
PacketUdpReceiver m_udp_receiver;
// Incoming packet buffers
jfjoch_packet_t m_packet_buffer[buffer_config::BUFFER_UDP_N_RECV_MSG];
iovec m_recv_buff_ptr[buffer_config::BUFFER_UDP_N_RECV_MSG];
mmsghdr m_msgs[buffer_config::BUFFER_UDP_N_RECV_MSG];
sockaddr_in m_sock_from[buffer_config::BUFFER_UDP_N_RECV_MSG];
bool packet_buffer_loaded_ = false;
int packet_buffer_n_packets_ = 0;
int packet_buffer_offset_ = 0;
inline void init_frame(ImageMetadata& frame_metadata, const int i_packet);
inline void copy_packet_to_buffers(ImageMetadata& metadata, char* frame_buffer, const int i_packet);
inline uint64_t m_process_packets(const int n_packets, ImageMetadata& metadata, char* frame_buffer);
public:
JFJochUdpReceiver(const uint16_t port, const int module_id);
virtual ~JFJochUdpReceiver();
uint64_t get_frame_from_udp(ImageMetadata& metadata, char* frame_buffer);
};
#endif //SF_DAQ_BUFFER_JOCHUDPRECEIVER_HPP
jfj-udp-recv/include/ImageStats.hpp → jfj-udp-recv/include/JfjFrameStats.hpp
View File
jfj-udp-recv/include/JfjFrameUdpReceiver.hpp
+31
View File
@@ -0,0 +1,31 @@
#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"
/** 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 udp_receiver_;
PacketBuffer<jfjoch_packet_t, buffer_config::BUFFER_UDP_N_RECV_MSG> m_buffer;
inline void init_frame(ImageMetadata& frame_metadata, jfjoch_packet_t& c_packet);
inline uint64_t process_packets(ImageMetadata& metadata, char* frame_buffer);
public:
JfjFrameUdpReceiver(const uint16_t port);
virtual ~JfjFrameUdpReceiver();
uint64_t get_frame_from_udp(ImageMetadata& metadata, char* frame_buffer);
};
#endif //SF_DAQ_BUFFER_JOCHUDPRECEIVER_HPP
jfj-udp-recv/include/PacketBuffer.hpp
+110
View File
@@ -0,0 +1,110 @@
#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**/
size_t size() const { return ( idx_write-idx_read ); }
size_t capacity() const { return m_capacity; }
bool is_full() const { return (idx_write >= m_capacity); }
bool is_empty() const { return (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**/
const 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_fom(TY& recv){
std::lock_guard<std::mutex> g_guard(m_mutex);
this->idx_write = recv.receive_many(this->msgs(), this->capacity());
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**/
size_t idx_write = 0;
size_t 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>
const 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/mockmain.cpp
+15
View File
@@ -0,0 +1,15 @@
#include "include/PacketBuffer.hpp"
struct DummyContainer{
uint64_t index;
uint64_t timestamp;
uint16_t data[32];
};
int main (int argc, char *argv[]) {
PacketBuffer<DummyContainer, 64> b;
}
jfj-udp-recv/src/JFJochUdpReceiver.cpp
-112
View File
@@ -1,112 +0,0 @@
#include <cstring>
#include <jungfraujoch.hpp>
#include "JFJochUdpReceiver.hpp"
using namespace std;
using namespace buffer_config;
JFJochUdpReceiver::JFJochUdpReceiver(const uint16_t port, const int module_id) : module_id_(module_id){
udp_receiver_.bind(port);
for (int i = 0; i < BUFFER_UDP_N_RECV_MSG; i++) {
m_recv_buff_ptr[i].iov_base = (void*) &(m_packet_buffer[i]);
m_recv_buff_ptr[i].iov_len = sizeof(jfjoch_packet_t);
msgs_[i].msg_hdr.msg_iov = &m_recv_buff_ptr[i];
msgs_[i].msg_hdr.msg_iovlen = 1;
msgs_[i].msg_hdr.msg_name = &m_sock_from[i];
msgs_[i].msg_hdr.msg_namelen = sizeof(sockaddr_in);
}
}
JFJochUdpReceiver::~JFJochUdpReceiver() {
m_udp_receiver.disconnect();
}
inline void JFJochUdpReceiver::init_frame(ImageMetadata& image_metadata, const int i_packet) {
image_metadata.pulse_id = m_packet_buffer[i_packet].bunchid;
image_metadata.frame_index = m_packet_buffer[i_packet].framenum;
image_metadata.daq_rec = m_packet_buffer[i_packet].debug;
image_metadata.is_good_image = 0;
}
inline void JFJochUdpReceiver::copy_packet_to_buffers(ImageMetadata& metadata, char* frame_buffer, const int idx_packet){
size_t buffer_offset = JUNGFRAU_DATA_BYTES_PER_PACKET * packet_buffer_[idx_packet].packetnum;
memcpy((void*) (frame_buffer + buffer_offset), m_packet_buffer[idx_packet].data, JUNGFRAU_DATA_BYTES_PER_PACKET);
metadata.n_recv_packets++;
}
/** Copy the contents of the packet buffer into a single assembled image
NOTE: In the jungfrau_packet, framenum is the trigger number
NOTE: Even partial frames are valid
**/
inline uint64_t JFJochUdpReceiver::m_process_packets(const int start_offset, ImageMetadata& metadata, char* frame_buffer){
for (int i_packet=start_offset; i_packet < packet_buffer_n_packets_; i_packet++) {
// First packet for this frame (sucks if this one is missed)
if (metadata.pulse_id == 0) {
init_frame(metadata, i_packet);
}
// Unexpected jump (if the last packet from the previous frame got lost)
if (metadata.frame_index != m_packet_buffer[i_packet].framenum) {
// Save queue status (lazy fifo queue)
m_packet_buffer_loaded_ = true;
m_packet_buffer_offset_ = i_packet;
// Even partial frames are valid?
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 == JFJ_N_PACKETS_PER_FRAME - 1) {
// Buffer is loaded only if this is not the last message.
if (i_packet+1 != packet_buffer_n_packets_) {
// Continue on next packet
m_packet_buffer_loaded = true;
m_packet_buffer_offset = i_packet + 1;
// If i_packet is the last packet the buffer is empty.
} else {
m_packet_buffer_loaded = true;
m_packet_buffer_offset = 0;
}
return metadata.pulse_id;
}
}
// We emptied the buffer.
m_packet_buffer_loaded = false;
m_packet_buffer_offset = 0;
return 0;
}
uint64_t JFJochUdpReceiver::get_frame_from_udp(ImageMetadata& 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 = m_process_packets(packet_buffer_offset_, metadata, frame_buffer);
if (pulse_id != 0) { return pulse_id; }
}
// Otherwise read a new one
while (true) {
packet_buffer_n_packets_ = udp_receiver_.receive_many(msgs_, BUFFER_UDP_N_RECV_MSG);
if (packet_buffer_n_packets_ > 0) {
auto pulse_id = m_process_packets(0, metadata, frame_buffer);
if (pulse_id != 0) { return pulse_id; }
}
}
}
jfj-udp-recv/src/ImageStats.cpp → jfj-udp-recv/src/JfjFrameStats.cpp
View File
jfj-udp-recv/src/JfjFrameUdpReceiver.cpp
+77
View File
@@ -0,0 +1,77 @@
#include <cstring>
#include <jungfrau.hpp>
#include "JfjFrameUdpReceiver.hpp"
using namespace std;
using namespace buffer_config;
JfjFrameUdpReceiver::JfjFrameUdpReceiver(const uint16_t port) {
udp_receiver_.bind(port);
}
JfjFrameUdpReceiver::~JfjFrameUdpReceiver() {
udp_receiver_.disconnect();
}
inline void JfjFrameUdpReceiver::init_frame(ImageMetadata& frame_metadata, const jfjoch_packet_t& c_packet) {
frame_metadata.pulse_id = c_packet.timestamp;
frame_metadata.frame_index = c_packet.framenum;
frame_metadata.daq_rec = (uint32_t) c_packet.debug;
frame_metadata.is_good_image = (int32_t) true;
}
inline uint64_t JfjFrameUdpReceiver::process_packets(ImageMetadata& 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;
frame_metadata.is_good_image = (int32_t) 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;
// Always copy metadata (otherwise problem when 0th packet gets lost)
this->init_frame(metadata, c_packet);
// Copy data to frame buffer
size_t offset = JUNGFRAU_DATA_BYTES_PER_PACKET * c_packet.packetnum;
memcpy( (void*) (frame_buffer + offset), c_packet.data, JUNGFRAU_DATA_BYTES_PER_PACKET);
metadata.n_recv_packets++;
// Last frame packet received. Frame finished.
if (c_packet.packetnum == JFJ_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(ImageMetadata& 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);
// 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; }
}
}
jfj-udp-recv/test/test_FrameUdpReceiver.cpp
+6 -15
View File
@@ -13,7 +13,6 @@ using namespace std;
TEST(BufferUdpReceiver, simple_recv)
{
auto n_packets = JF_N_PACKETS_PER_FRAME;
int source_id = 1234;
int n_frames = 5;
uint16_t udp_port = MOCK_UDP_PORT;
@@ -21,7 +20,7 @@ TEST(BufferUdpReceiver, simple_recv)
auto send_socket_fd = socket(AF_INET, SOCK_DGRAM, 0);
ASSERT_TRUE(send_socket_fd >= 0);
FrameUdpReceiver udp_receiver(udp_port, source_id);
JfjFrameUdpReceiver udp_receiver(udp_port);
auto handle = async(launch::async, [&](){
for (int i_frame=0; i_frame < n_frames; i_frame++){
@@ -57,7 +56,6 @@ TEST(BufferUdpReceiver, simple_recv)
ASSERT_EQ(metadata.daq_rec, i_frame + 10000);
// -1 because we skipped a packet.
ASSERT_EQ(metadata.n_recv_packets, n_packets);
ASSERT_EQ(metadata.module_id, source_id);
}
::close(send_socket_fd);
@@ -66,7 +64,6 @@ TEST(BufferUdpReceiver, simple_recv)
TEST(BufferUdpReceiver, missing_middle_packet)
{
auto n_packets = JF_N_PACKETS_PER_FRAME;
int source_id = 1234;
int n_frames = 3;
uint16_t udp_port = MOCK_UDP_PORT;
@@ -74,7 +71,7 @@ TEST(BufferUdpReceiver, missing_middle_packet)
auto send_socket_fd = socket(AF_INET, SOCK_DGRAM, 0);
ASSERT_TRUE(send_socket_fd >= 0);
FrameUdpReceiver udp_receiver(udp_port, source_id);
JfjFrameUdpReceiver udp_receiver(udp_port, source_id);
auto handle = async(launch::async, [&](){
for (int i_frame=0; i_frame < n_frames; i_frame++){
@@ -115,7 +112,6 @@ TEST(BufferUdpReceiver, missing_middle_packet)
ASSERT_EQ(metadata.daq_rec, i_frame + 10000);
// -1 because we skipped a packet.
ASSERT_EQ(metadata.n_recv_packets, n_packets - 1);
ASSERT_EQ(metadata.module_id, source_id);
}
::close(send_socket_fd);
@@ -124,7 +120,6 @@ TEST(BufferUdpReceiver, missing_middle_packet)
TEST(BufferUdpReceiver, missing_first_packet)
{
auto n_packets = JF_N_PACKETS_PER_FRAME;
int source_id = 1234;
int n_frames = 3;
uint16_t udp_port = MOCK_UDP_PORT;
@@ -132,7 +127,7 @@ TEST(BufferUdpReceiver, missing_first_packet)
auto send_socket_fd = socket(AF_INET, SOCK_DGRAM, 0);
ASSERT_TRUE(send_socket_fd >= 0);
FrameUdpReceiver udp_receiver(udp_port, source_id);
JfjFrameUdpReceiver udp_receiver(udp_port);
auto handle = async(launch::async, [&](){
for (int i_frame=0; i_frame < n_frames; i_frame++){
@@ -173,7 +168,6 @@ TEST(BufferUdpReceiver, missing_first_packet)
ASSERT_EQ(metadata.daq_rec, i_frame + 10000);
// -1 because we skipped a packet.
ASSERT_EQ(metadata.n_recv_packets, n_packets - 1);
ASSERT_EQ(metadata.module_id, source_id);
}
::close(send_socket_fd);
@@ -182,7 +176,6 @@ TEST(BufferUdpReceiver, missing_first_packet)
TEST(BufferUdpReceiver, missing_last_packet)
{
auto n_packets = JF_N_PACKETS_PER_FRAME;
int source_id = 1234;
int n_frames = 3;
uint16_t udp_port = MOCK_UDP_PORT;
@@ -190,7 +183,7 @@ TEST(BufferUdpReceiver, missing_last_packet)
auto send_socket_fd = socket(AF_INET, SOCK_DGRAM, 0);
ASSERT_TRUE(send_socket_fd >= 0);
FrameUdpReceiver udp_receiver(udp_port, source_id);
JfjFrameUdpReceiver udp_receiver(udp_port);
auto handle = async(launch::async, [&](){
for (int i_frame=0; i_frame < n_frames; i_frame++){
@@ -224,16 +217,14 @@ TEST(BufferUdpReceiver, missing_last_packet)
// 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());
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);
ASSERT_EQ(metadata.module_id, source_id);
}
::close(send_socket_fd);
}
}