 |
Belle II Software
release-05-01-25
|
10 #include <framework/pcore/zmq/connections/ZMQConfirmedConnection.h>
11 #include <framework/pcore/zmq/messages/ZMQMessageFactory.h>
23 log(
"last_received_message",
"");
24 log(
"total_number_messages", 0l);
25 log(
"registered_workers", 0l);
26 log(
"hello_messages", 0l);
27 log(
"dead_workers", 0l);
28 log(
"all_stop_messages", 0l);
29 log(
"sent_stop_messages", 0l);
30 log(
"last_stop_sent",
"");
31 log(
"received_stop_messages", 0l);
32 log(
"all_terminate_messages", 0l);
33 log(
"sent_terminate_messages", 0l);
34 log(
"last_terminate_sent",
"");
35 log(
"received_terminate_messages", 0l);
36 log(
"received_messages_after_stop", 0l);
37 log(
"last_received_event_message",
"");
38 log(
"last_clear",
"");
39 log(
"stop_overwrites", 0l);
40 log(
"last_stop_overwrite",
"");
42 log(
"data_size", 0.0);
43 log(
"received_events", 0l);
44 log(
"event_rate", 0.0);
52 auto message = ZMQMessageFactory::fromSocket<ZMQIdMessage>(
m_socket);
53 const auto fromIdentity = message->getIdentity();
55 logTime(
"last_received_message");
57 increment(
"total_number_messages_from[" + fromIdentity +
"]");
62 if (message->isMessage(EMessageTypes::c_helloMessage)) {
67 increment(
"hello_messages_from[" + fromIdentity +
"]");
69 }
else if (message->isMessage(EMessageTypes::c_deleteWorkerMessage)) {
72 B2DEBUG(10,
"Got message from " << message->getIdentity() <<
" to kill " << message->getMessagePartAsString<2>());
73 const std::string& killedIdentity = message->getMessagePartAsString<2>();
78 increment(
"dead_worker_messaged_from[" + fromIdentity +
"]");
81 B2ERROR(
"There is not a single worker registered anymore!");
107 B2ASSERT(
"Worker without proper registration!",
110 if (message->isMessage(EMessageTypes::c_lastEventMessage)) {
114 increment(
"total_received_stop_messages");
129 }
else if (message->isMessage(EMessageTypes::c_terminateMessage)) {
133 increment(
"total_received_terminate_messages");
140 logTime(
"last_terminate_sent");
151 B2ERROR(
"Received an event after having received stop messages from every worker. This is not a good sign! I will dismiss this event!");
152 increment(
"received_messages_after_stop");
157 const auto dataSize = message->getDataMessage().size();
159 average(
"data_size", dataSize);
160 average(
"data_size_from[" + fromIdentity +
"]", dataSize);
163 increment(
"received_events[" + fromIdentity +
"]");
166 timeit<200>(
"event_rate_from[" + fromIdentity +
"]");
168 logTime(
"last_received_event_message");
193 B2ERROR(
"Sending out a stop message although not all of the workers are finished already!");
195 logTime(
"last_stop_overwrite");
213 log(
"no_confirmation_message", 0l);
214 log(
"last_sent_event_message", 0l);
215 log(
"data_size", 0.0);
216 log(
"sent_events", 0l);
217 log(
"event_rate", 0.0);
218 log(
"timespan_waiting_for_confirmation", 0l);
232 auto current = std::chrono::system_clock::now();
237 }
else if (requireConfirmation) {
238 B2FATAL(
"Did not receive a confirmation message in time!");
240 B2WARNING(
"Did not receive a confirmation message in time!");
246 auto afterWaiting = std::chrono::system_clock::now();
253 const auto dataSize = message->getDataMessage().size();
255 average(
"data_size", dataSize);
259 logTime(
"last_sent_event_message");
269 auto acceptMessage = ZMQMessageFactory::fromSocket<ZMQNoIdMessage>(
m_socket);
270 B2ASSERT(
"Should be an accept message", acceptMessage->isMessage(EMessageTypes::c_confirmMessage));
ZMQConfirmedOutput(const std::string &outputAddress, const std::shared_ptr< ZMQParent > &parent)
Create a new confirmed output by connecting to the address.
static unsigned int poll(const std::vector< zmq::socket_t * > &socketList, int timeout)
Poll function.
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 ...
void handleEvent(std::unique_ptr< ZMQNoIdMessage > message, bool requireConfirmation=true, int maximalWaitTime=10000)
Send the message to the output (a message without a ID as there is only a single output).
static void send(std::unique_ptr< zmq::socket_t > &socket, AZMQMessage message)
Send a given message over the given created socket. You need to move in the message for zero-copy.
void logTime(const std::string &key)
Store the current time as a string under the given key.
std::shared_ptr< ZMQParent > m_parent
The shared ZMQParent instance.
unsigned int m_waitingForConfirmation
On how many confirmation messages are we still waiting?
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,...
void increment(const std::string &key)
Increment the value with the given key (only numerical values). If not present, set to 1.
Abstract base class for different kinds of events.
static auto createMessage(const std::string &msgIdentity, const EMessageTypes msgType, const std::unique_ptr< EvtMessage > &eventMessage)
Create an ID Message out of an identity, the type and an event message.
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_...
long m_timespanWaitingForConfirmation
Internal monitoring how long we were waiting for confirmation messages.
Specialized connection over a ZMQ socket.
void handleIncomingData()
Blocks until it can receive the (hopefully confirmation) message from the output.
std::unique_ptr< zmq::socket_t > m_socket
The memory of the socket. Needs to be initialized in a derived class.