Belle II Software development
ZMQHistogramToFileServer Class Reference

Final histogram app: receive histogram messages from all clients and react with a confirmation message. More...

#include <ZMQHistogramServer.h>

Inheritance diagram for ZMQHistogramToFileServer:
ZMQStandardApp< ZMQConfirmedInput, ZMQHistoServerToFile >

Public Member Functions

void initFromConsole (const std::string &description, int argc, char *argv[])
 Should be called before the main() function to initialize the connections using the paremeters given on command line.
 
void main ()
 Start the main loop polling on the output and monitoring connections and eventually also on the input if the output is ready.
 

Protected Member Functions

void initialize () final
 Initialize the two connections using the command line arguments.
 
void addOptions (po::options_description &desc) final
 Add the parameters to the cmd line arguments.
 
void handleExternalSignal (EMessageTypes type) final
 Merge on stop (even if not all clients have stopped) or clear the counters on start from the monitoring connection.
 
void handleInput () final
 Pass the message from the input connection to the histogram storage.
 
void handleTimeout () final
 Call the mergeAndSend function on timeout.
 
virtual void handleOutput ()
 Will get called for every message on the output connection. Can be overridden in a derived class. Calls handleIncomingData by default.
 
virtual void fillMonitoringJSON (std::stringstream &buffer) const
 Using the connections, fill up a buffer with the content to be monitored.
 
bool terminated () const
 Check if the main loop will be exited on next occasion. Can be set via the "m_terminate" flag.
 
void resetTimer ()
 Helper function to reset the start time and the remaining time.
 
void pollEvent (bool pollOnInput)
 Poll until a single event is retreived.
 

Protected Attributes

std::shared_ptr< ZMQParentm_parent
 Pointer to the ZMQParent to be used as base for all connections.
 
std::unique_ptr< ZMQConfirmedInputm_input
 Pointer to the input connection. Should be set in initialize.
 
std::unique_ptr< ZMQHistoServerToFilem_output
 Pointer to the output connection. Should be set in initialize.
 
std::unique_ptr< ZMQSimpleConnectionm_monitor
 Pointer to the monitoring connection. Should be set in initialize.
 
bool m_terminate
 Can be set by functions to terminate the main loop at the next possibility.
 
unsigned int m_timeout
 If set to a value != 0, will call handleTimeout with this frequency (in seconds).
 
bool m_monitorHasPriority
 Flag to break out of the polling loop to check for monitoring messages. Except for the finalcollector you probably do not want this.
 

Private Member Functions

bool checkTimer ()
 Helper function to check, if the timeout should happen.
 
void updateTimer ()
 Helper function to update the remaining time.
 
void handleMonitoring ()
 Handle an incoming message on the monitoring socket by either calling handleExternalSignal() or by passing on the monitor JSONs of the connections.
 

Private Attributes

std::string m_inputAddress
 Parameter: input address.
 
std::string m_sharedMemoryName = ""
 Parameter: name of the shared memory (or empty)
 
std::string m_rootFileName
 Parameter: name of the root file - can include {run_number} or {experiment_number}.
 
unsigned int m_maximalUncompressedBufferSize = 128'000'000
 Parameter: size of the temporary internal compression buffer.
 
HLTMainLoop m_mainLoop
 Internal signal handler.
 
std::string m_monitoringAddress
 Storage for the monitoring address for the cmd arguments.
 
int m_remainingTime
 Counter for the remaining time until a timeout happens.
 
std::chrono::system_clock::time_point m_start
 Start time for the timeout.
 

Detailed Description

Final histogram app: receive histogram messages from all clients and react with a confirmation message.

Periodically or on stop/terminate, merge them and write them out to a ROOT file and optionally to a memory file in shared memory. Apart from the connection-typical behavior, it reacts on termination messages by terminating. Behaves as a collector: When receiving a stop on monitoring, it merges the histograms (even if not all clients have sent a stop already). When receiving a start on monitoring, it clears the input and output stop counters and the stored histograms.

Definition at line 30 of file ZMQHistogramServer.h.

Member Function Documentation

◆ addOptions()

void addOptions ( po::options_description &  desc)
finalprotectedvirtual

Add the parameters to the cmd line arguments.

Reimplemented from ZMQStandardApp< ZMQConfirmedInput, ZMQHistoServerToFile >.

Definition at line 22 of file ZMQHistogramServer.cc.

23{
24 m_timeout = 30;
26 desc.add_options()
27 ("input", boost::program_options::value<std::string>(&m_inputAddress)->required(),
28 "where to read the events from")
29 ("sharedMemoryName",
30 boost::program_options::value<std::string>(&m_sharedMemoryName)->default_value(m_sharedMemoryName),
31 "name of the shared memory")
32 ("rootFileName", boost::program_options::value<std::string>(&m_rootFileName)->required(),
33 "name of the ROOT file - can include {run_number} or {experiment_number}")
34 ("timeout", boost::program_options::value<unsigned int>(&m_timeout)->default_value(m_timeout),
35 "how many seconds to wait between histogram dumps")
36 ("maximalUncompressedBufferSize",
37 boost::program_options::value<unsigned int>(&m_maximalUncompressedBufferSize)->default_value(
39 "size of the uncompress buffer");
40}
std::string m_rootFileName
Parameter: name of the root file - can include {run_number} or {experiment_number}.
std::string m_sharedMemoryName
Parameter: name of the shared memory (or empty)
std::string m_inputAddress
Parameter: input address.
unsigned int m_maximalUncompressedBufferSize
Parameter: size of the temporary internal compression buffer.
unsigned int m_timeout
If set to a value != 0, will call handleTimeout with this frequency (in seconds).
Definition: ZMQApp.h:75
virtual void addOptions(po::options_description &desc)
Override in a derived class to add the command line arguments. Do not forget to call this base functi...

◆ checkTimer()

bool checkTimer
privateinherited

Helper function to check, if the timeout should happen.

Definition at line 111 of file ZMQApp.details.h.

196 {
197 updateTimer();
198 return m_remainingTime == 0;
199 }
int m_remainingTime
Counter for the remaining time until a timeout happens.
Definition: ZMQApp.h:106
void updateTimer()
Helper function to update the remaining time.

◆ fillMonitoringJSON()

void fillMonitoringJSON ( std::stringstream &  buffer) const
protectedvirtualinherited

Using the connections, fill up a buffer with the content to be monitored.

Definition at line 92 of file ZMQApp.details.h.

257 {
258 buffer << "{" << std::endl;
259 buffer << "\"monitor\": " << m_monitor->getMonitoringJSON() << "," << std::endl;
260 buffer << "\"input\": " << m_input->getMonitoringJSON() << "," << std::endl;
261 buffer << "\"output\": " << m_output->getMonitoringJSON() << std::endl;
262 buffer << "}" << std::endl;
263 }
std::unique_ptr< ZMQHistoServerToFile > m_output
Pointer to the output connection. Should be set in initialize.
Definition: ZMQApp.h:69
std::unique_ptr< ZMQSimpleConnection > m_monitor
Pointer to the monitoring connection. Should be set in initialize.
Definition: ZMQApp.h:71
std::unique_ptr< ZMQConfirmedInput > m_input
Pointer to the input connection. Should be set in initialize.
Definition: ZMQApp.h:67

◆ handleExternalSignal()

void handleExternalSignal ( EMessageTypes  type)
finalprotectedvirtual

Merge on stop (even if not all clients have stopped) or clear the counters on start from the monitoring connection.

Reimplemented from ZMQStandardApp< ZMQConfirmedInput, ZMQHistoServerToFile >.

Definition at line 42 of file ZMQHistogramServer.cc.

43{
44 if (type == EMessageTypes::c_newRunMessage) {
45 m_input->clear();
46 m_output->clear();
47 return;
48 } else if (type == EMessageTypes::c_lastEventMessage) {
49 auto message = m_input->overwriteStopMessage();
50 if (message) {
51 m_output->handleEvent(std::move(message));
52 }
53 return;
54 }
55}

◆ handleInput()

void handleInput ( )
finalprotectedvirtual

Pass the message from the input connection to the histogram storage.

Reimplemented from ZMQStandardApp< ZMQConfirmedInput, ZMQHistoServerToFile >.

Definition at line 57 of file ZMQHistogramServer.cc.

58{
59 auto message = m_input->handleIncomingData();
60 if (message) {
61 if (message->isMessage(EMessageTypes::c_terminateMessage)) {
62 m_terminate = true;
63 }
64
65 m_output->handleEvent(std::move(message));
66 }
67}
bool m_terminate
Can be set by functions to terminate the main loop at the next possibility.
Definition: ZMQApp.h:73

◆ handleMonitoring()

void handleMonitoring
privateinherited

Handle an incoming message on the monitoring socket by either calling handleExternalSignal() or by passing on the monitor JSONs of the connections.

Definition at line 115 of file ZMQApp.details.h.

233 {
234 auto monitoringMessage = m_monitor->handleIncomingData();
235
236 if (monitoringMessage->isMessage(EMessageTypes::c_newRunMessage)) {
237 handleExternalSignal(EMessageTypes::c_newRunMessage);
238 return;
239 } else if (monitoringMessage->isMessage(EMessageTypes::c_lastEventMessage)) {
240 handleExternalSignal(EMessageTypes::c_lastEventMessage);
241 return;
242 } else if (monitoringMessage->isMessage(EMessageTypes::c_terminateMessage)) {
243 handleExternalSignal(EMessageTypes::c_terminateMessage);
244 return;
245 }
246
247 std::stringstream buffer;
248 fillMonitoringJSON(buffer);
249
250 auto message = ZMQMessageFactory::createMessage(monitoringMessage->getIdentity(),
251 EMessageTypes::c_confirmMessage, buffer.str());
252 m_monitor->handleEvent(std::move(message));
253 }
virtual void fillMonitoringJSON(std::stringstream &buffer) const
Using the connections, fill up a buffer with the content to be monitored.
virtual void handleExternalSignal(EMessageTypes)
Will get called for every signal message on the monitoring connection. Can be overridden in a derived...

◆ handleOutput()

void handleOutput
protectedvirtualinherited

Will get called for every message on the output connection. Can be overridden in a derived class. Calls handleIncomingData by default.

Definition at line 88 of file ZMQApp.details.h.

179 {
180 m_output->handleIncomingData();
181 }

◆ handleTimeout()

void handleTimeout ( )
finalprotectedvirtual

Call the mergeAndSend function on timeout.

Reimplemented from ZMQStandardApp< ZMQConfirmedInput, ZMQHistoServerToFile >.

Definition at line 69 of file ZMQHistogramServer.cc.

70{
71 m_output->mergeAndSend();
72}

◆ initFromConsole()

void initFromConsole ( const std::string &  description,
int  argc,
char *  argv[] 
)
inherited

Should be called before the main() function to initialize the connections using the paremeters given on command line.

Custom implementations should implement the addOptions function to pass the command line arguments correctly. Calls the initialize function (which should also be overridden).

Definition at line 54 of file ZMQApp.details.h.

22 {
23
24 po::options_description desc(description);
25 std::string connection_file;
26 int debugLevel(0);
27 desc.add_options()
28 ("connection-file", boost::program_options::value<std::string>(&connection_file),
29 "if given print the connection information for input/output and monitoring socket to the given filename "
30 "in json format")
31 ("debug", boost::program_options::value<int>(&debugLevel), "Enable debug logging");
32 addOptions(desc);
33
34 po::positional_options_description p;
35
36 po::variables_map vm;
37 try {
38 po::store(po::command_line_parser(argc, argv).options(desc).positional(p).run(), vm);
39 } catch (std::exception& e) {
40 B2FATAL(e.what());
41 }
42
43 if (vm.count("help")) {
44 std::cout << desc << std::endl;
45 exit(1);
46 }
47
48 try {
49 po::notify(vm);
50 } catch (std::exception& e) {
51 B2FATAL(e.what());
52 }
53
54 if (debugLevel > 0) {
55 auto& logging = LogSystem::Instance();
56 logging.getLogConfig()->setLogLevel(LogConfig::c_Debug);
57 logging.getLogConfig()->setDebugLevel(debugLevel);
58 B2DEBUG(1, "Enabled debug logging");
59 }
60
61 initialize();
62
63 if (not connection_file.empty()) {
64 B2DEBUG(1, "Write connection file" << LogVar("connection_file", connection_file));
65 nlohmann::json json;
66 try {
67 json["input"] = m_input->getEndPoint();
68 } catch (zmq::error_t& e) {
69 B2WARNING(e.what());
70 }
71 try {
72 json["output"] = m_output->getEndPoint();
73 } catch (zmq::error_t& e) {
74 B2WARNING(e.what());
75 }
76 std::ofstream connections(connection_file, std::ofstream::trunc);
77 if (!connections) {
78 B2FATAL("Cannot write connection file" << LogVar("connection_file", connection_file));
79 }
80 connections << std::setw(4) << json << std::endl;
81 }
82 }
Class to store variables with their name which were sent to the logging service.
virtual void initialize()
Override in a derived class to initialize the connections from the given command line arguments....

◆ initialize()

void initialize ( )
finalprotectedvirtual

Initialize the two connections using the command line arguments.

Reimplemented from ZMQStandardApp< ZMQConfirmedInput, ZMQHistoServerToFile >.

Definition at line 14 of file ZMQHistogramServer.cc.

15{
18 m_output.reset(
20}
Input part of a confirmed connection.
Add the common functionality to the histogram output classes.
std::shared_ptr< ZMQParent > m_parent
Pointer to the ZMQParent to be used as base for all connections.
Definition: ZMQApp.h:65

◆ main()

void main
inherited

Start the main loop polling on the output and monitoring connections and eventually also on the input if the output is ready.

Calls the functions handleExternalSignal, handleInput, handleOutput and handleTimeout as described in the documentation of this class.

Definition at line 61 of file ZMQApp.details.h.

86 {
87 resetTimer();
88
89 while (not terminated()) {
90 if (not m_output->isReady()) {
91 // if the output is not ready, we can not sent anything. So lets just poll on output
92 // and monitoring until it becomes ready
93 m_monitor->log("output_state", "not_ready");
94 pollEvent(false);
95 } else {
96 // if it is ready, we can also include the input socket as long as output stays ready
97 m_monitor->log("output_state", "ready");
98 pollEvent(true);
99 }
100 }
101 }
void pollEvent(bool pollOnInput)
Poll until a single event is retreived.
bool terminated() const
Check if the main loop will be exited on next occasion. Can be set via the "m_terminate" flag.
void resetTimer()
Helper function to reset the start time and the remaining time.

◆ pollEvent()

void pollEvent ( bool  pollOnInput)
protectedinherited

Poll until a single event is retreived.

Definition at line 98 of file ZMQApp.details.h.

105 {
106 auto reactToOutput = [this]() {
107 // Get all messages from output socket
108 while (ZMQConnection::hasMessage(m_output.get()) and not terminated()) {
109 handleOutput();
110 if (m_monitorHasPriority) {
111 break;
112 }
113 }
114 };
115
116 auto reactToMonitor = [this]() {
117 // Get all messages from monitoring socket
118 while (ZMQConnection::hasMessage(m_monitor.get()) and not terminated()) {
120 }
121 };
122
123 auto reactToInput = [this]() {
124 // Get all messages from input as long output is ready
125 while (ZMQConnection::hasMessage(m_input.get()) and m_output->isReady() and not terminated()) {
126 handleInput();
127 if (m_monitorHasPriority) {
128 break;
129 }
130 }
131 };
132
133 m_monitor->logTime("waiting_since");
134
135 if (pollOnInput) {
136 ZMQConnection::poll({{m_output.get(), reactToOutput}, {m_monitor.get(), reactToMonitor}, {m_input.get(), reactToInput}},
138 } else {
139 ZMQConnection::poll({{m_output.get(), reactToOutput}, {m_monitor.get(), reactToMonitor}}, m_remainingTime);
140 }
141
142 if (checkTimer() and not terminated()) {
143 B2ASSERT("There is no timeout set, but we still call the timeout() function? A bug!", m_timeout != 0);
144 m_monitor->increment("timeouts");
146 resetTimer();
147 } else {
148 updateTimer();
149 }
150 }
void handleMonitoring()
Handle an incoming message on the monitoring socket by either calling handleExternalSignal() or by pa...
virtual void handleInput()
Will get called for every message on the input connection. Can be overridden in a derived class....
virtual void handleOutput()
Will get called for every message on the output connection. Can be overridden in a derived class....
bool checkTimer()
Helper function to check, if the timeout should happen.
virtual void handleTimeout()
Will get called on a timeout. Can be overridden in a derived class. Empty by default.

◆ resetTimer()

void resetTimer
protectedinherited

Helper function to reset the start time and the remaining time.

Definition at line 96 of file ZMQApp.details.h.

220 {
221 // if there is no timeout, we should never set the remaining time
222 if (m_timeout == 0) {
223 m_remainingTime = -1;
224 return;
225 }
226
227 m_start = std::chrono::system_clock::now();
228 m_remainingTime = m_timeout * 1000;
229 }
std::chrono::system_clock::time_point m_start
Start time for the timeout.
Definition: ZMQApp.h:108

◆ terminated()

bool terminated
protectedinherited

Check if the main loop will be exited on next occasion. Can be set via the "m_terminate" flag.

Definition at line 94 of file ZMQApp.details.h.

190 {
191 return not m_mainLoop.isRunning() or m_terminate;
192 }

◆ updateTimer()

void updateTimer
privateinherited

Helper function to update the remaining time.

Definition at line 113 of file ZMQApp.details.h.

203 {
204 // if there is no timeout, we should never update the remaining time
205 if (m_timeout == 0) {
206 m_remainingTime = -1;
207 return;
208 }
209
210 auto currentTime = std::chrono::system_clock::now();
211 auto timeDifference = std::chrono::duration_cast<std::chrono::milliseconds>(currentTime - m_start);
212 m_remainingTime = m_timeout * 1000 - timeDifference.count();
213 if (m_remainingTime < 0) {
214 m_remainingTime = 0;
215 }
216 }

Member Data Documentation

◆ m_input

std::unique_ptr<ZMQConfirmedInput > m_input
protectedinherited

Pointer to the input connection. Should be set in initialize.

Definition at line 67 of file ZMQApp.h.

◆ m_inputAddress

std::string m_inputAddress
private

Parameter: input address.

Definition at line 45 of file ZMQHistogramServer.h.

◆ m_mainLoop

HLTMainLoop m_mainLoop
privateinherited

Internal signal handler.

Definition at line 102 of file ZMQApp.h.

◆ m_maximalUncompressedBufferSize

unsigned int m_maximalUncompressedBufferSize = 128'000'000
private

Parameter: size of the temporary internal compression buffer.

Definition at line 51 of file ZMQHistogramServer.h.

◆ m_monitor

std::unique_ptr<ZMQSimpleConnection> m_monitor
protectedinherited

Pointer to the monitoring connection. Should be set in initialize.

Definition at line 71 of file ZMQApp.h.

◆ m_monitorHasPriority

bool m_monitorHasPriority
protectedinherited

Flag to break out of the polling loop to check for monitoring messages. Except for the finalcollector you probably do not want this.

Definition at line 77 of file ZMQApp.h.

◆ m_monitoringAddress

std::string m_monitoringAddress
privateinherited

Storage for the monitoring address for the cmd arguments.

Definition at line 104 of file ZMQApp.h.

◆ m_output

std::unique_ptr<ZMQHistoServerToFile > m_output
protectedinherited

Pointer to the output connection. Should be set in initialize.

Definition at line 69 of file ZMQApp.h.

◆ m_parent

std::shared_ptr<ZMQParent> m_parent
protectedinherited

Pointer to the ZMQParent to be used as base for all connections.

Definition at line 65 of file ZMQApp.h.

◆ m_remainingTime

int m_remainingTime
privateinherited

Counter for the remaining time until a timeout happens.

Definition at line 106 of file ZMQApp.h.

◆ m_rootFileName

std::string m_rootFileName
private

Parameter: name of the root file - can include {run_number} or {experiment_number}.

Definition at line 49 of file ZMQHistogramServer.h.

◆ m_sharedMemoryName

std::string m_sharedMemoryName = ""
private

Parameter: name of the shared memory (or empty)

Definition at line 47 of file ZMQHistogramServer.h.

◆ m_start

std::chrono::system_clock::time_point m_start
privateinherited

Start time for the timeout.

Definition at line 108 of file ZMQApp.h.

◆ m_terminate

bool m_terminate
protectedinherited

Can be set by functions to terminate the main loop at the next possibility.

Definition at line 73 of file ZMQApp.h.

◆ m_timeout

unsigned int m_timeout
protectedinherited

If set to a value != 0, will call handleTimeout with this frequency (in seconds).

Definition at line 75 of file ZMQApp.h.


The documentation for this class was generated from the following files: