Line data Source code
1 : #ifndef TRGTOOLS_TAEMULATIONWORKER_CPP_
2 : #define TRGTOOLS_TAEMULATIONWORKER_CPP_
3 :
4 : #include "trgtools/TAEmulationWorker.hpp"
5 :
6 : namespace dunedaq::trgtools {
7 :
8 : uint16_t TAEmulationWorker::m_id_next = 0;
9 :
10 0 : TAEmulationWorker::TAEmulationWorker(std::vector<std::shared_ptr<hdf5libs::HDF5RawDataFile>> input_files,
11 : nlohmann::json config,
12 : std::pair<uint64_t, uint64_t> sliceid_range,
13 : bool run_parallel,
14 0 : bool quiet)
15 0 : : m_input_files(input_files)
16 0 : , m_sliceid_range(sliceid_range)
17 0 : , m_run_parallel(run_parallel)
18 0 : , m_quiet(quiet)
19 0 : , m_id(m_id_next++)
20 : {
21 0 : std::string algo_name = config["trigger_activity_plugin"][0];
22 0 : nlohmann::json algo_config = config["trigger_activity_config"][0];
23 :
24 : // Get the input file
25 : // Extract the run number etc
26 0 : std::vector<daqdataformats::run_number_t> run_numbers;
27 0 : std::vector<size_t> file_indices;
28 0 : for (const auto& input_file : input_files) {
29 0 : if (std::find(run_numbers.begin(),
30 : run_numbers.end(),
31 0 : input_file->get_attribute<daqdataformats::run_number_t>("run_number")) == run_numbers.end()) {
32 0 : run_numbers.push_back(input_file->get_attribute<daqdataformats::run_number_t>("run_number"));
33 : }
34 :
35 0 : if (std::find(file_indices.begin(),
36 : file_indices.end(),
37 0 : input_file->get_attribute<daqdataformats::run_number_t>("run_number")) == file_indices.end()) {
38 0 : file_indices.push_back(input_file->get_attribute<size_t>("file_index"));
39 : }
40 : }
41 :
42 0 : std::string application_name = m_input_files.front()->get_attribute<std::string>("application_name");
43 :
44 0 : if (!m_quiet) {
45 0 : fmt::print("Run Numbers: {}\nFile Indices: {}\nApp name: '{}'\n",
46 0 : fmt::join(run_numbers, ","),
47 0 : fmt::join(file_indices, ","),
48 : application_name);
49 : }
50 :
51 : // std::set of record IDs (pair of record number & sequence number)
52 0 : auto records = m_input_files.front()->get_all_record_ids();
53 :
54 : // Extract the number of TAMakers to create
55 0 : daqdataformats::TimeSlice first_timeslice = m_input_files.front()->get_timeslice(*records.begin());
56 0 : std::vector<daqdataformats::SourceID> valid_sources = get_valid_sourceids(first_timeslice);
57 0 : fmt::print("Number of makers to make: {}\n", valid_sources.size());
58 :
59 0 : for (const daqdataformats::SourceID& sid : valid_sources) {
60 : // Create TAMaker
61 0 : std::unique_ptr<triggeralgs::TriggerActivityMaker> ta_maker =
62 0 : triggeralgs::TriggerActivityFactory::get_instance()->build_maker(algo_name);
63 0 : ta_maker->configure(algo_config);
64 :
65 : // Add it to the enulators
66 0 : m_ta_emulators[sid] = std::make_unique<trgtools::TAEmulationUnit>();
67 0 : m_ta_emulators[sid]->set_maker(ta_maker);
68 :
69 : // Create a worker thread per emulator
70 0 : if (m_run_parallel) {
71 0 : m_thread_pool.emplace_back(&TAEmulationWorker::worker_thread, this);
72 : }
73 0 : }
74 0 : }
75 :
76 : std::vector<daqdataformats::SourceID>
77 0 : TAEmulationWorker::get_valid_sourceids(daqdataformats::TimeSlice& _timeslice)
78 : {
79 0 : const auto& fragments = _timeslice.get_fragments_ref();
80 :
81 0 : std::vector<daqdataformats::SourceID> ret;
82 0 : for (const auto& fragment : fragments) {
83 0 : if (fragment->get_fragment_type() != daqdataformats::FragmentType::kTriggerPrimitive) {
84 0 : continue;
85 : }
86 :
87 0 : daqdataformats::SourceID sourceid = fragment->get_element_id();
88 :
89 0 : ret.push_back(sourceid);
90 : }
91 :
92 0 : return ret;
93 0 : }
94 :
95 : hdf5libs::HDF5SourceIDHandler::source_id_geo_id_map_t
96 0 : TAEmulationWorker::get_sourceid_geoid_map()
97 : {
98 0 : if (!m_input_files.size()) {
99 0 : throw "Files not set yet!";
100 : }
101 :
102 0 : return m_input_files.front()->get_srcid_geoid_map();
103 : }
104 :
105 : void
106 0 : TAEmulationWorker::worker_thread()
107 : {
108 0 : while (true) {
109 : /// Get a task from the queue (with locking)
110 0 : std::function<void()> task;
111 0 : {
112 0 : std::unique_lock<std::mutex> lock(m_queue_mutex);
113 0 : m_condition.wait(lock, [this]() { return m_stop || !m_task_queue.empty(); });
114 :
115 0 : if (m_stop && m_task_queue.empty()) {
116 0 : return;
117 : }
118 :
119 0 : task = std::move(m_task_queue.front());
120 0 : m_task_queue.pop();
121 0 : }
122 :
123 : // Run & complete a task
124 0 : task();
125 :
126 : // Notify that task was completed
127 0 : {
128 0 : std::lock_guard<std::mutex> lock(m_queue_mutex);
129 0 : --m_active_tasks;
130 0 : if (m_active_tasks == 0) {
131 0 : m_task_complete_condition.notify_all();
132 : }
133 0 : }
134 0 : }
135 : }
136 :
137 : void
138 0 : TAEmulationWorker::process_tasks()
139 : {
140 : // Iterate over the input files
141 0 : for (auto& input_file : m_input_files) {
142 : // std::set of record IDs (pair of record number & sequence number)
143 0 : auto records = input_file->get_all_record_ids();
144 :
145 0 : for (const auto& record : records) {
146 0 : if (record.first < m_sliceid_range.first || record.first > m_sliceid_range.second) {
147 0 : if (!m_quiet)
148 0 : fmt::print(" Will not process RecordID {} because it's outside of our range!", record.first);
149 0 : continue;
150 : }
151 :
152 : // Get all the fragments
153 0 : daqdataformats::TimeSlice timeslice = input_file->get_timeslice(record);
154 0 : const auto& fragments = timeslice.get_fragments_ref();
155 :
156 : // Iterate over the fragments & process each fragment
157 0 : for (const auto& fragment : fragments) {
158 0 : daqdataformats::SourceID sid = fragment->get_element_id();
159 :
160 0 : if (!m_ta_emulators.contains(sid)) {
161 0 : continue;
162 : }
163 :
164 : // Pull tps out
165 0 : size_t n_tps = fragment->get_data_size() / SIZE_TP;
166 0 : if (!m_quiet) {
167 0 : fmt::print(" TP fragment size: {}\n", fragment->get_data_size());
168 0 : fmt::print(" Num TPs: {}\n", n_tps);
169 : }
170 :
171 : // Create a TP buffer
172 0 : std::vector<trgdataformats::TriggerPrimitive> tp_buffer;
173 : // Prepare the TP buffer, checking for time ordering
174 0 : tp_buffer.reserve(n_tps);
175 :
176 : // Populate the TP buffer
177 0 : trgdataformats::TriggerPrimitive* tp_array =
178 0 : static_cast<trgdataformats::TriggerPrimitive*>(fragment->get_data());
179 0 : uint64_t last_ts = 0;
180 0 : for (size_t tpid(0); tpid < n_tps; ++tpid) {
181 0 : auto& tp = tp_array[tpid];
182 0 : if (tp.time_start <= last_ts && !m_quiet) {
183 0 : fmt::print(" ERROR: {} {} ", +tp.time_start, last_ts);
184 : }
185 0 : tp_buffer.push_back(tp);
186 : }
187 :
188 0 : daqdataformats::FragmentHeader frag_hdr = fragment->get_header();
189 :
190 : // Customise the source id (add 1000 to id)
191 0 : frag_hdr.element_id = daqdataformats::SourceID{ daqdataformats::SourceID::Subsystem::kTrigger,
192 0 : fragment->get_element_id().id + 1000 };
193 :
194 : // Either enqueue the task if using parallel processing, or execute the task now
195 0 : if (m_run_parallel) {
196 0 : enqueue_task([this, sid, record, frag_hdr, tp_buffer = std::move(tp_buffer)]() mutable {
197 0 : this->process_task(sid, record.first, frag_hdr, std::move(tp_buffer));
198 0 : });
199 : } else {
200 0 : this->process_task(sid, record.first, frag_hdr, std::move(tp_buffer));
201 : }
202 0 : }
203 : // If running in parallel, wait to process entire slice before we move to
204 : // the next one
205 0 : if (m_run_parallel) {
206 0 : wait_to_complete_tasks();
207 : }
208 0 : }
209 :
210 0 : size_t total = 0;
211 0 : for (auto& [key, vec_tas] : m_tas) {
212 0 : total += vec_tas.size();
213 : }
214 0 : std::cout << "We have a total of " << total << " TAs!" << std::endl;
215 0 : }
216 0 : }
217 :
218 : void
219 0 : TAEmulationWorker::start_processing()
220 : {
221 0 : m_main_thread = std::thread(&TAEmulationWorker::process_tasks, this);
222 0 : }
223 :
224 : void
225 0 : TAEmulationWorker::enqueue_task(std::function<void()> task)
226 : {
227 0 : {
228 0 : std::lock_guard<std::mutex> lock(m_queue_mutex);
229 0 : m_task_queue.push(std::move(task));
230 0 : ++m_active_tasks;
231 0 : }
232 0 : m_condition.notify_one();
233 0 : }
234 :
235 : void
236 0 : TAEmulationWorker::wait_to_complete_tasks()
237 : {
238 0 : std::unique_lock<std::mutex> lock(m_queue_mutex);
239 0 : m_task_complete_condition.wait(lock, [this]() { return m_active_tasks == 0; });
240 0 : }
241 :
242 : void
243 0 : TAEmulationWorker::wait_to_complete_work()
244 : {
245 : // Wait for the main threads to join
246 0 : m_main_thread.join();
247 0 : fmt::print("TAEmulationWorker_{} work completed\n", m_id);
248 :
249 : // Wait for the tasks to complete
250 0 : if (m_run_parallel) {
251 0 : wait_to_complete_tasks();
252 :
253 0 : {
254 0 : std::lock_guard<std::mutex> lock(m_queue_mutex);
255 0 : m_stop = true;
256 0 : }
257 0 : m_condition.notify_all();
258 :
259 0 : fmt::print("m_stop issued\n");
260 0 : for (std::thread& thread : m_thread_pool) {
261 0 : thread.join();
262 : }
263 : }
264 0 : }
265 :
266 : void
267 0 : TAEmulationWorker::process_task(daqdataformats::SourceID _source_id,
268 : uint64_t _rec,
269 : daqdataformats::FragmentHeader _header,
270 : std::vector<trgdataformats::TriggerPrimitive>&& _tps)
271 : {
272 : // Get te last fragment
273 0 : std::unique_ptr<daqdataformats::Fragment> frag = m_ta_emulators[_source_id]->emulate_vector(_tps);
274 :
275 : // Don't do anything if no fragments found
276 0 : if (!frag) {
277 : return;
278 : }
279 :
280 : // Get all the TriggerActivities from the TA Emulator buffer
281 0 : std::vector<triggeralgs::TriggerActivity> ta_buffer = m_ta_emulators[_source_id]->get_last_output_buffer();
282 :
283 : // Don't continue if no TAs found
284 0 : size_t n_tas = ta_buffer.size();
285 0 : if (!n_tas) {
286 0 : return;
287 : }
288 :
289 0 : if (!m_quiet && n_tas) {
290 0 : fmt::print(" Found {} TAs!\n", n_tas);
291 : }
292 :
293 : // Set the fragment header & push into our output (with locking!)
294 0 : {
295 0 : if (m_run_parallel) {
296 0 : std::lock_guard<std::mutex> lock(m_savetps_mutex);
297 0 : }
298 0 : m_tas[_rec].reserve(m_tas[_rec].size() + ta_buffer.size());
299 0 : m_tas[_rec].insert(
300 0 : m_tas[_rec].end(), std::make_move_iterator(ta_buffer.begin()), std::make_move_iterator(ta_buffer.end()));
301 :
302 0 : frag->set_header_fields(_header);
303 0 : frag->set_type(daqdataformats::FragmentType::kTriggerActivity);
304 :
305 0 : m_ta_fragments[_rec].push_back(std::move(frag));
306 : }
307 0 : }
308 :
309 : std::map<uint64_t, std::vector<triggeralgs::TriggerActivity>>
310 0 : TAEmulationWorker::get_tas()
311 : {
312 0 : return std::move(m_tas);
313 : }
314 :
315 : std::map<uint64_t, std::vector<std::unique_ptr<daqdataformats::Fragment>>>
316 0 : TAEmulationWorker::get_frags()
317 : {
318 0 : return std::move(m_ta_fragments);
319 : }
320 :
321 : }; // namespace dunedaq::trgtools
322 :
323 : #endif // TRGTOOLS_TAEMULATIONWORKER_CXX_
|
