sf-daq/sf_daq_buffer
0
0
Fork 0
mirror of https://github.com/paulscherrerinstitute/sf_daq_buffer.git synced 2026-06-06 11:58:41 +02:00
Code Issues Packages Projects Releases Wiki Activity

Finished refactoring sf-buffer

Browse Source
This commit is contained in:
babic_a
2020-05-20 12:22:20 +02:00
parent b45b7d17fa
commit 6d210ed702
18 changed files with 1468 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sf-buffer/src/BufferBinaryWriter.cpp
+124
View File
@@ -0,0 +1,124 @@
#include "BufferBinaryWriter.hpp"
#include <unistd.h>
#include <iostream>
#include "date.h"
#include <cerrno>
#include <chrono>
#include <cstring>
#include <BufferUtils.hpp>
#include <fcntl.h>
#include <WriterUtils.hpp>
using namespace std;
BufferBinaryWriter::BufferBinaryWriter(
const string& device_name,
const string& root_folder) :
device_name_(device_name),
root_folder_(root_folder),
latest_filename_(root_folder + "/" + device_name + "/LATEST"),
current_output_filename_(""),
output_file_fd_(-1)
{
}
BufferBinaryWriter::~BufferBinaryWriter()
{
close_current_file();
}
void BufferBinaryWriter::write(uint64_t pulse_id, const BufferBinaryFormat* buffer)
{
auto current_frame_file =
BufferUtils::get_filename(root_folder_, device_name_, pulse_id);
if (current_frame_file != current_output_filename_) {
open_file(current_frame_file);
}
size_t n_bytes_offset =
BufferUtils::get_file_frame_index(pulse_id) * sizeof(BufferBinaryFormat);
auto lseek_result = lseek(output_file_fd_, n_bytes_offset, SEEK_SET);
if (lseek_result < 0) {
stringstream err_msg;
using namespace date;
using namespace chrono;
err_msg << "[" << system_clock::now() << "]";
err_msg << "[BinaryWriter::write]";
err_msg << " Error while lseek on file ";
err_msg << current_output_filename_;
err_msg << " for n_bytes_offset ";
err_msg << n_bytes_offset << ": ";
err_msg << strerror(errno) << endl;
throw runtime_error(err_msg.str());
}
auto n_bytes = ::write(output_file_fd_, buffer, sizeof(BufferBinaryFormat));
if (n_bytes < sizeof(BufferBinaryFormat)) {
stringstream err_msg;
using namespace date;
using namespace chrono;
err_msg << "[" << system_clock::now() << "]";
err_msg << "[BinaryWriter::write]";
err_msg << " Error while writing to file ";
err_msg << current_output_filename_ << ": ";
err_msg << strerror(errno) << endl;
throw runtime_error(err_msg.str());
}
}
void BufferBinaryWriter::open_file(const std::string& filename)
{
close_current_file();
WriterUtils::create_destination_folder(filename);
output_file_fd_ = ::open(filename.c_str(), O_WRONLY | O_CREAT,
S_IRWXU | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH);
if (output_file_fd_ < 0) {
stringstream err_msg;
using namespace date;
using namespace chrono;
err_msg << "[" << system_clock::now() << "]";
err_msg << "[BinaryWriter::open_file]";
err_msg << " Cannot create file ";
err_msg << filename << ": ";
err_msg << strerror(errno) << endl;
throw runtime_error(err_msg.str());
}
current_output_filename_ = filename;
}
void BufferBinaryWriter::close_current_file()
{
if (output_file_fd_ != -1) {
if (close(output_file_fd_) < 0) {
stringstream err_msg;
using namespace date;
using namespace chrono;
err_msg << "[" << system_clock::now() << "]";
err_msg << "[BinaryWriter::close_current_file]";
err_msg << " Error while closing file ";
err_msg << current_output_filename_ << ": ";
err_msg << strerror(errno) << endl;
throw runtime_error(err_msg.str());
}
output_file_fd_ = -1;
BufferUtils::update_latest_file(
latest_filename_, current_output_filename_);
}
current_output_filename_ = "";
}
sf-buffer/src/BufferH5Writer.cpp
+126
View File
@@ -0,0 +1,126 @@
#include <BufferUtils.hpp>
#include "BufferH5Writer.hpp"
#include <chrono>
#include <WriterUtils.hpp>
#include <cstring>
extern "C"
{
#include "H5DOpublic.h"
}
using namespace std;
using namespace core_buffer;
BufferH5Writer::BufferH5Writer(
const string& root_folder,
const string& device_name) :
root_folder_(root_folder),
device_name_(device_name),
LATEST_filename_(root_folder + "/" + device_name + "/LATEST"),
CURRENT_filename_(root_folder + "/" + device_name + "/CURRENT"),
output_filename_(""),
current_pulse_id_(0),
current_file_index_(0)
{
}
void BufferH5Writer::create_file(const string& filename)
{
h5_file_ = H5::H5File(filename, H5F_ACC_TRUNC);
output_filename_ = filename;
H5::DataSpace data_dspace(3, data_disk_dims, data_disk_dims);
H5::DSetCreatPropList data_dset_prop;
hsize_t data_dset_chunking[3] = {1, MODULE_Y_SIZE, MODULE_X_SIZE};
data_dset_prop.setChunk(3, data_dset_chunking);
current_image_dataset_ = h5_file_.createDataSet(
BUFFER_H5_FRAME_DATASET,
H5::PredType::NATIVE_UINT16,
data_dspace,
data_dset_prop);
H5::DataSpace meta_dspace(2, meta_disk_dims, meta_disk_dims);
H5::DSetCreatPropList meta_dset_prop;
hsize_t meta_dset_chunking[2] = {1, ModuleFrame_N_FIELDS};
meta_dset_prop.setChunk(2, meta_dset_chunking);
current_metadata_dataset_ = h5_file_.createDataSet(
BUFFER_H5_METADATA_DATASET,
H5::PredType::NATIVE_UINT64,
meta_dspace,
meta_dset_prop);
}
BufferH5Writer::~BufferH5Writer()
{
close_file();
}
void BufferH5Writer::close_file() {
current_image_dataset_.close();
current_metadata_dataset_.close();
h5_file_.close();
output_filename_ = "";
current_pulse_id_ = 0;
current_file_index_ = 0;
}
void BufferH5Writer::set_pulse_id(const uint64_t pulse_id)
{
current_pulse_id_ = pulse_id;
current_file_index_ = BufferUtils::get_file_frame_index(pulse_id);
auto new_output_filename = BufferUtils::get_filename(
root_folder_, device_name_, pulse_id);
if (new_output_filename != output_filename_){
if (h5_file_.getId() != -1) {
auto latest_filename = output_filename_;
close_file();
BufferUtils::update_latest_file(LATEST_filename_, latest_filename);
}
WriterUtils::create_destination_folder(new_output_filename);
create_file(new_output_filename);
BufferUtils::update_latest_file(CURRENT_filename_, output_filename_);
}
}
void BufferH5Writer::write(const ModuleFrame* metadata, const char* data)
{
hsize_t meta_buff_dims[1] = {ModuleFrame_N_FIELDS};
H5::DataSpace meta_buffer_space (1, meta_buff_dims);
H5::DataSpace meta_disk_space(2, meta_disk_dims);
hsize_t meta_count[] = {1, ModuleFrame_N_FIELDS};
hsize_t meta_start[] = {current_file_index_, 0};
meta_disk_space.selectHyperslab(H5S_SELECT_SET, meta_count, meta_start);
current_metadata_dataset_.write(
(char*) metadata,
H5::PredType::NATIVE_UINT64,
meta_buffer_space,
meta_disk_space);
hsize_t data_buff_dims[2] = {MODULE_Y_SIZE, MODULE_X_SIZE};
H5::DataSpace data_buffer_space (2, data_buff_dims);
H5::DataSpace data_disk_space(3, data_disk_dims);
hsize_t data_count[] = {1, MODULE_Y_SIZE, MODULE_X_SIZE};
hsize_t data_start[] = {current_file_index_, 0, 0};
data_disk_space.selectHyperslab(H5S_SELECT_SET, data_count, data_start);
current_image_dataset_.write(
data,
H5::PredType::NATIVE_UINT16,
data_buffer_space,
data_disk_space);
}
sf-buffer/src/BufferUdpReceiver.cpp
+137
View File
@@ -0,0 +1,137 @@
#include <cstring>
#include <jungfrau.hpp>
#include "BufferUdpReceiver.hpp"
using namespace std;
using namespace core_buffer;
BufferUdpReceiver::BufferUdpReceiver(
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);
}
}
BufferUdpReceiver::~BufferUdpReceiver() {
udp_receiver_.disconnect();
}
inline void BufferUdpReceiver::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 BufferUdpReceiver::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_received_packets++;
}
inline uint64_t BufferUdpReceiver::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.
} 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 ==
JUNGFRAU_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 BufferUdpReceiver::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_received_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/src/UdpReceiver.cpp
+89
View File
@@ -0,0 +1,89 @@
#include <netinet/in.h>
#include <iostream>
#include "UdpReceiver.hpp"
#include "jungfrau.hpp"
#include <unistd.h>
#include <cstring>
#include "buffer_config.hpp"
using namespace std;
using namespace core_buffer;
UdpReceiver::UdpReceiver() :
socket_fd_(-1)
{
}
UdpReceiver::~UdpReceiver()
{
disconnect();
}
void UdpReceiver::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 UdpReceiver::receive_many(mmsghdr* msgs, const size_t n_msgs)
{
return recvmmsg(socket_fd_, msgs, n_msgs, 0, 0);
}
bool UdpReceiver::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 UdpReceiver::disconnect()
{
close(socket_fd_);
socket_fd_ = -1;
}
sf-buffer/src/WriterUtils.cpp
+49
View File
@@ -0,0 +1,49 @@
#include <iostream>
#include <unistd.h>
#include "WriterUtils.hpp"
#include "date.h"
using namespace std;
void WriterUtils::set_process_effective_id(int user_id)
{
// TODO: use setfsuid and setfsgid
if (setegid(user_id)) {
stringstream err_msg;
using namespace date;
using namespace chrono;
err_msg << "[" << system_clock::now() << "]";
err_msg << "[WriterUtils::set_process_effective_id]";
err_msg << " Cannot set group_id to " << user_id << endl;
throw runtime_error(err_msg.str());
}
if (seteuid(user_id)) {
stringstream err_msg;
using namespace date;
using namespace chrono;
err_msg << "[" << system_clock::now() << "]";
err_msg << "[WriterUtils::set_process_effective_id]";
err_msg << " Cannot set user_id to " << user_id << endl;
throw runtime_error(err_msg.str());
}
}
void WriterUtils::create_destination_folder(const string& output_file)
{
auto file_separator_index = output_file.rfind('/');
if (file_separator_index != string::npos) {
string output_folder(output_file.substr(0, file_separator_index));
string create_folder_command("mkdir -p " + output_folder);
system(create_folder_command.c_str());
}
}
sf-buffer/src/main.cpp
+150
View File
@@ -0,0 +1,150 @@
#include <iostream>
#include <stdexcept>
#include <BufferH5Writer.hpp>
#include "zmq.h"
#include "buffer_config.hpp"
#include "jungfrau.hpp"
#include "BufferUdpReceiver.hpp"
#include <chrono>
#include <sstream>
#include <sys/resource.h>
#include <syscall.h>
#include <zconf.h>
using namespace std;
using namespace core_buffer;
int main (int argc, char *argv[]) {
if (argc != 5) {
cout << endl;
cout << "Usage: sf_buffer [device_name] [udp_port] [root_folder]";
cout << "[source_id]";
cout << endl;
cout << "\tdevice_name: Name to write to disk.";
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 device_name = string(argv[1]);
int udp_port = atoi(argv[2]);
string root_folder = string(argv[3]);
int source_id = atoi(argv[4]);
stringstream ipc_stream;
ipc_stream << BUFFER_LIVE_IPC_URL << source_id;
const auto ipc_address = ipc_stream.str();
auto ctx = zmq_ctx_new();
auto 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_ms = 0;
if (zmq_setsockopt(socket, ZMQ_LINGER, &linger_ms, sizeof(linger_ms)) != 0)
throw runtime_error(zmq_strerror(errno));
if (zmq_bind(socket, ipc_address.c_str()) != 0)
throw runtime_error(zmq_strerror(errno));
uint64_t stats_counter(0);
uint64_t n_missed_packets = 0;
uint64_t n_missed_frames = 0;
uint64_t n_corrupted_frames = 0;
uint64_t last_pulse_id = 0;
BufferH5Writer writer(root_folder, device_name);
BufferUdpReceiver receiver(udp_port, source_id);
pid_t tid;
tid = syscall(SYS_gettid);
int ret = setpriority(PRIO_PROCESS, tid, 0);
if (ret == -1) throw runtime_error("cannot set nice");
ModuleFrame metadata;
auto frame_buffer = new char[MODULE_N_BYTES * JUNGFRAU_N_MODULES];
size_t write_total_us = 0;
size_t write_max_us = 0;
size_t send_total_us = 0;
size_t send_max_us = 0;
while (true) {
auto pulse_id = receiver.get_frame_from_udp(metadata, frame_buffer);
auto start_time = chrono::steady_clock::now();
writer.set_pulse_id(pulse_id);
writer.write(&metadata, frame_buffer);
auto write_end_time = chrono::steady_clock::now();
auto write_us_duration = chrono::duration_cast<chrono::microseconds>(
write_end_time-start_time).count();
start_time = chrono::steady_clock::now();
zmq_send(socket, &metadata, sizeof(ModuleFrame), ZMQ_SNDMORE);
zmq_send(socket, frame_buffer, MODULE_N_BYTES, 0);
auto send_end_time = chrono::steady_clock::now();
auto send_us_duration = chrono::duration_cast<chrono::microseconds>(
send_end_time-start_time).count();
// TODO: Make real statistics, please.
stats_counter++;
write_total_us += write_us_duration;
send_total_us += send_us_duration;
if (write_us_duration > write_max_us) {
write_max_us = write_us_duration;
}
if (send_us_duration > send_max_us) {
send_max_us = send_us_duration;
}
if (metadata.n_received_packets < JUNGFRAU_N_PACKETS_PER_FRAME) {
n_missed_packets +=
JUNGFRAU_N_PACKETS_PER_FRAME - metadata.n_received_packets;
n_corrupted_frames++;
}
if (last_pulse_id>0) {
n_missed_frames += (pulse_id - last_pulse_id) - 1;
}
last_pulse_id = pulse_id;
if (stats_counter == STATS_MODULO) {
cout << "sf_buffer:device_name " << device_name;
cout << " sf_buffer:pulse_id " << pulse_id;
cout << " sf_buffer:n_missed_frames " << n_missed_frames;
cout << " sf_buffer:n_missed_packets " << n_missed_packets;
cout << " sf_buffer:n_corrupted_frames " << n_corrupted_frames;
cout << " sf_buffer:write_total_us " << write_total_us/STATS_MODULO;
cout << " sf_buffer:write_max_us " << write_max_us;
cout << " sf_buffer:send_total_us " << send_total_us/STATS_MODULO;
cout << " sf_buffer:send_max_us " << send_max_us;
cout << endl;
stats_counter = 0;
n_missed_packets = 0;
n_corrupted_frames = 0;
n_missed_frames = 0;
write_total_us = 0;
write_max_us = 0;
send_total_us = 0;
send_max_us = 0;
}
}
delete[] frame_buffer;
}