diff --git a/src/ProcessManager.cpp b/src/ProcessManager.cpp index a0c3a41..5503985 100644 --- a/src/ProcessManager.cpp +++ b/src/ProcessManager.cpp @@ -7,9 +7,83 @@ #include #include "RestApi.hpp" +#include "ProcessManager.hpp" +#include "config.hpp" +#include "H5Writer.hpp" using namespace std; +void ProcessManager::run_writer(WriterManager& manager, const H5Format& format, + ZmqReceiver& receiver, uint16_t rest_port) +{ + size_t n_slots = config::ring_buffer_n_slots; + RingBuffer ring_buffer(n_slots); + + #ifdef DEBUG_OUTPUT + cout << "[ProcessManager::run_writer] Running writer"; + cout << " and output_file " << manager.get_output_file(); + cout << " and n_slots " << n_slots; + cout << endl; + #endif + + boost::thread receiver_thread(receive_zmq, boost::ref(manager), boost::ref(ring_buffer), + boost::ref(receiver), boost::ref(format)); + boost::thread writer_thread(write_h5, boost::ref(manager), + boost::ref(format), boost::ref(ring_buffer)); + + RestApi::start_rest_api(manager, rest_port); + + #ifdef DEBUG_OUTPUT + cout << "[ProcessManager::run_writer] Rest API stopped." << endl; + #endif + + // In case SIGINT stopped the rest_api. + manager.stop(); + + receiver_thread.join(); + writer_thread.join(); + + #ifdef DEBUG_OUTPUT + cout << "[ProcessManager::run_writer] Writer properly stopped." << endl; + #endif +} + +void ProcessManager::receive_zmq(WriterManager& manager, RingBuffer& ring_buffer, + ZmqReceiver& receiver, const H5Format& format) +{ + while (manager.is_running()) { + + auto frame = receiver.receive(); + + // In case no message is available before the timeout, both pointers are NULL. + if (!frame.first){ + continue; + } + + auto frame_metadata = frame.first; + auto frame_data = frame.second; + + #ifdef DEBUG_OUTPUT + cout << "[ProcessManager::receive_zmq] Processing FrameMetadata"; + cout << " with frame_index " << frame_metadata->frame_index; + cout << " and frame_shape [" << frame_metadata->frame_shape[0] << ", " << frame_metadata->frame_shape[1] << "]"; + cout << " and endianness " << frame_metadata->endianness; + cout << " and type " << frame_metadata->type; + cout << " and frame_bytes_size " << frame_metadata->frame_bytes_size; + cout << "." << endl; + #endif + + // Commit the frame to the buffer. + ring_buffer.write(frame_metadata, frame_data); + + manager.received_frame(frame_metadata->frame_index); + } + + #ifdef DEBUG_OUTPUT + cout << "[ProcessManager::receive_zmq] Receiver thread stopped." << endl; + #endif +} + void ProcessManager::write_h5(WriterManager& manager, const H5Format& format, RingBuffer& ring_buffer) { H5Writer writer(manager.get_output_file()); @@ -47,7 +121,7 @@ void ProcessManager::write_h5(WriterManager& manager, const H5Format& format, Ri if (writer.is_file_open()) { #ifdef DEBUG_OUTPUT - cout << "[h5_zmq_writer::write] Writing file format." << endl; + cout << "[ProcessManager::write] Writing file format." << endl; #endif // Wait until all parameters are set or writer is killed. @@ -63,96 +137,21 @@ void ProcessManager::write_h5(WriterManager& manager, const H5Format& format, Ri try { H5FormatUtils::write_format(writer.get_h5_file(), format, parameters); } catch (const runtime_error& ex) { - cerr << "[h5_zmq_writer::write] Error while trying to write file format: "<< ex.what() << endl; + cerr << "[ProcessManager::write] Error while trying to write file format: "<< ex.what() << endl; } } } #ifdef DEBUG_OUTPUT - cout << "[h5_zmq_writer::write] Closing file " << manager.get_output_file() << endl; + cout << "[ProcessManager::write] Closing file " << manager.get_output_file() << endl; #endif writer.close_file(); #ifdef DEBUG_OUTPUT - cout << "[h5_zmq_writer::write] Writer thread stopped." << endl; + cout << "[ProcessManager::write] Writer thread stopped." << endl; #endif // Exit when writer thread has closed the file. exit(0); -} - -void ProcessManager::receive_zmq(WriterManager& manager, RingBuffer& ring_buffer, - ZmqReceiver& receiver, const H5Format& format) -{ - const auto& header_value_type = format.get_header_value_type(); - - while (manager.is_running()) { - - auto frame = receiver.receive(); - - // In case no message is available before the timeout, both pointers are NULL. - if (!frame.first){ - continue; - } - - auto frame_metadata = frame.first; - auto frame_data = frame.second; - - #ifdef DEBUG_OUTPUT - cout << "[h5_zmq_writer::receive_zmq] Processing FrameMetadata"; - cout << " with frame_index " << frame_metadata->frame_index; - cout << " and frame_shape [" << frame_metadata->frame_shape[0] << ", " << frame_metadata->frame_shape[1] << "]"; - cout << " and endianness " << frame_metadata->endianness; - cout << " and type " << frame_metadata->type; - cout << " and frame_bytes_size " << frame_metadata->frame_bytes_size; - cout << "." << endl; - #endif - - // Commit the frame to the buffer. - ring_buffer.write(frame_metadata, frame_data); - - manager.received_frame(frame_metadata->frame_index); - } - - #ifdef DEBUG_OUTPUT - cout << "[h5_zmq_writer::receive_zmq] Receiver thread stopped." << endl; - #endif -} - - - -void ProcessManager::run_writer(WriterManager& manager, const H5Format& format, - ZmqReceiver& receiver, uint16_t rest_port) -{ - size_t n_slots = config::ring_buffer_n_slots; - RingBuffer ring_buffer(n_slots); - - #ifdef DEBUG_OUTPUT - cout << "[h5_zmq_writer::run_writer] Running writer"; - cout << " and output_file " << manager.get_output_file(); - cout << " and n_slots " << n_slots; - cout << endl; - #endif - - boost::thread receiver_thread(receive_zmq, boost::ref(manager), boost::ref(ring_buffer), - boost::ref(receiver), boost::ref(format)); - boost::thread writer_thread(write_h5, boost::ref(manager), - boost::ref(format), boost::ref(ring_buffer)); - - RestApi::start_rest_api(manager, rest_port); - - #ifdef DEBUG_OUTPUT - cout << "[h5_zmq_writer::run_writer] Rest API stopped." << endl; - #endif - - // In case SIGINT stopped the rest_api. - manager.stop(); - - receiver_thread.join(); - writer_thread.join(); - - #ifdef DEBUG_OUTPUT - cout << "[h5_zmq_writer::run_writer] Writer properly stopped." << endl; - #endif } \ No newline at end of file