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

Finalize refactoring of sf_writer

Browse Source
This commit is contained in:
babic_a
2020-05-15 10:46:43 +02:00
parent a42e718880
commit 1f771bf444
1 changed files with 39 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sf-buffer/src/sf_writer.cpp
+39 -141
View File
@@ -6,52 +6,31 @@
#include <jungfrau.hpp>
#include <thread>
#include <chrono>
#include "SFWriter.hpp"
#include "WriterH5Writer.hpp"
#include <FastQueue.hpp>
#include <cstring>
#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<DetectorFrame>& queue,
void* ctx,
const uint64_t start_pulse_id,
FastQueue<ImageMetadata>& queue,
const uint64_t start_pulse_id,
const uint64_t stop_pulse_id)
{
try {
WriterZmqReceiver receiver(ctx, ipc_prefix, n_modules);
void *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(strerror(errno));
}
int linger = 0;
if (zmq_setsockopt(sockets[i], ZMQ_LINGER, &linger,
sizeof(linger)) != 0) {
throw runtime_error(strerror(errno));
}
stringstream ipc_addr;
ipc_addr << ipc_prefix << i;
const auto ipc = ipc_addr.str();
if (zmq_bind(sockets[i], ipc.c_str()) != 0) {
throw runtime_error(strerror(errno));
}
}
uint64_t current_pulse_id = start_pulse_id;
while (true) {
// "<= stop_pulse_id" because we include the last pulse_id.
for (uint64_t current_pulse_id=start_pulse_id;
current_pulse_id<=stop_pulse_id;
current_pulse_id++) {
auto slot_id = queue.reserve();
@@ -60,102 +39,19 @@ void receive_replay(
continue;
}
auto frame_meta_buffer = queue.get_metadata_buffer(slot_id);
auto frame_buffer = queue.get_data_buffer(slot_id);
auto image_metadata = queue.get_metadata_buffer(slot_id);
auto image_buffer = queue.get_data_buffer(slot_id);
for (
size_t i_buffer=0;
i_buffer<WRITER_N_FRAMES_BUFFER;
i_buffer++)
{
receiver.get_next_image(image_metadata, image_buffer);
frame_meta_buffer->is_good_frame[i_buffer] = true;
for (size_t i_module = 0; i_module < n_modules; i_module++) {
auto n_bytes_metadata = zmq_recv(
sockets[i_module],
module_meta_buffer.get(),
sizeof(ModuleFrame),
0);
if (n_bytes_metadata != sizeof(ModuleFrame)) {
throw runtime_error("Wrong number of metadata bytes.");
}
if (module_meta_buffer->pulse_id != current_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 " << current_pulse_id;
err_msg << " received ";
err_msg << module_meta_buffer->pulse_id << endl;
throw runtime_error(err_msg.str());
}
// Initialize buffers in first iteration for each pulse_id.
if (i_module == 0) {
frame_meta_buffer->pulse_id[i_buffer] =
module_meta_buffer->pulse_id;
frame_meta_buffer->frame_index[i_buffer] =
module_meta_buffer->frame_index;
frame_meta_buffer->daq_rec[i_buffer] =
module_meta_buffer->daq_rec;
frame_meta_buffer->n_received_packets[i_buffer] =
module_meta_buffer->n_received_packets;
if ( module_meta_buffer->n_received_packets != 128 ) frame_meta_buffer->is_good_frame[i_buffer] = false;
} else {
if (module_meta_buffer->pulse_id != frame_meta_buffer->pulse_id[i_buffer]) frame_meta_buffer->is_good_frame[i_buffer] = false;
if (module_meta_buffer->frame_index != frame_meta_buffer->frame_index[i_buffer]) frame_meta_buffer->is_good_frame[i_buffer] = false;
if (module_meta_buffer->daq_rec != frame_meta_buffer->daq_rec[i_buffer]) frame_meta_buffer->is_good_frame[i_buffer] = false;
if (module_meta_buffer->n_received_packets != 128 ) frame_meta_buffer->is_good_frame[i_buffer] = false;
}
if (frame_meta_buffer->pulse_id[i_buffer] !=
module_meta_buffer->pulse_id) {
throw runtime_error("Unexpected pulse_id received.");
}
// Offset due to frame in buffer.
size_t offset = MODULE_N_BYTES * n_modules * i_buffer;
// offset due to module in frame.
offset += MODULE_N_BYTES * i_module;
auto n_bytes_image = zmq_recv(
sockets[i_module],
(frame_buffer + offset),
MODULE_N_BYTES,
0);
if (n_bytes_image != MODULE_N_BYTES) {
throw runtime_error("Wrong number of data bytes.");
}
}
current_pulse_id++;
// received all frames, don't wait till all WRITER_N_FRAMES_BUFFER
if ( current_pulse_id > stop_pulse_id ) break;
if (image_metadata->pulse_id != current_pulse_id) {
throw runtime_error("Wrong pulse id from zmq receiver.");
}
queue.commit();
// break receiving loop
if ( current_pulse_id > stop_pulse_id ) break;
}
for (size_t i = 0; i < n_modules; i++) {
zmq_close(sockets[i]);
current_pulse_id++;
}
zmq_ctx_destroy(ctx);
} catch (const std::exception& e) {
using namespace date;
using namespace chrono;
@@ -190,23 +86,24 @@ int main (int argc, char *argv[])
size_t n_modules = 32;
FastQueue<DetectorFrame> queue(
n_modules * MODULE_N_BYTES * WRITER_N_FRAMES_BUFFER,
WRITER_RB_BUFFER_SLOTS);
auto compress_frame_size = bshuf_compress_lz4_bound(
MODULE_N_PIXELS, PIXEL_N_BYTES, MODULE_N_PIXELS);
FastQueue<ImageMetadata> queue(
compress_frame_size * n_modules, WRITER_FASTQUEUE_SLOTS);
auto ctx = zmq_ctx_new();
zmq_ctx_set (ctx, ZMQ_IO_THREADS, WRITER_ZMQ_IO_THREADS);
thread replay_receive_thread(
receive_replay, REPLAY_STREAM_IPC_URL, n_modules,
ref(queue), ctx, start_pulse_id, stop_pulse_id);
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;
SFWriter writer(output_file, n_frames, n_modules);
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;
@@ -215,6 +112,7 @@ int main (int argc, char *argv[])
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) {
auto slot_id = queue.read();
@@ -232,29 +130,29 @@ int main (int argc, char *argv[])
auto read_us_duration = chrono::duration_cast<chrono::microseconds>(
read_end_time-start_time).count();
if (metadata->pulse_id != current_pulse_id) {
throw runtime_error("Wrong pulse id from receiver thread.");
}
start_time = chrono::steady_clock::now();
writer.write(metadata, data);
queue.release();
current_pulse_id += WRITER_N_FRAMES_BUFFER;
// TODO: Some poor statistics.
stats_counter += WRITER_N_FRAMES_BUFFER;
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();
current_pulse_id++;
// 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;
if (read_us_duration > read_max_us) {
read_max_us = read_us_duration;
}
if (write_us_duration > write_max_us) {
write_max_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;