Line data Source code
1 : /**
2 : * @file TriggerRecordBuilderData.cpp TriggerRecordBuilderData Class Implementation
3 : *
4 : * The TriggerRecordBuilderData class represents the current state of a dataflow application's Trigger Record buffers
5 : * for use by the DFO.
6 : *
7 : * This is part of the DUNE DAQ Application Framework, copyright 2020.
8 : * Licensing/copyright details are in the COPYING file that you should have
9 : * received with this code.
10 : */
11 :
12 : #include "dfmodules/TriggerRecordBuilderData.hpp"
13 : #include "dfmodules/opmon/TRBuilderData.pb.h"
14 :
15 : #include "logging/Logging.hpp"
16 :
17 : #include <limits>
18 : #include <memory>
19 : #include <string>
20 : #include <utility>
21 :
22 : /**
23 : * @brief Name used by TRACE TLOG calls from this source file
24 : */
25 : #define TRACE_NAME "TRBData" // NOLINT
26 :
27 : namespace dunedaq {
28 : namespace dfmodules {
29 :
30 3 : TriggerRecordBuilderData::TriggerRecordBuilderData(std::string connection_name, size_t busy_threshold)
31 3 : : m_busy_threshold(busy_threshold)
32 3 : , m_free_threshold(busy_threshold)
33 3 : , m_is_busy(false)
34 3 : , m_in_error(false)
35 9 : , m_connection_name(connection_name)
36 : {
37 3 : }
38 :
39 3 : TriggerRecordBuilderData::TriggerRecordBuilderData(std::string connection_name,
40 : size_t busy_threshold,
41 3 : size_t free_threshold)
42 3 : : m_busy_threshold(busy_threshold)
43 3 : , m_free_threshold(busy_threshold)
44 3 : , m_is_busy(false)
45 3 : , m_in_error(false)
46 9 : , m_connection_name(connection_name)
47 : {
48 3 : if (busy_threshold < free_threshold)
49 1 : throw dfmodules::DFOThresholdsNotConsistent(ERS_HERE, busy_threshold, free_threshold);
50 7 : }
51 :
52 : std::shared_ptr<AssignedTriggerDecision>
53 8 : TriggerRecordBuilderData::extract_assignment(daqdataformats::trigger_number_t trigger_number)
54 : {
55 8 : std::shared_ptr<AssignedTriggerDecision> dec_ptr;
56 8 : auto lk = std::lock_guard<std::mutex>(m_assigned_trigger_decisions_mutex);
57 10 : for (auto it = m_assigned_trigger_decisions.begin(); it != m_assigned_trigger_decisions.end(); ++it) {
58 7 : if ((*it)->decision.trigger_number == trigger_number) {
59 5 : dec_ptr = *it;
60 5 : m_assigned_trigger_decisions.erase(it);
61 5 : break;
62 : }
63 : }
64 :
65 8 : if (m_assigned_trigger_decisions.size() < m_free_threshold.load())
66 6 : m_is_busy.store(false);
67 :
68 8 : return dec_ptr;
69 8 : }
70 :
71 : std::shared_ptr<AssignedTriggerDecision>
72 2 : TriggerRecordBuilderData::get_assignment(daqdataformats::trigger_number_t trigger_number) const
73 : {
74 2 : auto lk = std::lock_guard<std::mutex>(m_assigned_trigger_decisions_mutex);
75 2 : for (auto ptr : m_assigned_trigger_decisions) {
76 1 : if (ptr->decision.trigger_number == trigger_number) {
77 1 : return ptr;
78 : }
79 0 : }
80 :
81 1 : return nullptr;
82 2 : }
83 :
84 : std::shared_ptr<AssignedTriggerDecision>
85 6 : TriggerRecordBuilderData::complete_assignment(daqdataformats::trigger_number_t trigger_number,
86 : std::function<void(nlohmann::json&)> metadata_fun)
87 : {
88 :
89 6 : auto dec_ptr = extract_assignment(trigger_number);
90 :
91 6 : if (dec_ptr == nullptr)
92 2 : throw AssignedTriggerDecisionNotFound(ERS_HERE, trigger_number, m_connection_name);
93 :
94 4 : auto now = std::chrono::steady_clock::now();
95 4 : auto time = std::chrono::duration_cast<std::chrono::microseconds>(now - dec_ptr->assigned_time);
96 4 : {
97 4 : auto lk = std::lock_guard<std::mutex>(m_latency_info_mutex);
98 4 : m_latency_info.emplace_back(now, time);
99 :
100 4 : if (m_latency_info.size() > 1000)
101 0 : m_latency_info.pop_front();
102 4 : }
103 :
104 4 : if (metadata_fun)
105 1 : metadata_fun(m_metadata);
106 :
107 4 : ++m_complete_counter;
108 4 : auto completion_time = std::chrono::duration_cast<std::chrono::microseconds>(now - dec_ptr->assigned_time);
109 4 : if (completion_time.count() < m_min_complete_time.load())
110 2 : m_min_complete_time.store(completion_time.count());
111 4 : if (completion_time.count() > m_max_complete_time.load())
112 4 : m_max_complete_time.store(completion_time.count());
113 :
114 4 : opmon::TRCompleteInfo i;
115 4 : i.set_completion_time(completion_time.count());
116 4 : i.set_tr_number(dec_ptr->decision.trigger_number);
117 4 : i.set_run_number(dec_ptr->decision.run_number);
118 4 : i.set_trigger_type(dec_ptr->decision.trigger_type);
119 4 : publish(std::move(i), {}, opmonlib::to_level(opmonlib::EntryOpMonLevel::kEventDriven));
120 :
121 4 : return dec_ptr;
122 6 : }
123 :
124 : std::list<std::shared_ptr<AssignedTriggerDecision>>
125 3 : TriggerRecordBuilderData::flush()
126 : {
127 :
128 3 : auto lk = std::lock_guard<std::mutex>(m_assigned_trigger_decisions_mutex);
129 3 : std::list<std::shared_ptr<AssignedTriggerDecision>> ret;
130 :
131 4 : for (const auto& td : m_assigned_trigger_decisions) {
132 1 : ret.push_back(td);
133 : }
134 3 : m_assigned_trigger_decisions.clear();
135 :
136 3 : auto stat_lock = std::lock_guard<std::mutex>(m_latency_info_mutex);
137 3 : m_latency_info.clear();
138 3 : m_is_busy = false;
139 :
140 3 : m_in_error = false;
141 3 : m_metadata = nlohmann::json();
142 :
143 3 : return ret;
144 3 : }
145 :
146 : std::shared_ptr<AssignedTriggerDecision>
147 9 : TriggerRecordBuilderData::make_assignment(dfmessages::TriggerDecision decision)
148 : {
149 9 : return std::make_shared<AssignedTriggerDecision>(decision, m_connection_name);
150 : }
151 :
152 : void
153 10 : TriggerRecordBuilderData::add_assignment(std::shared_ptr<AssignedTriggerDecision> assignment)
154 : {
155 10 : auto lk = std::lock_guard<std::mutex>(m_assigned_trigger_decisions_mutex);
156 :
157 10 : if (is_in_error())
158 1 : throw NoSlotsAvailable(ERS_HERE, assignment->decision.trigger_number, m_connection_name);
159 :
160 9 : m_assigned_trigger_decisions.push_back(assignment);
161 9 : TLOG_DEBUG(13) << "Size of assigned_trigger_decision list is " << m_assigned_trigger_decisions.size();
162 :
163 9 : if (m_assigned_trigger_decisions.size() >= m_busy_threshold.load()) {
164 4 : m_is_busy.store(true);
165 : }
166 10 : }
167 :
168 : void
169 4 : TriggerRecordBuilderData::generate_opmon_data()
170 : {
171 4 : metric_t info;
172 4 : info.set_min_time_since_assignment(std::numeric_limits<time_counter_t>::max());
173 4 : info.set_max_time_since_assignment(0);
174 :
175 4 : time_counter_t time = 0;
176 :
177 4 : auto lk = std::unique_lock<std::mutex>(m_assigned_trigger_decisions_mutex);
178 4 : info.set_outstanding_decisions(m_assigned_trigger_decisions.size());
179 4 : auto current_time = std::chrono::steady_clock::now();
180 6 : for (const auto& dec_ptr : m_assigned_trigger_decisions) {
181 2 : auto us_since_assignment =
182 2 : std::chrono::duration_cast<std::chrono::microseconds>(current_time - dec_ptr->assigned_time);
183 2 : time += us_since_assignment.count();
184 2 : if (us_since_assignment.count() < info.min_time_since_assignment())
185 2 : info.set_min_time_since_assignment(us_since_assignment.count());
186 2 : if (us_since_assignment.count() > info.max_time_since_assignment())
187 1 : info.set_max_time_since_assignment(us_since_assignment.count());
188 : }
189 4 : lk.unlock();
190 :
191 4 : info.set_total_time_since_assignment(time);
192 :
193 : // estimate of the capcity
194 4 : auto completed_trigger_records = m_complete_counter.exchange(0);
195 4 : if (completed_trigger_records > 0) {
196 1 : m_last_average_time =
197 1 : 1e-6 * 0.5 * (m_min_complete_time.exchange(0) + m_max_complete_time.exchange(0)); // in seconds
198 : }
199 :
200 4 : if (m_last_average_time > 0.) {
201 : // prediction rate metrics
202 1 : info.set_capacity_rate(0.5 * (m_busy_threshold.load() + m_free_threshold.load()) / m_last_average_time);
203 : }
204 :
205 4 : publish(std::move(info));
206 4 : }
207 :
208 : std::chrono::microseconds
209 1 : TriggerRecordBuilderData::average_latency(std::chrono::steady_clock::time_point since) const
210 : {
211 1 : auto lk = std::lock_guard<std::mutex>(m_latency_info_mutex);
212 1 : std::chrono::microseconds sum = std::chrono::microseconds(0);
213 1 : size_t count = 0;
214 2 : for (auto it = m_latency_info.rbegin(); it != m_latency_info.rend(); ++it) {
215 1 : if (it->first < since)
216 : break;
217 :
218 1 : count++;
219 1 : sum += it->second;
220 : }
221 :
222 1 : return sum / count;
223 1 : }
224 :
225 : } // namespace dfmodules
226 : } // namespace dunedaq
|
