Belle II Software development
Public Member Functions | Protected Member Functions | Protected Attributes | Private Member Functions | Private Attributes | List of all members
ZMQHistogramToFileServer Class Reference
Packages » daq

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< ZMQParent > m_parent
 Pointer to the ZMQParent to be used as base for all connections.
 
std::unique_ptr< ZMQConfirmedInput > m_input
 Pointer to the input connection. Should be set in initialize.
 
std::unique_ptr< ZMQHistoServerToFile > m_output
 Pointer to the output connection. Should be set in initialize.
 
std::unique_ptr< ZMQSimpleConnection > m_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;
25 ZMQStandardApp::addOptions(desc);
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(
38 m_maximalUncompressedBufferSize),
39 "size of the uncompress buffer");
40}
Belle2::ZMQHistogramToFileServer::m_rootFileName
std::string m_rootFileName
Parameter: name of the root file - can include {run_number} or {experiment_number}...
Definition ZMQHistogramServer.h:49
Belle2::ZMQHistogramToFileServer::m_sharedMemoryName
std::string m_sharedMemoryName
Parameter: name of the shared memory (or empty)
Definition ZMQHistogramServer.h:47
Belle2::ZMQHistogramToFileServer::m_inputAddress
std::string m_inputAddress
Parameter: input address.
Definition ZMQHistogramServer.h:45
Belle2::ZMQHistogramToFileServer::m_maximalUncompressedBufferSize
unsigned int m_maximalUncompressedBufferSize
Parameter: size of the temporary internal compression buffer.
Definition ZMQHistogramServer.h:51
Belle2::ZMQStandardApp::addOptions
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...
Definition ZMQApp.details.h:160

◆ 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 }

◆ 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 }

◆ 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}
Belle2::ZMQConfirmedInput::clear
void clear()
Reset the counters for all received stop and terminate messages. Should be called on run start.
Definition ZMQConfirmedConnection.cc:179
Belle2::ZMQConfirmedInput::overwriteStopMessage
std::unique_ptr< ZMQIdMessage > overwriteStopMessage()
Manually overwrite the stop message counter and set it to have all stop messages received.
Definition ZMQConfirmedConnection.cc:196
Belle2::ZMQHistogramOutput::handleEvent
void handleEvent(std::unique_ptr< ZMQIdMessage > message)
Handle a new message to be "sent" (what this means is up to the base class) as described above.
Definition ZMQHistogramOutput.cc:24
Belle2::ZMQHistogramOutput::clear
void clear()
Forward a clear call to the base class and clear the stored messages. Should be called on run start.
Definition ZMQHistogramOutput.cc:98
Belle2::ZMQStandardApp< ZMQConfirmedInput, ZMQHistoServerToFile >::m_output
std::unique_ptr< ZMQHistoServerToFile > m_output
Definition ZMQApp.h:69

◆ 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}
Belle2::ZMQConfirmedInput::handleIncomingData
std::unique_ptr< ZMQIdMessage > handleIncomingData()
Block until a message can be received from one of the inputs.
Definition ZMQConfirmedConnection.cc:47

◆ 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 }

◆ 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}
Belle2::ZMQHistogramOutput::mergeAndSend
void mergeAndSend(EMessageTypes messageType=EMessageTypes::c_eventMessage)
Forward a merge call to the base class handing over the stored messages.
Definition ZMQHistogramOutput.cc:92

◆ 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 }

◆ 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{
16 ZMQStandardApp::initialize();
17 m_input.reset(new ZMQConfirmedInput(m_inputAddress, m_parent));
18 m_output.reset(
19 new ZMQHistoServerToFile(m_maximalUncompressedBufferSize, m_sharedMemoryName, m_rootFileName));
20}
Belle2::ZMQStandardApp::initialize
virtual void initialize()
Override in a derived class to initialize the connections from the given command line arguments....
Definition ZMQApp.details.h:153

◆ 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 }

◆ 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()) {
119 handleMonitoring();
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}},
137 m_remainingTime);
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");
145 handleTimeout();
146 resetTimer();
147 } else {
148 updateTimer();
149 }
150 }

◆ 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 }

◆ terminated()

bool terminated ( ) const
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: