Line data Source code
1 : /**
2 : * @file DFOModule.cpp DFOModule class implementation
3 : *
4 : * This is part of the DUNE DAQ Software Suite, copyright 2020.
5 : * Licensing/copyright details are in the COPYING file that you should have
6 : * received with this code.
7 : */
8 :
9 : #include "DFOModule.hpp"
10 : #include "dfmodules/CommonIssues.hpp"
11 :
12 : #include "dfmodules/opmon/DFOModule.pb.h"
13 :
14 : #include "appmodel/DFOModule.hpp"
15 : #include "confmodel/Connection.hpp"
16 : #include "iomanager/IOManager.hpp"
17 : #include "logging/Logging.hpp"
18 :
19 : #include <chrono>
20 : #include <cstdlib>
21 : #include <future>
22 : #include <limits>
23 : #include <list>
24 : #include <map>
25 : #include <memory>
26 : #include <string>
27 : #include <thread>
28 : #include <utility>
29 : #include <vector>
30 :
31 : /**
32 : * @brief Name used by TRACE TLOG calls from this source file
33 : */
34 : #define TRACE_NAME "DFOModule" // NOLINT
35 : enum
36 : {
37 : TLVL_ENTER_EXIT_METHODS = 5,
38 : TLVL_CONFIG = 7,
39 : TLVL_WORK_STEPS = 10,
40 : TLVL_TRIGDEC_RECEIVED = 21,
41 : TLVL_NOTIFY_TRIGGER = 22,
42 : TLVL_DISPATCH_TO_TRB = 23,
43 : TLVL_TDTOKEN_RECEIVED = 24
44 : };
45 :
46 : namespace dunedaq::dfmodules {
47 :
48 5 : DFOModule::DFOModule(const std::string& name)
49 : : dunedaq::appfwk::DAQModule(name)
50 5 : , m_queue_timeout(100)
51 10 : , m_run_number(0)
52 : {
53 5 : register_command("conf", &DFOModule::do_conf);
54 5 : register_command("start", &DFOModule::do_start);
55 5 : register_command("drain_dataflow", &DFOModule::do_stop);
56 5 : register_command("scrap", &DFOModule::do_scrap);
57 5 : }
58 :
59 : void
60 4 : DFOModule::init(std::shared_ptr<appfwk::ConfigurationManager> mcfg)
61 : {
62 4 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering init() method";
63 :
64 4 : auto mdal = mcfg->get_dal<appmodel::DFOModule>(get_name());
65 4 : if (!mdal) {
66 0 : throw appfwk::CommandFailed(ERS_HERE, "init", get_name(), "Unable to retrieve configuration object");
67 : }
68 4 : auto iom = iomanager::IOManager::get();
69 :
70 12 : for (auto con : mdal->get_inputs()) {
71 8 : if (con->get_data_type() == datatype_to_string<dfmessages::TriggerDecisionToken>()) {
72 4 : m_token_connection = con->UID();
73 : }
74 8 : if (con->get_data_type() == datatype_to_string<dfmessages::TriggerDecision>()) {
75 4 : m_td_connection = con->UID();
76 : }
77 : }
78 12 : for (auto con : mdal->get_outputs()) {
79 8 : if (con->get_data_type() == datatype_to_string<dfmessages::TriggerInhibit>()) {
80 4 : m_busy_sender = iom->get_sender<dfmessages::TriggerInhibit>(con->UID());
81 : }
82 8 : if (con->get_data_type() == datatype_to_string<dfmessages::TriggerDecision>()) {
83 4 : m_trb_conn_ids.push_back(con->UID());
84 : }
85 : }
86 :
87 4 : if (m_token_connection == "") {
88 0 : throw appfwk::MissingConnection(
89 0 : ERS_HERE, get_name(), datatype_to_string<dfmessages::TriggerDecisionToken>(), "input");
90 : }
91 4 : if (m_td_connection == "") {
92 0 : throw appfwk::MissingConnection(ERS_HERE, get_name(), datatype_to_string<dfmessages::TriggerDecision>(), "input");
93 : }
94 4 : if (m_busy_sender == nullptr) {
95 0 : throw appfwk::MissingConnection(ERS_HERE, get_name(), datatype_to_string<dfmessages::TriggerInhibit>(), "output");
96 : }
97 :
98 4 : m_dfo_conf = mdal->get_configuration();
99 : // these are just tests to check if the connections are ok
100 4 : iom->get_receiver<dfmessages::TriggerDecisionToken>(m_token_connection);
101 4 : iom->get_receiver<dfmessages::TriggerDecision>(m_td_connection);
102 :
103 4 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting init() method";
104 4 : }
105 :
106 : void
107 3 : DFOModule::do_conf(const CommandData_t&)
108 : {
109 3 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_conf() method";
110 :
111 3 : m_queue_timeout = std::chrono::milliseconds(m_dfo_conf->get_general_queue_timeout_ms());
112 3 : m_stop_timeout = std::chrono::milliseconds(m_dfo_conf->get_stop_timeout_ms());
113 3 : m_busy_threshold = m_dfo_conf->get_busy_threshold();
114 3 : m_free_threshold = m_dfo_conf->get_free_threshold();
115 :
116 3 : m_td_send_retries = m_dfo_conf->get_td_send_retries();
117 :
118 3 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_conf() method, there are "
119 3 : << m_dataflow_availability.size() << " TRB apps defined";
120 3 : }
121 :
122 : void
123 3 : DFOModule::do_start(const CommandData_t& payload)
124 : {
125 3 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_start() method";
126 :
127 3 : m_received_tokens = 0;
128 3 : m_run_number = payload.value<dunedaq::daqdataformats::run_number_t>("run", 0);
129 :
130 3 : m_running_status.store(true);
131 3 : m_last_notified_busy.store(false);
132 3 : m_last_assignement_it = m_dataflow_availability.end();
133 :
134 3 : m_last_token_received = m_last_td_received = std::chrono::steady_clock::now();
135 :
136 : // 19-Dec-2024, KAB: check that TriggerDecision senders are ready to send. This is done
137 : // so that the IOManager infrastructure fetches the necessary connection details from
138 : // the ConnectivityService at 'start' time, instead of the first time that the sender
139 : // is used to send a message. This avoids delays in the sending of the first TD in
140 : // the first data-taking run in a DAQ session. Such delays can lead to undesirable
141 : // system behavior like trigger inhibits.
142 3 : auto iom = iomanager::IOManager::get();
143 3 : if (m_busy_sender != nullptr) {
144 3 : bool is_ready = m_busy_sender->is_ready_for_sending(std::chrono::milliseconds(100));
145 3 : TLOG_DEBUG(0) << "The sender for TriggerInhibit messages " << (is_ready ? "is" : "is not") << " ready.";
146 : }
147 6 : for (auto trb_conn : m_trb_conn_ids) {
148 3 : auto sender = iom->get_sender<dfmessages::TriggerDecision>(trb_conn);
149 3 : if (sender != nullptr) {
150 3 : bool is_ready = sender->is_ready_for_sending(std::chrono::milliseconds(100));
151 3 : TLOG_DEBUG(0) << "The TriggerDecision sender for " << trb_conn << " " << (is_ready ? "is" : "is not")
152 3 : << " ready.";
153 : }
154 3 : }
155 3 : iom->add_callback<dfmessages::TriggerDecisionToken>(
156 3 : m_token_connection, std::bind(&DFOModule::receive_trigger_complete_token, this, std::placeholders::_1));
157 :
158 3 : iom->add_callback<dfmessages::TriggerDecision>(
159 3 : m_td_connection, std::bind(&DFOModule::receive_trigger_decision, this, std::placeholders::_1));
160 :
161 3 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_start() method";
162 3 : }
163 :
164 : void
165 3 : DFOModule::do_stop(const CommandData_t& /*args*/)
166 : {
167 3 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_stop() method";
168 :
169 3 : m_running_status.store(false);
170 :
171 3 : auto iom = iomanager::IOManager::get();
172 3 : iom->remove_callback<dfmessages::TriggerDecision>(m_td_connection);
173 :
174 3 : const int wait_steps = 20;
175 3 : auto step_timeout = m_stop_timeout / wait_steps;
176 3 : int step_counter = 0;
177 3 : while (!is_empty() && step_counter < wait_steps) {
178 0 : TLOG() << get_name() << ": stop delayed while waiting for " << used_slots() << " TDs to completed";
179 0 : std::this_thread::sleep_for(step_timeout);
180 0 : ++step_counter;
181 : }
182 :
183 3 : iom->remove_callback<dfmessages::TriggerDecisionToken>(m_token_connection);
184 :
185 3 : std::list<std::shared_ptr<AssignedTriggerDecision>> remnants;
186 5 : for (auto& app : m_dataflow_availability) {
187 2 : auto temp = app.second->flush();
188 2 : for (auto& td : temp) {
189 0 : remnants.push_back(td);
190 : }
191 2 : }
192 :
193 3 : for (auto& r : remnants) {
194 0 : ers::error(IncompleteTriggerDecision(ERS_HERE, r->decision.trigger_number, m_run_number));
195 : }
196 :
197 3 : std::lock_guard<std::mutex> guard(m_trigger_counters_mutex);
198 3 : m_trigger_counters.clear();
199 :
200 6 : TLOG() << get_name() << " successfully stopped";
201 3 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_stop() method";
202 3 : }
203 :
204 : void
205 3 : DFOModule::do_scrap(const CommandData_t& /*args*/)
206 : {
207 3 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_scrap() method";
208 :
209 3 : m_dataflow_availability.clear();
210 :
211 6 : TLOG() << get_name() << " successfully scrapped";
212 3 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_scrap() method";
213 3 : }
214 :
215 : void
216 5 : DFOModule::receive_trigger_decision(const dfmessages::TriggerDecision& decision)
217 : {
218 5 : TLOG_DEBUG(TLVL_TRIGDEC_RECEIVED) << get_name() << " Received TriggerDecision for trigger_number "
219 0 : << decision.trigger_number << " and run " << decision.run_number
220 5 : << " (current run is " << m_run_number << ")";
221 5 : if (decision.run_number != m_run_number) {
222 1 : ers::error(DFOModuleRunNumberMismatch(ERS_HERE, decision.run_number, m_run_number, "MLT", decision.trigger_number));
223 1 : return;
224 : }
225 :
226 4 : auto decision_received = std::chrono::steady_clock::now();
227 4 : ++m_received_decisions;
228 4 : auto trigger_types = unpack_types(decision.trigger_type);
229 8 : for (const auto t : trigger_types) {
230 4 : ++get_trigger_counter(t).received;
231 : }
232 :
233 4 : std::chrono::steady_clock::time_point decision_assigned;
234 4 : do {
235 :
236 4 : auto assignment = find_slot(decision);
237 :
238 4 : if (assignment == nullptr) { // this can happen if all application are in error state
239 0 : ers::error(UnableToAssign(ERS_HERE, decision.trigger_number));
240 0 : usleep(500);
241 0 : notify_trigger_if_needed();
242 0 : continue;
243 : }
244 :
245 4 : TLOG_DEBUG(TLVL_TRIGDEC_RECEIVED) << get_name() << " Slot found for trigger_number " << decision.trigger_number
246 0 : << " on connection " << assignment->connection_name
247 4 : << ", number of used slots is " << used_slots();
248 4 : decision_assigned = std::chrono::steady_clock::now();
249 4 : auto dispatch_successful = dispatch(assignment);
250 :
251 4 : if (dispatch_successful) {
252 3 : assign_trigger_decision(assignment);
253 3 : TLOG_DEBUG(TLVL_TRIGDEC_RECEIVED) << get_name() << " Assigned trigger_number " << decision.trigger_number
254 3 : << " to connection " << assignment->connection_name;
255 3 : break;
256 : } else {
257 1 : ers::error(TRBModuleAppUpdate(ERS_HERE, assignment->connection_name, "Could not send Trigger Decision"));
258 1 : m_dataflow_availability[assignment->connection_name]->set_in_error(true);
259 : }
260 :
261 5 : } while (m_running_status.load());
262 :
263 4 : notify_trigger_if_needed();
264 :
265 4 : m_waiting_for_decision +=
266 4 : std::chrono::duration_cast<std::chrono::microseconds>(decision_received - m_last_td_received).count();
267 4 : m_last_td_received = std::chrono::steady_clock::now();
268 4 : m_deciding_destination +=
269 4 : std::chrono::duration_cast<std::chrono::microseconds>(decision_assigned - decision_received).count();
270 4 : m_forwarding_decision +=
271 4 : std::chrono::duration_cast<std::chrono::microseconds>(m_last_td_received - decision_assigned).count();
272 4 : }
273 :
274 : std::shared_ptr<AssignedTriggerDecision>
275 4 : DFOModule::find_slot(const dfmessages::TriggerDecision& decision)
276 : {
277 :
278 : // this find_slot assings the decision with a round-robin logic
279 : // across all the available applications.
280 : // Applications in error are skipped.
281 : // we only probe the applications once.
282 : // if they are all unavailable the assignment is set to
283 : // the application with the lowest used slots
284 : // returning a nullptr will be considered as an error
285 : // from the upper level code
286 :
287 4 : std::shared_ptr<AssignedTriggerDecision> output = nullptr;
288 4 : auto minimum_occupied = m_dataflow_availability.end();
289 4 : size_t minimum = std::numeric_limits<size_t>::max();
290 4 : unsigned int counter = 0;
291 :
292 4 : auto candidate_it = m_last_assignement_it;
293 4 : if (candidate_it == m_dataflow_availability.end())
294 2 : candidate_it = m_dataflow_availability.begin();
295 :
296 8 : while (output == nullptr && counter < m_dataflow_availability.size()) {
297 :
298 4 : ++counter;
299 4 : ++candidate_it;
300 4 : if (candidate_it == m_dataflow_availability.end())
301 4 : candidate_it = m_dataflow_availability.begin();
302 :
303 : // get rid of the applications in error state
304 4 : if (candidate_it->second->is_in_error()) {
305 0 : continue;
306 : }
307 :
308 : // monitor
309 4 : auto slots = candidate_it->second->used_slots();
310 4 : if (slots < minimum) {
311 4 : minimum = slots;
312 4 : minimum_occupied = candidate_it;
313 : }
314 :
315 4 : if (candidate_it->second->is_busy())
316 1 : continue;
317 :
318 3 : output = candidate_it->second->make_assignment(decision);
319 3 : m_last_assignement_it = candidate_it;
320 : }
321 :
322 4 : if (!output) {
323 : // in this case all applications were busy
324 : // so we assign the decision to that with the lowest
325 : // number of assignments
326 1 : if (minimum_occupied != m_dataflow_availability.end()) {
327 1 : output = minimum_occupied->second->make_assignment(decision);
328 1 : m_last_assignement_it = minimum_occupied;
329 1 : ers::warning(AssignedToBusyApp(ERS_HERE, decision.trigger_number, minimum_occupied->first, minimum));
330 : }
331 : }
332 :
333 4 : if (output != nullptr) {
334 4 : TLOG_DEBUG(TLVL_WORK_STEPS) << "Assigned TriggerDecision with trigger number " << decision.trigger_number
335 4 : << " to TRB at connection " << output->connection_name;
336 : }
337 4 : return output;
338 0 : }
339 :
340 : void
341 6 : DFOModule::generate_opmon_data()
342 : {
343 :
344 6 : opmon::DFOInfo info;
345 6 : info.set_tokens_received(m_received_tokens.exchange(0));
346 6 : info.set_decisions_sent(m_sent_decisions.exchange(0));
347 6 : info.set_decisions_received(m_received_decisions.exchange(0));
348 6 : info.set_waiting_for_decision(m_waiting_for_decision.exchange(0));
349 6 : info.set_deciding_destination(m_deciding_destination.exchange(0));
350 6 : info.set_forwarding_decision(m_forwarding_decision.exchange(0));
351 6 : info.set_waiting_for_token(m_waiting_for_token.exchange(0));
352 6 : info.set_processing_token(m_processing_token.exchange(0));
353 6 : publish(std::move(info));
354 :
355 6 : std::lock_guard<std::mutex> guard(m_trigger_counters_mutex);
356 9 : for (auto& [type, counts] : m_trigger_counters) {
357 3 : opmon::TriggerInfo ti;
358 3 : ti.set_received(counts.received.exchange(0));
359 3 : ti.set_completed(counts.completed.exchange(0));
360 3 : auto name = dunedaq::trgdataformats::trigger_candidate_type_to_string(type);
361 6 : publish(std::move(ti), { { "type", name } });
362 3 : }
363 9 : }
364 :
365 : void
366 9 : DFOModule::receive_trigger_complete_token(const dfmessages::TriggerDecisionToken& token)
367 : {
368 9 : if (token.run_number == 0 && token.trigger_number == 0) {
369 2 : if (m_dataflow_availability.count(token.decision_destination) == 0) {
370 2 : TLOG_DEBUG(TLVL_CONFIG) << "Creating dataflow availability struct for uid " << token.decision_destination;
371 2 : auto entry = m_dataflow_availability[token.decision_destination] =
372 4 : std::make_shared<TriggerRecordBuilderData>(token.decision_destination, m_busy_threshold, m_free_threshold);
373 2 : register_node(token.decision_destination, entry);
374 2 : } else {
375 0 : TLOG() << TRBModuleAppUpdate(ERS_HERE, token.decision_destination, "Has reconnected");
376 0 : auto app_it = m_dataflow_availability.find(token.decision_destination);
377 0 : app_it->second->set_in_error(false);
378 : }
379 5 : return;
380 : }
381 :
382 7 : TLOG_DEBUG(TLVL_TDTOKEN_RECEIVED) << get_name() << " Received TriggerDecisionToken for trigger_number "
383 0 : << token.trigger_number << " and run " << token.run_number << " (current run is "
384 7 : << m_run_number << ")";
385 : // add a check to see if the application data found
386 7 : if (token.run_number != m_run_number) {
387 2 : std::ostringstream oss_source;
388 2 : oss_source << "TRB at connection " << token.decision_destination;
389 2 : ers::error(
390 4 : DFOModuleRunNumberMismatch(ERS_HERE, token.run_number, m_run_number, oss_source.str(), token.trigger_number));
391 2 : return;
392 2 : }
393 :
394 5 : auto app_it = m_dataflow_availability.find(token.decision_destination);
395 : // check if application data exists;
396 5 : if (app_it == m_dataflow_availability.end()) {
397 1 : ers::error(UnknownTokenSource(ERS_HERE, token.decision_destination));
398 1 : return;
399 : }
400 :
401 4 : ++m_received_tokens;
402 4 : auto callback_start = std::chrono::steady_clock::now();
403 :
404 4 : try {
405 4 : auto dec_ptr = app_it->second->complete_assignment(token.trigger_number, m_metadata_function);
406 3 : auto trigger_types = unpack_types(dec_ptr->decision.trigger_type);
407 6 : for (const auto t : trigger_types)
408 3 : ++get_trigger_counter(t).completed;
409 4 : } catch (AssignedTriggerDecisionNotFound const& err) {
410 1 : ers::error(err);
411 1 : }
412 :
413 4 : if (app_it->second->is_in_error()) {
414 0 : TLOG() << TRBModuleAppUpdate(ERS_HERE, token.decision_destination, "Has reconnected");
415 0 : app_it->second->set_in_error(false);
416 : }
417 :
418 4 : notify_trigger_if_needed();
419 :
420 4 : m_waiting_for_token +=
421 4 : std::chrono::duration_cast<std::chrono::microseconds>(callback_start - m_last_token_received).count();
422 4 : m_last_token_received = std::chrono::steady_clock::now();
423 4 : m_processing_token +=
424 4 : std::chrono::duration_cast<std::chrono::microseconds>(m_last_token_received - callback_start).count();
425 : }
426 :
427 : bool
428 8 : DFOModule::is_busy() const
429 : {
430 12 : for (auto& dfapp : m_dataflow_availability) {
431 8 : if (!dfapp.second->is_busy())
432 4 : return false;
433 : }
434 4 : return true;
435 : }
436 :
437 : bool
438 3 : DFOModule::is_empty() const
439 : {
440 5 : for (auto& dfapp : m_dataflow_availability) {
441 2 : if (dfapp.second->used_slots() != 0)
442 0 : return false;
443 : }
444 3 : return true;
445 : }
446 :
447 : size_t
448 0 : DFOModule::used_slots() const
449 : {
450 0 : size_t total = 0;
451 0 : for (auto& dfapp : m_dataflow_availability) {
452 0 : total += dfapp.second->used_slots();
453 : }
454 0 : return total;
455 : }
456 :
457 : void
458 8 : DFOModule::notify_trigger_if_needed() const
459 : {
460 : // 19-Dec-2024, KAB, ELF, MaR: combined the is_busy() and notify_trigger() calls in
461 : // a single method (notify_trigger_if_needed), and protected the contents of the new
462 : // method with a mutex, to avoid a race condition in which a given is_busy() result
463 : // is determined, but by the time that the value is sent to the MLT, the busy state
464 : // has changed.
465 8 : std::lock_guard<std::mutex> guard(m_notify_trigger_mutex);
466 :
467 8 : bool busy = is_busy();
468 8 : if (busy == m_last_notified_busy.load())
469 5 : return;
470 :
471 : bool wasSentSuccessfully = false;
472 :
473 3 : do {
474 3 : try {
475 3 : dfmessages::TriggerInhibit message{ busy, m_run_number };
476 3 : m_busy_sender->send(std::move(message), m_queue_timeout);
477 2 : wasSentSuccessfully = true;
478 2 : TLOG_DEBUG(TLVL_NOTIFY_TRIGGER) << get_name() << " Sent BUSY status " << busy << " to trigger in run "
479 2 : << m_run_number;
480 1 : } catch (const ers::Issue& excpt) {
481 1 : std::ostringstream oss_warn;
482 1 : oss_warn << "Send with sender \"" << m_busy_sender->get_name() << "\" failed";
483 1 : ers::warning(iomanager::OperationFailed(ERS_HERE, oss_warn.str(), excpt));
484 1 : }
485 :
486 5 : } while (!wasSentSuccessfully && m_running_status.load());
487 :
488 3 : m_last_notified_busy.store(busy);
489 8 : }
490 :
491 : bool
492 4 : DFOModule::dispatch(const std::shared_ptr<AssignedTriggerDecision>& assignment)
493 : {
494 :
495 4 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering dispatch() method. assignment->connection_name: "
496 4 : << assignment->connection_name;
497 :
498 4 : bool wasSentSuccessfully = false;
499 4 : int retries = m_td_send_retries;
500 4 : auto iom = iomanager::IOManager::get();
501 4 : do {
502 :
503 4 : try {
504 4 : auto decision_copy = dfmessages::TriggerDecision(assignment->decision);
505 8 : iom->get_sender<dfmessages::TriggerDecision>(assignment->connection_name)
506 4 : ->send(std::move(decision_copy), m_queue_timeout);
507 3 : wasSentSuccessfully = true;
508 3 : ++m_sent_decisions;
509 3 : TLOG_DEBUG(TLVL_DISPATCH_TO_TRB) << get_name() << " Sent TriggerDecision for trigger_number "
510 0 : << decision_copy.trigger_number << " to TRB at connection "
511 3 : << assignment->connection_name << " for run number " << decision_copy.run_number;
512 5 : } catch (const ers::Issue& excpt) {
513 1 : std::ostringstream oss_warn;
514 1 : oss_warn << "Send to connection \"" << assignment->connection_name << "\" failed";
515 1 : ers::warning(iomanager::OperationFailed(ERS_HERE, oss_warn.str(), excpt));
516 1 : }
517 :
518 4 : retries--;
519 :
520 4 : } while (!wasSentSuccessfully && m_running_status.load() && retries > 0);
521 :
522 4 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting dispatch() method";
523 4 : return wasSentSuccessfully;
524 4 : }
525 :
526 : void
527 3 : DFOModule::assign_trigger_decision(const std::shared_ptr<AssignedTriggerDecision>& assignment)
528 : {
529 3 : m_dataflow_availability[assignment->connection_name]->add_assignment(assignment);
530 3 : }
531 :
532 : } // namespace dunedaq::dfmodules
533 :
534 5 : DEFINE_DUNE_DAQ_MODULE(dunedaq::dfmodules::DFOModule)
|
