mirror of
https://github.com/paulscherrerinstitute/sf_daq_buffer.git
synced 2026-05-02 04:02:24 +02:00
Major project refactoring WIP
This commit is contained in:
@@ -1,124 +0,0 @@
|
||||
#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_ = "";
|
||||
}
|
||||
@@ -1,126 +0,0 @@
|
||||
#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);
|
||||
}
|
||||
@@ -1,137 +0,0 @@
|
||||
#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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,89 +0,0 @@
|
||||
#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;
|
||||
}
|
||||
@@ -1,49 +0,0 @@
|
||||
#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());
|
||||
}
|
||||
}
|
||||
@@ -1,150 +0,0 @@
|
||||
#include <iostream>
|
||||
#include <stdexcept>
|
||||
#include <RingBuffer.hpp>
|
||||
#include <BufferH5Writer.hpp>
|
||||
#include "zmq.h"
|
||||
#include "buffer_config.hpp"
|
||||
#include "jungfrau.hpp"
|
||||
#include "BufferUdpReceiver.hpp"
|
||||
|
||||
#include <sys/resource.h>
|
||||
#include <syscall.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(strerror (errno));
|
||||
|
||||
const int linger_ms = 0;
|
||||
if (zmq_setsockopt(socket, ZMQ_LINGER, &linger_ms, sizeof(linger_ms)) != 0)
|
||||
throw runtime_error(strerror (errno));
|
||||
|
||||
if (zmq_bind(socket, ipc_address.c_str()) != 0)
|
||||
throw runtime_error(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;
|
||||
}
|
||||
@@ -1,138 +0,0 @@
|
||||
#include "ReplayH5Reader.hpp"
|
||||
|
||||
#include "BufferUtils.hpp"
|
||||
#include <iostream>
|
||||
#include <chrono>
|
||||
#include <cstring>
|
||||
#include "date.h"
|
||||
|
||||
using namespace std;
|
||||
using namespace core_buffer;
|
||||
|
||||
void ReplayH5Reader::prepare_buffer_for_pulse(const uint64_t pulse_id)
|
||||
{
|
||||
auto pulse_filename = BufferUtils::get_filename(
|
||||
device_, channel_name_, pulse_id);
|
||||
|
||||
if (pulse_filename != current_filename_) {
|
||||
close_file();
|
||||
|
||||
current_filename_ = pulse_filename;
|
||||
current_file_ = H5::H5File(current_filename_, H5F_ACC_RDONLY);
|
||||
|
||||
dset_metadata_ = current_file_.openDataSet(BUFFER_H5_METADATA_DATASET);
|
||||
dset_frame_ = current_file_.openDataSet(BUFFER_H5_FRAME_DATASET);
|
||||
|
||||
// We always read the metadata for the entire file.
|
||||
hsize_t b_metadata_dims[2] =
|
||||
{FILE_MOD, ModuleFrame_N_FIELDS};
|
||||
H5::DataSpace b_m_space (2, b_metadata_dims);
|
||||
hsize_t b_m_count[] =
|
||||
{FILE_MOD, ModuleFrame_N_FIELDS};
|
||||
hsize_t b_m_start[] = {0, 0};
|
||||
b_m_space.selectHyperslab(H5S_SELECT_SET, b_m_count, b_m_start);
|
||||
|
||||
hsize_t f_metadata_dims[2] = {FILE_MOD, ModuleFrame_N_FIELDS};
|
||||
H5::DataSpace f_m_space (2, f_metadata_dims);
|
||||
hsize_t f_m_count[] =
|
||||
{FILE_MOD, ModuleFrame_N_FIELDS};
|
||||
hsize_t f_m_start[] = {0, 0};
|
||||
f_m_space.selectHyperslab(H5S_SELECT_SET, f_m_count, f_m_start);
|
||||
|
||||
dset_metadata_.read(&(metadata_buffer_[0]), H5::PredType::NATIVE_UINT64,
|
||||
b_m_space, f_m_space);
|
||||
|
||||
buffer_start_pulse_id_ = 0;
|
||||
buffer_end_pulse_id_ = 0;
|
||||
}
|
||||
|
||||
// End pulse_id is not included in the buffer.
|
||||
if ((pulse_id >= buffer_start_pulse_id_) &&
|
||||
(pulse_id < buffer_end_pulse_id_)) {
|
||||
return;
|
||||
}
|
||||
|
||||
buffer_start_pulse_id_ = pulse_id - (pulse_id % REPLAY_READ_BUFFER_SIZE);
|
||||
buffer_end_pulse_id_ = buffer_start_pulse_id_ + REPLAY_READ_BUFFER_SIZE;
|
||||
|
||||
auto start_index_in_file = BufferUtils::get_file_frame_index(
|
||||
buffer_start_pulse_id_);
|
||||
|
||||
hsize_t b_image_dims[3] =
|
||||
{REPLAY_READ_BUFFER_SIZE, MODULE_Y_SIZE, MODULE_X_SIZE};
|
||||
H5::DataSpace b_f_space (3, b_image_dims);
|
||||
hsize_t b_i_count[] =
|
||||
{REPLAY_READ_BUFFER_SIZE, MODULE_Y_SIZE, MODULE_X_SIZE};
|
||||
hsize_t b_i_start[] = {0, 0, 0};
|
||||
b_f_space.selectHyperslab(H5S_SELECT_SET, b_i_count, b_i_start);
|
||||
|
||||
hsize_t f_frame_dims[3] = {FILE_MOD, MODULE_Y_SIZE, MODULE_X_SIZE};
|
||||
H5::DataSpace f_f_space (3, f_frame_dims);
|
||||
hsize_t f_f_count[] =
|
||||
{REPLAY_READ_BUFFER_SIZE, MODULE_Y_SIZE, MODULE_X_SIZE};
|
||||
hsize_t f_f_start[] = {start_index_in_file, 0, 0};
|
||||
f_f_space.selectHyperslab(H5S_SELECT_SET, f_f_count, f_f_start);
|
||||
|
||||
dset_frame_.read(&(frame_buffer_[0]), H5::PredType::NATIVE_UINT16,
|
||||
b_f_space, f_f_space);
|
||||
}
|
||||
|
||||
ReplayH5Reader::ReplayH5Reader(
|
||||
const string device,
|
||||
const string channel_name) :
|
||||
device_(device),
|
||||
channel_name_(channel_name)
|
||||
{
|
||||
}
|
||||
|
||||
ReplayH5Reader::~ReplayH5Reader()
|
||||
{
|
||||
close_file();
|
||||
}
|
||||
|
||||
void ReplayH5Reader::close_file()
|
||||
{
|
||||
if (current_file_.getId() != -1) {
|
||||
dset_metadata_.close();
|
||||
dset_frame_.close();
|
||||
current_file_.close();
|
||||
}
|
||||
}
|
||||
|
||||
bool ReplayH5Reader::get_frame(
|
||||
const uint64_t pulse_id, ModuleFrame* metadata, char* frame_buffer)
|
||||
{
|
||||
prepare_buffer_for_pulse(pulse_id);
|
||||
|
||||
auto metadata_buffer_index = BufferUtils::get_file_frame_index(pulse_id);
|
||||
memcpy(metadata,
|
||||
&(metadata_buffer_[metadata_buffer_index]),
|
||||
sizeof(ModuleFrame));
|
||||
|
||||
auto frame_buffer_index = pulse_id - buffer_start_pulse_id_;
|
||||
memcpy(frame_buffer,
|
||||
&(frame_buffer_[frame_buffer_index * MODULE_N_BYTES]),
|
||||
MODULE_N_BYTES);
|
||||
|
||||
if (metadata->pulse_id == 0) {
|
||||
// Signal that there is no frame at this pulse_id.
|
||||
metadata->pulse_id = pulse_id;
|
||||
return false;
|
||||
|
||||
}else if (metadata->pulse_id != pulse_id) {
|
||||
stringstream err_msg;
|
||||
|
||||
using namespace date;
|
||||
using namespace chrono;
|
||||
err_msg << "[" << system_clock::now() << "]";
|
||||
err_msg << "[ReplayH5Reader::get_frame]";
|
||||
err_msg << " Corrupted file " << current_filename_;
|
||||
err_msg << " index_in_file " << metadata_buffer_index;
|
||||
err_msg << " expected pulse_id " << pulse_id;
|
||||
err_msg << " but read " << metadata->pulse_id << endl;
|
||||
|
||||
throw runtime_error(err_msg.str());
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
@@ -1,140 +0,0 @@
|
||||
#include <iostream>
|
||||
#include <thread>
|
||||
#include "jungfrau.hpp"
|
||||
|
||||
#include "zmq.h"
|
||||
#include "buffer_config.hpp"
|
||||
|
||||
#include <cstring>
|
||||
#include <ReplayH5Reader.hpp>
|
||||
#include "date.h"
|
||||
#include "bitshuffle/bitshuffle.h"
|
||||
|
||||
using namespace std;
|
||||
using namespace core_buffer;
|
||||
|
||||
void sf_replay (
|
||||
void* socket,
|
||||
const string& device,
|
||||
const string& channel_name,
|
||||
const uint64_t start_pulse_id,
|
||||
const uint64_t stop_pulse_id)
|
||||
{
|
||||
StreamModuleFrame metadata_buffer;
|
||||
auto frame_buffer = make_unique<uint16_t[]>(MODULE_N_PIXELS);
|
||||
|
||||
ReplayH5Reader file_reader(device, channel_name);
|
||||
|
||||
//TODO: Add statstics.
|
||||
uint64_t stats_counter = 0;
|
||||
|
||||
uint64_t total_read_us = 0;
|
||||
uint64_t max_read_us = 0;
|
||||
uint64_t total_send_us = 0;
|
||||
uint64_t max_send_us = 0;
|
||||
|
||||
// "<= stop_pulse_id" because we include the stop_pulse_id in the file.
|
||||
for (
|
||||
uint64_t curr_pulse_id = start_pulse_id;
|
||||
curr_pulse_id <= stop_pulse_id;
|
||||
curr_pulse_id++) {
|
||||
|
||||
auto start_time = chrono::steady_clock::now();
|
||||
|
||||
metadata_buffer.is_frame_present = file_reader.get_frame(
|
||||
curr_pulse_id,
|
||||
&(metadata_buffer.metadata),
|
||||
(char*)(frame_buffer.get()));
|
||||
|
||||
metadata_buffer.data_n_bytes = MODULE_N_BYTES;
|
||||
|
||||
auto end_time = chrono::steady_clock::now();
|
||||
auto read_us_duration = chrono::duration_cast<chrono::microseconds>(
|
||||
end_time-start_time).count();
|
||||
|
||||
start_time = chrono::steady_clock::now();
|
||||
|
||||
zmq_send(socket,
|
||||
&metadata_buffer,
|
||||
sizeof(StreamModuleFrame),
|
||||
ZMQ_SNDMORE);
|
||||
zmq_send(socket,
|
||||
(char*)(frame_buffer.get()),
|
||||
metadata_buffer.data_n_bytes,
|
||||
0);
|
||||
|
||||
end_time = chrono::steady_clock::now();
|
||||
auto send_us_duration = chrono::duration_cast<chrono::microseconds>(
|
||||
end_time-start_time).count();
|
||||
|
||||
// TODO: Make proper stastistics.
|
||||
stats_counter++;
|
||||
total_read_us += read_us_duration;
|
||||
max_read_us = max(max_read_us, (uint64_t)read_us_duration);
|
||||
|
||||
total_send_us += send_us_duration;
|
||||
max_send_us = max(max_send_us, (uint64_t)send_us_duration);
|
||||
|
||||
if (stats_counter == STATS_MODULO) {
|
||||
cout << "sf_replay:avg_read_us " << total_read_us/STATS_MODULO;
|
||||
cout << " sf_replay:max_read_us " << max_read_us;
|
||||
cout << " sf_replay:avg_send_us " << total_send_us/STATS_MODULO;
|
||||
cout << " sf_replay:max_send_us " << max_send_us;
|
||||
|
||||
cout << endl;
|
||||
|
||||
stats_counter = 0;
|
||||
total_read_us = 0;
|
||||
max_read_us = 0;
|
||||
total_send_us = 0;
|
||||
max_send_us = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
int main (int argc, char *argv[]) {
|
||||
|
||||
if (argc != 6) {
|
||||
cout << endl;
|
||||
cout << "Usage: sf_replay [device]";
|
||||
cout << " [channel_name] [source_id] [start_pulse_id] [stop_pulse_id]";
|
||||
cout << endl;
|
||||
cout << "\tdevice: Name of detector." << endl;
|
||||
cout << "\tchannel_name: M00-M31 for JF16M." << endl;
|
||||
cout << "\tsource_id: Module index" << endl;
|
||||
cout << "\tstart_pulse_id: Start pulse_id of retrieval." << endl;
|
||||
cout << "\tstop_pulse_id: Stop pulse_id of retrieval." << endl;
|
||||
cout << endl;
|
||||
|
||||
exit(-1);
|
||||
}
|
||||
|
||||
const string device = string(argv[1]);
|
||||
const string channel_name = string(argv[2]);
|
||||
const auto source_id = (uint16_t) atoi(argv[3]);
|
||||
const auto start_pulse_id = (uint64_t) atoll(argv[4]);
|
||||
const auto stop_pulse_id = (uint64_t) atoll(argv[5]);
|
||||
|
||||
stringstream ipc_stream;
|
||||
ipc_stream << REPLAY_STREAM_IPC_URL << (int)source_id;
|
||||
const auto ipc_address = ipc_stream.str();
|
||||
|
||||
auto ctx = zmq_ctx_new();
|
||||
auto socket = zmq_socket(ctx, ZMQ_PUSH);
|
||||
|
||||
const int sndhwm = REPLAY_SNDHWM;
|
||||
if (zmq_setsockopt(socket, ZMQ_SNDHWM, &sndhwm, sizeof(sndhwm)) != 0)
|
||||
throw runtime_error(strerror (errno));
|
||||
|
||||
const int linger_ms = -1;
|
||||
if (zmq_setsockopt(socket, ZMQ_LINGER, &linger_ms, sizeof(linger_ms)) != 0)
|
||||
throw runtime_error(strerror (errno));
|
||||
|
||||
if (zmq_bind(socket, ipc_address.c_str()) != 0)
|
||||
throw runtime_error(strerror (errno));
|
||||
|
||||
sf_replay(socket, device, channel_name, start_pulse_id, stop_pulse_id);
|
||||
|
||||
zmq_close(socket);
|
||||
zmq_ctx_destroy(ctx);
|
||||
}
|
||||
@@ -1,223 +0,0 @@
|
||||
#include "LiveRecvModule.hpp"
|
||||
#include "date.h"
|
||||
#include <iostream>
|
||||
#include <cstring>
|
||||
#include "zmq.h"
|
||||
#include "buffer_config.hpp"
|
||||
|
||||
using namespace std;
|
||||
using namespace core_buffer;
|
||||
|
||||
LiveRecvModule::LiveRecvModule(
|
||||
FastQueue<ModuleFrameBuffer>& queue_,
|
||||
const size_t n_modules,
|
||||
void* ctx_,
|
||||
const string& ipc_prefix) :
|
||||
queue_(queue_),
|
||||
n_modules_(n_modules),
|
||||
ctx_(ctx_),
|
||||
ipc_prefix_(ipc_prefix),
|
||||
is_receiving_(true)
|
||||
{
|
||||
receiving_thread_ = thread(&LiveRecvModule::receive_thread, this);
|
||||
}
|
||||
|
||||
LiveRecvModule::~LiveRecvModule()
|
||||
{
|
||||
stop();
|
||||
}
|
||||
|
||||
void LiveRecvModule::stop()
|
||||
{
|
||||
is_receiving_ = false;
|
||||
receiving_thread_.join();
|
||||
}
|
||||
|
||||
void* LiveRecvModule::connect_socket(size_t module_id)
|
||||
{
|
||||
void* sock = zmq_socket(ctx_, ZMQ_SUB);
|
||||
if (sock == nullptr) {
|
||||
throw runtime_error(zmq_strerror(errno));
|
||||
}
|
||||
|
||||
int rcvhwm = STREAM_RCVHWM;
|
||||
if (zmq_setsockopt(sock, ZMQ_RCVHWM, &rcvhwm, sizeof(rcvhwm)) != 0) {
|
||||
throw runtime_error(zmq_strerror(errno));
|
||||
}
|
||||
int linger = 0;
|
||||
if (zmq_setsockopt(sock, ZMQ_LINGER, &linger, sizeof(linger)) != 0) {
|
||||
throw runtime_error(zmq_strerror(errno));
|
||||
}
|
||||
|
||||
stringstream ipc_addr;
|
||||
ipc_addr << ipc_prefix_ << module_id;
|
||||
const auto ipc = ipc_addr.str();
|
||||
|
||||
if (zmq_connect(sock, ipc.c_str()) != 0) {
|
||||
throw runtime_error(zmq_strerror(errno));
|
||||
}
|
||||
|
||||
if (zmq_setsockopt(sock, ZMQ_SUBSCRIBE, "", 0) != 0) {
|
||||
throw runtime_error(zmq_strerror(errno));
|
||||
}
|
||||
|
||||
return sock;
|
||||
}
|
||||
|
||||
void LiveRecvModule::recv_single_module(
|
||||
void* socket, ModuleFrame* metadata, char* data)
|
||||
{
|
||||
auto n_bytes_metadata = zmq_recv(
|
||||
socket,
|
||||
metadata,
|
||||
sizeof(ModuleFrame),
|
||||
0);
|
||||
|
||||
if (n_bytes_metadata == -1) {
|
||||
throw runtime_error(zmq_strerror(errno));
|
||||
}else if (n_bytes_metadata != sizeof(ModuleFrame)) {
|
||||
throw runtime_error("Stream header of wrong size.");
|
||||
}
|
||||
|
||||
if (metadata->pulse_id == 0) {
|
||||
throw runtime_error("Received invalid pulse_id=0.");
|
||||
}
|
||||
|
||||
auto n_bytes_image = zmq_recv(
|
||||
socket,
|
||||
data,
|
||||
MODULE_N_BYTES,
|
||||
0);
|
||||
|
||||
if (n_bytes_image == -1) {
|
||||
throw runtime_error(zmq_strerror(errno));
|
||||
} else if (n_bytes_image != MODULE_N_BYTES) {
|
||||
throw runtime_error("Stream data of wrong size.");
|
||||
}
|
||||
}
|
||||
|
||||
uint64_t LiveRecvModule::align_modules(
|
||||
const vector<void*>& sockets, ModuleFrameBuffer *metadata, char *data)
|
||||
{
|
||||
uint64_t max_pulse_id = 0;
|
||||
|
||||
// First pass - determine current max_pulse_id.
|
||||
for (size_t i_module = 0; i_module < n_modules_; i_module++) {
|
||||
auto& module_metadata = metadata->module[i_module];
|
||||
max_pulse_id = max(max_pulse_id, module_metadata.pulse_id);
|
||||
}
|
||||
|
||||
// Second pass - align all receivers to max_pulse_id.
|
||||
for (size_t i_module = 0; i_module < n_modules_; i_module++) {
|
||||
auto& module_metadata = metadata->module[i_module];
|
||||
|
||||
size_t diff_to_max = max_pulse_id - module_metadata.pulse_id;
|
||||
for (size_t i = 0; i < diff_to_max; i++) {
|
||||
recv_single_module(
|
||||
sockets[i_module],
|
||||
&module_metadata,
|
||||
data + (MODULE_N_BYTES * i_module));
|
||||
}
|
||||
|
||||
if (module_metadata.pulse_id != max_pulse_id) {
|
||||
throw runtime_error("Cannot align pulse_ids.");
|
||||
}
|
||||
}
|
||||
|
||||
return max_pulse_id;
|
||||
}
|
||||
|
||||
void LiveRecvModule::receive_thread()
|
||||
{
|
||||
try {
|
||||
|
||||
vector<void*> sockets(n_modules_);
|
||||
|
||||
for (size_t i = 0; i < n_modules_; i++) {
|
||||
sockets[i] = connect_socket(i);
|
||||
}
|
||||
|
||||
auto slot_id = queue_.reserve();
|
||||
if (slot_id == -1) throw runtime_error("This cannot really happen");
|
||||
|
||||
auto metadata = queue_.get_metadata_buffer(slot_id);
|
||||
auto data = queue_.get_data_buffer(slot_id);
|
||||
|
||||
// First buffer load for alignment.
|
||||
for (size_t i_module = 0; i_module < n_modules_; i_module++) {
|
||||
auto& module_metadata = metadata->module[i_module];
|
||||
|
||||
recv_single_module(
|
||||
sockets[i_module],
|
||||
&module_metadata,
|
||||
data + (MODULE_N_BYTES * i_module));
|
||||
}
|
||||
|
||||
auto current_pulse_id = align_modules(sockets, metadata, data);
|
||||
|
||||
queue_.commit();
|
||||
current_pulse_id++;
|
||||
|
||||
while(is_receiving_.load(memory_order_relaxed)) {
|
||||
auto slot_id = queue_.reserve();
|
||||
|
||||
if (slot_id == -1){
|
||||
this_thread::sleep_for(chrono::milliseconds(5));
|
||||
continue;
|
||||
}
|
||||
|
||||
metadata = queue_.get_metadata_buffer(slot_id);
|
||||
data = queue_.get_data_buffer(slot_id);
|
||||
|
||||
bool sync_needed = false;
|
||||
for (size_t i_module = 0; i_module < n_modules_; i_module++) {
|
||||
auto& module_metadata = metadata->module[i_module];
|
||||
|
||||
recv_single_module(
|
||||
sockets[i_module],
|
||||
&module_metadata,
|
||||
data + (MODULE_N_BYTES * i_module));
|
||||
|
||||
if (module_metadata.pulse_id != current_pulse_id) {
|
||||
sync_needed = true;
|
||||
}
|
||||
}
|
||||
|
||||
if (sync_needed) {
|
||||
auto start_time = chrono::steady_clock::now();
|
||||
|
||||
auto new_pulse_id = align_modules(sockets, metadata, data);
|
||||
auto lost_pulses = new_pulse_id - current_pulse_id;
|
||||
current_pulse_id = new_pulse_id;
|
||||
|
||||
auto end_time = chrono::steady_clock::now();
|
||||
auto us_duration = chrono::duration_cast<chrono::microseconds>(
|
||||
end_time-start_time).count();
|
||||
|
||||
cout << "sf_stream:sync_lost_pulses " << lost_pulses;
|
||||
cout << " sf_stream::sync_us " << us_duration;
|
||||
cout << endl;
|
||||
}
|
||||
|
||||
queue_.commit();
|
||||
current_pulse_id++;
|
||||
}
|
||||
|
||||
for (size_t i = 0; i < n_modules_; i++) {
|
||||
zmq_close(sockets[i]);
|
||||
}
|
||||
|
||||
} catch (const std::exception& e) {
|
||||
is_receiving_ = false;
|
||||
|
||||
using namespace date;
|
||||
using namespace chrono;
|
||||
|
||||
cout << "[" << system_clock::now() << "]";
|
||||
cout << "[LiveRecvModule::receive_thread]";
|
||||
cout << " Stopped because of exception: " << endl;
|
||||
cout << e.what() << endl;
|
||||
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -1,221 +0,0 @@
|
||||
#include <iostream>
|
||||
#include <stdexcept>
|
||||
#include "buffer_config.hpp"
|
||||
|
||||
#include <string>
|
||||
#include <jungfrau.hpp>
|
||||
#include <thread>
|
||||
#include <chrono>
|
||||
#include "WriterH5Writer.hpp"
|
||||
#include <FastQueue.hpp>
|
||||
#include <cstring>
|
||||
#include <zmq.h>
|
||||
#include <LiveRecvModule.hpp>
|
||||
#include "date.h"
|
||||
#include <jsoncpp/json/json.h>
|
||||
|
||||
using namespace std;
|
||||
using namespace core_buffer;
|
||||
|
||||
int main (int argc, char *argv[])
|
||||
{
|
||||
if (argc != 5) {
|
||||
cout << endl;
|
||||
cout << "Usage: sf_stream ";
|
||||
cout << " [streamvis_address] [reduction_factor_streamvis]";
|
||||
cout << " [live_analysis_address] [reduction_factor_live_analysis]";
|
||||
cout << endl;
|
||||
cout << "\tstreamvis_address: address to streamvis, example tcp://129.129.241.42:9007" << endl;
|
||||
cout << "\treduction_factor_streamvis: 1 out of N (example 10) images to send to streamvis. For remaining send metadata." << endl;
|
||||
cout << "\tlive_analysis_address: address to live_analysis, example tcp://129.129.241.42:9107" << endl;
|
||||
cout << "\treduction_factor_live_analysis: 1 out of N (example 10) images to send to live analysis. For remaining send metadata. N<=1 - send every image" << endl;
|
||||
cout << endl;
|
||||
|
||||
exit(-1);
|
||||
}
|
||||
|
||||
string streamvis_address = string(argv[1]);
|
||||
int reduction_factor_streamvis = (int) atoll(argv[2]);
|
||||
string live_analysis_address = string(argv[3]);
|
||||
int reduction_factor_live_analysis = (uint64_t) atoll(argv[4]);
|
||||
|
||||
size_t n_modules = 32;
|
||||
FastQueue<ModuleFrameBuffer> queue(
|
||||
n_modules * MODULE_N_BYTES,
|
||||
STREAM_FASTQUEUE_SLOTS);
|
||||
|
||||
auto ctx = zmq_ctx_new();
|
||||
zmq_ctx_set (ctx, ZMQ_IO_THREADS, STREAM_ZMQ_IO_THREADS);
|
||||
|
||||
LiveRecvModule recv_module(queue, n_modules, ctx, BUFFER_LIVE_IPC_URL);
|
||||
|
||||
// 0mq sockets to streamvis and live analysis
|
||||
void *socket_streamvis = zmq_socket(ctx, ZMQ_PUB);
|
||||
if (zmq_bind(socket_streamvis, streamvis_address.c_str()) != 0) {
|
||||
throw runtime_error(strerror(errno));
|
||||
}
|
||||
void *socket_live = zmq_socket(ctx, ZMQ_PUB);
|
||||
if (zmq_bind(socket_live, live_analysis_address.c_str()) != 0) {
|
||||
throw runtime_error(strerror(errno));
|
||||
}
|
||||
|
||||
uint16_t data_empty [] = { 0, 0, 0, 0};
|
||||
Json::Value header;
|
||||
Json::StreamWriterBuilder builder;
|
||||
|
||||
// TODO: Remove stats trash.
|
||||
int stats_counter = 0;
|
||||
|
||||
size_t read_total_us = 0;
|
||||
size_t read_max_us = 0;
|
||||
|
||||
while (true) {
|
||||
|
||||
auto start_time = chrono::steady_clock::now();
|
||||
|
||||
auto slot_id = queue.read();
|
||||
|
||||
if(slot_id == -1) {
|
||||
this_thread::sleep_for(chrono::milliseconds(
|
||||
core_buffer::RB_READ_RETRY_INTERVAL_MS));
|
||||
continue;
|
||||
}
|
||||
|
||||
auto metadata = queue.get_metadata_buffer(slot_id);
|
||||
auto data = queue.get_data_buffer(slot_id);
|
||||
|
||||
auto read_end_time = chrono::steady_clock::now();
|
||||
auto read_us_duration = chrono::duration_cast<chrono::microseconds>(
|
||||
read_end_time-start_time).count();
|
||||
|
||||
uint64_t pulse_id = 0;
|
||||
uint64_t frame_index = 0;
|
||||
uint64_t daq_rec = 0;
|
||||
bool is_good_frame = true;
|
||||
|
||||
for (size_t i_module = 0; i_module < n_modules; i_module++) {
|
||||
// TODO: Place this tests in the appropriate spot.
|
||||
auto& module_metadata = metadata->module[i_module];
|
||||
if (i_module == 0) {
|
||||
pulse_id = module_metadata.pulse_id;
|
||||
frame_index = module_metadata.frame_index;
|
||||
daq_rec = module_metadata.daq_rec;
|
||||
|
||||
if ( module_metadata.n_received_packets != 128 ) is_good_frame = false;
|
||||
} else {
|
||||
if (module_metadata.pulse_id != pulse_id) is_good_frame = false;
|
||||
|
||||
if (module_metadata.frame_index != frame_index) is_good_frame = false;
|
||||
|
||||
if (module_metadata.daq_rec != daq_rec) is_good_frame = false;
|
||||
|
||||
if (module_metadata.n_received_packets != 128 ) is_good_frame = false;
|
||||
}
|
||||
}
|
||||
|
||||
//Here we need to send to streamvis and live analysis metadata(probably need to operate still on them) and data(not every frame)
|
||||
|
||||
header["frame"] = (Json::Value::UInt64)frame_index;
|
||||
header["is_good_frame"] = is_good_frame;
|
||||
header["daq_rec"] = (Json::Value::UInt64)daq_rec;
|
||||
header["pulse_id"] = (Json::Value::UInt64)pulse_id;
|
||||
|
||||
//this needs to be re-read from external source
|
||||
header["pedestal_file"] = "/sf/bernina/data/p17534/res/JF_pedestals/pedestal_20200423_1018.JF07T32V01.res.h5";
|
||||
header["gain_file"] = "/sf/bernina/config/jungfrau/gainMaps/JF07T32V01/gains.h5";
|
||||
|
||||
header["number_frames_expected"] = 10000;
|
||||
header["run_name"] = to_string(uint64_t(pulse_id/10000)*10000);
|
||||
|
||||
// detector name should come as parameter to sf_stream
|
||||
header["detector_name"] = "JF07T32V01";
|
||||
|
||||
header["htype"] = "array-1.0";
|
||||
header["type"] = "uint16";
|
||||
|
||||
int send_streamvis = 0;
|
||||
if ( reduction_factor_streamvis > 1 ) {
|
||||
send_streamvis = rand() % reduction_factor_streamvis;
|
||||
}
|
||||
if ( send_streamvis == 0 ) {
|
||||
header["shape"][0] = 16384;
|
||||
header["shape"][1] = 1024;
|
||||
} else{
|
||||
header["shape"][0] = 2;
|
||||
header["shape"][1] = 2;
|
||||
}
|
||||
|
||||
string text_header = Json::writeString(builder, header);
|
||||
|
||||
zmq_send(socket_streamvis,
|
||||
text_header.c_str(),
|
||||
text_header.size(),
|
||||
ZMQ_SNDMORE);
|
||||
|
||||
if ( send_streamvis == 0 ) {
|
||||
zmq_send(socket_streamvis,
|
||||
(char*)data,
|
||||
core_buffer::MODULE_N_BYTES*n_modules,
|
||||
0);
|
||||
} else {
|
||||
zmq_send(socket_streamvis,
|
||||
(char*)data_empty,
|
||||
8,
|
||||
0);
|
||||
}
|
||||
|
||||
//same for live analysis
|
||||
int send_live_analysis = 0;
|
||||
if ( reduction_factor_live_analysis > 1 ) {
|
||||
send_live_analysis = rand() % reduction_factor_live_analysis;
|
||||
}
|
||||
if ( send_live_analysis == 0 ) {
|
||||
header["shape"][0] = 16384;
|
||||
header["shape"][1] = 1024;
|
||||
} else{
|
||||
header["shape"][0] = 2;
|
||||
header["shape"][1] = 2;
|
||||
}
|
||||
|
||||
text_header = Json::writeString(builder, header);
|
||||
|
||||
zmq_send(socket_live,
|
||||
text_header.c_str(),
|
||||
text_header.size(),
|
||||
ZMQ_SNDMORE);
|
||||
|
||||
if ( send_live_analysis == 0 ) {
|
||||
zmq_send(socket_live,
|
||||
(char*)data,
|
||||
core_buffer::MODULE_N_BYTES*n_modules,
|
||||
0);
|
||||
} else {
|
||||
zmq_send(socket_live,
|
||||
(char*)data_empty,
|
||||
8,
|
||||
0);
|
||||
}
|
||||
|
||||
queue.release();
|
||||
|
||||
// TODO: Some poor statistics.
|
||||
stats_counter++;
|
||||
read_total_us += read_us_duration;
|
||||
|
||||
if (read_us_duration > read_max_us) {
|
||||
read_max_us = read_us_duration;
|
||||
}
|
||||
|
||||
if (stats_counter == STATS_MODULO) {
|
||||
cout << "sf_stream:read_us " << read_total_us / STATS_MODULO;
|
||||
cout << " sf_stream:read_max_us " << read_max_us;
|
||||
cout << endl;
|
||||
|
||||
stats_counter = 0;
|
||||
read_total_us = 0;
|
||||
read_max_us = 0;
|
||||
}
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
@@ -1,68 +0,0 @@
|
||||
#include "BufferedFastQueue.hpp"
|
||||
#include <thread>
|
||||
|
||||
using namespace std;
|
||||
using namespace core_buffer;
|
||||
|
||||
BufferedFastQueue::BufferedFastQueue(
|
||||
FastQueue<ImageMetadataBuffer>& queue,
|
||||
const size_t buffer_n_pulses,
|
||||
const size_t n_modules) :
|
||||
buffer_n_pulses_(buffer_n_pulses),
|
||||
queue_(queue),
|
||||
n_modules_(n_modules)
|
||||
{
|
||||
while ((current_slot_id_ = queue_.reserve()) == -1){
|
||||
this_thread::sleep_for(
|
||||
chrono::milliseconds(RB_READ_RETRY_INTERVAL_MS));
|
||||
}
|
||||
|
||||
queue_meta_buffer_ = queue_.get_metadata_buffer(current_slot_id_);
|
||||
queue_meta_buffer_->n_pulses_in_buffer = 0;
|
||||
queue_data_buffer_ = queue_.get_data_buffer(current_slot_id_);
|
||||
}
|
||||
|
||||
ImageMetadata* BufferedFastQueue::get_metadata_buffer()
|
||||
{
|
||||
return &image_metadata_;
|
||||
}
|
||||
|
||||
char* BufferedFastQueue::get_data_buffer()
|
||||
{
|
||||
auto index = queue_meta_buffer_->n_pulses_in_buffer;
|
||||
auto image_size = MODULE_N_BYTES * n_modules_;
|
||||
|
||||
return queue_data_buffer_ + (index * image_size);
|
||||
}
|
||||
|
||||
void BufferedFastQueue::commit()
|
||||
{
|
||||
auto index = queue_meta_buffer_->n_pulses_in_buffer;
|
||||
|
||||
queue_meta_buffer_->pulse_id[index] = image_metadata_.pulse_id;
|
||||
queue_meta_buffer_->frame_index[index] = image_metadata_.frame_index;
|
||||
queue_meta_buffer_->daq_rec[index] = image_metadata_.daq_rec;
|
||||
queue_meta_buffer_->is_good_frame[index] = image_metadata_.is_good_frame;
|
||||
queue_meta_buffer_->data_n_bytes[index] = image_metadata_.data_n_bytes;
|
||||
|
||||
queue_meta_buffer_->n_pulses_in_buffer++;
|
||||
|
||||
if (queue_meta_buffer_->n_pulses_in_buffer == buffer_n_pulses_) {
|
||||
queue_.commit();
|
||||
|
||||
while ((current_slot_id_ = queue_.reserve()) == -1){
|
||||
this_thread::sleep_for(
|
||||
chrono::milliseconds(RB_READ_RETRY_INTERVAL_MS));
|
||||
}
|
||||
|
||||
queue_meta_buffer_ = queue_.get_metadata_buffer(current_slot_id_);
|
||||
queue_meta_buffer_->n_pulses_in_buffer = 0;
|
||||
queue_data_buffer_ = queue_.get_data_buffer(current_slot_id_);
|
||||
}
|
||||
}
|
||||
|
||||
void BufferedFastQueue::finalize() {
|
||||
if (queue_meta_buffer_->n_pulses_in_buffer > 0) {
|
||||
queue_.commit();
|
||||
}
|
||||
}
|
||||
@@ -1,109 +0,0 @@
|
||||
#include <stdexcept>
|
||||
#include <WriterH5Writer.hpp>
|
||||
#include <jungfrau.hpp>
|
||||
#include "FastQueue.hpp"
|
||||
|
||||
using namespace std;
|
||||
|
||||
template <class T>
|
||||
FastQueue<T>::FastQueue(
|
||||
const size_t slot_data_n_bytes,
|
||||
const uint16_t n_slots) :
|
||||
slot_n_bytes_(slot_data_n_bytes + sizeof(T)),
|
||||
n_slots_(n_slots)
|
||||
{
|
||||
buffer_ = new char[slot_n_bytes_ * n_slots_];
|
||||
buffer_status_ = new atomic_int[n_slots];
|
||||
|
||||
// TODO: Are atomic variables initialized?
|
||||
for (size_t i=0; i < n_slots_; i++) {
|
||||
buffer_status_[i] = 0;
|
||||
}
|
||||
|
||||
write_slot_id_ = 0;
|
||||
read_slot_id_ = 0;
|
||||
}
|
||||
|
||||
template <class T>
|
||||
FastQueue<T>::~FastQueue()
|
||||
{
|
||||
delete[] buffer_;
|
||||
delete[] buffer_status_;
|
||||
}
|
||||
|
||||
template<class T>
|
||||
T* FastQueue<T>::get_metadata_buffer(const int slot_id)
|
||||
{
|
||||
return (T*)(buffer_ + (slot_id * slot_n_bytes_));
|
||||
}
|
||||
|
||||
template<class T>
|
||||
char* FastQueue<T>::get_data_buffer(const int slot_id)
|
||||
{
|
||||
return (char*)(buffer_ + (slot_id * slot_n_bytes_) + sizeof(T));
|
||||
}
|
||||
|
||||
template<class T>
|
||||
int FastQueue<T>::reserve()
|
||||
{
|
||||
int expected = SLOT_STATUS::EMPTY;
|
||||
// If (buffer_status==SLOT_EMPTY) buffer_status=SLOT_RESERVED.
|
||||
bool slot_reserved =
|
||||
buffer_status_[write_slot_id_].compare_exchange_strong(
|
||||
expected, SLOT_STATUS::RESERVED);
|
||||
|
||||
if (!slot_reserved) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
return write_slot_id_;
|
||||
}
|
||||
|
||||
template<class T>
|
||||
void FastQueue<T>::commit()
|
||||
{
|
||||
int expected = SLOT_STATUS::RESERVED;
|
||||
// If (buffer_status==SLOT_RESERVED) buffer_status=SLOT_READY.
|
||||
bool slot_ready =
|
||||
buffer_status_[write_slot_id_].compare_exchange_strong(
|
||||
expected, SLOT_STATUS::READY);
|
||||
|
||||
if (!slot_ready) {
|
||||
throw runtime_error("Slot should be reserved first.");
|
||||
}
|
||||
|
||||
write_slot_id_++;
|
||||
write_slot_id_ %= n_slots_;
|
||||
}
|
||||
|
||||
template<class T>
|
||||
int FastQueue<T>::read()
|
||||
{
|
||||
if (buffer_status_[read_slot_id_] != SLOT_STATUS::READY) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
return read_slot_id_;
|
||||
}
|
||||
|
||||
template<class T>
|
||||
void FastQueue<T>::release()
|
||||
{
|
||||
int expected = SLOT_STATUS::READY;
|
||||
// If (buffer_status==SLOT_RESERVED) buffer_status=SLOT_READY.
|
||||
bool slot_empty =
|
||||
buffer_status_[read_slot_id_].compare_exchange_strong(
|
||||
expected, SLOT_STATUS::EMPTY);
|
||||
|
||||
if (!slot_empty) {
|
||||
throw runtime_error("Slot should be ready first.");
|
||||
}
|
||||
|
||||
read_slot_id_++;
|
||||
read_slot_id_ %= n_slots_;
|
||||
}
|
||||
|
||||
template class FastQueue<ImageMetadata>;
|
||||
template class FastQueue<ImageMetadataBuffer>;
|
||||
template class FastQueue<ModuleFrame>;
|
||||
template class FastQueue<ModuleFrameBuffer>;
|
||||
@@ -1,162 +0,0 @@
|
||||
#include "WriterH5Writer.hpp"
|
||||
#include <sstream>
|
||||
|
||||
|
||||
//extern "C"
|
||||
//{
|
||||
// #include "H5DOpublic.h"
|
||||
// #include <bitshuffle/bshuf_h5filter.h>
|
||||
//}
|
||||
|
||||
using namespace std;
|
||||
using namespace core_buffer;
|
||||
|
||||
WriterH5Writer::WriterH5Writer(
|
||||
const string& output_file,
|
||||
const size_t n_frames,
|
||||
const size_t n_modules) :
|
||||
n_frames_(n_frames),
|
||||
n_modules_(n_modules),
|
||||
current_write_index_(0)
|
||||
{
|
||||
|
||||
// bshuf_register_h5filter();
|
||||
|
||||
file_ = H5::H5File(output_file, H5F_ACC_TRUNC);
|
||||
|
||||
hsize_t image_dataset_dims[3] =
|
||||
{n_frames_, n_modules * MODULE_Y_SIZE, MODULE_X_SIZE};
|
||||
|
||||
H5::DataSpace image_dataspace(3, image_dataset_dims);
|
||||
|
||||
hsize_t image_dataset_chunking[3] =
|
||||
{1, n_modules * MODULE_Y_SIZE, MODULE_X_SIZE};
|
||||
H5::DSetCreatPropList image_dataset_properties;
|
||||
image_dataset_properties.setChunk(3, image_dataset_chunking);
|
||||
|
||||
// // block_size, compression type
|
||||
// uint compression_prop[] =
|
||||
// {MODULE_N_PIXELS, //block size
|
||||
// BSHUF_H5_COMPRESS_LZ4}; // Compression type
|
||||
//
|
||||
// H5Pset_filter(image_dataset_properties.getId(),
|
||||
// BSHUF_H5FILTER,
|
||||
// H5Z_FLAG_MANDATORY,
|
||||
// 2,
|
||||
// &(compression_prop[0]));
|
||||
|
||||
image_dataset_ = file_.createDataSet(
|
||||
"image",
|
||||
H5::PredType::NATIVE_UINT16,
|
||||
image_dataspace,
|
||||
image_dataset_properties);
|
||||
|
||||
hsize_t metadata_dataset_dims[] = {n_frames_, 1};
|
||||
H5::DataSpace metadata_dataspace(2, metadata_dataset_dims);
|
||||
|
||||
// Chunk cannot be larger than n_frames.
|
||||
auto metadata_chunk_size = WRITER_METADATA_CHUNK_N_IMAGES;
|
||||
if (n_frames < metadata_chunk_size) {
|
||||
metadata_chunk_size = n_frames;
|
||||
}
|
||||
|
||||
hsize_t metadata_dataset_chunking[] = {metadata_chunk_size, 1};
|
||||
H5::DSetCreatPropList metadata_dataset_properties;
|
||||
metadata_dataset_properties.setChunk(2, metadata_dataset_chunking);
|
||||
|
||||
pulse_id_dataset_ = file_.createDataSet(
|
||||
"pulse_id",
|
||||
H5::PredType::NATIVE_UINT64,
|
||||
metadata_dataspace,
|
||||
metadata_dataset_properties);
|
||||
|
||||
frame_index_dataset_ = file_.createDataSet(
|
||||
"frame_index",
|
||||
H5::PredType::NATIVE_UINT64,
|
||||
metadata_dataspace,
|
||||
metadata_dataset_properties);
|
||||
|
||||
daq_rec_dataset_ = file_.createDataSet(
|
||||
"daq_rec",
|
||||
H5::PredType::NATIVE_UINT32,
|
||||
metadata_dataspace,
|
||||
metadata_dataset_properties);
|
||||
|
||||
is_good_frame_dataset_ = file_.createDataSet(
|
||||
"is_good_frame",
|
||||
H5::PredType::NATIVE_UINT8,
|
||||
metadata_dataspace,
|
||||
metadata_dataset_properties);
|
||||
|
||||
}
|
||||
|
||||
WriterH5Writer::~WriterH5Writer()
|
||||
{
|
||||
close_file();
|
||||
}
|
||||
|
||||
void WriterH5Writer::close_file()
|
||||
{
|
||||
image_dataset_.close();
|
||||
pulse_id_dataset_.close();
|
||||
frame_index_dataset_.close();
|
||||
daq_rec_dataset_.close();
|
||||
is_good_frame_dataset_.close();
|
||||
|
||||
file_.close();
|
||||
}
|
||||
|
||||
void WriterH5Writer::write(
|
||||
const ImageMetadataBuffer* metadata, const char* data)
|
||||
{
|
||||
auto n_images_in_buffer = metadata->n_pulses_in_buffer;
|
||||
|
||||
hsize_t b_i_dims[3] = {
|
||||
n_images_in_buffer,
|
||||
MODULE_Y_SIZE*n_modules_,
|
||||
MODULE_X_SIZE};
|
||||
H5::DataSpace b_i_space(3, b_i_dims);
|
||||
|
||||
hsize_t f_i_dims[3] = {n_frames_,
|
||||
MODULE_Y_SIZE * n_modules_,
|
||||
MODULE_X_SIZE};
|
||||
H5::DataSpace f_i_space(3, f_i_dims);
|
||||
|
||||
hsize_t i_count[] = {n_images_in_buffer,
|
||||
MODULE_Y_SIZE*n_modules_,
|
||||
MODULE_X_SIZE};
|
||||
hsize_t i_start[] = {current_write_index_, 0, 0};
|
||||
f_i_space.selectHyperslab(H5S_SELECT_SET, i_count, i_start);
|
||||
|
||||
image_dataset_.write(
|
||||
data, H5::PredType::NATIVE_UINT16,
|
||||
b_i_space, f_i_space);
|
||||
|
||||
hsize_t b_m_dims[2] = {n_images_in_buffer, 1};
|
||||
H5::DataSpace b_m_space (2, b_m_dims);
|
||||
|
||||
hsize_t f_m_dims[] = {n_frames_, 1};
|
||||
H5::DataSpace f_m_space(2, f_m_dims);
|
||||
|
||||
hsize_t meta_count[] = {n_images_in_buffer, 1};
|
||||
hsize_t meta_start[] = {current_write_index_, 0};
|
||||
f_m_space.selectHyperslab(H5S_SELECT_SET, meta_count, meta_start);
|
||||
|
||||
pulse_id_dataset_.write(
|
||||
&(metadata->pulse_id), H5::PredType::NATIVE_UINT64,
|
||||
b_m_space, f_m_space);
|
||||
|
||||
frame_index_dataset_.write(
|
||||
&(metadata->frame_index), H5::PredType::NATIVE_UINT64,
|
||||
b_m_space, f_m_space);
|
||||
|
||||
daq_rec_dataset_.write(
|
||||
&(metadata->daq_rec), H5::PredType::NATIVE_UINT32,
|
||||
b_m_space, f_m_space);
|
||||
|
||||
is_good_frame_dataset_.write(
|
||||
&(metadata->is_good_frame), H5::PredType::NATIVE_UINT8,
|
||||
b_m_space, f_m_space);
|
||||
|
||||
current_write_index_++;
|
||||
}
|
||||
@@ -1,138 +0,0 @@
|
||||
#include "WriterZmqReceiver.hpp"
|
||||
#include "zmq.h"
|
||||
#include "date.h"
|
||||
#include <chrono>
|
||||
#include <sstream>
|
||||
|
||||
using namespace std;
|
||||
using namespace core_buffer;
|
||||
|
||||
WriterZmqReceiver::WriterZmqReceiver(
|
||||
void *ctx,
|
||||
const string &ipc_prefix,
|
||||
const size_t n_modules) :
|
||||
n_modules_(n_modules),
|
||||
sockets_(n_modules)
|
||||
{
|
||||
|
||||
for (size_t i = 0; i < n_modules; i++) {
|
||||
sockets_[i] = zmq_socket(ctx, ZMQ_PULL);
|
||||
|
||||
int rcvhwm = WRITER_RCVHWM;
|
||||
if (zmq_setsockopt(sockets_[i], ZMQ_RCVHWM, &rcvhwm,
|
||||
sizeof(rcvhwm)) != 0) {
|
||||
throw runtime_error(zmq_strerror(errno));
|
||||
}
|
||||
int linger = 0;
|
||||
if (zmq_setsockopt(sockets_[i], ZMQ_LINGER, &linger,
|
||||
sizeof(linger)) != 0) {
|
||||
throw runtime_error(zmq_strerror(errno));
|
||||
}
|
||||
|
||||
stringstream ipc_addr;
|
||||
ipc_addr << ipc_prefix << i;
|
||||
const auto ipc = ipc_addr.str();
|
||||
|
||||
if (zmq_connect(sockets_[i], ipc.c_str()) != 0) {
|
||||
throw runtime_error(zmq_strerror(errno));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
WriterZmqReceiver::~WriterZmqReceiver()
|
||||
{
|
||||
for (size_t i = 0; i < n_modules_; i++) {
|
||||
zmq_close(sockets_[i]);
|
||||
}
|
||||
}
|
||||
|
||||
void WriterZmqReceiver::get_next_image(
|
||||
const uint64_t pulse_id,
|
||||
ImageMetadata* image_metadata,
|
||||
char* image_buffer)
|
||||
{
|
||||
// Init the image metadata.
|
||||
image_metadata->pulse_id = pulse_id;
|
||||
image_metadata->frame_index = 0;
|
||||
image_metadata->daq_rec = 0;
|
||||
image_metadata->data_n_bytes = 0;
|
||||
image_metadata->is_good_frame = 1;
|
||||
bool image_metadata_init = false;
|
||||
|
||||
size_t image_buffer_offset = 0;
|
||||
|
||||
for (size_t i_module = 0; i_module < n_modules_; i_module++) {
|
||||
|
||||
auto n_bytes_metadata = zmq_recv(
|
||||
sockets_[i_module],
|
||||
&frame_metadata,
|
||||
sizeof(StreamModuleFrame),
|
||||
0);
|
||||
|
||||
if (n_bytes_metadata != sizeof(StreamModuleFrame)) {
|
||||
throw runtime_error("Wrong number of metadata bytes.");
|
||||
}
|
||||
|
||||
// sf_replay should always send the right pulse_id.
|
||||
if (frame_metadata.metadata.pulse_id != pulse_id) {
|
||||
stringstream err_msg;
|
||||
|
||||
using namespace date;
|
||||
using namespace chrono;
|
||||
err_msg << "[" << system_clock::now() << "]";
|
||||
err_msg << "[sf_writer::receive_replay]";
|
||||
err_msg << " Read unexpected pulse_id. ";
|
||||
err_msg << " Expected " << pulse_id;
|
||||
err_msg << " received ";
|
||||
err_msg << frame_metadata.metadata.pulse_id;
|
||||
err_msg << " from i_module " << i_module << endl;
|
||||
|
||||
throw runtime_error(err_msg.str());
|
||||
}
|
||||
|
||||
if (!frame_metadata.is_frame_present) {
|
||||
image_metadata->is_good_frame = 0;
|
||||
|
||||
// Init the image metadata with the first valid frame.
|
||||
} else if (!image_metadata_init) {
|
||||
image_metadata_init = true;
|
||||
|
||||
image_metadata->frame_index =
|
||||
frame_metadata.metadata.frame_index;
|
||||
image_metadata->daq_rec =
|
||||
frame_metadata.metadata.daq_rec;
|
||||
}
|
||||
|
||||
// Once the image is not good, we don't care to re-flag it.
|
||||
if (image_metadata->is_good_frame == 1) {
|
||||
if (frame_metadata.metadata.frame_index !=
|
||||
image_metadata->frame_index) {
|
||||
image_metadata->is_good_frame = 0;
|
||||
}
|
||||
|
||||
if (frame_metadata.metadata.daq_rec !=
|
||||
image_metadata->daq_rec) {
|
||||
image_metadata->is_good_frame = 0;
|
||||
}
|
||||
|
||||
if (frame_metadata.metadata.n_received_packets !=
|
||||
JUNGFRAU_N_PACKETS_PER_FRAME) {
|
||||
image_metadata->is_good_frame = 0;
|
||||
}
|
||||
}
|
||||
|
||||
auto n_bytes_image = zmq_recv(
|
||||
sockets_[i_module],
|
||||
(image_buffer + image_buffer_offset),
|
||||
frame_metadata.data_n_bytes,
|
||||
0);
|
||||
|
||||
if (n_bytes_image != frame_metadata.data_n_bytes) {
|
||||
throw runtime_error("Wrong number of data bytes.");
|
||||
}
|
||||
|
||||
image_buffer_offset += n_bytes_image;
|
||||
}
|
||||
|
||||
image_metadata->data_n_bytes = image_buffer_offset;
|
||||
}
|
||||
@@ -1,179 +0,0 @@
|
||||
#include <iostream>
|
||||
#include <stdexcept>
|
||||
#include "buffer_config.hpp"
|
||||
#include "zmq.h"
|
||||
#include <string>
|
||||
#include <jungfrau.hpp>
|
||||
#include <thread>
|
||||
#include <chrono>
|
||||
#include "WriterH5Writer.hpp"
|
||||
#include <FastQueue.hpp>
|
||||
#include <cstring>
|
||||
#include <BufferedFastQueue.hpp>
|
||||
#include "date.h"
|
||||
#include "bitshuffle/bitshuffle.h"
|
||||
#include "WriterZmqReceiver.hpp"
|
||||
|
||||
using namespace std;
|
||||
using namespace core_buffer;
|
||||
|
||||
void receive_replay(
|
||||
void* ctx,
|
||||
const string ipc_prefix,
|
||||
const size_t n_modules,
|
||||
FastQueue<ImageMetadataBuffer>& queue,
|
||||
const uint64_t start_pulse_id,
|
||||
const uint64_t stop_pulse_id)
|
||||
{
|
||||
try {
|
||||
WriterZmqReceiver receiver(ctx, ipc_prefix, n_modules);
|
||||
BufferedFastQueue buffered_queue(
|
||||
queue, WRITER_DATA_CACHE_N_IMAGES, n_modules);
|
||||
|
||||
uint64_t current_pulse_id=start_pulse_id;
|
||||
|
||||
// "<= stop_pulse_id" because we include the last pulse_id.
|
||||
while(current_pulse_id<=stop_pulse_id) {
|
||||
|
||||
auto image_metadata = buffered_queue.get_metadata_buffer();
|
||||
auto image_buffer = buffered_queue.get_data_buffer();
|
||||
|
||||
receiver.get_next_image(
|
||||
current_pulse_id, image_metadata, image_buffer);
|
||||
|
||||
if (image_metadata->pulse_id != current_pulse_id) {
|
||||
throw runtime_error("Wrong pulse id from zmq receiver.");
|
||||
}
|
||||
|
||||
buffered_queue.commit();
|
||||
current_pulse_id++;
|
||||
}
|
||||
|
||||
buffered_queue.finalize();
|
||||
|
||||
} catch (const std::exception& e) {
|
||||
using namespace date;
|
||||
using namespace chrono;
|
||||
|
||||
cout << "[" << system_clock::now() << "]";
|
||||
cout << "[sf_writer::receive_replay]";
|
||||
cout << " Stopped because of exception: " << endl;
|
||||
cout << e.what() << endl;
|
||||
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
int main (int argc, char *argv[])
|
||||
{
|
||||
if (argc != 4) {
|
||||
cout << endl;
|
||||
cout << "Usage: sf_writer ";
|
||||
cout << " [output_file] [start_pulse_id] [stop_pulse_id]";
|
||||
cout << endl;
|
||||
cout << "\toutput_file: Complete path to the output file." << endl;
|
||||
cout << "\tstart_pulse_id: Start pulse_id of retrieval." << endl;
|
||||
cout << "\tstop_pulse_id: Stop pulse_id of retrieval." << endl;
|
||||
cout << endl;
|
||||
|
||||
exit(-1);
|
||||
}
|
||||
|
||||
string output_file = string(argv[1]);
|
||||
uint64_t start_pulse_id = (uint64_t) atoll(argv[2]);
|
||||
uint64_t stop_pulse_id = (uint64_t) atoll(argv[3]);
|
||||
|
||||
size_t n_modules = 32;
|
||||
|
||||
FastQueue<ImageMetadataBuffer> queue(
|
||||
MODULE_N_BYTES * n_modules * WRITER_DATA_CACHE_N_IMAGES,
|
||||
WRITER_FASTQUEUE_N_SLOTS);
|
||||
|
||||
auto ctx = zmq_ctx_new();
|
||||
zmq_ctx_set (ctx, ZMQ_IO_THREADS, WRITER_ZMQ_IO_THREADS);
|
||||
|
||||
thread replay_receive_thread(receive_replay,
|
||||
ctx, REPLAY_STREAM_IPC_URL, n_modules,
|
||||
ref(queue), start_pulse_id, stop_pulse_id);
|
||||
|
||||
size_t n_frames = stop_pulse_id - start_pulse_id + 1;
|
||||
WriterH5Writer writer(output_file, n_frames, n_modules);
|
||||
|
||||
// TODO: Remove stats trash.
|
||||
int stats_counter = 0;
|
||||
size_t read_total_us = 0;
|
||||
size_t write_total_us = 0;
|
||||
size_t read_max_us = 0;
|
||||
size_t write_max_us = 0;
|
||||
|
||||
auto start_time = chrono::steady_clock::now();
|
||||
|
||||
auto current_pulse_id = start_pulse_id;
|
||||
// "<= stop_pulse_id" because we include the last pulse_id.
|
||||
while (current_pulse_id <= stop_pulse_id) {
|
||||
|
||||
int slot_id; ;
|
||||
while((slot_id = queue.read()) == -1) {
|
||||
this_thread::sleep_for(chrono::milliseconds(
|
||||
RB_READ_RETRY_INTERVAL_MS));
|
||||
}
|
||||
|
||||
auto metadata = queue.get_metadata_buffer(slot_id);
|
||||
auto data = queue.get_data_buffer(slot_id);
|
||||
|
||||
auto read_end_time = chrono::steady_clock::now();
|
||||
auto read_us_duration = chrono::duration_cast<chrono::microseconds>(
|
||||
read_end_time-start_time).count();
|
||||
|
||||
// Verify that all pulse_ids are correct.
|
||||
for (int i=0; i<metadata->n_pulses_in_buffer; i++) {
|
||||
if (metadata->pulse_id[i] != current_pulse_id) {
|
||||
throw runtime_error("Wrong pulse id from receiver thread.");
|
||||
}
|
||||
|
||||
current_pulse_id++;
|
||||
}
|
||||
|
||||
start_time = chrono::steady_clock::now();
|
||||
|
||||
writer.write(metadata, data);
|
||||
|
||||
auto write_end_time = chrono::steady_clock::now();
|
||||
auto write_us_duration = chrono::duration_cast<chrono::microseconds>(
|
||||
write_end_time-start_time).count();
|
||||
|
||||
queue.release();
|
||||
|
||||
// TODO: Some poor statistics.
|
||||
stats_counter++;
|
||||
|
||||
read_total_us += read_us_duration;
|
||||
read_max_us = max(read_max_us, (uint64_t)read_us_duration);
|
||||
|
||||
write_total_us += write_us_duration;
|
||||
write_max_us = max(write_max_us, (uint64_t)write_us_duration);
|
||||
|
||||
// if (stats_counter == STATS_MODULO) {
|
||||
cout << "sf_writer:read_us " << read_total_us / STATS_MODULO;
|
||||
cout << " sf_writer:read_max_us " << read_max_us;
|
||||
cout << " sf_writer:write_us " << write_total_us / STATS_MODULO;
|
||||
cout << " sf_writer:write_max_us " << write_max_us;
|
||||
|
||||
cout << endl;
|
||||
|
||||
stats_counter = 0;
|
||||
read_total_us = 0;
|
||||
read_max_us = 0;
|
||||
write_total_us = 0;
|
||||
write_max_us = 0;
|
||||
// }
|
||||
|
||||
start_time = chrono::steady_clock::now();
|
||||
}
|
||||
|
||||
writer.close_file();
|
||||
|
||||
//wait till receive thread is finished
|
||||
replay_receive_thread.join();
|
||||
return 0;
|
||||
}
|
||||
Reference in New Issue
Block a user