Line data Source code
1 : /**
2 : * @file TRBModule.hpp
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 : #ifndef DFMODULES_PLUGINS_TRIGGERRECORDBUILDER_HPP_
10 : #define DFMODULES_PLUGINS_TRIGGERRECORDBUILDER_HPP_
11 :
12 : #include "appmodel/ReadoutApplication.hpp"
13 : #include "appmodel/SmartDaqApplication.hpp"
14 : #include "appmodel/TRBConf.hpp"
15 : #include "daqdataformats/Fragment.hpp"
16 : #include "daqdataformats/SourceID.hpp"
17 : #include "daqdataformats/TriggerRecord.hpp"
18 : #include "daqdataformats/Types.hpp"
19 : #include "dfmessages/DataRequest.hpp"
20 : #include "dfmessages/TRMonRequest.hpp"
21 : #include "dfmessages/TriggerDecision.hpp"
22 : #include "dfmessages/Types.hpp"
23 :
24 : #include "appfwk/DAQModule.hpp"
25 : #include "iomanager/Receiver.hpp"
26 : #include "iomanager/Sender.hpp"
27 : #include "logging/Logging.hpp" // NOTE: if ISSUES ARE DECLARED BEFORE include logging/Logging.hpp, TLOG_DEBUG<<issue wont work.
28 : #include "utilities/WorkerThread.hpp"
29 :
30 : #include "dfmodules/opmon/TRBModule.pb.h"
31 :
32 : #include <chrono>
33 : #include <list>
34 : #include <map>
35 : #include <memory>
36 : #include <mutex>
37 : #include <string>
38 : #include <tuple>
39 : #include <utility>
40 : #include <vector>
41 :
42 : namespace dunedaq {
43 :
44 : namespace dfmodules {
45 :
46 : /**
47 : * @brief TriggerId is a little class that defines a unique identifier for a
48 : * trigger decision/record It also provides an operator < to be used by map to
49 : * optimise bookkeeping
50 : */
51 : struct TriggerId
52 : {
53 :
54 : TriggerId() = default;
55 :
56 0 : explicit TriggerId(const dfmessages::TriggerDecision& td,
57 : daqdataformats::sequence_number_t s = daqdataformats::TypeDefaults::s_invalid_sequence_number)
58 0 : : trigger_number(td.trigger_number)
59 0 : , sequence_number(s)
60 0 : , run_number(td.run_number)
61 : {
62 0 : ;
63 0 : }
64 0 : explicit TriggerId(daqdataformats::Fragment& f)
65 0 : : trigger_number(f.get_trigger_number())
66 0 : , sequence_number(f.get_sequence_number())
67 0 : , run_number(f.get_run_number())
68 : {
69 0 : ;
70 0 : }
71 :
72 : daqdataformats::trigger_number_t trigger_number;
73 : daqdataformats::sequence_number_t sequence_number;
74 : daqdataformats::run_number_t run_number;
75 :
76 0 : bool operator<(const TriggerId& other) const noexcept
77 : {
78 0 : return std::tuple(trigger_number, sequence_number, run_number) <
79 0 : std::tuple(other.trigger_number, other.sequence_number, other.run_number);
80 : }
81 :
82 0 : friend std::ostream& operator<<(std::ostream& out, const TriggerId& id) noexcept
83 : {
84 0 : out << id.trigger_number << '-' << id.sequence_number << '/' << id.run_number;
85 0 : return out;
86 : }
87 :
88 0 : friend TraceStreamer& operator<<(TraceStreamer& out, const TriggerId& id) noexcept
89 : {
90 0 : return out << id.trigger_number << '.' << id.sequence_number << "/" << id.run_number;
91 : }
92 :
93 : friend std::istream& operator>>(std::istream& in, TriggerId& id)
94 : {
95 : char t1, t2;
96 : in >> id.trigger_number >> t1 >> id.sequence_number >> t2 >> id.run_number;
97 : return in;
98 : }
99 : };
100 :
101 : } // namespace dfmodules
102 :
103 : /**
104 : * @brief Unexpected trigger decision
105 : */
106 0 : ERS_DECLARE_ISSUE(dfmodules, ///< Namespace
107 : UnexpectedTriggerDecision, ///< Issue class name
108 : "Unexpected Trigger Decisions: " << trigger << '/' << decision_run << " while in run " << current_run,
109 : ((daqdataformats::trigger_number_t)trigger) ///< Message parameters
110 : ((daqdataformats::run_number_t)decision_run) ///< Message parameters
111 : ((daqdataformats::run_number_t)current_run) ///< Message parameters
112 : )
113 :
114 : /**
115 : * @brief Timed out Trigger Decision
116 : */
117 0 : ERS_DECLARE_ISSUE(dfmodules, ///< Namespace
118 : TimedOutTriggerDecision, ///< Issue class name
119 : "trigger id: " << trigger_id << " generate at: " << trigger_timestamp << " timed out", ///< Message
120 : ((dfmodules::TriggerId)trigger_id) ///< Message parameters
121 : ((daqdataformats::timestamp_t)trigger_timestamp) ///< Message parameters
122 : )
123 :
124 : /**
125 : * @brief Unexpected fragment
126 : */
127 0 : ERS_DECLARE_ISSUE(dfmodules, ///< Namespace
128 : UnexpectedFragment, ///< Issue class name
129 : "Unexpected Fragment for triggerID " << trigger_id << ", type " << fragment_type << ", " << source_id,
130 : ((dfmodules::TriggerId)trigger_id) ///< Message parameters
131 : ((daqdataformats::fragment_type_t)fragment_type) ///< Message parameters
132 : ((daqdataformats::SourceID)source_id) ///< Message parameters
133 : )
134 :
135 : /**
136 : * @brief Duplicate trigger decision
137 : */
138 0 : ERS_DECLARE_ISSUE(dfmodules, ///< Namespace
139 : DuplicatedTriggerDecision, ///< Issue class name
140 : "Duplicated trigger ID " << trigger_id,
141 : ((dfmodules::TriggerId)trigger_id) ///< Message parameters
142 : )
143 :
144 : /**
145 : * @brief Abandoned TR
146 : */
147 0 : ERS_DECLARE_ISSUE(dfmodules, ///< Namespace
148 : AbandonedTriggerDecision, ///< Issue class name
149 : "trigger ID " << trigger_id << " could not be sent to writing and it's lost",
150 : ((dfmodules::TriggerId)trigger_id) ///< Message parameters
151 : )
152 :
153 : /**
154 : * @brief Incomplete TR
155 : */
156 0 : ERS_DECLARE_ISSUE(dfmodules, ///< Namespace
157 : IncompleteTriggerRecord, ///< Issue class name
158 : "sending incomplete TriggerRecord downstream "
159 : << optional_stop_time_phrase << " (trigger/run_number=" << id << ", " << num_frags_present << " of "
160 : << num_components_requested << " fragments included)",
161 : ((std::string)optional_stop_time_phrase)((dfmodules::TriggerId)id)((int)num_frags_present)(
162 : (int)num_components_requested) ///< Message parameters
163 : )
164 :
165 : /**
166 : * @brief Missing connection ID
167 : */
168 0 : ERS_DECLARE_ISSUE(dfmodules, ///< Namespace
169 : MissingConnectionID, ///< Issue class name
170 : "No connection ID was found for connection name \""
171 : << conn_name << "\" in the conn_ref list that was provided at 'init' time.",
172 : ((std::string)conn_name) ///< Message parameters
173 : )
174 :
175 : namespace dfmodules {
176 :
177 : /**
178 : * @brief TRBModule is the Module that collects Trigger
179 : TriggersDecisions, sends the corresponding data requests and collects Fragment
180 : to form a complete Trigger Record. The TR then sent out possibly to a write
181 : module
182 : */
183 : class TRBModule : public dunedaq::appfwk::DAQModule
184 : {
185 : public:
186 : /**
187 : * @brief TRBModule Constructor
188 : * @param name Instance name for this TRBModule instance
189 : */
190 : explicit TRBModule(const std::string& name);
191 :
192 : TRBModule(const TRBModule&) = delete; ///< TRBModule is not copy-constructible
193 : TRBModule& operator=(const TRBModule&) = delete; ///< TRBModule is not copy-assignable
194 : TRBModule(TRBModule&&) = delete; ///< TRBModule is not move-constructible
195 : TRBModule& operator=(TRBModule&&) = delete; ///< TRBModule is not move-assignable
196 :
197 : void init(std::shared_ptr<appfwk::ConfigurationManager> mcfg) override;
198 :
199 : void generate_opmon_data() override;
200 :
201 : protected:
202 : using trigger_decision_receiver_t = iomanager::ReceiverConcept<dfmessages::TriggerDecision>;
203 : using data_req_sender_t = iomanager::SenderConcept<dfmessages::DataRequest>;
204 : using fragment_receiver_t = iomanager::ReceiverConcept<std::unique_ptr<daqdataformats::Fragment>>;
205 :
206 : using trigger_record_ptr_t = std::unique_ptr<daqdataformats::TriggerRecord>;
207 : using trigger_record_sender_t = iomanager::SenderConcept<trigger_record_ptr_t>;
208 :
209 : void trigger_decision_callback(dfmessages::TriggerDecision& td);
210 : void fragments_callback(std::unique_ptr<daqdataformats::Fragment>& frag);
211 :
212 : trigger_record_ptr_t extract_trigger_record(const TriggerId&);
213 : // build_trigger_record will allocate memory and then orphan it to the caller
214 : // via the returned pointer Plese note that the method will destroy the memory
215 : // saved in the bookkeeping map
216 :
217 : unsigned int create_trigger_records_and_dispatch(const dfmessages::TriggerDecision&);
218 :
219 : bool dispatch_data_requests(dfmessages::DataRequest, const daqdataformats::SourceID&);
220 :
221 : bool send_trigger_record(const TriggerId&);
222 : // this creates a trigger record and send it
223 :
224 : bool check_stale_requests();
225 : // it returns true when there are changes in the book = a TR timed out
226 :
227 : void flush_trigger_records();
228 :
229 : private:
230 : // Commands
231 : void do_conf(const CommandData_t&);
232 : void do_scrap(const CommandData_t&);
233 : void do_start(const CommandData_t&);
234 : void do_stop(const CommandData_t&);
235 :
236 : // Monitoring callback
237 : void tr_requested(const dfmessages::TRMonRequest&);
238 :
239 : // Threading
240 : std::atomic<bool> m_stop_requested;
241 :
242 : // Configuration
243 : const appmodel::TRBConf* m_trb_conf;
244 : std::chrono::milliseconds m_tr_queue_timeout;
245 : std::chrono::milliseconds m_dreq_queue_timeout;
246 : std::string m_reply_connection;
247 : size_t m_max_open_trigger_records;
248 : daqdataformats::SourceID m_this_trb_source_id;
249 :
250 : // Input Connections
251 : std::shared_ptr<trigger_decision_receiver_t> m_trigger_decision_input;
252 : std::shared_ptr<fragment_receiver_t> m_fragment_input;
253 :
254 : // Output connections
255 : std::shared_ptr<trigger_record_sender_t> m_trigger_record_output;
256 : mutable std::mutex m_map_sourceid_connections_mutex;
257 : std::map<daqdataformats::SourceID, std::shared_ptr<data_req_sender_t>>
258 : m_map_sourceid_connections; ///< Mappinng between SourceID and connections
259 :
260 : // bookeeping
261 : using clock_type = std::chrono::steady_clock;
262 : std::mutex m_trigger_records_mutex;
263 : clock_type::time_point m_last_bookkeeping{};
264 : std::map<TriggerId, std::pair<clock_type::time_point, trigger_record_ptr_t>> m_trigger_records;
265 : std::condition_variable m_open_trigger_record_cv;
266 :
267 : // Data request properties
268 : daqdataformats::timestamp_diff_t m_max_sequence_length;
269 :
270 : // Run information
271 : std::unique_ptr<const daqdataformats::run_number_t> m_run_number = nullptr;
272 :
273 : // Monitoring related variables
274 : std::mutex m_mon_mutex;
275 : std::shared_ptr<iomanager::ReceiverConcept<dfmessages::TRMonRequest>> m_mon_receiver;
276 : std::list<dfmessages::TRMonRequest> m_mon_requests;
277 :
278 : // book related metrics
279 : using metric_counter_type = uint64_t; // decltype(triggerrecordbuilderinfo::Info::pending_trigger_decisions);
280 : mutable std::atomic<metric_counter_type> m_trigger_decisions_counter = { 0 }; // currently
281 : mutable std::atomic<metric_counter_type> m_fragment_counter = { 0 }; // currently
282 : mutable std::atomic<metric_counter_type> m_pending_fragment_counter = { 0 }; // currently
283 :
284 : mutable std::atomic<metric_counter_type> m_timed_out_trigger_records = { 0 }; // in the run
285 : mutable std::atomic<metric_counter_type> m_unexpected_fragments = { 0 }; // in the run
286 : mutable std::atomic<metric_counter_type> m_unexpected_trigger_decisions = { 0 }; // in the run
287 : mutable std::atomic<metric_counter_type> m_lost_fragments = { 0 }; // in the run
288 : mutable std::atomic<metric_counter_type> m_invalid_requests = { 0 }; // in the run
289 : mutable std::atomic<metric_counter_type> m_duplicated_trigger_ids = { 0 }; // in the run
290 : mutable std::atomic<metric_counter_type> m_abandoned_trigger_records = { 0 }; // in the run
291 :
292 : mutable std::atomic<metric_counter_type> m_received_trigger_decisions = { 0 }; // in between calls
293 : mutable std::atomic<metric_counter_type> m_received_fragments = { 0 }; // in between calls
294 : mutable std::atomic<metric_counter_type> m_generated_trigger_records = { 0 }; // in between calls
295 : mutable std::atomic<metric_counter_type> m_generated_data_requests = { 0 }; // in between calls
296 : mutable std::atomic<metric_counter_type> m_data_waiting_time = { 0 }; // in between calls
297 : mutable std::atomic<metric_counter_type> m_trigger_decision_width = { 0 }; // in between calls
298 : mutable std::atomic<metric_counter_type> m_data_request_width = { 0 }; // in between calls
299 : mutable std::atomic<metric_counter_type> m_td_processing_us = { 0 }; // in between calls
300 : mutable std::atomic<metric_counter_type> m_fragment_processing_us = { 0 }; // in between calls
301 :
302 : mutable std::atomic<metric_counter_type> m_trmon_request_counter = { 0 };
303 : mutable std::atomic<metric_counter_type> m_trmon_sent_counter = { 0 };
304 :
305 : // time thresholds
306 : using duration_type = std::chrono::microseconds;
307 : duration_type m_old_trigger_threshold;
308 : duration_type m_trigger_timeout;
309 : };
310 : } // namespace dfmodules
311 : } // namespace dunedaq
312 :
313 : #endif // DFMODULES_PLUGINS_TRIGGERRECORDBUILDER_HPP_
|
