From e9e7fdf2a405013ddf6efff2d8c61c1856e1181d Mon Sep 17 00:00:00 2001 From: Babicaa Date: Fri, 26 Apr 2019 12:55:46 +0200 Subject: [PATCH] Finished refactoring WriterManager --- lib/src/WriterManager.cpp | 262 ++++++++++++++++++++++++++++++++++---- lib/src/WriterManager.hpp | 31 ++++- 2 files changed, 265 insertions(+), 28 deletions(-) diff --git a/lib/src/WriterManager.cpp b/lib/src/WriterManager.cpp index f2785e6..9f6a17c 100644 --- a/lib/src/WriterManager.cpp +++ b/lib/src/WriterManager.cpp @@ -1,8 +1,11 @@ #include #include -#include "WriterManager.hpp" #include +#include "WriterManager.hpp" +#include "MetadataBuffer.hpp" +#include "BufferedWriter.hpp" +#include "config.hpp" using namespace std; @@ -12,14 +15,16 @@ void writer_utils::set_process_id(int user_id) #ifdef DEBUG_OUTPUT using namespace date; cout << "[" << std::chrono::system_clock::now() << "]"; - cout << "[writer_utils::set_process_id] Setting process uid to " << user_id << endl; + cout << "[writer_utils::set_process_id] Setting process user to "; + cout << user_id << endl; #endif if (setegid(user_id)) { stringstream error_message; using namespace date; error_message << "[" << std::chrono::system_clock::now() << "]"; - error_message << "[writer_utils::set_process_id] Cannot set group_id to " << user_id << endl; + error_message << "[writer_utils::set_process_id] Cannot set group_id to "; + error_message << user_id << endl; throw runtime_error(error_message.str()); } @@ -28,7 +33,8 @@ void writer_utils::set_process_id(int user_id) stringstream error_message; using namespace date; error_message << "[" << std::chrono::system_clock::now() << "]"; - error_message << "[writer_utils::set_process_id] Cannot set user_id to " << user_id << endl; + error_message << "[writer_utils::set_process_id] Cannot set user_id to "; + error_message << user_id << endl; throw runtime_error(error_message.str()); } @@ -42,18 +48,33 @@ void writer_utils::create_destination_folder(const string& output_file) string output_folder(output_file.substr(0, file_separator_index)); using namespace date; cout << "[" << std::chrono::system_clock::now() << "]"; - cout << "[writer_utils::create_destination_folder] Creating folder " << output_folder << endl; + cout << "[writer_utils::create_destination_folder] Creating folder "; + cout << output_folder << endl; string create_folder_command("mkdir -p " + output_folder); system(create_folder_command.c_str()); } } -WriterManager::WriterManager(): - logs(10), - writing_flag(false), running_flag(true), - n_frames_to_receive(0), n_frames_to_write(0) +WriterManager::WriterManager( + RingBuffer& ring_buffer, + const H5Format& format, + std::shared_ptr header_values_type, + hsize_t frames_per_file): + ring_buffer(ring_buffer), + format(format), + header_values_type(header_values_type), + logs(10) { + running_flag = true; + + writing_flag = false; + receiving_flag = false; + + n_frames_to_receive = 0; + n_frames_to_write = 0; + + #ifdef DEBUG_OUTPUT using namespace date; cout << "[" << std::chrono::system_clock::now() << @@ -81,7 +102,7 @@ string WriterManager::get_status() } else if (running_flag) { return "ready"; } else { - return "Error.. I guess. This shouldn't be possible? Are you sure you are using it correctly?"; + return "Error.. I guess. This shouldn't be possible?"; } } @@ -123,12 +144,16 @@ void WriterManager::start(const string output_file, n_frames_to_receive = n_frames; receiving_flag = true; - boost::thread writer_thread(&ProcessManager::write_h5, this, output_file, n_frames); + + writing_thread = boost::thread(&WriterManager::write_h5, + this, + output_file, + n_frames); //TODO: Sent this event somewhere? } -bool WriterManager::is_running() +bool WriterManager::is_running() const { return running_flag.load(); } @@ -168,17 +193,206 @@ void WriterManager::writing_completed() { //TODO: Send this event somewhere somehow? } -void WriterManager::writing_error(string error_message) { - writing_flag = false; +void WriterManager::writing_error(string error) { - - #ifdef DEBUG_OUTPUT - stringstream output_message; - using namespace date; - output_message << "[" << std::chrono::system_clock::now() << "]"; - output_message << "[WriterManager::writing_error] Error while writing: "; - output_message << error_message << endl; - #endif - - // TODO: Send this error somewhere? } + +void WriterManager::write_h5_format(H5::H5File& file) { + + try { + H5FormatUtils::write_format(file, format, {}); + } catch (const runtime_error& ex) { + using namespace date; + using namespace chrono; + + cout << "[" << system_clock::now() << "]"; + cout << "[ProcessManager::write_h5_format] Error while"; + cout << " trying to write file format: "<< ex.what() << endl; + } +} + +void WriterManager::write_h5(string output_file, uint64_t n_frames) +{ + try { + + size_t metadata_buffer_size = + frames_per_file != 0 ? frames_per_file : n_frames; + + auto metadata_buffer = unique_ptr( + new MetadataBuffer(metadata_buffer_size, + header_values_type)); + + auto writer = get_buffered_writer( + output_file, + n_frames, + move(metadata_buffer), + frames_per_file, + config::dataset_increase_step); + + writer->create_file(); + + auto raw_frames_dataset_name = config::raw_image_dataset_name; + + uint64_t last_pulse_id = 0; + + while(is_writing() || !ring_buffer.is_empty()) { + + if (ring_buffer.is_empty()) { + boost::this_thread::sleep_for( + boost::chrono::milliseconds( + config::ring_buffer_read_retry_interval)); + continue; + } + + const pair< shared_ptr, char* > received_data = + ring_buffer.read(); + + // NULL pointer means that the ringbuffer->read() timeouted. + if(!received_data.first) { + continue; + } + + // The acquisition stops when there are no more frames to write. + if (!write_frame()) { + break; + } + + // Write file format before rolling to next file. + if (!writer->is_data_for_current_file( + received_data.first->frame_index)) { + + #ifdef DEBUG_OUTPUT + using namespace date; + using namespace chrono; + + cout << "[" << system_clock::now() << "]"; + cout << "[PSIWriter::write_h5] Frame index "; + cout << received_data.first->frame_index; + cout << " does not belong to current file. "; + cout << " Write format before switching file." << endl; + #endif + + writer->write_metadata_to_file(); + + write_h5_format(writer->get_h5_file()); + } + + #ifdef PERF_OUTPUT + using namespace date; + using namespace chrono; + + auto start_time_frame = system_clock::now(); + #endif + + // Write image data. + writer->write_data(raw_frames_dataset_name, + received_data.first->frame_index, + received_data.second, + received_data.first->frame_shape, + received_data.first->frame_bytes_size, + received_data.first->type, + received_data.first->endianness); + + #ifdef PERF_OUTPUT + using namespace date; + using namespace chrono; + + auto frame_time_difference = system_clock::now() - start_time_frame; + + auto frame_diff_ms = + duration(frame_time_difference).count(); + + cout << "[" << system_clock::now() << "]"; + cout << "[PSIWriter::write_h5] Frame index "; + cout << received_data.first->frame_index; + cout << " written in " << frame_diff_ms << " ms." << endl; + #endif + + ring_buffer.release(received_data.first->buffer_slot_index); + + #ifdef PERF_OUTPUT + using namespace date; + using namespace chrono; + + auto start_time_metadata = system_clock::now(); + #endif + + // Write image metadata if mapping specified. + if (header_values_type) { + + for (const auto& header_type : *header_values_type) { + + auto& name = header_type.first; + auto value = received_data.first->header_values.at(name); + + // TODO: Ugly hack until we get the start sequence in bsread. + if (name == "pulse_id") { + if (!last_pulse_id) { + last_pulse_id = *(reinterpret_cast(value.get())); + //notify_first_pulse_id(last_pulse_id); + } else { + last_pulse_id = *(reinterpret_cast(value.get())); + } + } + + writer->cache_metadata(name, received_data.first->frame_index, value.get()); + } + } + + #ifdef PERF_OUTPUT + using namespace date; + using namespace chrono; + + auto metadata_time_difference = system_clock::now() - start_time_metadata; + auto metadata_diff_ms = duration(metadata_time_difference).count(); + + cout << "[" << system_clock::now() << "]"; + cout << "[ProcessManager::write_h5] Frame metadata index "; + cout << received_data.first->frame_index << " written in " << metadata_diff_ms << " ms." << endl; + #endif + } + + // Send the last_pulse_id only if it was set. + if (last_pulse_id) { + //notify_last_pulse_id(last_pulse_id); + } + + if (writer->is_file_open()) { + #ifdef DEBUG_OUTPUT + using namespace date; + using namespace chrono; + + cout << "[" << system_clock::now() << "]"; + cout << "[ProcessManager::write] Writing file format." << endl; + #endif + + writer->write_metadata_to_file(); + + write_h5_format(writer->get_h5_file()); + } + + #ifdef DEBUG_OUTPUT + using namespace date; + using namespace chrono; + + cout << "[" << system_clock::now() << "]"; + cout << "[ProcessManager::write] Closing file " << get_output_file() << endl; + #endif + + writer->close_file(); + + #ifdef DEBUG_OUTPUT + using namespace date; + using namespace chrono; + + cout << "[" << system_clock::now() << "]"; + cout << "[ProcessManager::write] Writer thread stopped." << endl; + #endif + + writing_completed(); + + } catch (const exception& ex) { + writing_error(ex.what()); + } +} + diff --git a/lib/src/WriterManager.hpp b/lib/src/WriterManager.hpp index 82487a5..d53ae16 100644 --- a/lib/src/WriterManager.hpp +++ b/lib/src/WriterManager.hpp @@ -6,9 +6,13 @@ #include #include #include +#include #include #include "date.h" #include + +#include "ZmqReceiver.hpp" +#include "RingBuffer.hpp" #include "H5Format.hpp" namespace writer_utils { @@ -40,8 +44,28 @@ class WriterManager std::atomic n_frames_to_receive; std::atomic n_frames_to_write; + + protected: + RingBuffer& ring_buffer; + const H5Format& format; + hsize_t frames_per_file; + + boost::thread writing_thread; + + typedef std::unordered_map header_map; + std::shared_ptr header_values_type = NULL; + + void write_h5(std::string output_file, + uint64_t n_frames); + void write_h5_format(H5::H5File& file); + + public: - WriterManager(); + WriterManager(RingBuffer& ring_buffer, + const H5Format& format, + std::shared_ptr header_values_type, + hsize_t frames_per_file=0); + virtual ~WriterManager(); void start(std::string output_file, int n_frames, int user_id); @@ -54,16 +78,15 @@ class WriterManager bool receive_frame(); // True if the process should conitnue. bool is_running() const; + bool is_writing() const; // Return True if the frame is to be written, False otherwise. bool write_frame(); // True if the writing should continue. - bool is_writing() const; - // Signal that the writing has completed. void writing_completed(); - void writing_error(std::string error_message); + void writing_error(std::string error); }; #endif