diff --git a/sf-buffer/src/sf_writer.cpp b/sf-buffer/src/sf_writer.cpp index 496851e..2df16ce 100644 --- a/sf-buffer/src/sf_writer.cpp +++ b/sf-buffer/src/sf_writer.cpp @@ -6,52 +6,31 @@ #include #include #include -#include "SFWriter.hpp" +#include "WriterH5Writer.hpp" #include #include #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& queue, - void* ctx, - const uint64_t start_pulse_id, + FastQueue& 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_bufferis_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 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 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( 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( 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;