Line data Source code
1 : /**
2 : * @file TPStreamWriterModule.cpp TPStreamWriterModule 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 "TPStreamWriterModule.hpp"
10 : #include "dfmodules/CommonIssues.hpp"
11 : #include "dfmodules/TPBundleHandler.hpp"
12 : #include "dfmodules/opmon/TPStreamWriter.pb.h"
13 :
14 : #include "appmodel/DataStoreConf.hpp"
15 : #include "appmodel/TPStreamWriterModule.hpp"
16 : #include "confmodel/Connection.hpp"
17 : #include "confmodel/Session.hpp"
18 : #include "daqdataformats/Fragment.hpp"
19 : #include "daqdataformats/Types.hpp"
20 : #include "iomanager/IOManager.hpp"
21 : #include "logging/Logging.hpp"
22 : #include "rcif/cmd/Nljs.hpp"
23 :
24 : #include "boost/date_time/posix_time/posix_time.hpp"
25 :
26 : #include <chrono>
27 : #include <memory>
28 : #include <sstream>
29 : #include <string>
30 : #include <utility>
31 : #include <vector>
32 :
33 : enum
34 : {
35 : TLVL_ENTER_EXIT_METHODS = 5,
36 : TLVL_CONFIG = 7,
37 : };
38 :
39 : namespace dunedaq {
40 : namespace dfmodules {
41 :
42 0 : TPStreamWriterModule::TPStreamWriterModule(const std::string& name)
43 : : dunedaq::appfwk::DAQModule(name)
44 0 : , m_thread(std::bind(&TPStreamWriterModule::do_work, this, std::placeholders::_1))
45 0 : , m_queue_timeout(100)
46 0 : , m_data_storage_is_enabled(true)
47 : {
48 0 : register_command("conf", &TPStreamWriterModule::do_conf);
49 0 : register_command("start", &TPStreamWriterModule::do_start);
50 0 : register_command("stop", &TPStreamWriterModule::do_stop);
51 0 : register_command("scrap", &TPStreamWriterModule::do_scrap);
52 0 : }
53 :
54 : void
55 0 : TPStreamWriterModule::init(std::shared_ptr<appfwk::ConfigurationManager> mcfg)
56 : {
57 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering init() method";
58 0 : auto mdal = mcfg->get_dal<appmodel::TPStreamWriterModule>(get_name());
59 0 : if (!mdal) {
60 0 : throw appfwk::CommandFailed(ERS_HERE, "init", get_name(), "Unable to retrieve configuration object");
61 : }
62 0 : assert(mdal->get_inputs().size() == 1);
63 0 : m_module_configuration = mcfg;
64 0 : m_tpset_source = iomanager::IOManager::get()->get_receiver<trigger::TPSet>(mdal->get_inputs()[0]->UID());
65 0 : m_writer_identifier = mdal->get_writer_identifier();
66 0 : m_tp_writer_conf = mdal->get_configuration();
67 0 : m_source_id = mdal->get_source_id();
68 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting init() method";
69 0 : }
70 :
71 : void
72 0 : TPStreamWriterModule::generate_opmon_data()
73 : {
74 0 : opmon::TPStreamWriterInfo info;
75 :
76 0 : info.set_heartbeat_tpsets_received(m_heartbeat_tpsets.exchange(0));
77 0 : info.set_tpsets_with_tps_received(m_tpsets_with_tps.exchange(0));
78 0 : info.set_tps_received(m_tps_received.exchange(0));
79 0 : info.set_tps_written(m_tps_written.exchange(0));
80 0 : info.set_tps_discarded(m_tps_discarded.exchange(0));
81 0 : info.set_total_tps_received(m_total_tps_received.load());
82 0 : info.set_total_tps_written(m_total_tps_written.load());
83 0 : info.set_total_tps_discarded(m_total_tps_discarded.load());
84 0 : info.set_tardy_timeslice_max_seconds(m_tardy_timeslice_max_seconds.exchange(0.0));
85 0 : info.set_timeslices_written(m_timeslices_written.exchange(0));
86 0 : info.set_bytes_output(m_bytes_output.exchange(0));
87 :
88 0 : publish(std::move(info));
89 0 : }
90 :
91 : void
92 0 : TPStreamWriterModule::do_conf(const CommandData_t&)
93 : {
94 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_conf() method";
95 0 : m_accumulation_interval_ticks = m_tp_writer_conf->get_tp_accumulation_interval();
96 0 : m_accumulation_inactivity_time_before_write = std::chrono::milliseconds(
97 0 : static_cast<int>(1000 * m_tp_writer_conf->get_tp_accumulation_inactivity_time_before_write_sec()));
98 0 : m_warn_user_when_tardy_tps_are_discarded = m_tp_writer_conf->get_warn_user_when_tardy_tps_are_discarded();
99 0 : m_accumulation_interval_seconds = ((double)m_accumulation_interval_ticks) / 62500000.0;
100 :
101 : // create the DataStore instance here
102 0 : try {
103 0 : m_data_writer = make_data_store(m_tp_writer_conf->get_data_store_params()->get_type(),
104 0 : m_tp_writer_conf->get_data_store_params()->UID(),
105 0 : m_module_configuration,
106 0 : m_writer_identifier);
107 0 : register_node("data_writer", m_data_writer);
108 0 : } catch (const ers::Issue& excpt) {
109 0 : throw UnableToConfigure(ERS_HERE, get_name(), excpt);
110 0 : }
111 :
112 : // ensure that we have a valid dataWriter instance
113 0 : if (m_data_writer.get() == nullptr) {
114 0 : throw InvalidDataWriterModule(ERS_HERE, get_name());
115 : }
116 :
117 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_conf() method";
118 0 : }
119 :
120 : void
121 0 : TPStreamWriterModule::do_start(const CommandData_t& payload)
122 : {
123 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_start() method";
124 0 : rcif::cmd::StartParams start_params = payload.get<rcif::cmd::StartParams>();
125 0 : m_data_storage_is_enabled = (!start_params.disable_data_storage);
126 0 : m_run_number = start_params.run;
127 0 : m_total_tps_received.store(0);
128 0 : m_total_tps_written.store(0);
129 0 : m_total_tps_discarded.store(0);
130 :
131 : // 06-Mar-2022, KAB: added this call to allow DataStore to prepare for the run.
132 : // I've put this call fairly early in this method because it could throw an
133 : // exception and abort the run start. And, it seems sensible to avoid starting
134 : // threads, etc. if we throw an exception.
135 0 : if (m_data_storage_is_enabled) {
136 0 : try {
137 0 : m_data_writer->prepare_for_run(m_run_number, (start_params.production_vs_test == "TEST"));
138 0 : } catch (const ers::Issue& excpt) {
139 0 : throw UnableToStart(ERS_HERE, get_name(), m_run_number, excpt);
140 0 : }
141 : }
142 :
143 0 : m_thread.start_working_thread(get_name());
144 :
145 0 : TLOG() << get_name() << " successfully started for run number " << m_run_number;
146 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_start() method";
147 0 : }
148 :
149 : void
150 0 : TPStreamWriterModule::do_stop(const CommandData_t& /*payload*/)
151 : {
152 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_stop() method";
153 0 : m_thread.stop_working_thread();
154 :
155 : // 06-Mar-2022, KAB: added this call to allow DataStore to finish up with this run.
156 : // I've put this call fairly late in this method so that any draining of queues
157 : // (or whatever) can take place before we finalize things in the DataStore.
158 0 : if (m_data_storage_is_enabled) {
159 0 : try {
160 0 : m_data_writer->finish_with_run(m_run_number);
161 0 : } catch (const std::exception& excpt) {
162 0 : ers::error(ProblemDuringStop(ERS_HERE, get_name(), m_run_number, excpt));
163 0 : }
164 : }
165 :
166 0 : TLOG() << get_name() << " successfully stopped for run number " << m_run_number;
167 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_stop() method";
168 0 : }
169 :
170 : void
171 0 : TPStreamWriterModule::do_scrap(const CommandData_t& /*payload*/)
172 : {
173 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_scrap() method";
174 :
175 : // clear/reset the DataStore instance here
176 0 : m_data_writer.reset();
177 :
178 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_scrap() method";
179 0 : }
180 :
181 : void
182 0 : TPStreamWriterModule::do_work(std::atomic<bool>& running_flag)
183 : {
184 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_work() method";
185 :
186 0 : using namespace std::chrono;
187 0 : size_t n_tpset_received = 0;
188 0 : auto start_time = steady_clock::now();
189 0 : daqdataformats::timestamp_t first_timestamp = 0;
190 0 : daqdataformats::timestamp_t last_timestamp = 0;
191 :
192 0 : TPBundleHandler tp_bundle_handler(
193 0 : m_accumulation_interval_ticks, m_run_number, m_accumulation_inactivity_time_before_write);
194 :
195 0 : bool possible_pending_data = true;
196 0 : size_t largest_timeslice_number = 0;
197 0 : while (running_flag.load() || possible_pending_data) {
198 0 : trigger::TPSet tpset;
199 0 : try {
200 0 : tpset = m_tpset_source->receive(m_queue_timeout);
201 0 : ++n_tpset_received;
202 :
203 0 : if (tpset.type == trigger::TPSet::Type::kHeartbeat) {
204 0 : ++m_heartbeat_tpsets;
205 0 : continue;
206 : }
207 :
208 0 : TLOG_DEBUG(21) << "Number of TPs in TPSet is " << tpset.objects.size() << ", Source ID is " << tpset.origin
209 0 : << ", seqno is " << tpset.seqno << ", start timestamp is " << tpset.start_time
210 0 : << ", run number is " << tpset.run_number << ", slice id is "
211 0 : << (tpset.start_time / m_accumulation_interval_ticks);
212 :
213 : // 30-Mar-2022, KAB: added test for matching run number. This is to avoid getting
214 : // confused by TPSets that happen to be leftover in transit from one run to the
215 : // next (which we have observed in v2.10.x systems).
216 0 : if (tpset.run_number != m_run_number) {
217 0 : TLOG_DEBUG(22) << "Discarding TPSet with invalid run number " << tpset.run_number << " (current is "
218 0 : << m_run_number << "), Source ID is " << tpset.origin << ", seqno is " << tpset.seqno;
219 0 : continue;
220 0 : }
221 0 : ++m_tpsets_with_tps;
222 :
223 0 : size_t num_tps_in_tpset = tpset.objects.size();
224 0 : tp_bundle_handler.add_tpset(std::move(tpset));
225 0 : m_tps_received += num_tps_in_tpset;
226 0 : m_total_tps_received += num_tps_in_tpset;
227 0 : possible_pending_data = true;
228 0 : } catch (iomanager::ConnectionInstanceNotFound&) {
229 : // sleep for a little bit; and indicate no pending data, in case we never get a connection
230 : // and the run ends - we don't want to believe that there is pending data in that case.
231 0 : usleep(1000 * m_queue_timeout.count());
232 0 : possible_pending_data = false;
233 0 : } catch (iomanager::TimeoutExpired&) {
234 : // nothing special to do here, we'll simply let the rest of the code in this
235 : // while loop do its job
236 0 : }
237 :
238 0 : std::vector<std::unique_ptr<daqdataformats::TimeSlice>> list_of_timeslices;
239 0 : if (running_flag.load()) {
240 0 : list_of_timeslices = tp_bundle_handler.get_properly_aged_timeslices();
241 : } else {
242 0 : list_of_timeslices = tp_bundle_handler.get_all_remaining_timeslices();
243 0 : possible_pending_data = false;
244 : }
245 :
246 : // keep track of the largest timeslice number (for reporting on tardy ones)
247 0 : for (auto& timeslice_ptr : list_of_timeslices) {
248 0 : largest_timeslice_number = std::max(timeslice_ptr->get_header().timeslice_number, largest_timeslice_number);
249 : }
250 :
251 : // attempt to write out each TimeSlice
252 0 : for (auto& timeslice_ptr : list_of_timeslices) {
253 0 : daqdataformats::SourceID sid(daqdataformats::SourceID::Subsystem::kTRBuilder, m_source_id);
254 0 : timeslice_ptr->set_element_id(sid);
255 :
256 0 : if (m_data_storage_is_enabled) {
257 :
258 : // write the TSH and the fragments as a set of data blocks
259 0 : bool should_retry = true;
260 : size_t retry_wait_usec = 1000;
261 0 : do {
262 0 : should_retry = false;
263 0 : size_t number_of_tps =
264 0 : (timeslice_ptr->get_sum_of_fragment_payload_sizes() / sizeof(trgdataformats::TriggerPrimitive));
265 0 : try {
266 0 : m_data_writer->write(*timeslice_ptr);
267 0 : ++m_timeslices_written;
268 0 : m_bytes_output += timeslice_ptr->get_total_size_bytes();
269 0 : m_tps_written += number_of_tps;
270 0 : m_total_tps_written += number_of_tps;
271 0 : } catch (const RetryableDataStoreProblem& excpt) {
272 0 : should_retry = true;
273 0 : ers::error(DataWritingProblem(ERS_HERE,
274 0 : get_name(),
275 0 : timeslice_ptr->get_header().timeslice_number,
276 0 : timeslice_ptr->get_header().run_number,
277 0 : excpt));
278 0 : usleep(retry_wait_usec);
279 0 : retry_wait_usec = std::min(retry_wait_usec * 2, 1000000UL);
280 0 : } catch (const IgnorableDataStoreProblem& excpt) {
281 0 : int timeslice_number_diff = largest_timeslice_number - timeslice_ptr->get_header().timeslice_number;
282 0 : double seconds_too_late = m_accumulation_interval_seconds * timeslice_number_diff;
283 0 : m_tardy_timeslice_max_seconds = std::max(m_tardy_timeslice_max_seconds.load(), seconds_too_late);
284 0 : m_tps_discarded += number_of_tps;
285 0 : m_total_tps_discarded += number_of_tps;
286 0 : if (m_warn_user_when_tardy_tps_are_discarded) {
287 0 : std::ostringstream sid_list;
288 0 : bool first_frag = true;
289 0 : for (auto const& frag_ptr : timeslice_ptr->get_fragments_ref()) {
290 0 : if (first_frag) {
291 : first_frag = false;
292 : } else {
293 0 : sid_list << ",";
294 : }
295 0 : sid_list << frag_ptr->get_element_id().to_string();
296 : }
297 0 : ers::warning(TardyTPsDiscarded(
298 0 : ERS_HERE, get_name(), sid_list.str(), timeslice_ptr->get_header().timeslice_number, seconds_too_late));
299 0 : }
300 0 : } catch (const std::exception& excpt) {
301 0 : m_tps_discarded += number_of_tps;
302 0 : m_total_tps_discarded += number_of_tps;
303 0 : ers::error(DataWritingProblem(ERS_HERE,
304 0 : get_name(),
305 0 : timeslice_ptr->get_header().timeslice_number,
306 0 : timeslice_ptr->get_header().run_number,
307 0 : excpt));
308 0 : }
309 0 : } while (should_retry && running_flag.load());
310 : } // if (m_data_storage_is_enabled) {
311 : }
312 :
313 0 : if (first_timestamp == 0) {
314 0 : first_timestamp = tpset.start_time;
315 : }
316 0 : last_timestamp = tpset.start_time;
317 0 : } // while(running)
318 :
319 0 : auto end_time = steady_clock::now();
320 0 : auto time_ms = duration_cast<milliseconds>(end_time - start_time).count();
321 0 : float rate_hz = 1e3 * static_cast<float>(n_tpset_received) / time_ms;
322 0 : float inferred_clock_frequency = 1e3 * (last_timestamp - first_timestamp) / time_ms;
323 :
324 0 : TLOG() << "Received " << n_tpset_received << " TPSets in " << time_ms << "ms. " << rate_hz
325 0 : << " TPSet/s. Inferred clock frequency " << inferred_clock_frequency << "Hz";
326 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_work() method";
327 0 : } // NOLINT Function length
328 :
329 : } // namespace dfmodules
330 : } // namespace dunedaq
331 :
332 0 : DEFINE_DUNE_DAQ_MODULE(dunedaq::dfmodules::TPStreamWriterModule)
|
