Belle II Software  release-08-01-10
ZMQHistoServerToFileOutput Class Reference

Output histograms into a ROOT file and shared memory after merging. More...

#include <ZMQHistogramConnection.h>

Inheritance diagram for ZMQHistoServerToFileOutput:
Collaboration diagram for ZMQHistoServerToFileOutput:

Public Types

using ReactorFunction = std::function< void(void)>
 Typedef of a function which will be called if a connection has a message.
 

Public Member Functions

 ZMQHistoServerToFileOutput (unsigned int maximalUncompressedBufferSize, const std::string &dqmFileName, const std::string &rootFileName)
 Create a new connection initializing the DQMMemFile.
 
void mergeAndSend (const std::map< std::string, HistogramMapping > &storedMessages, const std::optional< unsigned int > &experiment, const std::optional< unsigned int > &run, EMessageTypes messageType)
 Merge the given histograms into a single set of histograms and store them to file/shm.
 
void clear ()
 Clear the shared memory.
 
std::vector< zmq::socket_t * > getSockets () const final
 No sockets to poll on, so return an empty list.
 
void handleIncomingData ()
 There should be never incoming data, so raise an exception if called anyways.
 
std::string getEndPoint () const
 Return the connection string.
 
virtual bool isReady () const
 Return true of this connection is able to send messages right now. Can be overloaded in derived classes.
 
virtual std::string getMonitoringJSON () const
 Convert the stored monitoring values to a JSON string ready for sending out via a message.
 
template<class AClass >
void log (const std::string &key, const AClass &value)
 Store a value under a certain key. Different types of values can be stored, namely long, double or string. Mixtures are not allowed for a given key.
 
void increment (const std::string &key)
 Increment the value with the given key (only numerical values). If not present, set to 1.
 
void decrement (const std::string &key)
 Decrement the value with the given key (only numerical values). If not present, set to -1.
 
template<size_t MAX_SIZE = 100>
void average (const std::string &key, double value)
 Instead of storeing the double value directly under the given key, store the average of the last MAX_SIZE values.
 
template<size_t AVERAGE_SIZE = 2000>
void timeit (const std::string &key)
 Measure the rate of calls with the same key every AVERAGE_SIZE calls (and also display the last time AVERAGE_SIZE was reached under <key>_last_measurement)
 
void logTime (const std::string &key)
 Store the current time as a string under the given key.
 

Static Public Member Functions

static bool poll (const std::map< const ZMQConnection *, ReactorFunction > &connectionList, int timeout)
 Poll on the given connections and call the attached function if a messages comes in. More...
 
static bool hasMessage (const ZMQConnection *connection)
 Check if the given connection as an incoming message (right now, no waiting).
 

Private Attributes

DqmSharedMemm_sharedMemory = nullptr
 The SHM file. Please note that we do not call its destructor on purpose.
 
std::string m_dqmMemFileName
 Name of the shared memory.
 
std::string m_rootFileName
 Output file name (possible with placeholders)
 
std::map< std::string, std::variant< long, double, std::string > > m_monitoring
 Internal storage of all stored values.
 
std::unordered_map< std::string, std::tuple< std::vector< double >, size_t > > m_averages
 Internal storage of the previous values when calculating averages.
 
std::unordered_map< std::string, std::tuple< unsigned long, std::chrono::system_clock::time_point > > m_timeCounters
 Internal storage how often the timeit function for a given key was called and when it has last reached MAX_SIZE.
 

Detailed Description

Output histograms into a ROOT file and shared memory after merging.

This "connection" class needs to be wrapped with a ZMQHistogramOutput before using it as a real output connection (which adds the common functionalities).

The ROOT and shared memory are written using ROOTs own serialization. The shared memory is herby filled with the content of a temporary TMemFile. On clear the shared memory is cleared (ROOT files stay what they are).

Nothing special happens on stop or terminate other than that a merge is performed.

The given file path for the root file can include placeholders "{experiment_number}" and "{run_number}" which will be replaced accordingly. If no root file output name or no shared memory file for the DQMMemFile is given, this feature is not used.

Please note that we do not call the destructor of the shared memory on purpose as this leads to problems.

Definition at line 51 of file ZMQHistogramConnection.h.

Member Function Documentation

◆ poll()

bool poll ( const std::map< const ZMQConnection *, ReactorFunction > &  connectionList,
int  timeout 
)
staticinherited

Poll on the given connections and call the attached function if a messages comes in.

If after timeout milliseconds still no message is received, return anyways. If timeout is 0, do not wait. If timeout is -1, wait infinitely.

Returns true if a message was received on any socket, false otherwise. Attention: in case of an interrupted system call (e.g. because a signal was received) the function might return anyways with a negative result even before the timeout!

Definition at line 27 of file ZMQConnection.cc.

28 {
29  std::vector<const ReactorFunction*> socketMapping;
30  std::vector<zmq::pollitem_t> pollItems;
31 
32  // zmq needs a special format for its polling, so create it here.
33  for (const auto& [connection, function] : connectionList) {
34  auto sockets = connection->getSockets();
35  for (zmq::socket_t* socket : sockets) {
36  zmq::pollitem_t pollItem;
37  pollItem.socket = static_cast<void*>(*socket);
38  pollItem.events = ZMQ_POLLIN;
39  pollItem.revents = 0;
40  pollItems.push_back(std::move(pollItem));
41 
42  // but keep reference to the original function, so we can call the correct one later
43  socketMapping.push_back(&function);
44  }
45  }
46 
47  if (pollItems.empty()) {
48  return false;
49  }
50 
51  try {
52  zmq::poll(&pollItems[0], pollItems.size(), timeout);
53 
54  bool anySocket = false;
55  unsigned int counter = 0;
56  for (const auto& pollItem : pollItems) {
57  if (pollItem.revents & ZMQ_POLLIN) {
58  anySocket = true;
59  const auto* functionPtr = socketMapping.at(counter);
60  const auto function = *functionPtr;
61  function();
62  }
63  counter++;
64  }
65 
66  return anySocket;
67  } catch (zmq::error_t& error) {
68  if (error.num() == EINTR) {
69  // Could happen if there was an interrupt, return false so the caller knows the time did not pass already
70  return false;
71  } else {
72  // cannot handle, rethrow exception
73  throw;
74  }
75  }
76 }

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