Line data Source code
1 : /**
2 : * @file TRBModule.cpp TRBModule 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 "TRBModule.hpp"
10 : #include "dfmodules/CommonIssues.hpp"
11 :
12 : #include "appmodel/NetworkConnectionDescriptor.hpp"
13 : #include "appmodel/NetworkConnectionRule.hpp"
14 : #include "appmodel/ReadoutApplication.hpp"
15 : #include "appmodel/SourceIDConf.hpp"
16 : #include "appmodel/SourceIDToNetworkConnection.hpp"
17 : #include "appmodel/TRBModule.hpp"
18 : #include "appmodel/TriggerApplication.hpp"
19 : #include "confmodel/Application.hpp"
20 : #include "confmodel/Connection.hpp"
21 : #include "confmodel/DetectorStream.hpp"
22 : #include "confmodel/DetectorToDaqConnection.hpp"
23 : #include "confmodel/NetworkConnection.hpp"
24 : #include "confmodel/Session.hpp"
25 : #include "dfmessages/TriggerRecord_serialization.hpp"
26 : #include "logging/Logging.hpp"
27 :
28 : #include "iomanager/IOManager.hpp"
29 :
30 : #include <algorithm>
31 : #include <chrono>
32 : #include <cmath>
33 : #include <cstdlib>
34 : #include <limits>
35 : #include <memory>
36 : #include <numeric>
37 : #include <string>
38 : #include <thread>
39 : #include <utility>
40 : #include <vector>
41 :
42 : /**
43 : * @brief TRACE debug levels used in this source file
44 : */
45 : enum
46 : {
47 : TLVL_ENTER_EXIT_METHODS = 5,
48 : TLVL_INIT = 8,
49 : TLVL_WORK_STEPS = 10,
50 : TLVL_BOOKKEEPING = 15,
51 : TLVL_DISPATCH_DATAREQ = 21,
52 : TLVL_FRAGMENT_RECEIVE = 22
53 : };
54 :
55 : namespace dunedaq {
56 : namespace dfmodules {
57 :
58 : using daqdataformats::TriggerRecordStatusBits;
59 :
60 0 : TRBModule::TRBModule(const std::string& name)
61 : : dunedaq::appfwk::DAQModule(name)
62 0 : , m_stop_requested(true)
63 0 : , m_tr_queue_timeout(100)
64 0 : , m_dreq_queue_timeout(100)
65 : {
66 :
67 0 : register_command("conf", &TRBModule::do_conf);
68 0 : register_command("scrap", &TRBModule::do_scrap);
69 0 : register_command("start", &TRBModule::do_start);
70 0 : register_command("stop", &TRBModule::do_stop);
71 0 : }
72 :
73 : void
74 0 : TRBModule::init(std::shared_ptr<appfwk::ConfigurationManager> mcfg)
75 : {
76 :
77 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering init() method";
78 :
79 : //--------------------------------
80 : // Get single queues
81 : //---------------------------------
82 :
83 0 : auto mdal = mcfg->get_dal<appmodel::TRBModule>(get_name());
84 0 : if (!mdal) {
85 0 : throw appfwk::CommandFailed(ERS_HERE, "init", get_name(), "Unable to retrieve configuration object");
86 : }
87 :
88 0 : auto iom = iomanager::IOManager::get();
89 0 : for (auto con : mdal->get_inputs()) {
90 0 : if (con->get_data_type() == datatype_to_string<dfmessages::TriggerDecision>()) {
91 0 : m_trigger_decision_input = iom->get_receiver<dfmessages::TriggerDecision>(con->UID());
92 : }
93 0 : if (con->get_data_type() == datatype_to_string<std::unique_ptr<daqdataformats::Fragment>>()) {
94 0 : m_fragment_input = iom->get_receiver<std::unique_ptr<daqdataformats::Fragment>>(con->UID());
95 :
96 : // save the data fragment receiver global connection name for later, when it gets
97 : // copied into the DataRequests so that data producers know where to send their fragments
98 0 : m_reply_connection = con->UID();
99 : }
100 0 : if (con->get_data_type() == datatype_to_string<dfmessages::TRMonRequest>()) {
101 0 : m_mon_receiver = iom->get_receiver<dfmessages::TRMonRequest>(con->UID());
102 : }
103 : }
104 :
105 0 : if (m_trigger_decision_input == nullptr) {
106 0 : throw InvalidQueueFatalError(ERS_HERE, get_name(), "TriggerDecision Input queue");
107 : }
108 0 : if (m_fragment_input == nullptr) {
109 0 : throw InvalidQueueFatalError(ERS_HERE, get_name(), "Fragment Input queue");
110 : }
111 :
112 0 : m_trigger_record_output =
113 0 : iom->get_sender<std::unique_ptr<daqdataformats::TriggerRecord>>(mdal->get_trigger_record_output()->UID());
114 :
115 0 : for (auto con : mdal->get_request_connections()) {
116 0 : for (auto source_id : con->get_source_ids()) {
117 :
118 : // find the queue for sourceid_req in the map
119 0 : std::unique_lock<std::mutex> lk(m_map_sourceid_connections_mutex);
120 0 : daqdataformats::SourceID sid;
121 0 : sid.subsystem = daqdataformats::SourceID::string_to_subsystem(source_id->get_subsystem());
122 0 : sid.id = source_id->get_sid();
123 0 : auto it_req = m_map_sourceid_connections.find(sid);
124 0 : if (it_req == m_map_sourceid_connections.end() || it_req->second == nullptr) {
125 0 : m_map_sourceid_connections[sid] = get_iom_sender<dfmessages::DataRequest>(con->get_netconn()->UID());
126 : }
127 0 : lk.unlock();
128 0 : }
129 : }
130 :
131 0 : m_trb_conf = mdal->get_configuration();
132 :
133 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting init() method";
134 0 : }
135 :
136 : void
137 0 : TRBModule::generate_opmon_data()
138 : {
139 :
140 0 : opmon::TRBInfo i;
141 :
142 : // status metrics
143 0 : i.set_pending_trigger_decisions(m_trigger_decisions_counter.load());
144 0 : i.set_fragments_in_the_book(m_fragment_counter.load());
145 0 : i.set_pending_fragments(m_pending_fragment_counter.load());
146 :
147 : // operation metrics
148 0 : i.set_received_trigger_decisions(m_received_trigger_decisions.exchange(0));
149 0 : i.set_generated_trigger_records(m_generated_trigger_records.exchange(0));
150 0 : i.set_generated_data_requests(m_generated_data_requests.exchange(0));
151 0 : i.set_received_fragments(m_received_fragments.exchange(0));
152 0 : i.set_data_waiting_time(m_data_waiting_time.exchange(0));
153 0 : i.set_data_request_width(m_data_request_width.exchange(0));
154 0 : i.set_trigger_decision_width(m_trigger_decision_width.exchange(0));
155 0 : i.set_received_trmon_requests(m_trmon_request_counter.exchange(0));
156 0 : i.set_sent_trmon(m_trmon_sent_counter.exchange(0));
157 :
158 0 : i.set_td_processing_us(m_td_processing_us.exchange(0));
159 0 : i.set_fragment_processing_us(m_fragment_processing_us.exchange(0));
160 :
161 0 : publish(std::move(i));
162 :
163 0 : opmon::TRBErrors err;
164 : // error counters
165 0 : err.set_timed_out_trigger_records(m_timed_out_trigger_records.load());
166 0 : err.set_abandoned_trigger_records(m_abandoned_trigger_records.load());
167 0 : err.set_unexpected_fragments(m_unexpected_fragments.load());
168 0 : err.set_unexpected_trigger_decisions(m_unexpected_trigger_decisions.load());
169 0 : err.set_lost_fragments(m_lost_fragments.load());
170 0 : err.set_invalid_requests(m_invalid_requests.load());
171 0 : err.set_duplicated_trigger_ids(m_duplicated_trigger_ids.load());
172 :
173 0 : publish(std::move(err));
174 0 : }
175 :
176 : void
177 0 : TRBModule::do_conf(const CommandData_t&)
178 : {
179 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_conf() method";
180 :
181 0 : m_trigger_timeout = std::chrono::milliseconds(m_trb_conf->get_trigger_record_timeout_ms());
182 :
183 0 : m_tr_queue_timeout = std::chrono::milliseconds(m_trb_conf->get_tr_queue_timeout());
184 0 : m_dreq_queue_timeout = std::chrono::milliseconds(m_trb_conf->get_request_queue_timeout());
185 :
186 0 : TLOG() << get_name() << ": timeouts (ms): TR = " << m_tr_queue_timeout.count()
187 0 : << ", DReq = " << m_dreq_queue_timeout.count();
188 0 : m_max_sequence_length = m_trb_conf->get_max_sequence_length_ticks();
189 0 : TLOG() << get_name() << ": Max time window is " << m_max_sequence_length;
190 :
191 0 : m_this_trb_source_id.subsystem = daqdataformats::SourceID::Subsystem::kTRBuilder;
192 0 : m_this_trb_source_id.id = m_trb_conf->get_source_id();
193 :
194 0 : m_max_open_trigger_records = m_trb_conf->get_maximum_open_trigger_records();
195 :
196 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_conf() method";
197 0 : }
198 :
199 : void
200 0 : TRBModule::do_scrap(const CommandData_t& /*args*/)
201 : {
202 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_scrap() method";
203 :
204 0 : TLOG() << get_name() << " successfully scrapped";
205 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_scrap() method";
206 0 : }
207 :
208 : void
209 0 : TRBModule::do_start(const CommandData_t& args)
210 : {
211 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_start() method";
212 :
213 : // clean books from possible previous memory
214 0 : m_trigger_records.clear();
215 0 : m_trigger_decisions_counter.store(0);
216 0 : m_unexpected_trigger_decisions.store(0);
217 0 : m_pending_fragment_counter.store(0);
218 0 : m_generated_trigger_records.store(0);
219 0 : m_fragment_counter.store(0);
220 0 : m_timed_out_trigger_records.store(0);
221 0 : m_abandoned_trigger_records.store(0);
222 0 : m_unexpected_fragments.store(0);
223 0 : m_lost_fragments.store(0);
224 0 : m_invalid_requests.store(0);
225 0 : m_duplicated_trigger_ids.store(0);
226 :
227 : // 19-Dec-2024, KAB: check that DataRequest senders are ready to send. This is done so
228 : // that the IOManager infrastructure fetches the necessary connection details from
229 : // the ConnectivityService at 'start' time, instead of the first time that the sender
230 : // is used to send a message. This avoids delays in the sending of the first request in
231 : // the first data-taking run in a DAQ session. Such delays can lead to undesirable
232 : // system behavior like trigger inhibits.
233 0 : {
234 0 : std::unique_lock<std::mutex> lk(m_map_sourceid_connections_mutex);
235 0 : for (const auto& sid_sender : m_map_sourceid_connections) {
236 0 : std::shared_ptr<data_req_sender_t> sender = sid_sender.second;
237 0 : if (sender != nullptr) {
238 0 : bool is_ready = sender->is_ready_for_sending(std::chrono::milliseconds(100));
239 0 : TLOG_DEBUG(0) << "The DataRequest sender for " << sid_sender.first << " " << (is_ready ? "is" : "is not")
240 0 : << " ready.";
241 : }
242 0 : }
243 0 : }
244 :
245 0 : m_run_number.reset(new const daqdataformats::run_number_t(args.at("run").get<daqdataformats::run_number_t>()));
246 0 : m_stop_requested = false;
247 :
248 : // Register the callback to receive monitoring requests
249 0 : if (m_mon_receiver) {
250 0 : m_mon_requests.clear();
251 0 : m_mon_receiver->add_callback(std::bind(&TRBModule::tr_requested, this, std::placeholders::_1));
252 : }
253 :
254 0 : m_fragment_input->add_callback(std::bind(&TRBModule::fragments_callback, this, std::placeholders::_1));
255 0 : m_trigger_decision_input->add_callback(std::bind(&TRBModule::trigger_decision_callback, this, std::placeholders::_1));
256 :
257 0 : TLOG() << get_name() << " successfully started";
258 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_start() method";
259 0 : }
260 :
261 : void
262 0 : TRBModule::do_stop(const CommandData_t& /*args*/)
263 : {
264 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_stop() method";
265 : // Unregister the monitoring requests callback
266 :
267 0 : if (m_mon_receiver) {
268 0 : m_mon_receiver->remove_callback();
269 : }
270 :
271 0 : m_trigger_decision_input->remove_callback();
272 0 : m_fragment_input->remove_callback();
273 :
274 0 : m_stop_requested = true;
275 :
276 0 : flush_trigger_records();
277 :
278 0 : TLOG() << get_name() << " successfully stopped";
279 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_stop() method";
280 0 : }
281 :
282 : void
283 0 : TRBModule::tr_requested(const dfmessages::TRMonRequest& req)
284 : {
285 0 : ++m_trmon_request_counter;
286 :
287 : // Ignore requests that don't belong to the ongoing run
288 0 : if (req.run_number != *m_run_number)
289 0 : return;
290 :
291 : // Add requests to pending requests
292 : // To be done: choose a concurrent container implementation.
293 0 : const std::lock_guard<std::mutex> lock(m_mon_mutex);
294 0 : m_mon_requests.push_back(req);
295 0 : }
296 :
297 : void
298 0 : TRBModule::flush_trigger_records()
299 : {
300 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Starting draining phase ";
301 0 : std::chrono::steady_clock::time_point t1 = std::chrono::steady_clock::now();
302 :
303 : // //-------------------------------------------------
304 : // // Here we drain what has been left from the running condition
305 : // //--------------------------------------------------
306 :
307 : // create all possible trigger record
308 0 : std::vector<TriggerId> triggers;
309 0 : for (const auto& entry : m_trigger_records) {
310 0 : triggers.push_back(entry.first);
311 : }
312 :
313 : // create the trigger record and send it
314 0 : for (const auto& t : triggers) {
315 0 : send_trigger_record(t);
316 : }
317 :
318 0 : std::chrono::steady_clock::time_point t2 = std::chrono::steady_clock::now();
319 :
320 0 : std::chrono::duration<double> time_span = std::chrono::duration_cast<std::chrono::duration<double>>(t2 - t1);
321 :
322 0 : std::ostringstream oss_summ;
323 0 : oss_summ << ": Exiting the do_work() method, " << m_trigger_records.size() << " remaining Trigger Records"
324 0 : << std::endl
325 0 : << "Draining took : " << time_span.count() << " s";
326 0 : TLOG() << ProgressUpdate(ERS_HERE, get_name(), oss_summ.str());
327 :
328 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_work() method";
329 0 : } // NOLINT(readability/fn_size)
330 :
331 : void
332 0 : TRBModule::fragments_callback(std::unique_ptr<daqdataformats::Fragment>& temp_fragment)
333 : {
334 :
335 0 : auto start_time = std::chrono::steady_clock::now();
336 :
337 0 : TLOG_DEBUG(TLVL_FRAGMENT_RECEIVE) << get_name() << " Received fragment for trigger/sequence_number "
338 0 : << temp_fragment->get_trigger_number() << "."
339 0 : << temp_fragment->get_sequence_number() << " from "
340 0 : << temp_fragment->get_element_id();
341 :
342 0 : TriggerId temp_id(*temp_fragment);
343 0 : std::vector<TriggerId> complete;
344 0 : bool requested = false;
345 :
346 0 : { // Begin mutex block
347 0 : std::lock_guard<std::mutex> lk(m_trigger_records_mutex);
348 0 : auto it = m_trigger_records.find(temp_id);
349 :
350 0 : if (it != m_trigger_records.end()) {
351 :
352 : // check if the fragment has a Source Id that was desired
353 0 : daqdataformats::TriggerRecordHeader& header = it->second.second->get_header_ref();
354 :
355 0 : for (size_t i = 0; i < header.get_num_requested_components(); ++i) {
356 :
357 0 : const daqdataformats::ComponentRequest& request = header.at(i);
358 0 : if (request.component == temp_fragment->get_element_id()) {
359 : requested = true;
360 : break;
361 : }
362 :
363 : } // request loop
364 :
365 : } // if there is a corresponding trigger ID entry in the boook
366 :
367 0 : if (requested) {
368 0 : it->second.second->add_fragment(std::move(temp_fragment));
369 0 : ++m_fragment_counter;
370 0 : --m_pending_fragment_counter;
371 : } else {
372 0 : ers::error(UnexpectedFragment(
373 0 : ERS_HERE, temp_id, temp_fragment->get_fragment_type_code(), temp_fragment->get_element_id()));
374 0 : ++m_unexpected_fragments;
375 : }
376 :
377 : // Only do bookkeeping periodically
378 0 : if (std::chrono::duration_cast<std::chrono::milliseconds>(clock_type::now() - m_last_bookkeeping).count() > 1000) {
379 : //-------------------------------------------------
380 : // Check if trigger records are complete or timedout
381 : // and create dedicated record
382 : //--------------------------------------------------
383 0 : TLOG_DEBUG(TLVL_BOOKKEEPING) << "Bookeeping status: " << m_trigger_records.size()
384 0 : << " trigger records in progress ";
385 :
386 0 : for (const auto& tr : m_trigger_records) {
387 :
388 0 : auto comp_size = tr.second.second->get_fragments_ref().size();
389 0 : auto requ_size = tr.second.second->get_header_ref().get_num_requested_components();
390 0 : std::ostringstream message;
391 0 : message << tr.first << " with " << comp_size << '/' << requ_size << " components";
392 :
393 0 : if (comp_size == requ_size) {
394 :
395 0 : message << ": complete";
396 0 : complete.push_back(tr.first);
397 : }
398 :
399 0 : TLOG_DEBUG(TLVL_BOOKKEEPING) << message.str();
400 0 : m_last_bookkeeping = clock_type::now();
401 :
402 0 : } // loop over TRs to check if they are complete
403 0 : } else if (requested && it != m_trigger_records.end()) {
404 :
405 0 : auto comp_size = it->second.second->get_fragments_ref().size();
406 0 : auto requ_size = it->second.second->get_header_ref().get_num_requested_components();
407 :
408 0 : if (comp_size == requ_size) {
409 0 : complete.push_back(temp_id);
410 : }
411 : }
412 0 : } // End mutex block
413 : //------------------------------------------------
414 : // Create TriggerRecords and send them
415 : //-----------------------------------------------
416 :
417 0 : for (const auto& id : complete) {
418 :
419 0 : send_trigger_record(id);
420 :
421 : } // loop over compled trigger id
422 :
423 0 : check_stale_requests();
424 :
425 0 : ++m_received_fragments;
426 :
427 0 : auto end_time = std::chrono::steady_clock::now();
428 0 : m_fragment_processing_us += std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time).count();
429 0 : }
430 :
431 : void
432 0 : TRBModule::trigger_decision_callback(dfmessages::TriggerDecision& td)
433 : {
434 :
435 0 : auto start_time = std::chrono::steady_clock::now();
436 :
437 0 : if (td.run_number != *m_run_number) {
438 0 : ers::error(UnexpectedTriggerDecision(ERS_HERE, td.trigger_number, td.run_number, *m_run_number));
439 0 : ++m_unexpected_trigger_decisions;
440 0 : return;
441 : }
442 :
443 0 : ++m_received_trigger_decisions;
444 :
445 0 : create_trigger_records_and_dispatch(td);
446 : // check_stale_requests();
447 :
448 0 : auto end_time = std::chrono::steady_clock::now();
449 0 : m_td_processing_us += std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time).count();
450 : }
451 :
452 : TRBModule::trigger_record_ptr_t
453 0 : TRBModule::extract_trigger_record(const TriggerId& id)
454 : {
455 0 : std::unique_lock<std::mutex> lk(m_trigger_records_mutex);
456 0 : auto it = m_trigger_records.extract(id);
457 0 : m_open_trigger_record_cv.notify_one();
458 0 : lk.unlock();
459 :
460 0 : if (it.empty())
461 0 : return nullptr;
462 :
463 0 : trigger_record_ptr_t temp = std::move(it.mapped().second);
464 :
465 0 : auto time = clock_type::now();
466 0 : auto duration = time - it.mapped().first;
467 :
468 0 : m_data_waiting_time += std::chrono::duration_cast<duration_type>(duration).count();
469 :
470 0 : --m_trigger_decisions_counter;
471 0 : m_fragment_counter -= temp->get_fragments_ref().size();
472 :
473 0 : auto missing_fragments = temp->get_header_ref().get_num_requested_components() - temp->get_fragments_ref().size();
474 :
475 0 : if (missing_fragments > 0) {
476 :
477 0 : m_lost_fragments += missing_fragments;
478 0 : m_pending_fragment_counter -= missing_fragments;
479 0 : temp->get_header_ref().set_status_bit(TriggerRecordStatusBits::kIncomplete, true);
480 :
481 0 : ers::error(IncompleteTriggerRecord(ERS_HERE,
482 0 : (m_stop_requested.load() ? "at Stop time " : ""),
483 : id,
484 0 : temp->get_fragments_ref().size(),
485 0 : temp->get_header_ref().get_num_requested_components()));
486 : }
487 :
488 0 : return temp;
489 0 : }
490 :
491 : unsigned int
492 0 : TRBModule::create_trigger_records_and_dispatch(const dfmessages::TriggerDecision& td)
493 : {
494 :
495 0 : unsigned int new_tr_counter = 0;
496 :
497 : // check the whole time window
498 0 : daqdataformats::timestamp_t begin = std::numeric_limits<daqdataformats::timestamp_t>::max();
499 0 : daqdataformats::timestamp_t end = 0;
500 :
501 0 : for (const auto& component : td.components) {
502 0 : if (component.window_begin < begin)
503 0 : begin = component.window_begin;
504 0 : if (component.window_end > end)
505 0 : end = component.window_end;
506 : }
507 :
508 0 : daqdataformats::timestamp_diff_t tot_width = end - begin;
509 0 : daqdataformats::sequence_number_t max_sequence_number =
510 0 : (m_max_sequence_length > 0 && tot_width > 0) ? ((tot_width - 1) / m_max_sequence_length) : 0;
511 :
512 0 : TLOG_DEBUG(TLVL_WORK_STEPS) << get_name() << ": trig_number " << td.trigger_number << ": run_number " << td.run_number
513 0 : << ": trig_timestamp " << td.trigger_timestamp << " will have " << max_sequence_number + 1
514 0 : << " sequences";
515 :
516 0 : m_trigger_decision_width += tot_width;
517 :
518 : // create the trigger records
519 0 : for (daqdataformats::sequence_number_t sequence = 0; sequence <= max_sequence_number; ++sequence) {
520 :
521 0 : daqdataformats::timestamp_t slice_begin = begin + sequence * m_max_sequence_length;
522 0 : daqdataformats::timestamp_t slice_end =
523 0 : m_max_sequence_length > 0 ? std::min(slice_begin + m_max_sequence_length, end) : end;
524 :
525 0 : TLOG_DEBUG(TLVL_WORK_STEPS) << get_name() << ": trig_number " << td.trigger_number << ", sequence " << sequence
526 0 : << " ts=" << slice_begin << ":" << slice_end << " (TR " << begin << ":" << end << ")";
527 :
528 : // create the components cropped in time
529 0 : decltype(td.components) slice_components;
530 0 : for (const auto& component : td.components) {
531 :
532 0 : if (component.window_begin > slice_end)
533 0 : continue;
534 0 : if (component.window_end < slice_begin)
535 0 : continue;
536 :
537 0 : daqdataformats::timestamp_t new_begin = std::max(slice_begin, component.window_begin);
538 0 : daqdataformats::timestamp_t new_end = std::min(slice_end, component.window_end);
539 :
540 0 : daqdataformats::ComponentRequest temp(component.component, new_begin, new_end);
541 0 : slice_components.push_back(temp);
542 :
543 0 : m_data_request_width += new_end - new_begin;
544 :
545 : } // loop over component in trigger decision
546 :
547 : // Pleae note that the system could generate empty sequences
548 : // The code keeps them.
549 :
550 : // create the book entry
551 0 : TriggerId slice_id(td, sequence);
552 0 : {
553 0 : std::unique_lock<std::mutex> lk(m_trigger_records_mutex);
554 0 : m_open_trigger_record_cv.wait(lk, [&] { return m_trigger_records.size() < m_max_open_trigger_records; });
555 0 : auto it = m_trigger_records.find(slice_id);
556 0 : if (it != m_trigger_records.end()) {
557 0 : ers::error(DuplicatedTriggerDecision(ERS_HERE, slice_id));
558 0 : ++m_duplicated_trigger_ids;
559 0 : continue;
560 : }
561 :
562 : // create trigger record for the slice
563 0 : auto& entry = m_trigger_records[slice_id] = std::make_pair(clock_type::now(), trigger_record_ptr_t());
564 0 : trigger_record_ptr_t& trp = entry.second;
565 0 : trp.reset(new daqdataformats::TriggerRecord(slice_components));
566 0 : daqdataformats::TriggerRecord& tr = *trp;
567 :
568 0 : tr.get_header_ref().set_trigger_number(td.trigger_number);
569 0 : tr.get_header_ref().set_sequence_number(sequence);
570 0 : tr.get_header_ref().set_max_sequence_number(max_sequence_number);
571 0 : tr.get_header_ref().set_run_number(td.run_number);
572 0 : tr.get_header_ref().set_trigger_timestamp(td.trigger_timestamp);
573 0 : tr.get_header_ref().set_trigger_type(td.trigger_type);
574 0 : tr.get_header_ref().set_element_id(m_this_trb_source_id);
575 :
576 0 : m_trigger_decisions_counter++;
577 0 : m_pending_fragment_counter += slice_components.size();
578 0 : ++new_tr_counter;
579 0 : }
580 :
581 : // create and send the requests
582 0 : TLOG_DEBUG(TLVL_WORK_STEPS) << get_name() << ": Trigger Decision components: " << td.components.size()
583 0 : << ", slice components: " << slice_components.size();
584 0 : for (const auto& component : slice_components) {
585 :
586 0 : dfmessages::DataRequest dataReq;
587 0 : dataReq.trigger_number = td.trigger_number;
588 0 : dataReq.sequence_number = sequence;
589 0 : dataReq.run_number = td.run_number;
590 0 : dataReq.trigger_timestamp = td.trigger_timestamp;
591 0 : dataReq.readout_type = td.readout_type;
592 0 : dataReq.request_information = component;
593 0 : dataReq.data_destination = m_reply_connection;
594 0 : TLOG_DEBUG(TLVL_WORK_STEPS) << get_name() << ": TR " << slice_id << ": trig_timestamp "
595 0 : << dataReq.trigger_timestamp << ": SourceID " << component.component << ": window ["
596 0 : << dataReq.request_information.window_begin << ", "
597 0 : << dataReq.request_information.window_end << ']';
598 :
599 0 : dispatch_data_requests(std::move(dataReq), component.component);
600 :
601 0 : } // loop loop over component in the slice
602 :
603 0 : } // sequence loop
604 :
605 0 : return new_tr_counter;
606 : }
607 :
608 : bool
609 0 : TRBModule::dispatch_data_requests(dfmessages::DataRequest dr, const daqdataformats::SourceID& sid)
610 :
611 : {
612 :
613 : // find the queue for sourceid_req in the map
614 0 : std::unique_lock<std::mutex> lk(m_map_sourceid_connections_mutex);
615 0 : std::shared_ptr<data_req_sender_t> sender = nullptr;
616 0 : auto it_req = m_map_sourceid_connections.find(sid);
617 0 : if (it_req == m_map_sourceid_connections.end() || it_req->second == nullptr) {
618 :
619 : // if sourceid request is not valid. then print error and continue
620 0 : ers::error(
621 0 : dunedaq::dfmodules::DRSenderLookupFailed(ERS_HERE, sid, dr.run_number, dr.trigger_number, dr.sequence_number));
622 0 : ++m_invalid_requests;
623 0 : return false; // lk goes out of scope, is destroyed
624 : } else {
625 : // get the queue from map element
626 0 : sender = it_req->second;
627 : }
628 0 : lk.unlock();
629 :
630 0 : if (sender == nullptr) {
631 : // if sender lookup failed, report error and continue
632 0 : ers::error(
633 0 : dunedaq::dfmodules::DRSenderLookupFailed(ERS_HERE, sid, dr.run_number, dr.trigger_number, dr.sequence_number));
634 0 : ++m_invalid_requests;
635 0 : return false;
636 : }
637 :
638 : bool wasSentSuccessfully = false;
639 0 : do {
640 0 : TLOG_DEBUG(TLVL_DISPATCH_DATAREQ) << get_name() << ": Pushing the DataRequest from trigger/sequence number "
641 0 : << dr.trigger_number << "." << dr.sequence_number
642 0 : << " onto connection :" << sender->get_name();
643 :
644 : // send data request into the corresponding connection
645 0 : try {
646 0 : sender->send(std::move(dr), m_dreq_queue_timeout);
647 0 : wasSentSuccessfully = true;
648 0 : ++m_generated_data_requests;
649 0 : } catch (const ers::Issue& excpt) {
650 0 : std::ostringstream oss_warn;
651 0 : oss_warn << "Send to connection \"" << sender->get_name() << "\" failed";
652 0 : ers::warning(iomanager::OperationFailed(ERS_HERE, oss_warn.str(), excpt));
653 0 : }
654 0 : } while (!wasSentSuccessfully && !m_stop_requested.load());
655 :
656 : return wasSentSuccessfully;
657 0 : }
658 :
659 : bool
660 0 : TRBModule::send_trigger_record(const TriggerId& id)
661 : {
662 :
663 0 : trigger_record_ptr_t temp_record(extract_trigger_record(id));
664 :
665 : // Send to monitoring, if needed
666 :
667 0 : if (m_mon_receiver) {
668 0 : const std::lock_guard<std::mutex> lock(m_mon_mutex);
669 0 : auto it = m_mon_requests.begin();
670 0 : std::set<std::string> sent_destinations;
671 :
672 0 : while (it != m_mon_requests.end()) {
673 : // Only sent TR to each monitor once
674 0 : if (sent_destinations.count(it->data_destination)) {
675 0 : ++it;
676 0 : continue;
677 : }
678 :
679 : // send TR to mon if correct trigger type
680 0 : if ((it->trigger_type_mask & temp_record->get_header_data().trigger_type) != 0) {
681 0 : auto iom = iomanager::IOManager::get();
682 : bool wasSentSuccessfully = false;
683 0 : do {
684 0 : try {
685 : // HACK to copy the trigger record so we can send it off to monitoring
686 0 : auto trigger_record_bytes =
687 0 : serialization::serialize(temp_record, serialization::SerializationType::kMsgPack);
688 0 : trigger_record_ptr_t record_copy = serialization::deserialize<trigger_record_ptr_t>(trigger_record_bytes);
689 0 : iom->get_sender<trigger_record_ptr_t>(it->data_destination)
690 0 : ->send(std::move(record_copy), m_tr_queue_timeout);
691 0 : ++m_trmon_sent_counter;
692 0 : wasSentSuccessfully = true;
693 0 : } catch (const ers::Issue& excpt) {
694 0 : std::ostringstream oss_warn;
695 0 : oss_warn << "Sending TR to connection \"" << it->data_destination << "\" failed";
696 0 : ers::warning(iomanager::OperationFailed(ERS_HERE, oss_warn.str(), excpt));
697 0 : }
698 0 : } while (!m_stop_requested.load() && !wasSentSuccessfully);
699 0 : sent_destinations.insert(it->data_destination);
700 0 : it = m_mon_requests.erase(it);
701 0 : } else {
702 0 : ++it;
703 : }
704 : }
705 0 : } // if m_mon_receiver
706 :
707 : bool wasSentSuccessfully = false;
708 0 : do {
709 0 : try {
710 0 : m_trigger_record_output->send(std::move(temp_record), m_tr_queue_timeout);
711 0 : wasSentSuccessfully = true;
712 0 : ++m_generated_trigger_records;
713 0 : } catch (const ers::Issue& excpt) {
714 0 : ers::warning(excpt);
715 0 : }
716 0 : } while (!m_stop_requested.load() && !wasSentSuccessfully); // push while loop
717 :
718 0 : if (!wasSentSuccessfully) {
719 0 : ++m_abandoned_trigger_records;
720 0 : m_lost_fragments += temp_record->get_fragments_ref().size();
721 0 : ers::error(dunedaq::dfmodules::AbandonedTriggerDecision(ERS_HERE, id));
722 : }
723 :
724 0 : return wasSentSuccessfully;
725 0 : }
726 :
727 : bool
728 0 : TRBModule::check_stale_requests()
729 : {
730 :
731 0 : bool book_updates = false;
732 :
733 : // -----------------------------------------------
734 : // optionally send over stale trigger records
735 : // -----------------------------------------------
736 :
737 0 : if (m_trigger_timeout.count() > 0) {
738 :
739 0 : std::vector<TriggerId> stale_triggers;
740 0 : {
741 0 : std::lock_guard<std::mutex> lk(m_trigger_records_mutex);
742 0 : for (auto it = m_trigger_records.begin(); it != m_trigger_records.end(); ++it) {
743 :
744 0 : daqdataformats::TriggerRecord& tr = *it->second.second;
745 :
746 0 : auto tr_time = clock_type::now() - it->second.first;
747 :
748 0 : if (tr_time > m_trigger_timeout) {
749 :
750 0 : ers::error(TimedOutTriggerDecision(ERS_HERE, it->first, tr.get_header_ref().get_trigger_timestamp()));
751 :
752 : // mark trigger record for seding
753 0 : stale_triggers.push_back(it->first);
754 0 : ++m_timed_out_trigger_records;
755 :
756 0 : book_updates = true;
757 : }
758 :
759 : } // trigger record loop
760 0 : }
761 : // create the trigger record and send it
762 0 : for (const auto& t : stale_triggers) {
763 0 : send_trigger_record(t);
764 : }
765 :
766 0 : } // m_trigger_timeout > 0
767 :
768 0 : return book_updates;
769 : }
770 :
771 : } // namespace dfmodules
772 : } // namespace dunedaq
773 :
774 0 : DEFINE_DUNE_DAQ_MODULE(dunedaq::dfmodules::TRBModule)
|
