Standard distributor app: receive data via a raw connection (e.g. More...

#include <ZMQDistributor.h>

Inheritance diagram for ZMQDistributor:

Public Member Functions
void	initFromConsole (const std::string &description, int argc, char *argv[])
	Should be called before the main() function to initialize the connections using the paremeters given on command line.

void	main ()
	Start the main loop polling on the output and monitoring connections and eventually also on the input if the output is ready.

Protected Member Functions
void	initialize () final
	Initialize the two connections using the command line arguments.

void	addOptions (po::options_description &desc) final
	Add the parameters to the cmd line arguments.

void	handleExternalSignal (EMessageTypes type) final
	Handle stop, start and terminate messages as described above.

void	handleInput () final
	Pass the message from the input connection to the output connection (only data messages)

void	handleTimeout () final
	When a timeout is set (= we are waiting for all messages after a stop), send a stop message once we have not seen any more events.

virtual void	handleOutput ()
	Will get called for every message on the output connection. Can be overridden in a derived class. Calls handleIncomingData by default.

virtual void	fillMonitoringJSON (std::stringstream &buffer) const
	Using the connections, fill up a buffer with the content to be monitored.

bool	terminated () const
	Check if the main loop will be exited on next occasion. Can be set via the "m_terminate" flag.

void	resetTimer ()
	Helper function to reset the start time and the remaining time.

void	pollEvent (bool pollOnInput)
	Poll until a single event is retreived.

Protected Attributes
std::shared_ptr< ZMQParent >	m_parent
	Pointer to the ZMQParent to be used as base for all connections.

std::unique_ptr< ZMQRawInput >	m_input
	Pointer to the input connection. Should be set in initialize.

std::unique_ptr< ZMQLoadBalancedOutput >	m_output
	Pointer to the output connection. Should be set in initialize.

std::unique_ptr< ZMQSimpleConnection >	m_monitor
	Pointer to the monitoring connection. Should be set in initialize.

bool	m_terminate
	Can be set by functions to terminate the main loop at the next possibility.

unsigned int	m_timeout
	If set to a value != 0, will call handleTimeout with this frequency (in seconds).

bool	m_monitorHasPriority
	Flag to break out of the polling loop to check for monitoring messages. Except for the finalcollector you probably do not want this.

Private Member Functions
bool	checkTimer ()
	Helper function to check, if the timeout should happen.

void	updateTimer ()
	Helper function to update the remaining time.

void	handleMonitoring ()
	Handle an incoming message on the monitoring socket by either calling handleExternalSignal() or by passing on the monitor JSONs of the connections.

Private Attributes
std::string	m_inputAddress
	Parameter: input address.

std::string	m_outputAddress
	Parameter: output address.

unsigned int	m_maximalBufferSize = 80'000'000
	Parameter: buffer size for storing input messages.

bool	m_expressRecoMode = false
	Parameter: Do not wait for a ready worker if set to true, but dismiss the incoming event.

unsigned int	m_stopWaitingTime = 2
	Parameter: how long to wait after no events come anymore.

HLTMainLoop	m_mainLoop
	Internal signal handler.

std::string	m_monitoringAddress
	Storage for the monitoring address for the cmd arguments.

int	m_remainingTime
	Counter for the remaining time until a timeout happens.

std::chrono::system_clock::time_point	m_start
	Start time for the timeout.

Detailed Description

Standard distributor app: receive data via a raw connection (e.g.

from event builder) and send them out to any ready worker (via a load-balanced connection). If expressRecoMode is not set, wait until a worker is ready (otherwise dismiss events). As there are not signal messages on input, does not react on anything from input. When receiving a stop on monitoring, it starts a counter. Once there has not been any event for N second, it sends a stop to all clients. When receiving a start on monitoring, it clears the output. When receiving a terminate on monitoring, it sends out a terminate message to all clients and terminated itself.

Definition at line 33 of file ZMQDistributor.h.

Member Function Documentation

◆ addOptions()

void addOptions ( po::options_description & desc )

finalprotectedvirtual

Add the parameters to the cmd line arguments.

Reimplemented from ZMQStandardApp< ZMQRawInput, ZMQLoadBalancedOutput >.

Definition at line 13 of file ZMQDistributor.cc.

{
  ZMQStandardApp::addOptions(desc);
  desc.add_options()
  ("input", boost::program_options::value<std::string>(&m_inputAddress)->required(),
   "where to read the events from")
  ("output", boost::program_options::value<std::string>(&m_outputAddress)->required(),
   "where to send the events to")
  ("expressRecoMode", boost::program_options::bool_switch(&m_expressRecoMode)->default_value(m_expressRecoMode),
   "express reco mode: dismiss events if no worker is ready and send out event messages (instead of raw messages)")
  ("maximalBufferSize",
   boost::program_options::value<unsigned int>(&m_maximalBufferSize)->default_value(m_maximalBufferSize),
   "size of the input buffer")
  ("stopWaitingTime",
   boost::program_options::value<unsigned int>(&m_stopWaitingTime)->default_value(m_stopWaitingTime),
   "how long to wait after no events come anymore");
}

◆ checkTimer()

bool checkTimer

privateinherited

Helper function to check, if the timeout should happen.

Definition at line 111 of file ZMQApp.details.h.

  {
    updateTimer();
    return m_remainingTime == 0;
  }

◆ fillMonitoringJSON()

void fillMonitoringJSON ( std::stringstream & buffer ) const

protectedvirtualinherited

Using the connections, fill up a buffer with the content to be monitored.

Definition at line 92 of file ZMQApp.details.h.

  {
    buffer << "{" << std::endl;
    buffer << "\"monitor\": " << m_monitor->getMonitoringJSON() << "," << std::endl;
    buffer << "\"input\": " << m_input->getMonitoringJSON() << "," << std::endl;
    buffer << "\"output\": " << m_output->getMonitoringJSON() << std::endl;
    buffer << "}" << std::endl;
  }

◆ handleExternalSignal()

void handleExternalSignal ( EMessageTypes type )

finalprotectedvirtual

Handle stop, start and terminate messages as described above.

Reimplemented from ZMQStandardApp< ZMQRawInput, ZMQLoadBalancedOutput >.

Definition at line 38 of file ZMQDistributor.cc.

{
  if (type == EMessageTypes::c_newRunMessage) {
    m_input->clear();
    m_output->clear();
  } else if (type == EMessageTypes::c_lastEventMessage) {
    m_timeout = m_stopWaitingTime;
    resetTimer();
  } else if (type == EMessageTypes::c_terminateMessage) {
    m_output->handleEvent(ZMQMessageFactory::createMessage(EMessageTypes::c_terminateMessage));
    m_input->clear();
    m_terminate = true;
  }
}

◆ handleInput()

void handleInput ( )

finalprotectedvirtual

Pass the message from the input connection to the output connection (only data messages)

Reimplemented from ZMQStandardApp< ZMQRawInput, ZMQLoadBalancedOutput >.

Definition at line 62 of file ZMQDistributor.cc.

{
  auto messages = m_input->handleIncomingData();
 
  for (auto&& message : messages) {
    // So there has been a message, make sure to reset the timer for waiting (if no timer is set this will just return)
    resetTimer();
 
    EMessageTypes messageType = EMessageTypes::c_rawDataMessage;
    if (m_expressRecoMode) {
      messageType = EMessageTypes::c_eventMessage;
    }
 
    auto outputMessage = ZMQMessageFactory::createMessage(messageType, std::move(message));
 
    // We know that the output is ready for the first message, but we do not know anything about any other messages, so lets
    // be safe and poll the output if it is not ready so far
    while (not m_output->isReady() and not terminated()) {
      pollEvent(false);
    }
    if (terminated()) {
      return;
    }
    m_output->handleEvent(std::move(outputMessage));
  }
}

◆ handleMonitoring()

void handleMonitoring

privateinherited

Handle an incoming message on the monitoring socket by either calling handleExternalSignal() or by passing on the monitor JSONs of the connections.

Definition at line 115 of file ZMQApp.details.h.

  {
    auto monitoringMessage = m_monitor->handleIncomingData();
 
    if (monitoringMessage->isMessage(EMessageTypes::c_newRunMessage)) {
      handleExternalSignal(EMessageTypes::c_newRunMessage);
      return;
    } else if (monitoringMessage->isMessage(EMessageTypes::c_lastEventMessage)) {
      handleExternalSignal(EMessageTypes::c_lastEventMessage);
      return;
    } else if (monitoringMessage->isMessage(EMessageTypes::c_terminateMessage)) {
      handleExternalSignal(EMessageTypes::c_terminateMessage);
      return;
    }
 
    std::stringstream buffer;
    fillMonitoringJSON(buffer);
 
    auto message = ZMQMessageFactory::createMessage(monitoringMessage->getIdentity(),
                                                    EMessageTypes::c_confirmMessage, buffer.str());
    m_monitor->handleEvent(std::move(message));
  }

◆ handleOutput()

void handleOutput

protectedvirtualinherited

Will get called for every message on the output connection. Can be overridden in a derived class. Calls handleIncomingData by default.

Definition at line 88 of file ZMQApp.details.h.

  {
    m_output->handleIncomingData();
  }

◆ handleTimeout()

void handleTimeout ( )

finalprotectedvirtual

When a timeout is set (= we are waiting for all messages after a stop), send a stop message once we have not seen any more events.

Reimplemented from ZMQStandardApp< ZMQRawInput, ZMQLoadBalancedOutput >.

Definition at line 53 of file ZMQDistributor.cc.

{
  m_output->handleEvent(ZMQMessageFactory::createMessage(EMessageTypes::c_lastEventMessage));
 
  // We do not want to send out another stop message, so reset the counter
  m_timeout = 0;
  resetTimer();
}

◆ initFromConsole()

void initFromConsole	(	const std::string &	description,
		int	argc,
		char *	argv[]
	)

inherited

Should be called before the main() function to initialize the connections using the paremeters given on command line.

Custom implementations should implement the addOptions function to pass the command line arguments correctly. Calls the initialize function (which should also be overridden).

Definition at line 54 of file ZMQApp.details.h.

  {
 
    po::options_description desc(description);
    std::string connection_file;
    int debugLevel(0);
    desc.add_options()
    ("connection-file", boost::program_options::value<std::string>(&connection_file),
     "if given print the connection information for input/output and monitoring socket to the given filename "
     "in json format")
    ("debug", boost::program_options::value<int>(&debugLevel), "Enable debug logging");
    addOptions(desc);
 
    po::positional_options_description p;
 
    po::variables_map vm;
    try {
      po::store(po::command_line_parser(argc, argv).options(desc).positional(p).run(), vm);
    } catch (std::exception& e) {
      B2FATAL(e.what());
    }
 
    if (vm.count("help")) {
      std::cout << desc << std::endl;
      exit(1);
    }
 
    try {
      po::notify(vm);
    } catch (std::exception& e) {
      B2FATAL(e.what());
    }
 
    if (debugLevel > 0) {
      auto& logging = LogSystem::Instance();
      logging.getLogConfig()->setLogLevel(LogConfig::c_Debug);
      logging.getLogConfig()->setDebugLevel(debugLevel);
      B2DEBUG(1, "Enabled debug logging");
    }
 
    initialize();
 
    if (not connection_file.empty()) {
      B2DEBUG(1, "Write connection file" << LogVar("connection_file", connection_file));
      nlohmann::json json;
      try {
        json["input"] = m_input->getEndPoint();
      } catch (zmq::error_t& e) {
        B2WARNING(e.what());
      }
      try {
        json["output"] = m_output->getEndPoint();
      } catch (zmq::error_t& e) {
        B2WARNING(e.what());
      }
      std::ofstream connections(connection_file, std::ofstream::trunc);
      if (!connections) {
        B2FATAL("Cannot write connection file" << LogVar("connection_file", connection_file));
      }
      connections << std::setw(4) << json << std::endl;
    }
  }

◆ initialize()

void initialize ( )

finalprotectedvirtual

Initialize the two connections using the command line arguments.

Reimplemented from ZMQStandardApp< ZMQRawInput, ZMQLoadBalancedOutput >.

Definition at line 31 of file ZMQDistributor.cc.

{
  ZMQStandardApp::initialize();
  m_input.reset(new ZMQRawInput(m_inputAddress, m_maximalBufferSize, m_expressRecoMode, m_parent));
  m_output.reset(new ZMQLoadBalancedOutput(m_outputAddress, m_expressRecoMode, m_parent));
}

◆ main()

void main

inherited

Start the main loop polling on the output and monitoring connections and eventually also on the input if the output is ready.

Calls the functions handleExternalSignal, handleInput, handleOutput and handleTimeout as described in the documentation of this class.

Definition at line 61 of file ZMQApp.details.h.

  {
    resetTimer();
 
    while (not terminated()) {
      if (not m_output->isReady()) {
        // if the output is not ready, we can not sent anything. So lets just poll on output
        // and monitoring until it becomes ready
        m_monitor->log("output_state", "not_ready");
        pollEvent(false);
      } else {
        // if it is ready, we can also include the input socket as long as output stays ready
        m_monitor->log("output_state", "ready");
        pollEvent(true);
      }
    }
  }

◆ pollEvent()

void pollEvent ( bool pollOnInput )

protectedinherited

Poll until a single event is retreived.

Definition at line 98 of file ZMQApp.details.h.

  {
    auto reactToOutput = [this]() {
      // Get all messages from output socket
      while (ZMQConnection::hasMessage(m_output.get()) and not terminated()) {
        handleOutput();
        if (m_monitorHasPriority) {
          break;
        }
      }
    };
 
    auto reactToMonitor = [this]() {
      // Get all messages from monitoring socket
      while (ZMQConnection::hasMessage(m_monitor.get()) and not terminated()) {
        handleMonitoring();
      }
    };
 
    auto reactToInput = [this]() {
      // Get all messages from input as long output is ready
      while (ZMQConnection::hasMessage(m_input.get()) and m_output->isReady() and not terminated()) {
        handleInput();
        if (m_monitorHasPriority) {
          break;
        }
      }
    };
 
    m_monitor->logTime("waiting_since");
 
    if (pollOnInput) {
      ZMQConnection::poll({{m_output.get(), reactToOutput}, {m_monitor.get(), reactToMonitor}, {m_input.get(), reactToInput}},
      m_remainingTime);
    } else {
      ZMQConnection::poll({{m_output.get(), reactToOutput}, {m_monitor.get(), reactToMonitor}}, m_remainingTime);
    }
 
    if (checkTimer() and not terminated()) {
      B2ASSERT("There is no timeout set, but we still call the timeout() function? A bug!", m_timeout != 0);
      m_monitor->increment("timeouts");
      handleTimeout();
      resetTimer();
    } else {
      updateTimer();
    }
  }

◆ resetTimer()

void resetTimer

protectedinherited

Helper function to reset the start time and the remaining time.

Definition at line 96 of file ZMQApp.details.h.

  {
    // if there is no timeout, we should never set the remaining time
    if (m_timeout == 0) {
      m_remainingTime = -1;
      return;
    }
 
    m_start = std::chrono::system_clock::now();
    m_remainingTime = m_timeout * 1000;
  }

◆ terminated()

bool terminated

protectedinherited

Check if the main loop will be exited on next occasion. Can be set via the "m_terminate" flag.

Definition at line 94 of file ZMQApp.details.h.

  {
    return not m_mainLoop.isRunning() or m_terminate;
  }

◆ updateTimer()

void updateTimer

privateinherited

Helper function to update the remaining time.

Definition at line 113 of file ZMQApp.details.h.

  {
    // if there is no timeout, we should never update the remaining time
    if (m_timeout == 0) {
      m_remainingTime = -1;
      return;
    }
 
    auto currentTime = std::chrono::system_clock::now();
    auto timeDifference = std::chrono::duration_cast<std::chrono::milliseconds>(currentTime - m_start);
    m_remainingTime = m_timeout * 1000 - timeDifference.count();
    if (m_remainingTime < 0) {
      m_remainingTime = 0;
    }
  }

Member Data Documentation

◆ m_expressRecoMode

bool m_expressRecoMode = false

private

Parameter: Do not wait for a ready worker if set to true, but dismiss the incoming event.

Definition at line 54 of file ZMQDistributor.h.

◆ m_input

std::unique_ptr<ZMQRawInput > m_input

protectedinherited

Pointer to the input connection. Should be set in initialize.

Definition at line 67 of file ZMQApp.h.

◆ m_inputAddress

std::string m_inputAddress

private

Parameter: input address.

Definition at line 48 of file ZMQDistributor.h.

◆ m_mainLoop

HLTMainLoop m_mainLoop

privateinherited

Internal signal handler.

Definition at line 102 of file ZMQApp.h.

◆ m_maximalBufferSize

unsigned int m_maximalBufferSize = 80'000'000

private

Parameter: buffer size for storing input messages.

Definition at line 52 of file ZMQDistributor.h.

◆ m_monitor

std::unique_ptr<ZMQSimpleConnection> m_monitor

protectedinherited

Pointer to the monitoring connection. Should be set in initialize.

Definition at line 71 of file ZMQApp.h.

◆ m_monitorHasPriority

bool m_monitorHasPriority

protectedinherited

Flag to break out of the polling loop to check for monitoring messages. Except for the finalcollector you probably do not want this.

Definition at line 77 of file ZMQApp.h.

◆ m_monitoringAddress

std::string m_monitoringAddress

privateinherited

Storage for the monitoring address for the cmd arguments.

Definition at line 104 of file ZMQApp.h.

◆ m_output

std::unique_ptr<ZMQLoadBalancedOutput > m_output

protectedinherited

Pointer to the output connection. Should be set in initialize.

Definition at line 69 of file ZMQApp.h.

◆ m_outputAddress

std::string m_outputAddress

private

Parameter: output address.

Definition at line 50 of file ZMQDistributor.h.

◆ m_parent

std::shared_ptr<ZMQParent> m_parent

protectedinherited

Pointer to the ZMQParent to be used as base for all connections.

Definition at line 65 of file ZMQApp.h.

◆ m_remainingTime

int m_remainingTime

privateinherited

Counter for the remaining time until a timeout happens.

Definition at line 106 of file ZMQApp.h.

◆ m_start

std::chrono::system_clock::time_point m_start

privateinherited

Start time for the timeout.

Definition at line 108 of file ZMQApp.h.

◆ m_stopWaitingTime

unsigned int m_stopWaitingTime = 2

private

Parameter: how long to wait after no events come anymore.

Definition at line 56 of file ZMQDistributor.h.

◆ m_terminate

bool m_terminate

protectedinherited

Can be set by functions to terminate the main loop at the next possibility.

Definition at line 73 of file ZMQApp.h.

◆ m_timeout

unsigned int m_timeout

protectedinherited

If set to a value != 0, will call handleTimeout with this frequency (in seconds).

Definition at line 75 of file ZMQApp.h.

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

daq/hbasf2/apps/include/ZMQDistributor.h
daq/hbasf2/apps/src/ZMQDistributor.cc

Public Member Functions

Protected Member Functions

Protected Attributes

Private Member Functions

Private Attributes

Detailed Description

Member Function Documentation

◆ addOptions()

◆ checkTimer()

◆ fillMonitoringJSON()

◆ handleExternalSignal()

◆ handleInput()

◆ handleMonitoring()

◆ handleOutput()

◆ handleTimeout()

◆ initFromConsole()

◆ initialize()

◆ main()

◆ pollEvent()

◆ resetTimer()

◆ terminated()

◆ updateTimer()

Member Data Documentation

◆ m_expressRecoMode

◆ m_input

◆ m_inputAddress

◆ m_mainLoop

◆ m_maximalBufferSize

◆ m_monitor

◆ m_monitorHasPriority

◆ m_monitoringAddress

◆ m_output

◆ m_outputAddress

◆ m_parent

◆ m_remainingTime

◆ m_start

◆ m_stopWaitingTime

◆ m_terminate

◆ m_timeout