ZMQRawInput Class Reference

Input connection allowing to speak with non-zmq sockets via a ZMQ_STREAM socket. More...

#include <ZMQRawConnection.h>

Inheritance diagram for ZMQRawInput:

Collaboration diagram for ZMQRawInput:

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
	ZMQRawInput (const std::string &inputAddress, unsigned int maximalBufferSize, bool receiveEventMessages, const std::shared_ptr< ZMQParent > &parent)
	Create a new raw input connection. The bind or connect behavior is chosen according to the given address.
std::vector< zmq::message_t >	handleIncomingData ()
	Block until a TCP packet can be received from the socket. More...
void	clear ()
	Reset the internal buffer and counter.
std::vector< zmq::socket_t * >	getSockets () const final
	The socket used for polling is just the stored socket.
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).

Protected Attributes
std::shared_ptr< ZMQParent >	m_parent
	The shared ZMQParent instance.
std::unique_ptr< zmq::socket_t >	m_socket
	The memory of the socket. Needs to be initialized in a derived class.

Private Attributes
unsigned int	m_maximalBufferSize
	Parameter for the maximal buffer size. If this size is reached, a FATAL will be issued.
std::vector< char >	m_buffer
	Internal storage for the buffer.
size_t	m_writeAddress = 0
	Where in the buffer are we currently writing to.
unsigned int	m_currentSize = 0
	How large should the full message be? The information is from the first int of the message.
bool	m_receiveEventMessages
	Parameter to receive event messages (see above)
std::string	m_inputIdentity = ""
	Internal storage of the connected socket to check if we get messages from multiple ones.
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

Input connection allowing to speak with non-zmq sockets via a ZMQ_STREAM socket.

Can only speak with a single socket.

On receiving a TC packet, it is stored in an internal buffer. Only if the message is complete it is returned wrapped up as a zmq:message_t.

Two message formats are understood:

• if receiveEventMessages is set to false, the first int of the message must be the size L of the message, in units of sizeof(int) = words. (so full message size in bytes = L * sizeof(int)) Please note that this size also includes this first int, so the real data message has a size in bytes of L * sizeof(int) - sizeof(int)
• if it is true, the first int of the message must be the ntohl-converted size of the message in bytes without this first size int.

The two message formats are chosen to be compatible with the HLT implementation.

Definition at line 51 of file ZMQRawConnection.h.

Member Function Documentation

handleIncomingData()

std::vector< zmq::message_t > handleIncomingData

(

)

Block until a TCP packet can be received from the socket.

Returns only full messages as zmq:message_t. Please note that this can be

• 0 messages if there has not been a full message so far
• 1 message if there was
• more than 1 if the message size is smaller than the TCP package size (~8kB)

Definition at line 48 of file ZMQRawConnection.cc.

{
  std::vector<zmq::message_t> receivedMessages;
 
  // We will always get one or two parts. The first one is the identity..
  zmq::message_t identity;
  m_socket->recv(&identity);
  std::string identityString(identity.data<char>(), identity.size());
  B2ASSERT("The message is incomplete!", m_socket->getsockopt<int>(ZMQ_RCVMORE) == 1);
  B2ASSERT("The app can only handle a single connection!",
           m_inputIdentity == identityString or m_inputIdentity.empty());
 
  // ... and the second one is the message itself.
  const size_t remainingSpace = m_maximalBufferSize - m_writeAddress;
  const size_t receivedBytes = m_socket->recv(&m_buffer[m_writeAddress], remainingSpace);
  B2ASSERT("The message is longer than expected! Increase the buffer size.", m_socket->getsockopt<int>(ZMQ_RCVMORE) == 0);
  if (receivedBytes == 0) {
    // Empty message means the client connected or disconnected
    if (m_inputIdentity.empty()) {
      m_inputIdentity = identityString;
      log("socket_state", "connected");
      increment("socket_connects");
    } else {
      m_inputIdentity = "";
      log("socket_state", "disconnected");
      increment("socket_disconnects");
    }
    return receivedMessages;
  }
  // We can maximal write `remainingSpace` into the buffer. If the message was longer, ZMQ will just cut it.
  // This means we are loosing data and the buffer size should be increased.
  if (receivedBytes > remainingSpace) {
    B2FATAL("The size of the buffer is too small! " << receivedBytes << " > " << remainingSpace);
  }
  average("average_received_byte_packages", receivedBytes);
 
  // `m_writeAddress` always points to the index on where we will write next.
  // As we have written `receivedBytes` we need to advance
  m_writeAddress += receivedBytes;
 
  log("write_address", static_cast<long>(m_writeAddress));
 
  // If the current buffer is smaller than an int, we can not get the size
  while (m_writeAddress >= sizeof(int)) {
    if (m_currentSize == 0) {
      // we do not know the size of the data package already, so lets get it from the buffer.
      // It is always in the first sizeof(int) of the data
      memcpy(&m_currentSize, &m_buffer[0], sizeof(int));
 
      // Here the two different message formats differ
      if (m_receiveEventMessages) {
        m_currentSize = ntohl(m_currentSize);
      }
      B2ASSERT("Strange size in the data!", m_currentSize > 0);
      if (m_receiveEventMessages) {
        m_currentSize += sizeof(int);
      } else {
        m_currentSize *= sizeof(int);
      }
 
      log("current_size", static_cast<long>(m_currentSize));
    }
    if (m_writeAddress >= m_currentSize) {
      // Now we know the size already, and we have enough data received so we have actually the full
      // data. We can build the total message.
      average("data_size", m_currentSize);
      increment("received_events");
      timeit("event_rate");
 
      // Again, here the two different message formats differ. One includes the first int with the length...
      unsigned int startAddress = 0;
      if (m_receiveEventMessages) {
        // .. and the other does not
        startAddress = sizeof(int);
      }
      zmq::message_t dataMessage(&m_buffer[startAddress], static_cast<size_t>(m_currentSize - startAddress));
 
      memmove(&m_buffer[0], &m_buffer[m_currentSize], m_writeAddress - m_currentSize);
      m_writeAddress -= m_currentSize;
      m_currentSize = 0;
 
      log("write_address", static_cast<long>(m_writeAddress));
      log("current_size", static_cast<long>(m_currentSize));
 
      receivedMessages.push_back(std::move(dataMessage));
    } else {
      // We did not receive a full message up to now
      break;
    }
  }
 
  average("average_number_of_events_per_package", receivedMessages.size());
  return receivedMessages;
}

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 29 of file ZMQConnection.cc.

---

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

• framework/pcore/zmq/connections/include/ZMQRawConnection.h
• framework/pcore/zmq/connections/src/ZMQRawConnection.cc