diff --git a/sf-buffer/src/sf_writer.cpp b/sf-buffer/src/sf_writer.cpp index bf02051..98abebf 100644 --- a/sf-buffer/src/sf_writer.cpp +++ b/sf-buffer/src/sf_writer.cpp @@ -7,11 +7,119 @@ #include #include #include "SFWriter.hpp" +#include #include using namespace std; using namespace core_buffer; +void receive_replay( + const string ipc_prefix, + const size_t n_modules, + FastQueue& queue, + void* ctx) +{ + try { + + void *sockets[n_modules]; + for (size_t i = 0; i < n_modules; i++) { + sockets[i] = zmq_socket(ctx, ZMQ_PULL); + int rcvhwm = REPLAY_BLOCK_SIZE; + 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)); + } + } + + auto module_meta_buffer = make_unique(); + + while (true) { + + auto slot_id = queue.reserve(); + + if (slot_id == -1){ + this_thread::sleep_for(chrono::milliseconds(5)); + continue; + } + + auto frame_meta_buffer = queue.get_metadata_buffer(slot_id); + auto frame_buffer = queue.get_data_buffer(slot_id); + + for (size_t i = 0; i < n_modules; i++) { + auto n_bytes_metadata = zmq_recv( + sockets[i], + module_meta_buffer.get(), + sizeof(ModuleFrame), + 0); + + if (n_bytes_metadata != sizeof(ModuleFrame)) { + // TODO: Make nicer expcetion. + throw runtime_error(strerror(errno)); + } + + // Initialize buffers in first iteration for each pulse_id. + if (i == 0) { + frame_meta_buffer->pulse_id = + module_meta_buffer->pulse_id; + frame_meta_buffer->frame_index = + module_meta_buffer->frame_index; + frame_meta_buffer->daq_rec = + module_meta_buffer->daq_rec; + frame_meta_buffer->n_received_packets = + module_meta_buffer->n_received_packets; + } + + if (frame_meta_buffer->pulse_id != + module_meta_buffer->pulse_id) { + throw runtime_error("Unexpected pulse_id received."); + } + + auto n_bytes_image = zmq_recv( + sockets[i], + (frame_buffer + (MODULE_N_PIXELS * i)), + MODULE_N_BYTES, + 0); + + if (n_bytes_image != MODULE_N_BYTES) { + // TODO: Make nicer expcetion. + throw runtime_error("Unexpected number of bytes in image."); + } + } + + queue.commit(); + } + + for (size_t i = 0; i < n_modules; i++) { + zmq_close(sockets[i]); + } + + zmq_ctx_destroy(ctx); + } catch (const std::exception& e) { + using namespace date; + using namespace chrono; + + cout << "[" << system_clock::now() << "]"; + cout << "[sf_h5_writer::receive_replay]"; + cout << " Stopped because of exception: " << endl; + cout << e.what() << endl; + + throw; + } +} + int main (int argc, char *argv[]) { if (argc != 4) { @@ -33,32 +141,20 @@ int main (int argc, char *argv[]) size_t n_modules = 32; + FastQueue queue( + n_modules * MODULE_N_BYTES, + WRITER_RB_BUFFER_SLOTS); + string ipc_prefix = "ipc://sf-replay-"; auto ctx = zmq_ctx_new(); zmq_ctx_set (ctx, ZMQ_IO_THREADS, WRITER_ZMQ_IO_THREADS); - void *sockets[n_modules]; - for (size_t i = 0; i < n_modules; i++) { - sockets[i] = zmq_socket(ctx, ZMQ_PULL); - int rcvhwm = REPLAY_BLOCK_SIZE; - 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)); - } - } + thread replay_receive_thread( + receive_replay, + ipc_prefix, + n_modules, + ref(queue), + ctx); size_t n_frames = stop_pulse_id - start_pulse_id; SFWriter writer(output_file, n_frames, n_modules); @@ -71,54 +167,35 @@ int main (int argc, char *argv[]) size_t read_max_us = 0; size_t write_max_us = 0; - auto module_meta_buffer = make_unique(); - auto frame_meta_buffer = make_unique(); - auto frame_buffer = make_unique(MODULE_N_BYTES * n_modules); - auto start_time = chrono::steady_clock::now(); auto current_pulse_id = start_pulse_id; while (current_pulse_id <= stop_pulse_id) { - for (size_t i = 0; i < n_modules; i++) { - auto n_bytes_metadata = zmq_recv( - sockets[i], - module_meta_buffer.get(), - sizeof(ModuleFrame), - 0); + auto slot_id = queue.read(); - if (n_bytes_metadata != sizeof(ModuleFrame)) { - // TODO: Make nicer expcetion. - throw runtime_error(strerror(errno)); - } + if(slot_id == -1) { + this_thread::sleep_for(chrono::milliseconds( + config::ring_buffer_read_retry_interval)); + continue; + } - // Initialize buffers in first iteration for each pulse_id. - if (i == 0) { - frame_meta_buffer->pulse_id = - module_meta_buffer->pulse_id; - frame_meta_buffer->frame_index = - module_meta_buffer->frame_index; - frame_meta_buffer->daq_rec = - module_meta_buffer->daq_rec; - frame_meta_buffer->n_received_packets = - module_meta_buffer->n_received_packets; - } + auto metadata = queue.get_metadata_buffer(slot_id); + auto data = queue.get_data_buffer(slot_id); - if (frame_meta_buffer->pulse_id != - module_meta_buffer->pulse_id) { - throw runtime_error("Unexpected pulse_id received."); - } + if (metadata->pulse_id != current_pulse_id) { + stringstream err_msg; - auto n_bytes_image = zmq_recv( - sockets[i], - ((frame_buffer.get()) + (MODULE_N_PIXELS * i)), - MODULE_N_BYTES, - 0); + using namespace date; + using namespace chrono; + err_msg << "[" << system_clock::now() << "]"; + err_msg << "[sf_writer::main]"; + err_msg << " Read unexpected pulse_id. "; + err_msg << " Expected " << current_pulse_id; + err_msg << " received " << metadata->pulse_id; + err_msg << endl; - if (n_bytes_image != MODULE_N_BYTES) { - // TODO: Make nicer exception. - throw runtime_error("Unexpected number of bytes in image."); - } + throw runtime_error(err_msg.str()); } auto read_end_time = chrono::steady_clock::now(); @@ -127,7 +204,8 @@ int main (int argc, char *argv[]) start_time = chrono::steady_clock::now(); - writer.write(frame_meta_buffer.get(), frame_buffer.get()); + writer.write(metadata, data); + queue.release(); current_pulse_id++; // TODO: Some poor statistics. @@ -167,10 +245,5 @@ int main (int argc, char *argv[]) writer.close_file(); - for (size_t i = 0; i < n_modules; i++) { - zmq_close(sockets[i]); - } - zmq_ctx_destroy(ctx); - return 0; }