Line data Source code
1 : /**
2 : * @file TCProcessor.hpp TPC TP specific Task based raw processor
3 : *
4 : * This is part of the DUNE DAQ , copyright 2023.
5 : * Licensing/copyright details are in the COPYING file that you should have
6 : * received with this code.
7 : */
8 : #include "trigger/TCProcessor.hpp" // NOLINT(build/include)
9 :
10 : #include "iomanager/Sender.hpp"
11 : #include "logging/Logging.hpp"
12 :
13 : #include "datahandlinglibs/DataHandlingIssues.hpp"
14 : #include "datahandlinglibs/FrameErrorRegistry.hpp"
15 : #include "datahandlinglibs/ReadoutLogging.hpp"
16 : #include "datahandlinglibs/models/IterableQueueModel.hpp"
17 : #include "trigger/TCWrapper.hpp"
18 : #include "triggeralgs/TriggerCandidate.hpp"
19 :
20 : #include "appmodel/TCDataProcessor.hpp"
21 : #include "appmodel/TriggerDataHandlerModule.hpp"
22 :
23 : using dunedaq::datahandlinglibs::logging::TLVL_BOOKKEEPING;
24 : using dunedaq::datahandlinglibs::logging::TLVL_TAKE_NOTE;
25 :
26 : // THIS SHOULDN'T BE HERE!!!!! But it is necessary.....
27 : DUNE_DAQ_TYPESTRING(dunedaq::trigger::TCWrapper, "TriggerCandidate")
28 :
29 : namespace dunedaq {
30 : namespace trigger {
31 :
32 0 : TCProcessor::TCProcessor(std::unique_ptr<datahandlinglibs::FrameErrorRegistry>& error_registry,
33 0 : bool post_processing_enabled)
34 0 : : datahandlinglibs::TaskRawDataProcessorModel<TCWrapper>(error_registry, post_processing_enabled)
35 : {
36 0 : }
37 :
38 0 : TCProcessor::~TCProcessor() {}
39 :
40 : void
41 0 : TCProcessor::start(const appfwk::DAQModule::CommandData_t& args)
42 : {
43 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Entering start() method";
44 :
45 0 : m_running_flag.store(true);
46 0 : m_send_trigger_decisions_thread = std::thread(&TCProcessor::send_trigger_decisions, this);
47 0 : pthread_setname_np(m_send_trigger_decisions_thread.native_handle(), "mlt-dec"); // TODO: originally mlt-trig-dec
48 :
49 : // Reset stats
50 0 : m_tds_created_count.store(0);
51 0 : m_tds_sent_count.store(0);
52 0 : m_tds_dropped_count.store(0);
53 0 : m_tds_failed_bitword_count.store(0);
54 0 : m_tds_cleared_count.store(0);
55 : // per TC
56 0 : m_tc_received_count.store(0);
57 0 : m_tds_created_tc_count.store(0);
58 0 : m_tds_sent_tc_count.store(0);
59 0 : m_tds_dropped_tc_count.store(0);
60 0 : m_tds_failed_bitword_tc_count.store(0);
61 0 : m_tds_cleared_tc_count.store(0);
62 0 : m_tc_ignored_count.store(0);
63 0 : inherited::start(args);
64 :
65 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Exiting start() method";
66 0 : }
67 :
68 : void
69 0 : TCProcessor::stop(const appfwk::DAQModule::CommandData_t& args)
70 : {
71 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Entering stop() method";
72 :
73 0 : inherited::stop(args);
74 0 : m_running_flag.store(false);
75 :
76 : // Make sure condition_variable knows we flipped running flag
77 0 : {
78 0 : std::lock_guard<std::mutex> lock(m_td_vector_mutex);
79 0 : m_cv.notify_all();
80 0 : }
81 :
82 : // Wait for the TD-sending thread to stop
83 0 : m_send_trigger_decisions_thread.join();
84 :
85 : // Drop all TDs in vectors at run stage change. Have to do this
86 : // after joining m_send_trigger_decisions_thread so we don't
87 : // concurrently access the vectors
88 0 : clear_td_vectors();
89 :
90 0 : print_opmon_stats();
91 :
92 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Exiting stop() method";
93 0 : }
94 :
95 : void
96 0 : TCProcessor::conf(const appmodel::DataHandlerModule* cfg)
97 : {
98 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Entering conf() method";
99 :
100 0 : auto mtrg = cfg->cast<appmodel::TriggerDataHandlerModule>();
101 0 : if (mtrg == nullptr) {
102 0 : throw(InvalidConfiguration(ERS_HERE, "Provided null TriggerDataHandlerModule configuration!"));
103 : }
104 0 : for (auto output : mtrg->get_outputs()) {
105 0 : try {
106 0 : if (output->get_data_type() == "TriggerDecision") {
107 0 : m_td_sink = get_iom_sender<dfmessages::TriggerDecision>(output->UID());
108 : }
109 0 : } catch (const ers::Issue& excpt) {
110 0 : ers::error(datahandlinglibs::ResourceQueueError(ERS_HERE, "td", "DefaultRequestHandlerModel", excpt));
111 0 : }
112 : }
113 :
114 0 : auto dp = mtrg->get_module_configuration()->get_data_processor();
115 0 : auto proc_conf = dp->cast<appmodel::TCDataProcessor>();
116 :
117 : // Add all Source IDs to mandatoy links for now...
118 0 : for (auto const& link : mtrg->get_mandatory_source_ids()) {
119 0 : m_mandatory_links.push_back(
120 0 : dfmessages::SourceID{ daqdataformats::SourceID::string_to_subsystem(link->get_subsystem()), link->get_sid() });
121 : }
122 0 : for (auto const& link : mtrg->get_enabled_source_ids()) {
123 0 : m_mandatory_links.push_back(
124 0 : dfmessages::SourceID{ daqdataformats::SourceID::string_to_subsystem(link->get_subsystem()), link->get_sid() });
125 : }
126 :
127 : // TODO: Group links!
128 : // m_group_links_data = conf->get_groups_links();
129 0 : parse_group_links(m_group_links_data);
130 0 : print_group_links();
131 0 : m_total_group_links = m_group_links.size();
132 0 : TLOG_DEBUG(3) << "Total group links: " << m_total_group_links;
133 :
134 0 : m_tc_merging = proc_conf->get_merge_overlapping_tcs();
135 0 : m_ignore_tc_pileup = proc_conf->get_ignore_overlapping_tcs();
136 0 : m_buffer_timeout = proc_conf->get_buffer_timeout();
137 0 : m_send_timed_out_tds = (m_ignore_tc_pileup) ? false : proc_conf->get_td_out_of_timeout();
138 0 : m_td_readout_limit = proc_conf->get_td_readout_limit();
139 0 : m_ignored_tc_types = proc_conf->get_ignore_tc();
140 0 : m_ignoring_tc_types = !m_ignored_tc_types.empty();
141 :
142 : // Trigger bitwords
143 0 : std::vector<const appmodel::TriggerBitword*> bitwords = proc_conf->get_trigger_bitwords();
144 0 : m_use_bitwords = !bitwords.empty();
145 0 : if (m_use_bitwords) {
146 0 : set_trigger_bitwords(bitwords);
147 0 : print_trigger_bitwords();
148 : }
149 0 : TLOG_DEBUG(3) << "Use bitwords: " << m_use_bitwords;
150 0 : TLOG_DEBUG(3) << "Allow merging: " << m_tc_merging;
151 0 : TLOG_DEBUG(3) << "Ignore pileup: " << m_ignore_tc_pileup;
152 0 : TLOG_DEBUG(3) << "Buffer timeout: " << m_buffer_timeout;
153 0 : TLOG_DEBUG(3) << "Should send timed out TDs: " << m_send_timed_out_tds;
154 0 : TLOG_DEBUG(3) << "TD readout limit: " << m_td_readout_limit;
155 :
156 : // ROI map
157 0 : m_roi_conf_data = proc_conf->get_roi_group_conf();
158 0 : m_use_roi_readout = !m_roi_conf_data.empty();
159 0 : if (m_use_roi_readout) {
160 0 : parse_roi_conf(m_roi_conf_data);
161 0 : print_roi_conf(m_roi_conf);
162 : }
163 0 : TLOG_DEBUG(3) << "Use ROI readout?: " << m_use_roi_readout;
164 :
165 : // Custom readout map
166 0 : m_readout_window_map_data = proc_conf->get_tc_readout_map();
167 0 : m_use_readout_map = !m_readout_window_map_data.empty();
168 0 : if (m_use_readout_map) {
169 0 : parse_readout_map(m_readout_window_map_data);
170 0 : print_readout_map(m_readout_window_map);
171 : }
172 0 : TLOG_DEBUG(3) << "Use readout map: " << m_use_readout_map;
173 :
174 : // Ignoring TC types
175 0 : TLOG_DEBUG(3) << "Ignoring TC types: " << m_ignoring_tc_types;
176 0 : if (m_ignoring_tc_types) {
177 0 : TLOG_DEBUG(3) << "TC types to ignore: ";
178 0 : for (std::vector<unsigned int>::iterator it = m_ignored_tc_types.begin(); it != m_ignored_tc_types.end();) {
179 0 : TLOG_DEBUG(3) << *it;
180 0 : ++it;
181 : }
182 : }
183 0 : m_latency_monitoring.store(dp->get_latency_monitoring());
184 0 : inherited::add_postprocess_task(std::bind(&TCProcessor::make_td, this, std::placeholders::_1));
185 :
186 0 : inherited::conf(mtrg);
187 :
188 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Exiting conf() method";
189 0 : }
190 :
191 : void
192 0 : TCProcessor::scrap(const appfwk::DAQModule::CommandData_t& args)
193 : {
194 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Entering scrap() method";
195 :
196 0 : m_mandatory_links.clear();
197 0 : m_group_links.clear();
198 0 : m_roi_conf.clear();
199 0 : m_roi_conf_data.clear();
200 0 : m_roi_conf_ids.clear();
201 0 : m_roi_conf_probs.clear();
202 0 : m_roi_conf_probs_c.clear();
203 0 : m_pending_tds.clear();
204 0 : m_readout_window_map_data.clear();
205 0 : m_readout_window_map.clear();
206 0 : m_ignored_tc_types.clear();
207 :
208 0 : m_td_sink.reset();
209 :
210 0 : m_group_links_data.clear();
211 :
212 0 : inherited::scrap(args);
213 :
214 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Exiting scrap() method";
215 0 : }
216 :
217 : void
218 0 : TCProcessor::generate_opmon_data()
219 : {
220 0 : opmon::TCProcessorInfo info;
221 :
222 0 : info.set_tds_created_count(m_tds_created_count.load());
223 0 : info.set_tds_sent_count(m_tds_sent_count.load());
224 0 : info.set_tds_dropped_count(m_tds_dropped_count.load());
225 0 : info.set_tds_failed_bitword_count(m_tds_failed_bitword_count.load());
226 0 : info.set_tds_cleared_count(m_tds_cleared_count.load());
227 0 : info.set_tc_received_count(m_tc_received_count.load());
228 0 : info.set_tc_ignored_count(m_tc_ignored_count.load());
229 0 : info.set_tds_created_tc_count(m_tds_created_tc_count.load());
230 0 : info.set_tds_sent_tc_count(m_tds_sent_tc_count.load());
231 0 : info.set_tds_dropped_tc_count(m_tds_dropped_tc_count.load());
232 0 : info.set_tds_failed_bitword_tc_count(m_tds_failed_bitword_tc_count.load());
233 0 : info.set_tds_cleared_tc_count(m_tds_cleared_tc_count.load());
234 :
235 0 : this->publish(std::move(info));
236 :
237 0 : if (m_latency_monitoring.load() && m_running_flag.load()) {
238 0 : opmon::TriggerLatency lat_info;
239 :
240 0 : lat_info.set_latency_in(m_latency_instance.get_latency_in());
241 0 : lat_info.set_latency_out(m_latency_instance.get_latency_out());
242 :
243 0 : this->publish(std::move(lat_info));
244 0 : }
245 0 : }
246 :
247 : /**
248 : * Pipeline Stage 2.: put valid TCs in a vector for grouping and forming of TDs
249 : * */
250 : void
251 0 : TCProcessor::make_td(const TCWrapper* tcw)
252 : {
253 :
254 0 : auto tc = tcw->candidate;
255 0 : if (m_latency_monitoring.load())
256 0 : m_latency_instance.update_latency_in(tc.time_start);
257 0 : m_tc_received_count++;
258 :
259 0 : if ((m_use_readout_map) && (m_readout_window_map.count(tc.type))) {
260 0 : TLOG_DEBUG(3) << "Got TC of type " << static_cast<int>(tc.type) << ", timestamp " << tc.time_candidate
261 0 : << ", start/end " << tc.time_start << "/" << tc.time_end << ", readout start/end "
262 0 : << tc.time_candidate - m_readout_window_map[tc.type].first << "/"
263 0 : << tc.time_candidate + m_readout_window_map[tc.type].second;
264 : } else {
265 0 : TLOG_DEBUG(3) << "Got TC of type " << static_cast<int>(tc.type) << ", timestamp " << tc.time_candidate
266 0 : << ", start/end " << tc.time_start << "/" << tc.time_end;
267 : }
268 :
269 : // Option to ignore TC types (if given by config)
270 0 : if (m_ignoring_tc_types == true && check_trigger_type_ignore(static_cast<unsigned int>(tc.type)) == true) {
271 0 : TLOG_DEBUG(3) << " Ignore TC type: " << static_cast<unsigned int>(tc.type);
272 0 : m_tc_ignored_count++;
273 :
274 : /*FIXME: comment out this block: if a TC is to be ignored it shall just be ignored!
275 : if (m_tc_merging) {
276 : // Still need to check for overlap with existing TD, if overlaps, include in the TD, but don't extend
277 : // readout
278 : std::lock_guard<std::mutex> lock(m_td_vector_mutex);
279 : add_tc_ignored(*tc);
280 : }
281 : */
282 : } else {
283 0 : std::lock_guard<std::mutex> lock(m_td_vector_mutex);
284 0 : add_tc(tc);
285 0 : m_cv.notify_one();
286 0 : TLOG_DEBUG(10) << "pending tds size: " << m_pending_tds.size();
287 0 : }
288 0 : m_last_processed_daq_ts = tc.time_start;
289 0 : return;
290 0 : }
291 :
292 : dfmessages::TriggerDecision
293 0 : TCProcessor::create_decision(const PendingTD& pending_td)
294 : {
295 0 : m_earliest_tc_index = get_earliest_tc_index(pending_td);
296 0 : TLOG_DEBUG(5) << "earliest TC index: " << m_earliest_tc_index;
297 :
298 0 : if (pending_td.contributing_tcs.size() > 1) {
299 0 : TLOG_DEBUG(5) << "!!! TD created from " << pending_td.contributing_tcs.size() << " TCs !!!";
300 : }
301 :
302 0 : dfmessages::TriggerDecision decision;
303 0 : decision.trigger_number = 0; // filled by MLT
304 0 : decision.run_number = 0; // filled by MLT
305 0 : decision.trigger_timestamp = pending_td.contributing_tcs[m_earliest_tc_index].time_candidate;
306 0 : decision.readout_type = dfmessages::ReadoutType::kLocalized;
307 :
308 0 : TDBitset td_bitword = get_TD_bitword(pending_td);
309 0 : TLOG_DEBUG(5) << "[MLT] TD has bitword: " << td_bitword << " "
310 0 : << static_cast<dfmessages::trigger_type_t>(td_bitword.to_ulong());
311 0 : decision.trigger_type = static_cast<dfmessages::trigger_type_t>(td_bitword.to_ulong()); // m_trigger_type;
312 :
313 : // decision.trigger_type = 1; // m_trigger_type;
314 :
315 0 : TLOG_DEBUG(3) << ", TC detid: " << pending_td.contributing_tcs[m_earliest_tc_index].detid
316 0 : << ", TC type: " << static_cast<int>(pending_td.contributing_tcs[m_earliest_tc_index].type)
317 0 : << ", TC cont number: " << pending_td.contributing_tcs.size()
318 0 : << ", DECISION trigger type: " << decision.trigger_type
319 0 : << ", DECISION timestamp: " << decision.trigger_timestamp
320 0 : << ", request window begin: " << pending_td.readout_start
321 0 : << ", request window end: " << pending_td.readout_end;
322 :
323 0 : std::vector<dfmessages::ComponentRequest> requests =
324 0 : create_all_decision_requests(m_mandatory_links, pending_td.readout_start, pending_td.readout_end);
325 0 : add_requests_to_decision(decision, requests);
326 :
327 0 : if (!m_use_roi_readout) {
328 0 : for (const auto& [key, value] : m_group_links) {
329 0 : std::vector<dfmessages::ComponentRequest> group_requests =
330 0 : create_all_decision_requests(value, pending_td.readout_start, pending_td.readout_end);
331 0 : add_requests_to_decision(decision, group_requests);
332 0 : }
333 : } else { // using ROI readout
334 0 : roi_readout_make_requests(decision);
335 : }
336 :
337 0 : m_tds_created_count++;
338 0 : m_tds_created_tc_count += pending_td.contributing_tcs.size();
339 :
340 0 : return decision;
341 0 : }
342 :
343 : void
344 0 : TCProcessor::send_trigger_decisions()
345 : {
346 : // A unique lock that can be locked and unlocked
347 0 : std::unique_lock<std::mutex> lock(m_td_vector_mutex);
348 :
349 0 : while (m_running_flag) {
350 : // TODO: think about better implementation (notify?, something event driven)
351 0 : m_cv.wait_for(lock, std::chrono::microseconds(100));
352 0 : auto ready_tds = get_ready_tds(m_pending_tds);
353 0 : TLOG_DEBUG(10) << "ready tds: " << ready_tds.size() << ", updated pending tds: " << m_pending_tds.size();
354 :
355 0 : for (std::vector<PendingTD>::iterator it = ready_tds.begin(); it != ready_tds.end();) {
356 0 : call_tc_decision(*it);
357 0 : ++it;
358 : }
359 0 : }
360 0 : }
361 :
362 : void
363 0 : TCProcessor::call_tc_decision(const TCProcessor::PendingTD& pending_td)
364 : {
365 :
366 0 : if (m_use_bitwords) {
367 : // Check trigger bitwords
368 0 : TDBitset td_bitword = get_TD_bitword(pending_td);
369 0 : if (!check_trigger_bitwords(td_bitword)) {
370 : // Don't process further if the bitword check failed
371 0 : m_tds_failed_bitword_count++;
372 0 : m_tds_failed_bitword_tc_count += pending_td.contributing_tcs.size();
373 0 : return;
374 : }
375 : }
376 :
377 0 : dfmessages::TriggerDecision decision = create_decision(pending_td);
378 0 : auto tn = decision.trigger_number;
379 0 : auto td_ts = decision.trigger_timestamp;
380 :
381 0 : if (m_latency_monitoring.load())
382 0 : m_latency_instance.update_latency_out(pending_td.contributing_tcs.front().time_start);
383 0 : if (!m_td_sink->try_send(std::move(decision), iomanager::Sender::s_no_block)) {
384 0 : ers::warning(TDDropped(ERS_HERE, tn, td_ts));
385 0 : m_tds_dropped_count++;
386 0 : m_tds_dropped_tc_count += pending_td.contributing_tcs.size();
387 : } else {
388 0 : m_tds_sent_count++;
389 0 : m_tds_sent_tc_count += pending_td.contributing_tcs.size();
390 : }
391 0 : }
392 :
393 : void
394 0 : TCProcessor::add_tc(const triggeralgs::TriggerCandidate tc)
395 : {
396 0 : bool tc_dealt = false;
397 0 : int64_t tc_wallclock_arrived =
398 0 : std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
399 :
400 0 : if (m_tc_merging || m_ignore_tc_pileup) {
401 :
402 0 : for (std::vector<PendingTD>::iterator it = m_pending_tds.begin(); it != m_pending_tds.end();) {
403 : // Don't deal with TC here if there's no overlap
404 0 : if (!check_overlap(tc, *it)) {
405 0 : it++;
406 0 : continue;
407 : }
408 :
409 : // If overlap and ignoring, we drop the TC and flag it as dealt with.
410 0 : if (m_ignore_tc_pileup) {
411 0 : m_tds_dropped_tc_count++;
412 0 : tc_dealt = true;
413 0 : TLOG_DEBUG(3) << "TC overlapping with a previous TD, dropping!";
414 0 : break;
415 : }
416 :
417 : // If we're here, TC merging must be on, in which case we're actually
418 : // going to merge the TC into the TD.
419 0 : it->contributing_tcs.push_back(tc);
420 0 : if ((m_use_readout_map) && (m_readout_window_map.count(tc.type))) {
421 0 : TLOG_DEBUG(3) << "TC with start/end times " << tc.time_candidate - m_readout_window_map[tc.type].first << "/"
422 0 : << tc.time_candidate + m_readout_window_map[tc.type].second
423 0 : << " overlaps with pending TD with start/end times " << it->readout_start << "/"
424 0 : << it->readout_end;
425 0 : it->readout_start = ((tc.time_candidate - m_readout_window_map[tc.type].first) >= it->readout_start)
426 0 : ? it->readout_start
427 0 : : (tc.time_candidate - m_readout_window_map[tc.type].first);
428 0 : it->readout_end = ((tc.time_candidate + m_readout_window_map[tc.type].second) >= it->readout_end)
429 0 : ? (tc.time_candidate + m_readout_window_map[tc.type].second)
430 : : it->readout_end;
431 : } else {
432 0 : TLOG_DEBUG(3) << "TC with start/end times " << tc.time_start << "/" << tc.time_end
433 0 : << " overlaps with pending TD with start/end times " << it->readout_start << "/"
434 0 : << it->readout_end;
435 0 : it->readout_start = (tc.time_start >= it->readout_start) ? it->readout_start : tc.time_start;
436 0 : it->readout_end = (tc.time_end >= it->readout_end) ? tc.time_end : it->readout_end;
437 : }
438 0 : it->walltime_expiration = tc_wallclock_arrived + m_buffer_timeout;
439 0 : tc_dealt = true;
440 0 : break;
441 : }
442 : }
443 :
444 : // Don't do anything else if we've already dealt with the TC
445 0 : if (tc_dealt) {
446 0 : return;
447 : }
448 :
449 : // Create a new TD out of the TC
450 0 : PendingTD td_candidate;
451 0 : td_candidate.contributing_tcs.push_back(tc);
452 0 : if ((m_use_readout_map) && (m_readout_window_map.count(tc.type))) {
453 0 : td_candidate.readout_start = tc.time_candidate - m_readout_window_map[tc.type].first;
454 0 : td_candidate.readout_end = tc.time_candidate + m_readout_window_map[tc.type].second;
455 : } else {
456 0 : td_candidate.readout_start = tc.time_start;
457 0 : td_candidate.readout_end = tc.time_end;
458 : }
459 0 : td_candidate.walltime_expiration = tc_wallclock_arrived + m_buffer_timeout;
460 0 : m_pending_tds.push_back(td_candidate);
461 0 : }
462 :
463 : void
464 0 : TCProcessor::add_tc_ignored(const triggeralgs::TriggerCandidate tc)
465 : {
466 0 : for (std::vector<PendingTD>::iterator it = m_pending_tds.begin(); it != m_pending_tds.end();) {
467 0 : if (check_overlap(tc, *it)) {
468 0 : if ((m_use_readout_map) && (m_readout_window_map.count(tc.type))) {
469 0 : TLOG_DEBUG(3) << "!Ignored! TC with start/end times " << tc.time_candidate - m_readout_window_map[tc.type].first
470 0 : << "/" << tc.time_candidate + m_readout_window_map[tc.type].second
471 0 : << " overlaps with pending TD with start/end times " << it->readout_start << "/"
472 0 : << it->readout_end;
473 : } else {
474 0 : TLOG_DEBUG(3) << "!Ignored! TC with start/end times " << tc.time_start << "/" << tc.time_end
475 0 : << " overlaps with pending TD with start/end times " << it->readout_start << "/"
476 0 : << it->readout_end;
477 : }
478 0 : it->contributing_tcs.push_back(tc);
479 : break;
480 : }
481 0 : ++it;
482 : }
483 0 : return;
484 : }
485 :
486 : bool
487 0 : TCProcessor::check_overlap(const triggeralgs::TriggerCandidate& tc, const PendingTD& pending_td)
488 : {
489 0 : if ((m_use_readout_map) && (m_readout_window_map.count(tc.type))) {
490 0 : return !(((tc.time_candidate + m_readout_window_map[tc.type].second) < pending_td.readout_start) ||
491 0 : ((tc.time_candidate - m_readout_window_map[tc.type].first > pending_td.readout_end)));
492 : } else {
493 0 : return !((tc.time_end < pending_td.readout_start) || (tc.time_start > pending_td.readout_end));
494 : }
495 : }
496 :
497 : std::vector<TCProcessor::PendingTD>
498 0 : TCProcessor::get_ready_tds(std::vector<PendingTD>& pending_tds)
499 : {
500 0 : std::vector<PendingTD> return_tds;
501 0 : for (std::vector<PendingTD>::iterator it = pending_tds.begin(); it != pending_tds.end();) {
502 0 : auto timestamp_now =
503 0 : std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch())
504 0 : .count();
505 0 : if (timestamp_now >= it->walltime_expiration) {
506 0 : return_tds.push_back(*it);
507 0 : it = pending_tds.erase(it);
508 0 : } else if (check_td_readout_length(*it)) { // Also pass on TDs with (too) long readout window
509 0 : return_tds.push_back(*it);
510 0 : it = pending_tds.erase(it);
511 : } else {
512 0 : ++it;
513 : }
514 : }
515 0 : return return_tds;
516 0 : }
517 :
518 : int
519 0 : TCProcessor::get_earliest_tc_index(const PendingTD& pending_td)
520 : {
521 0 : int earliest_tc_index = -1;
522 0 : triggeralgs::timestamp_t earliest_tc_time;
523 0 : for (int i = 0; i < static_cast<int>(pending_td.contributing_tcs.size()); i++) {
524 0 : if (earliest_tc_index == -1) {
525 0 : earliest_tc_time = pending_td.contributing_tcs[i].time_candidate;
526 0 : earliest_tc_index = i;
527 : } else {
528 0 : if (pending_td.contributing_tcs[i].time_candidate < earliest_tc_time) {
529 0 : earliest_tc_time = pending_td.contributing_tcs[i].time_candidate;
530 0 : earliest_tc_index = i;
531 : }
532 : }
533 : }
534 0 : return earliest_tc_index;
535 : }
536 :
537 : bool
538 0 : TCProcessor::check_td_readout_length(const PendingTD& pending_td)
539 : {
540 0 : bool td_too_long = false;
541 0 : if (static_cast<int64_t>(pending_td.readout_end - pending_td.readout_start) >= m_td_readout_limit) {
542 0 : td_too_long = true;
543 0 : TLOG_DEBUG(3) << "Too long readout window: " << (pending_td.readout_end - pending_td.readout_start)
544 0 : << ", sending immediate TD!";
545 : }
546 0 : return td_too_long;
547 : }
548 :
549 : void
550 0 : TCProcessor::clear_td_vectors()
551 : {
552 0 : std::lock_guard<std::mutex> lock(m_td_vector_mutex);
553 0 : m_tds_cleared_count += m_pending_tds.size();
554 : // Use std::accumulate to sum up the sizes of all contributing_tcs vectors
555 0 : size_t tds_cleared_tc_count =
556 0 : std::accumulate(m_pending_tds.begin(), m_pending_tds.end(), 0, [](size_t sum, const PendingTD& ptd) {
557 0 : return sum + ptd.contributing_tcs.size();
558 0 : });
559 0 : m_tds_cleared_tc_count += tds_cleared_tc_count;
560 0 : m_pending_tds.clear();
561 0 : }
562 : bool
563 0 : TCProcessor::check_trigger_type_ignore(unsigned int tc_type)
564 : {
565 0 : bool ignore = false;
566 0 : for (std::vector<unsigned int>::iterator it = m_ignored_tc_types.begin(); it != m_ignored_tc_types.end();) {
567 0 : if (tc_type == *it) {
568 : ignore = true;
569 : break;
570 : }
571 0 : ++it;
572 : }
573 0 : return ignore;
574 : }
575 :
576 : void
577 0 : TCProcessor::print_trigger_bitwords()
578 : {
579 0 : TLOG_DEBUG(3) << "Configured trigger words:";
580 0 : for (const auto& bitword : m_trigger_bitwords) {
581 0 : TLOG_DEBUG(3) << bitword;
582 : }
583 0 : }
584 :
585 : bool
586 0 : TCProcessor::check_trigger_bitwords(const TDBitset& td_bitword) const
587 : {
588 0 : bool trigger_check = false;
589 0 : for (const auto& bitword : m_trigger_bitwords) {
590 0 : TLOG_DEBUG(15) << "TD word: " << td_bitword << ", bitword: " << bitword;
591 0 : trigger_check = ((td_bitword & bitword) == bitword);
592 0 : TLOG_DEBUG(15) << "&: " << (td_bitword & bitword);
593 0 : TLOG_DEBUG(15) << "trigger?: " << trigger_check;
594 0 : if (trigger_check == true)
595 : break;
596 : }
597 0 : return trigger_check;
598 : }
599 :
600 : void
601 0 : TCProcessor::set_trigger_bitwords(const std::vector<const appmodel::TriggerBitword*>& _bitwords)
602 : {
603 0 : for (const appmodel::TriggerBitword* bitword : _bitwords) {
604 0 : TDBitset temp_bitword;
605 :
606 0 : for (const std::string& tctype_str : bitword->get_bitword()) {
607 0 : TCType tc_type = static_cast<TCType>(dunedaq::trgdataformats::string_to_trigger_candidate_type(tctype_str));
608 :
609 0 : if (tc_type == TCType::kUnknown) {
610 0 : throw(InvalidConfiguration(ERS_HERE, "Provided an unknown/non-existent TC type as a trigger bitword!"));
611 : }
612 :
613 0 : temp_bitword.set(static_cast<uint64_t>(tc_type));
614 : }
615 :
616 0 : m_trigger_bitwords.push_back(temp_bitword);
617 : }
618 0 : }
619 :
620 : void
621 0 : TCProcessor::parse_readout_map(const std::vector<const appmodel::TCReadoutMap*>& data)
622 : {
623 0 : for (auto readout_type : data) {
624 0 : TCType tc_type =
625 0 : static_cast<TCType>(dunedaq::trgdataformats::string_to_trigger_candidate_type(readout_type->get_tc_type_name()));
626 :
627 : // Throw error if unknown TC type
628 0 : if (tc_type == TCType::kUnknown) {
629 0 : throw(InvalidConfiguration(ERS_HERE, "Provided an unknown TC type in the TCReadoutMap for the TCProcessor"));
630 : }
631 :
632 0 : m_readout_window_map[tc_type] = { readout_type->get_time_before(), readout_type->get_time_after() };
633 : }
634 0 : return;
635 : }
636 : void
637 0 : TCProcessor::print_readout_map(std::map<TCType, std::pair<triggeralgs::timestamp_t, triggeralgs::timestamp_t>> map)
638 : {
639 0 : TLOG_DEBUG(3) << "MLT TD Readout map:";
640 0 : for (auto const& [key, val] : map) {
641 0 : TLOG_DEBUG(3) << "Type: " << static_cast<int>(key) << ", before: " << val.first << ", after: " << val.second;
642 : }
643 0 : return;
644 : }
645 :
646 : void
647 0 : TCProcessor::parse_group_links(const nlohmann::json& data)
648 : {
649 0 : for (auto group : data) {
650 0 : const nlohmann::json& temp_links_data = group["links"];
651 0 : std::vector<dfmessages::SourceID> temp_links;
652 0 : for (auto link : temp_links_data) {
653 0 : temp_links.push_back(
654 0 : dfmessages::SourceID{ daqdataformats::SourceID::string_to_subsystem(link["subsystem"]), link["element"] });
655 0 : }
656 0 : m_group_links.insert({ group["group"], temp_links });
657 0 : }
658 0 : return;
659 : }
660 :
661 : void
662 0 : TCProcessor::print_group_links()
663 : {
664 0 : TLOG_DEBUG(3) << "MLT Group Links:";
665 0 : for (auto const& [key, val] : m_group_links) {
666 0 : TLOG_DEBUG(3) << "Group: " << key;
667 0 : for (auto const& link : val) {
668 0 : TLOG_DEBUG(3) << link;
669 : }
670 : }
671 0 : TLOG_DEBUG(3) << " ";
672 0 : return;
673 : }
674 : dfmessages::ComponentRequest
675 0 : TCProcessor::create_request_for_link(dfmessages::SourceID link,
676 : triggeralgs::timestamp_t start,
677 : triggeralgs::timestamp_t end)
678 : {
679 0 : dfmessages::ComponentRequest request;
680 0 : request.component = link;
681 0 : request.window_begin = start;
682 0 : request.window_end = end;
683 :
684 0 : TLOG_DEBUG(10) << "setting request start: " << request.window_begin;
685 0 : TLOG_DEBUG(10) << "setting request end: " << request.window_end;
686 :
687 0 : return request;
688 : }
689 :
690 : std::vector<dfmessages::ComponentRequest>
691 0 : TCProcessor::create_all_decision_requests(std::vector<dfmessages::SourceID> links,
692 : triggeralgs::timestamp_t start,
693 : triggeralgs::timestamp_t end)
694 : {
695 0 : std::vector<dfmessages::ComponentRequest> requests;
696 0 : for (auto link : links) {
697 0 : requests.push_back(create_request_for_link(link, start, end));
698 : }
699 0 : return requests;
700 0 : }
701 :
702 : void
703 0 : TCProcessor::add_requests_to_decision(dfmessages::TriggerDecision& decision,
704 : std::vector<dfmessages::ComponentRequest> requests)
705 : {
706 0 : for (auto request : requests) {
707 0 : decision.components.push_back(request);
708 : }
709 0 : }
710 :
711 : void
712 0 : TCProcessor::parse_roi_conf(const std::vector<const appmodel::ROIGroupConf*>& data)
713 : {
714 0 : int counter = 0;
715 0 : float run_sum = 0;
716 0 : for (auto group : data) {
717 0 : roi_group temp_roi_group;
718 0 : temp_roi_group.n_links = group->get_number_of_link_groups();
719 0 : temp_roi_group.prob = group->get_probability();
720 0 : temp_roi_group.time_window = group->get_time_window();
721 0 : temp_roi_group.mode = group->get_groups_selection_mode();
722 0 : m_roi_conf.insert({ counter, temp_roi_group });
723 0 : m_roi_conf_ids.push_back(counter);
724 0 : m_roi_conf_probs.push_back(group->get_probability());
725 0 : run_sum += static_cast<float>(group->get_probability());
726 0 : m_roi_conf_probs_c.push_back(run_sum);
727 0 : counter++;
728 0 : }
729 0 : return;
730 : }
731 :
732 : void
733 0 : TCProcessor::print_roi_conf(std::map<int, roi_group> roi_conf)
734 : {
735 0 : TLOG_DEBUG(3) << "ROI CONF";
736 0 : for (const auto& [key, value] : roi_conf) {
737 0 : TLOG_DEBUG(3) << "ID: " << key;
738 0 : TLOG_DEBUG(3) << "n links: " << value.n_links;
739 0 : TLOG_DEBUG(3) << "prob: " << value.prob;
740 0 : TLOG_DEBUG(3) << "time: " << value.time_window;
741 0 : TLOG_DEBUG(3) << "mode: " << value.mode;
742 : }
743 0 : TLOG_DEBUG(3) << " ";
744 0 : return;
745 : }
746 :
747 : float
748 0 : TCProcessor::get_random_num_float(float limit)
749 : {
750 0 : float rnd = (double)rand() / RAND_MAX;
751 0 : return rnd * (limit);
752 : }
753 :
754 : int
755 0 : TCProcessor::pick_roi_group_conf()
756 : {
757 0 : float rnd_num = get_random_num_float(m_roi_conf_probs_c.back());
758 0 : for (int i = 0; i < static_cast<int>(m_roi_conf_probs_c.size()); i++) {
759 0 : if (rnd_num < m_roi_conf_probs_c[i]) {
760 : return i;
761 : }
762 : }
763 : return -1;
764 : }
765 :
766 : int
767 0 : TCProcessor::get_random_num_int()
768 : {
769 0 : int range = m_total_group_links;
770 0 : int rnd = rand() % range;
771 0 : return rnd;
772 : }
773 : void
774 0 : TCProcessor::roi_readout_make_requests(dfmessages::TriggerDecision& decision)
775 : {
776 : // Get configuration at random (weighted)
777 0 : int group_pick = pick_roi_group_conf();
778 0 : if (group_pick != -1) {
779 0 : roi_group this_group = m_roi_conf[m_roi_conf_ids[group_pick]];
780 0 : std::vector<dfmessages::SourceID> links;
781 :
782 : // If mode is random, pick groups to request at random
783 0 : if (this_group.mode == "kRandom") {
784 0 : TLOG_DEBUG(10) << "RAND";
785 0 : std::set<int> groups;
786 0 : while (static_cast<int>(groups.size()) < this_group.n_links) {
787 0 : groups.insert(get_random_num_int());
788 : }
789 0 : for (auto r_id : groups) {
790 0 : links.insert(links.end(), m_group_links[r_id].begin(), m_group_links[r_id].end());
791 : }
792 : // Otherwise, read sequntially by IDs, starting at 0
793 0 : } else {
794 0 : TLOG_DEBUG(10) << "SEQ";
795 0 : int r_id = 0;
796 0 : while (r_id < this_group.n_links) {
797 0 : links.insert(links.end(), m_group_links[r_id].begin(), m_group_links[r_id].end());
798 0 : r_id++;
799 : }
800 : }
801 :
802 0 : TLOG_DEBUG(10) << "TD timestamp: " << decision.trigger_timestamp;
803 0 : TLOG_DEBUG(10) << "group window: " << this_group.time_window;
804 :
805 : // Once the components are prepared, create requests and append them to decision
806 0 : std::vector<dfmessages::ComponentRequest> requests = create_all_decision_requests(
807 0 : links, decision.trigger_timestamp - this_group.time_window, decision.trigger_timestamp + this_group.time_window);
808 0 : add_requests_to_decision(decision, requests);
809 0 : links.clear();
810 0 : }
811 0 : return;
812 : }
813 :
814 : TCProcessor::TDBitset
815 0 : TCProcessor::get_TD_bitword(const PendingTD& ready_td) const
816 : {
817 : // get only unique types
818 0 : std::vector<int> tc_types;
819 0 : for (auto tc : ready_td.contributing_tcs) {
820 0 : tc_types.push_back(static_cast<int>(tc.type));
821 0 : }
822 0 : tc_types.erase(std::unique(tc_types.begin(), tc_types.end()), tc_types.end());
823 :
824 : // form TD bitword
825 0 : TDBitset td_bitword;
826 0 : for (auto tc_type : tc_types) {
827 0 : td_bitword.set(tc_type);
828 : }
829 0 : return td_bitword;
830 0 : }
831 :
832 : void
833 0 : TCProcessor::print_opmon_stats()
834 : {
835 0 : TLOG() << "TCProcessor opmon counters summary:";
836 0 : TLOG() << "------------------------------";
837 0 : TLOG() << "TDs created: \t\t\t" << m_tds_created_count << " \t(" << m_tds_created_tc_count << " TCs)";
838 0 : TLOG() << "TDs sent: \t\t\t" << m_tds_sent_count << " \t(" << m_tds_sent_tc_count << " TCs)";
839 0 : TLOG() << "TDs dropped: \t\t\t" << m_tds_dropped_count << " \t(" << m_tds_dropped_tc_count << " TCs)";
840 0 : TLOG() << "TDs failed bitword check: \t" << m_tds_failed_bitword_count << " \t(" << m_tds_failed_bitword_tc_count
841 0 : << " TCs)";
842 0 : TLOG() << "TDs cleared: \t\t\t" << m_tds_cleared_count << " \t(" << m_tds_cleared_tc_count << " TCs)";
843 0 : TLOG() << "------------------------------";
844 0 : TLOG() << "TCs received: \t" << m_tc_received_count;
845 0 : TLOG() << "TCs ignored: \t" << m_tc_ignored_count;
846 0 : TLOG();
847 0 : }
848 :
849 : } // namespace fdreadoutlibs
850 : } // namespace dunedaq
|
