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

Refactoring ProcessManager

Browse Source
This commit is contained in:
babic_a
2018-02-12 10:27:45 +01:00
parent 205770647a
commit 3a4167d6fc
1 changed files with 78 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ProcessManager.cpp
+78 -79
View File
@@ -7,9 +7,83 @@
#include <boost/thread.hpp>
#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
}