ZMQHistoServerToZMQOutput Class Reference

Send the histograms as compressed byte stream via a ZMQConfirmedOutput connection after merging with all the properties of a normal confirmed output. More...

#include <ZMQHistogramConnection.h>

Inheritance diagram for ZMQHistoServerToZMQOutput:

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
	ZMQHistoServerToZMQOutput (const std::string &outputAddress, const std::shared_ptr< ZMQParent > &parent)
	Initialize the ZMQConfirmedOutput with the given address.
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 histograms and send them via the connection. Stop/Terminate messages are sent after that.
void	clear ()
	Nothing to do on clear.
std::string	getMonitoringJSON () const final
	The monitoring JSON is just passed from the ZMQConfirmedOutput.
void	handleIncomingData ()
	The input data handling is passed from the ZMQConfirmedOutput.
bool	isReady () const override
	The is ready is passed from the ZMQConfirmedOutput.
std::vector< zmq::socket_t * >	getSockets () const final
	The sockets are passed from ZMQConfirmedOutput.
std::string	getEndPoint () const
	Return the connection string.
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 storing 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.
static bool	hasMessage (const ZMQConnection *connection)
	Check if the given connection as an incoming message (right now, no waiting).

Private Attributes
ZMQConfirmedOutput	m_output
	The output connection used for sending the histograms.
std::vector< char >	m_outputBuffer
	Buffer used for compression.
unsigned int	m_maximalCompressedSize = 100'000'000
	Maximal size of the compression buffer.
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

Send the histograms as compressed byte stream via a ZMQConfirmedOutput connection after merging with all the properties of a normal confirmed output.

Stop and terminate messages are sent as normal (after performing a last merge and sending it). As additional message the event meta data is passed.

This connection allows to built hierarchies of histogram servers with ZMQ.

Definition at line 80 of file ZMQHistogramConnection.h.

Member Typedef Documentation

ReactorFunction

using ReactorFunction = std::function<void(void)>

inherited

Typedef of a function which will be called if a connection has a message.

Definition at line 34 of file ZMQConnection.h.

Constructor & Destructor Documentation

ZMQHistoServerToZMQOutput()

ZMQHistoServerToZMQOutput	(	const std::string &	outputAddress,
		const std::shared_ptr< ZMQParent > &	parent
	)

Initialize the ZMQConfirmedOutput with the given address.

Definition at line 112 of file ZMQHistogramConnection.cc.

                                                                                                                           :
  m_output(outputAddress, parent)
{
  m_output.log("histogram_merges", 0l);
  m_output.log("last_merged_histograms", 0l);
  m_output.log("average_merged_histograms", 0l);
  m_output.log("last_merge", "");
  m_output.log("size_before_compression", 0.0);
  m_output.log("size_after_compression", 0.0);
}

Member Function Documentation

clear()

void clear ( )

inline

Nothing to do on clear.

Definition at line 90 of file ZMQHistogramConnection.h.

{}

decrement()

void decrement ( const std::string &  key )

inherited

Decrement the value with the given key (only numerical values). If not present, set to -1.

Definition at line 37 of file ZMQLogger.cc.

{
  std::visit(Decrementor{}, m_monitoring[key]);
}

getEndPoint()

std::string getEndPoint ( ) const

inline

Return the connection string.

Definition at line 101 of file ZMQHistogramConnection.h.

{ return m_output.getEndPoint(); }

getMonitoringJSON()

std::string getMonitoringJSON ( ) const

inlinefinalvirtual

The monitoring JSON is just passed from the ZMQConfirmedOutput.

Reimplemented from ZMQLogger.

Definition at line 93 of file ZMQHistogramConnection.h.

{ return m_output.getMonitoringJSON(); }

getSockets()

std::vector< zmq::socket_t * > getSockets ( ) const

inlinefinalvirtual

The sockets are passed from ZMQConfirmedOutput.

Implements ZMQConnection.

Definition at line 99 of file ZMQHistogramConnection.h.

{ return m_output.getSockets(); }

handleIncomingData()

void handleIncomingData ( )

inline

The input data handling is passed from the ZMQConfirmedOutput.

Definition at line 95 of file ZMQHistogramConnection.h.

{ m_output.handleIncomingData(); }

hasMessage()

bool hasMessage ( const ZMQConnection *  connection )

staticinherited

Check if the given connection as an incoming message (right now, no waiting).

Definition at line 20 of file ZMQConnection.cc.

{
  // Just poll with 0 timeout and no reaction function. Hacky trick to reduce code duplication
  const auto emptyFunction = []() {};
  return ZMQConnection::poll({{connection, emptyFunction}}, 0);
}

increment()

void increment ( const std::string &  key )

inherited

Increment the value with the given key (only numerical values). If not present, set to 1.

Definition at line 32 of file ZMQLogger.cc.

{
  std::visit(Incrementor{}, m_monitoring[key]);
}

isReady()

bool isReady ( ) const

inlineoverridevirtual

The is ready is passed from the ZMQConfirmedOutput.

Reimplemented from ZMQConnection.

Definition at line 97 of file ZMQHistogramConnection.h.

{ return m_output.isReady(); }

logTime()

void logTime ( const std::string &  key )

inherited

Store the current time as a string under the given key.

Definition at line 42 of file ZMQLogger.cc.

{
  auto current = std::chrono::system_clock::now();
  auto displayTime = std::chrono::system_clock::to_time_t(current);
  log(key, std::ctime(&displayTime));
}

