ZMQConfirmedInput Class Reference

Input part of a confirmed connection. More...

#include <ZMQConfirmedConnection.h>

Inheritance diagram for ZMQConfirmedInput:

Collaboration diagram for ZMQConfirmedInput:

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
	ZMQConfirmedInput (const std::string &inputAddress, const std::shared_ptr< ZMQParent > &parent)
	Create a new confirmed output by binding to the address.
std::unique_ptr< ZMQIdMessage >	handleIncomingData ()
	Block until a message can be received from one of the inputs. More...
void	clear ()
	Reset the counters for all received stop and terminate messages. Should be called on run start.
std::unique_ptr< ZMQIdMessage >	overwriteStopMessage ()
	Manually overwrite the stop message counter and set it to have all stop messages received.
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
std::set< std::string >	m_receivedStopMessages
	The set of input identities which have already sent a stop message.
bool	m_allStopMessages = false
	Have we received all stop messages?
std::set< std::string >	m_receivedTerminateMessages
	The set of input identities which have already sent a terminate message.
bool	m_allTerminateMessages = false
	Have we received all terminante messages?
std::set< std::string >	m_registeredWorkersInput
	The set of all registered inputs.
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 part of a confirmed connection.

In a confirmed connection every message sent by the input to the output is confirmed by a confirmation message by the output. The output can handle multiple inputs, each input can only be connected to a single output.

The input part keeps track of the registered workers and if it has received all stop and/or terminate messages. Stop/Terminate messages are only passed on if received from all registered workers.

If the input receives a message from one of the outputs, it sends back a confirmation message. Depending on the message type, it does several things:

• if the message is a hello message (the input registered itself), it adds the input identity to the registered workers. Nothing is returned to the caller in this case.
• if the message is a delete message (someone unregistered a worker), it removes the specified input identity in the message from the registered workers. Nothing is returned to the caller in this case. (corner case: if it turns out the worker to be unregistered was the last one missing for a full stop or terminate message from all workers, this message is passed on)
• if the message is a stop message, add it to the received stop messages. Only if all registered workers have send a stop message pass it on.
• same for terminate messages
• in all other cases just pass on the message.

Internally, a ZMQ_ROUTER is used in bind-mode. The order of starting output and input does not play any role.

Definition at line 65 of file ZMQConfirmedConnection.h.

Member Function Documentation

handleIncomingData()

std::unique_ptr< ZMQIdMessage > handleIncomingData

(

)

Block until a message can be received from one of the inputs.

React as described in the general description of this class. Only if a message is to be passed on actually returns a message (in all other cases a nullptr).

Definition at line 50 of file ZMQConfirmedConnection.cc.

