ZMQLoadBalancedOutput Class Reference

Output part of a load-balanced connection. More...

#include <ZMQLoadBalancedConnection.h>

Inheritance diagram for ZMQLoadBalancedOutput:

Collaboration diagram for ZMQLoadBalancedOutput:

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
	ZMQLoadBalancedOutput (const std::string &outputAddress, bool lax, const std::shared_ptr< ZMQParent > &parent)
	Create a new load-balanced output and bind to the given address.
void	handleEvent (std::unique_ptr< ZMQNoIdMessage > message)
	Send the given message (without identity) to the next input in the ready list. More...
void	handleIncomingData ()
	Block until a ready message from an input is received and add it to the ready queue.
void	clear ()
	Clear the counter for sent stop and terminate messages. Should be called on run start.
bool	isReady () const final
	If lax mode is disabled, the output is ready if at least a single input is ready. Else always.
std::vector< zmq::socket_t * >	getSockets () const final
	The socket used for polling is just the stored socket.
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::deque< std::string >	m_readyWorkers
	List of identities of ready inputs in LIFO order.
std::set< std::string >	m_allWorkers
	All ever registered inputs.
bool	m_sentStopMessages = false
	Did we already sent a stop message?
bool	m_sentTerminateMessages = false
	Did we already sent a terminate message?
bool	m_lax = false
	Parameter to enable lax mode.
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::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 part of a load-balanced connection.

Multiple inputs can connect to this output. The output keeps a list of all received ready messages and sends events on requests always to the next input in the list, which creates some sort of load-balancing.

Stop and terminate messages are sent all all inputs that have sent at least a single ready so far. There is no unregistration happening, so dead inputs will still be served with both events and stop/terminate messages. For stop/terminate messages this is no problem and event messages will only be sent as long as there are ready messages.

After a stop message is sent all additional incoming events will be dismissed.

If no input has send any ready message, the output is not ready and needs to be polled for new ready messages. However, there is also a "lax" mode which makes the events being silently discarded if no input is ready.

Internally a ZMQ_ROUTER in bind mode is used.

Definition at line 81 of file ZMQLoadBalancedConnection.h.

Member Function Documentation

handleEvent()

void handleEvent

(

std::unique_ptr< ZMQNoIdMessage >

message

)

Send the given message (without identity) to the next input in the ready list.

If it is a stop or terminate message send the message to all inputs which have ever sent a ready message. If it is an event message, sent it only to the next ready input. If there is no ready input it either (a) throws and exception if lax mode is not enabled. Make sure to only sent events if the output is ready (which indicates exactly this: an input is ready). Or (b) discard the event without warning if lax mode is enabled.

Definition at line 83 of file ZMQLoadBalancedConnection.cc.

{
  if (message->isMessage(EMessageTypes::c_lastEventMessage)) {
    // Tell all workers to stop this run
    for (auto worker : m_allWorkers) {
      auto sendMessage = ZMQMessageFactory::createMessage(worker, EMessageTypes::c_lastEventMessage);
      ZMQParent::send(m_socket, std::move(sendMessage));
    }
    m_sentStopMessages = true;
    log("all_stop_messages", static_cast<long>(m_sentStopMessages));
    increment("sent_stop_messages");
    logTime("last_stop_sent");
    return;
  } else if (message->isMessage(EMessageTypes::c_terminateMessage)) {
    // Tell all workers to terminate
    for (auto worker : m_allWorkers) {
      auto sendMessage = ZMQMessageFactory::createMessage(worker, EMessageTypes::c_terminateMessage);
      ZMQParent::send(m_socket, std::move(sendMessage));
    }
    m_sentTerminateMessages = true;
    log("all_terminate_messages", static_cast<long>(m_sentTerminateMessages));
    increment("sent_terminate_messages");
    logTime("last_terminate_sent");
    return;
  }
 
  if (m_lax and m_readyWorkers.empty()) {
    // There is no one that can handle the event in the moment, dismiss it (if lax is true)
    increment("dismissed_events");
    return;
  }
  if (m_sentStopMessages) {
    B2ERROR("Received events after stop! I will dismiss this event.");
    increment("dismissed_events");
    return;
  }
 
  const auto dataSize = message->getDataMessage().size();
 
  B2ASSERT("Must be > 0", not m_readyWorkers.empty());
  auto nextWorker = m_readyWorkers.front();
  m_readyWorkers.pop_front();
 
  average("data_size", dataSize);
  average("data_size_to[" + nextWorker + "]", dataSize);
 
  increment("sent_events");
  increment("sent_events[" + nextWorker + "]");
 
  timeit("event_rate");
  timeit<200>("event_rate_to[" + nextWorker + "]");
 
  m_socket->send(ZMQMessageHelper::createZMQMessage(nextWorker), ZMQ_SNDMORE);
  ZMQParent::send(m_socket, std::move(message));
 
  log("ready_queue_size", static_cast<long>(m_readyWorkers.size()));
  decrement("ready_messages[" + nextWorker + "]");
}

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/ZMQLoadBalancedConnection.h
• framework/pcore/zmq/connections/src/ZMQLoadBalancedConnection.cc