mergeAndSend()

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 histograms and send them via the connection. Stop/Terminate messages are sent after that.

Definition at line 123 of file ZMQHistogramConnection.cc.

{
  if (messageType == EMessageTypes::c_lastEventMessage) {
    // merge one last time
    mergeAndSend(storedMessages, experiment, run, EMessageTypes::c_eventMessage);
    // and then send out a stop signal by ourself
    auto message = ZMQMessageFactory::createMessage(EMessageTypes::c_lastEventMessage);
    m_output.handleEvent(std::move(message));
    return;
  } else if (messageType == EMessageTypes::c_terminateMessage) {
    // merge one last time
    mergeAndSend(storedMessages, experiment, run, EMessageTypes::c_eventMessage);
    // and send out a terminate message
    auto message = ZMQMessageFactory::createMessage(EMessageTypes::c_terminateMessage);
    m_output.handleEvent(std::move(message));
    return;
  }
 
  B2ASSERT("This should be an event message!", messageType == EMessageTypes::c_eventMessage);
 
  // Makes no sense to send out an empty event
  if (storedMessages.empty()) {
    return;
  }
 
  B2ASSERT("Experiment and run must be set at this stage", experiment and run);
 
  m_output.increment("histogram_merges");
 
  HistogramMapping mergeHistograms;
 
  m_output.log("last_merged_histograms", static_cast<long>(storedMessages.size()));
  m_output.average("average_merged_histograms", static_cast<double>(storedMessages.size()));
  m_output.logTime("last_merge");
 
  for (const auto& keyValue : storedMessages) {
    const auto& histogram = keyValue.second;
    mergeHistograms += histogram;
  }
 
  auto eventMessage = mergeHistograms.toMessage();
 
  if (m_outputBuffer.empty()) {
    m_outputBuffer.resize(m_maximalCompressedSize, 0);
  }
 
  m_output.average("size_before_compression", eventMessage->size());
  int size = m_maximalCompressedSize;
  size = LZ4_compress_default(eventMessage->buffer(), &m_outputBuffer[0], eventMessage->size(), size);
  B2ASSERT("Compression failed", size > 0);
  m_output.average("size_after_compression", size);
 
  zmq::message_t message(&m_outputBuffer[0], size);
 
  EventMetaData eventMetaData(0, *run, *experiment);
  auto eventInformationString = TBufferJSON::ToJSON(&eventMetaData);
  zmq::message_t additionalEventMessage(eventInformationString.Data(), eventInformationString.Length());
 
  auto zmqMessage = ZMQMessageFactory::createMessage(EMessageTypes::c_compressedDataMessage, std::move(message),
                                                     std::move(additionalEventMessage));
  m_output.handleEvent(std::move(zmqMessage), true, 20000);
}

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.

{
  std::vector<const ReactorFunction*> socketMapping;
  std::vector<zmq::pollitem_t> pollItems;
 
  // zmq needs a special format for its polling, so create it here.
  for (const auto& [connection, function] : connectionList) {
    auto sockets = connection->getSockets();
    for (zmq::socket_t* socket : sockets) {
      zmq::pollitem_t pollItem;
      pollItem.socket = static_cast<void*>(*socket);
      pollItem.events = ZMQ_POLLIN;
      pollItem.revents = 0;
      pollItems.push_back(std::move(pollItem));
 
      // but keep reference to the original function, so we can call the correct one later
      socketMapping.push_back(&function);
    }
  }
 
  if (pollItems.empty()) {
    return false;
  }
 
  try {
    zmq::poll(&pollItems[0], pollItems.size(), timeout);
 
    bool anySocket = false;
    unsigned int counter = 0;
    for (const auto& pollItem : pollItems) {
      if (pollItem.revents & ZMQ_POLLIN) {
        anySocket = true;
        const auto* functionPtr = socketMapping.at(counter);
        const auto function = *functionPtr;
        function();
      }
      counter++;
    }
 
    return anySocket;
  } catch (zmq::error_t& error) {
    if (error.num() == EINTR) {
      // Could happen if there was an interrupt, return false so the caller knows the time did not pass already
      return false;
    } else {
      // cannot handle, rethrow exception
      throw;
    }
  }
}

Member Data Documentation

m_averages

std::unordered_map<std::string, std::tuple<std::vector<double>, size_t> > m_averages

privateinherited

Internal storage of the previous values when calculating averages.

Definition at line 60 of file ZMQLogger.h.

m_maximalCompressedSize

unsigned int m_maximalCompressedSize = 100'000'000

private

Maximal size of the compression buffer.

Definition at line 109 of file ZMQHistogramConnection.h.

m_monitoring

std::map<std::string, std::variant<long, double, std::string> > m_monitoring

privateinherited

Internal storage of all stored values.

Definition at line 58 of file ZMQLogger.h.

m_output

ZMQConfirmedOutput m_output

private

The output connection used for sending the histograms.

Definition at line 104 of file ZMQHistogramConnection.h.

m_outputBuffer

std::vector<char> m_outputBuffer

private

Buffer used for compression.

Definition at line 107 of file ZMQHistogramConnection.h.

m_timeCounters

std::unordered_map<std::string, std::tuple<unsigned long, std::chrono::system_clock::time_point> > m_timeCounters

privateinherited

Internal storage how often the timeit function for a given key was called and when it has last reached MAX_SIZE.

Definition at line 62 of file ZMQLogger.h.

---

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

• daq/hbasf2/connections/include/ZMQHistogramConnection.h
• daq/hbasf2/connections/src/ZMQHistogramConnection.cc