{
  auto message = ZMQMessageFactory::fromSocket<ZMQIdMessage>(m_socket);
  const auto fromIdentity = message->getIdentity();
 
  logTime("last_received_message");
  increment("total_number_messages");
  increment("total_number_messages_from[" + fromIdentity + "]");
 
  auto confirmMessage = ZMQMessageFactory::createMessage(fromIdentity, EMessageTypes::c_confirmMessage);
  ZMQParent::send(m_socket, std::move(confirmMessage));
 
  if (message->isMessage(EMessageTypes::c_helloMessage)) {
    // a hello message makes us register the worker identity - which is the identity of the sender of the message
    m_registeredWorkersInput.emplace(fromIdentity);
    log("registered_workers", static_cast<long>(m_registeredWorkersInput.size()));
    increment("hello_messages");
    increment("hello_messages_from[" + fromIdentity + "]");
    return {};
  } else if (message->isMessage(EMessageTypes::c_deleteWorkerMessage)) {
    // a delete message makes us forget about the worker identity. The identity is taken from the message data
    // making it possible to delete other workers.
    B2DEBUG(10, "Got message from " << message->getIdentity() << " to kill " << message->getMessagePartAsString<2>());
    const std::string& killedIdentity = message->getMessagePartAsString<2>();
    m_registeredWorkersInput.erase(killedIdentity);
 
    log("registered_workers", static_cast<long>(m_registeredWorkersInput.size()));
    increment("dead_workers");
    increment("dead_worker_messaged_from[" + fromIdentity + "]");
 
    if (m_registeredWorkersInput.empty()) {
      B2ERROR("There is not a single worker registered anymore!");
      return {};
    }
 
    // Corner case: could be that this was the one we were waiting for
    if (not m_allStopMessages and m_receivedStopMessages == m_registeredWorkersInput) {
      m_allStopMessages = true;
      log("all_stop_messages", static_cast<long>(m_allStopMessages));
      increment("sent_stop_messages");
      logTime("last_stop_sent");
 
      return ZMQMessageFactory::createMessage(killedIdentity, EMessageTypes::c_lastEventMessage);
    }
    // Corner case: could be that this was the one we were waiting for
    if (not m_allTerminateMessages and m_receivedTerminateMessages == m_registeredWorkersInput) {
      m_allTerminateMessages = true;
      log("all_terminate_messages", static_cast<long>(m_allTerminateMessages));
      increment("sent_terminate_messages");
      logTime("last_terminate_sent");
 
      return ZMQMessageFactory::createMessage(killedIdentity, EMessageTypes::c_terminateMessage);
    }
 
    return {};
  }
 
  B2ASSERT("Worker without proper registration!",
           m_registeredWorkersInput.find(fromIdentity) != m_registeredWorkersInput.end());
 
  if (message->isMessage(EMessageTypes::c_lastEventMessage)) {
    // Increment the stop messages
    m_receivedStopMessages.emplace(fromIdentity);
    log("received_stop_messages", static_cast<long>(m_receivedStopMessages.size()));
    increment("total_received_stop_messages");
 
    if (not m_allStopMessages and m_receivedStopMessages == m_registeredWorkersInput) {
      // But only return this if everyone has sent a stop message already
      m_allStopMessages = true;
      log("all_stop_messages", static_cast<long>(m_allStopMessages));
      increment("sent_stop_messages");
      logTime("last_stop_sent");
 
      return ZMQMessageFactory::createMessage(fromIdentity, EMessageTypes::c_lastEventMessage);
    }
 
    // Whatever we return here will be carried on to the application and eventually also to the output.
    // This means as we are not passing the stop message now, we return nothing.
    return {};
  } else if (message->isMessage(EMessageTypes::c_terminateMessage)) {
    // Increment the terminate messages
    m_receivedTerminateMessages.emplace(fromIdentity);
    log("received_terminate_messages", static_cast<long>(m_receivedTerminateMessages.size()));
    increment("total_received_terminate_messages");
 
    if (not m_allTerminateMessages and m_receivedTerminateMessages == m_registeredWorkersInput) {
      // But only return this if everyone has sent a terminate message already
      m_allTerminateMessages = true;
      log("all_terminate_messages", static_cast<long>(m_allTerminateMessages));
      increment("sent_terminate_messages");
      logTime("last_terminate_sent");
 
      return ZMQMessageFactory::createMessage(fromIdentity, EMessageTypes::c_terminateMessage);
    }
 
    // Whatever we return here will be carried on to the application and eventually also to the output.
    // This means as we are not passing the stop message now, we return nothing.
    return {};
  }
 
  if (m_allStopMessages) {
    B2ERROR("Received an event after having received stop messages from every worker. This is not a good sign! I will dismiss this event!");
    increment("received_messages_after_stop");
    return {};
  }
 
  // Now it can only be a plain normal data message, so just pass it on
  const auto dataSize = message->getDataMessage().size();
 
  average("data_size", dataSize);
  average("data_size_from[" + fromIdentity + "]", dataSize);
 
  increment("received_events");
  increment("received_events[" + fromIdentity + "]");
 
  timeit("event_rate");
  timeit<200>("event_rate_from[" + fromIdentity + "]");
 
  logTime("last_received_event_message");
 
  return message;
}

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.

{
  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;
    }
  }
}

---

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

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