Line data Source code
1 : #include "artdaq/DAQdata/Globals.hh" // include these 2 first -
2 : #define TRACE_NAME (app_name + "_BoardReaderCore").c_str()
3 :
4 : #include "artdaq-core/Core/MonitoredQuantity.hh"
5 : #include "artdaq-core/Data/Fragment.hh"
6 : #include "artdaq-core/Plugins/makeFragmentGenerator.hh"
7 : #include "artdaq-core/Utilities/ExceptionHandler.hh"
8 : #include "artdaq/Application/BoardReaderCore.hh"
9 : #include "artdaq/Application/TaskType.hh"
10 : #include "artdaq/Generators/makeCommandableFragmentGenerator.hh"
11 :
12 : #include "cetlib_except/exception.h"
13 : #include "fhiclcpp/ParameterSet.h"
14 :
15 : #include <boost/lexical_cast.hpp>
16 :
17 : #include <pthread.h>
18 : #include <sched.h>
19 : #include <algorithm>
20 : #include <memory>
21 : #include <thread>
22 :
23 : const std::string artdaq::BoardReaderCore::
24 : FRAGMENTS_PROCESSED_STAT_KEY("BoardReaderCoreFragmentsProcessed");
25 : const std::string artdaq::BoardReaderCore::
26 : INPUT_WAIT_STAT_KEY("BoardReaderCoreInputWaitTime");
27 : const std::string artdaq::BoardReaderCore::BUFFER_WAIT_STAT_KEY("BoardReaderCoreBufferWaitTime");
28 : const std::string artdaq::BoardReaderCore::REQUEST_WAIT_STAT_KEY("BoardReaderCoreRequestWaitTime");
29 : const std::string artdaq::BoardReaderCore::
30 : OUTPUT_WAIT_STAT_KEY("BoardReaderCoreOutputWaitTime");
31 : const std::string artdaq::BoardReaderCore::
32 : FRAGMENTS_PER_READ_STAT_KEY("BoardReaderCoreFragmentsPerRead");
33 :
34 : std::unique_ptr<artdaq::DataSenderManager> artdaq::BoardReaderCore::sender_ptr_ = nullptr;
35 :
36 0 : artdaq::BoardReaderCore::BoardReaderCore(Commandable& parent_application)
37 0 : : parent_application_(parent_application)
38 : /*, local_group_comm_(local_group_comm)*/
39 0 : , generator_ptr_(nullptr)
40 0 : , run_id_(art::RunID::flushRun())
41 0 : , fragment_count_(0)
42 0 : , stop_requested_(false)
43 0 : , pause_requested_(false)
44 : {
45 0 : TLOG(TLVL_DEBUG + 32) << "Constructor";
46 0 : statsHelper_.addMonitoredQuantityName(FRAGMENTS_PROCESSED_STAT_KEY);
47 0 : statsHelper_.addMonitoredQuantityName(INPUT_WAIT_STAT_KEY);
48 0 : statsHelper_.addMonitoredQuantityName(BUFFER_WAIT_STAT_KEY);
49 0 : statsHelper_.addMonitoredQuantityName(REQUEST_WAIT_STAT_KEY);
50 0 : statsHelper_.addMonitoredQuantityName(OUTPUT_WAIT_STAT_KEY);
51 0 : statsHelper_.addMonitoredQuantityName(FRAGMENTS_PER_READ_STAT_KEY);
52 0 : }
53 :
54 0 : artdaq::BoardReaderCore::~BoardReaderCore()
55 : {
56 0 : TLOG(TLVL_DEBUG + 32) << "Destructor";
57 0 : TLOG(TLVL_DEBUG + 32) << "Stopping Request Receiver BEGIN";
58 0 : request_receiver_ptr_.reset(nullptr);
59 0 : TLOG(TLVL_DEBUG + 32) << "Stopping Request Receiver END";
60 0 : }
61 :
62 0 : void artdaq::BoardReaderCore::recordInitFailure_(const std::string& stage, const std::string& detail, fhicl::ParameterSet const& fr_pset)
63 : {
64 0 : std::string message = "Exception thrown during initialization of " + stage;
65 0 : if (!detail.empty()) { message += ": " + detail; }
66 :
67 0 : last_init_error_ = message;
68 0 : TLOG(TLVL_ERROR) << message;
69 0 : TLOG(TLVL_DEBUG + 32) << "FHiCL parameter set used to initialize the " << stage << " which threw an exception: " << fr_pset.to_string();
70 0 : }
71 :
72 0 : bool artdaq::BoardReaderCore::initialize(fhicl::ParameterSet const& pset, uint64_t /*unused*/, uint64_t /*unused*/)
73 : {
74 0 : last_init_error_.clear();
75 :
76 0 : TLOG(TLVL_DEBUG + 32) << "initialize method called with "
77 0 : << "ParameterSet = \"" << pset.to_string() << "\".";
78 :
79 : // pull out the relevant parts of the ParameterSet
80 0 : fhicl::ParameterSet daq_pset;
81 : try
82 : {
83 0 : daq_pset = pset.get<fhicl::ParameterSet>("daq");
84 : }
85 0 : catch (...)
86 : {
87 0 : TLOG(TLVL_ERROR)
88 0 : << "Unable to find the DAQ parameters in the initialization "
89 0 : << "ParameterSet: \"" + pset.to_string() + "\".";
90 0 : return false;
91 0 : }
92 0 : fhicl::ParameterSet fr_pset;
93 : try
94 : {
95 0 : fr_pset = daq_pset.get<fhicl::ParameterSet>("fragment_receiver");
96 0 : data_pset_ = fr_pset;
97 : }
98 0 : catch (...)
99 : {
100 0 : TLOG(TLVL_ERROR)
101 0 : << "Unable to find the fragment_receiver parameters in the DAQ "
102 0 : << "initialization ParameterSet: \"" + daq_pset.to_string() + "\".";
103 0 : return false;
104 0 : }
105 :
106 : // pull out the Metric part of the ParameterSet
107 0 : fhicl::ParameterSet metric_pset;
108 : try
109 : {
110 0 : metric_pset = daq_pset.get<fhicl::ParameterSet>("metrics");
111 : }
112 0 : catch (...)
113 0 : {} // OK if there's no metrics table defined in the FHiCL
114 :
115 0 : if (metric_pset.is_empty())
116 : {
117 0 : TLOG(TLVL_INFO) << "No metric plugins appear to be defined";
118 : }
119 : try
120 : {
121 0 : metricMan->initialize(metric_pset, app_name);
122 : }
123 0 : catch (...)
124 : {
125 0 : ExceptionHandler(ExceptionHandlerRethrow::no,
126 : "Error loading metrics in BoardReaderCore::initialize()");
127 0 : }
128 :
129 0 : if (daq_pset.has_key("rank"))
130 : {
131 0 : if (my_rank >= 0 && daq_pset.get<int>("rank") != my_rank)
132 : {
133 0 : TLOG(TLVL_WARNING) << "BoardReader rank specified at startup is different than rank specified at configure! Using rank received at configure!";
134 : }
135 0 : my_rank = daq_pset.get<int>("rank");
136 : }
137 0 : if (my_rank == -1)
138 : {
139 0 : TLOG(TLVL_ERROR) << "BoardReader rank not specified at startup or in configuration! Aborting";
140 0 : throw cet::exception("RankNotSpecifiedError") << "BoardReader rank not specified at startup or in configuration! Aborting";
141 : }
142 :
143 : // create the requested CommandableFragmentGenerator
144 0 : auto frag_gen_name = fr_pset.get<std::string>("generator", "");
145 0 : if (frag_gen_name.length() == 0)
146 : {
147 0 : TLOG(TLVL_ERROR)
148 0 : << "No fragment generator (parameter name = \"generator\") was "
149 0 : << "specified in the fragment_receiver ParameterSet. The "
150 0 : << "DAQ initialization PSet was \"" << daq_pset.to_string() << "\".";
151 0 : return false;
152 : }
153 :
154 : // The remaining initialization steps can all throw. A single handler reports
155 : // which step failed, tracked by init_stage.
156 0 : std::string init_stage = "fragment generator of type \"" + frag_gen_name + "\"";
157 : try
158 : {
159 0 : auto gen = artdaq::makeFragmentGenerator(frag_gen_name, fr_pset);
160 0 : auto* commandable = dynamic_cast<artdaq::CommandableFragmentGenerator*>(gen.get());
161 0 : if (commandable == nullptr)
162 : {
163 0 : last_init_error_ = "Generator \"" + frag_gen_name + "\" is not a CommandableFragmentGenerator and cannot be used in a BoardReader";
164 0 : TLOG(TLVL_ERROR) << last_init_error_ << ". Use DEFINE_ARTDAQ_COMMANDABLE_GENERATOR for BoardReader-compatible generators.";
165 0 : return false;
166 : }
167 0 : gen.release();
168 0 : generator_ptr_.reset(commandable);
169 :
170 0 : init_stage = "Fragment Buffer";
171 0 : fragment_buffer_ptr_ = std::make_shared<FragmentBuffer>(fr_pset);
172 :
173 0 : init_stage = "request receiver";
174 0 : std::shared_ptr<RequestBuffer> request_buffer = std::make_shared<RequestBuffer>();
175 0 : request_receiver_ptr_.reset(new RequestReceiver(fr_pset, request_buffer));
176 0 : generator_ptr_->SetRequestBuffer(request_buffer);
177 0 : generator_ptr_->SetFragmentBuffer(fragment_buffer_ptr_);
178 0 : fragment_buffer_ptr_->SetRequestBuffer(request_buffer);
179 0 : }
180 0 : catch (const std::exception& e)
181 : {
182 0 : recordInitFailure_(init_stage, e.what(), fr_pset);
183 0 : return false;
184 0 : }
185 0 : catch (...)
186 : {
187 0 : recordInitFailure_(init_stage, "unknown exception type", fr_pset);
188 0 : return false;
189 0 : }
190 0 : metricMan->setPrefix(generator_ptr_->metricsReportingInstanceName());
191 :
192 0 : rt_priority_ = fr_pset.get<int>("rt_priority", 0);
193 :
194 : // fetch the monitoring parameters and create the MonitoredQuantity instances
195 0 : statsHelper_.createCollectors(fr_pset, 100, 30.0, 60.0, FRAGMENTS_PROCESSED_STAT_KEY);
196 :
197 : // check if we should skip the sequence ID test...
198 0 : skip_seqId_test_ = (fr_pset.get<bool>("skip_seqID_test", false) || generator_ptr_->fragmentIDs().size() > 1 || fragment_buffer_ptr_->request_mode() != RequestMode::Ignored);
199 :
200 0 : verbose_ = fr_pset.get<bool>("verbose", true);
201 :
202 0 : return true;
203 0 : }
204 :
205 0 : bool artdaq::BoardReaderCore::start(art::RunID id, uint64_t timeout, uint64_t timestamp)
206 : {
207 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Starting run " << id.run();
208 0 : stop_requested_.store(false);
209 0 : pause_requested_.store(false);
210 :
211 0 : fragment_count_ = 0;
212 0 : prev_seq_id_ = 0;
213 0 : statsHelper_.resetStatistics();
214 :
215 0 : fragment_buffer_ptr_->Reset(false);
216 :
217 0 : metricMan->do_start();
218 :
219 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Pausing for 1s";
220 0 : usleep(1000000);
221 :
222 0 : generator_ptr_->StartCmd(id.run(), timeout, timestamp);
223 0 : run_id_ = id;
224 :
225 0 : request_receiver_ptr_->SetRunNumber(static_cast<uint32_t>(id.run()));
226 0 : request_receiver_ptr_->startRequestReception();
227 :
228 0 : running_ = true;
229 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Completed the Start transition (Started run) for run " << run_id_.run()
230 0 : << ", timeout = " << timeout << ", timestamp = " << timestamp;
231 0 : return true;
232 : }
233 :
234 0 : bool artdaq::BoardReaderCore::stop(uint64_t timeout, uint64_t timestamp)
235 : {
236 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Stopping run " << run_id_.run() << " after " << fragment_count_ << " fragments.";
237 0 : stop_requested_.store(true);
238 :
239 0 : TLOG(TLVL_DEBUG + 32) << "Stopping Request reception BEGIN";
240 0 : request_receiver_ptr_->stopRequestReception();
241 0 : TLOG(TLVL_DEBUG + 32) << "Stopping Request reception END";
242 :
243 0 : TLOG(TLVL_DEBUG + 32) << "Stopping CommandableFragmentGenerator BEGIN";
244 0 : generator_ptr_->StopCmd(timeout, timestamp);
245 0 : TLOG(TLVL_DEBUG + 32) << "Stopping CommandableFragmentGenerator END";
246 :
247 0 : TLOG(TLVL_DEBUG + 32) << "Stopping FragmentBuffer";
248 0 : fragment_buffer_ptr_->Stop();
249 :
250 0 : TLOG(TLVL_DEBUG + 32) << "Stopping DataSenderManager";
251 0 : if (sender_ptr_)
252 : {
253 0 : sender_ptr_->StopSender();
254 : }
255 :
256 0 : running_ = false;
257 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Completed the Stop transition for run " << run_id_.run();
258 0 : return true;
259 : }
260 :
261 0 : bool artdaq::BoardReaderCore::pause(uint64_t timeout, uint64_t timestamp)
262 : {
263 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Pausing run " << run_id_.run() << " after " << fragment_count_ << " fragments.";
264 0 : pause_requested_.store(true);
265 0 : generator_ptr_->PauseCmd(timeout, timestamp);
266 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Completed the Pause transition for run " << run_id_.run();
267 0 : return true;
268 : }
269 :
270 0 : bool artdaq::BoardReaderCore::resume(uint64_t timeout, uint64_t timestamp)
271 : {
272 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Resuming run " << run_id_.run();
273 0 : pause_requested_.store(false);
274 0 : metricMan->do_start();
275 0 : generator_ptr_->ResumeCmd(timeout, timestamp);
276 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Completed the Resume transition for run " << run_id_.run();
277 0 : return true;
278 : }
279 :
280 0 : bool artdaq::BoardReaderCore::shutdown(uint64_t /*unused*/)
281 : {
282 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Starting Shutdown transition";
283 0 : generator_ptr_->joinThreads(); // Cleanly shut down the CommandableFragmentGenerator
284 0 : generator_ptr_.reset(nullptr);
285 0 : metricMan->shutdown();
286 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Completed Shutdown transition";
287 0 : return true;
288 : }
289 :
290 0 : bool artdaq::BoardReaderCore::soft_initialize(fhicl::ParameterSet const& pset, uint64_t timeout, uint64_t timestamp)
291 : {
292 0 : TLOG(TLVL_DEBUG + 32) << "soft_initialize method called with "
293 0 : << "ParameterSet = \"" << pset.to_string()
294 0 : << "\". Forwarding to initialize.";
295 0 : return initialize(pset, timeout, timestamp);
296 : }
297 :
298 0 : bool artdaq::BoardReaderCore::reinitialize(fhicl::ParameterSet const& pset, uint64_t timeout, uint64_t timestamp)
299 : {
300 0 : TLOG(TLVL_DEBUG + 32) << "reinitialize method called with "
301 0 : << "ParameterSet = \"" << pset.to_string()
302 0 : << "\". Forwarding to initalize.";
303 0 : return initialize(pset, timeout, timestamp);
304 : }
305 :
306 0 : void artdaq::BoardReaderCore::receive_fragments()
307 : {
308 0 : if (rt_priority_ > 0)
309 : {
310 : #pragma GCC diagnostic push
311 : #pragma GCC diagnostic ignored "-Wmissing-field-initializers"
312 0 : sched_param s_param = {};
313 0 : s_param.sched_priority = rt_priority_;
314 0 : if (pthread_setschedparam(pthread_self(), SCHED_RR, &s_param))
315 0 : TLOG(TLVL_WARNING) << "setting realtime priority failed";
316 : #pragma GCC diagnostic pop
317 : }
318 :
319 : // try-catch block here?
320 :
321 : // how to turn RT PRI off?
322 0 : if (rt_priority_ > 0)
323 : {
324 : #pragma GCC diagnostic push
325 : #pragma GCC diagnostic ignored "-Wmissing-field-initializers"
326 0 : sched_param s_param = {};
327 0 : s_param.sched_priority = rt_priority_;
328 0 : int status = pthread_setschedparam(pthread_self(), SCHED_RR, &s_param);
329 0 : if (status != 0)
330 : {
331 0 : TLOG(TLVL_ERROR)
332 0 : << "Failed to set realtime priority to " << rt_priority_
333 0 : << ", return code = " << status;
334 : }
335 : #pragma GCC diagnostic pop
336 : }
337 :
338 0 : TLOG(TLVL_DEBUG + 32) << "Waiting for first fragment.";
339 : artdaq::MonitoredQuantityStats::TIME_POINT_T startTime, after_input, after_buffer;
340 0 : artdaq::FragmentPtrs frags;
341 :
342 0 : receiver_thread_active_ = true;
343 :
344 0 : auto wait_start = std::chrono::steady_clock::now();
345 0 : while (!running_ && TimeUtils::GetElapsedTime(wait_start) < start_transition_timeout_)
346 : {
347 0 : usleep(10000);
348 : }
349 0 : if (!running_)
350 : {
351 0 : TLOG(TLVL_ERROR) << "Timeout (" << start_transition_timeout_ << " s) while waiting for Start after receive_fragments thread started!";
352 0 : receiver_thread_active_ = false;
353 : }
354 :
355 0 : while (receiver_thread_active_)
356 : {
357 0 : startTime = artdaq::MonitoredQuantity::getCurrentTime();
358 :
359 0 : TLOG(TLVL_DEBUG + 35) << "receive_fragments getNext start";
360 0 : receiver_thread_active_ = generator_ptr_->getNext(frags);
361 0 : TLOG(TLVL_DEBUG + 35) << "receive_fragments getNext done (receiver_thread_active_=" << receiver_thread_active_ << ")";
362 :
363 : // 08-May-2015, KAB & JCF: if the generator getNext() method returns false
364 : // (which indicates that the data flow has stopped) *and* the reason that
365 : // it has stopped is because there was an exception that wasn't handled by
366 : // the experiment-specific FragmentGenerator class, we move to the
367 : // InRunError state so that external observers (e.g. RunControl or
368 : // DAQInterface) can see that there was a problem.
369 0 : if (!receiver_thread_active_ && generator_ptr_ && generator_ptr_->exception())
370 : {
371 0 : parent_application_.in_run_failure();
372 : }
373 :
374 0 : after_input = artdaq::MonitoredQuantity::getCurrentTime();
375 :
376 0 : if (!receiver_thread_active_) { break; }
377 :
378 0 : size_t fragments_processed = frags.size();
379 0 : if (fragments_processed > 0)
380 : {
381 0 : TLOG(TLVL_DEBUG + 35) << "receive_fragments AddFragmentsToBuffer start";
382 0 : fragment_buffer_ptr_->AddFragmentsToBuffer(std::move(frags));
383 0 : TLOG(TLVL_DEBUG + 35) << "receive_fragments AddFragmentsToBuffer done";
384 :
385 0 : after_buffer = artdaq::MonitoredQuantity::getCurrentTime();
386 :
387 0 : double input_wait_time = after_input - startTime;
388 0 : double buffer_wait_time = after_buffer - after_input;
389 0 : TLOG(TLVL_DEBUG + 34) << "receive_fragments INPUT_WAIT=" << input_wait_time << ", BUFFER_WAIT=" << buffer_wait_time << ", frags.size()=" << fragments_processed;
390 0 : statsHelper_.addSample(INPUT_WAIT_STAT_KEY, input_wait_time);
391 0 : TLOG(TLVL_DEBUG + 34) << "receive_fragments input wait metric=" << input_wait_time / static_cast<double>(fragments_processed);
392 0 : metricMan->sendMetric("Input Wait Time", input_wait_time / static_cast<double>(fragments_processed), "seconds/fragment", 3, MetricMode::Average | MetricMode::Accumulate | MetricMode::Maximum | MetricMode::Minimum);
393 0 : statsHelper_.addSample(BUFFER_WAIT_STAT_KEY, buffer_wait_time);
394 0 : TLOG(TLVL_DEBUG + 34) << "receive_fragments buffer wait metric=" << buffer_wait_time / static_cast<double>(fragments_processed);
395 0 : metricMan->sendMetric("Buffer Wait Time", buffer_wait_time / static_cast<double>(fragments_processed), "seconds/fragment", 3, MetricMode::Average | MetricMode::Accumulate | MetricMode::Maximum | MetricMode::Minimum);
396 0 : metricMan->sendMetric("Frags Per Read", fragments_processed, "fragments/read", 4, MetricMode::Average | MetricMode::Maximum);
397 : }
398 :
399 0 : statsHelper_.addSample(FRAGMENTS_PER_READ_STAT_KEY, fragments_processed);
400 0 : frags.clear();
401 :
402 0 : bool readyToReport = statsHelper_.readyToReport();
403 0 : if (readyToReport)
404 : {
405 0 : TLOG(TLVL_INFO) << buildStatisticsString_();
406 : }
407 : }
408 :
409 : // 11-May-2015, KAB: call MetricManager::do_stop whenever we exit the
410 : // processing fragments loop so that metrics correctly go to zero when
411 : // there is no data flowing
412 0 : metricMan->do_stop();
413 :
414 0 : TLOG(TLVL_DEBUG + 32) << "receive_fragments loop end";
415 0 : }
416 0 : void artdaq::BoardReaderCore::send_fragments()
417 : {
418 0 : if (rt_priority_ > 0)
419 : {
420 : #pragma GCC diagnostic push
421 : #pragma GCC diagnostic ignored "-Wmissing-field-initializers"
422 0 : sched_param s_param = {};
423 0 : s_param.sched_priority = rt_priority_;
424 0 : if (pthread_setschedparam(pthread_self(), SCHED_RR, &s_param) != 0)
425 : {
426 0 : TLOG(TLVL_WARNING) << "setting realtime priority failed";
427 : }
428 : #pragma GCC diagnostic pop
429 : }
430 :
431 : // try-catch block here?
432 :
433 : // how to turn RT PRI off?
434 0 : if (rt_priority_ > 0)
435 : {
436 : #pragma GCC diagnostic push
437 : #pragma GCC diagnostic ignored "-Wmissing-field-initializers"
438 0 : sched_param s_param = {};
439 0 : s_param.sched_priority = rt_priority_;
440 0 : int status = pthread_setschedparam(pthread_self(), SCHED_RR, &s_param);
441 0 : if (status != 0)
442 : {
443 0 : TLOG(TLVL_ERROR)
444 0 : << "Failed to set realtime priority to " << rt_priority_
445 0 : << ", return code = " << status;
446 : }
447 : #pragma GCC diagnostic pop
448 : }
449 :
450 0 : TLOG(TLVL_DEBUG + 32) << "Initializing DataSenderManager. my_rank=" << my_rank;
451 0 : sender_ptr_ = std::make_unique<artdaq::DataSenderManager>(data_pset_);
452 :
453 0 : TLOG(TLVL_DEBUG + 32) << "Waiting for first fragment.";
454 : artdaq::MonitoredQuantityStats::TIME_POINT_T startTime;
455 : double delta_time;
456 0 : artdaq::FragmentPtrs frags;
457 0 : auto targetFragCount = generator_ptr_->fragmentIDs().size();
458 :
459 0 : sender_thread_active_ = true;
460 :
461 0 : auto wait_start = std::chrono::steady_clock::now();
462 0 : while (!running_ && TimeUtils::GetElapsedTime(wait_start) < start_transition_timeout_)
463 : {
464 0 : usleep(10000);
465 : }
466 0 : if (!running_)
467 : {
468 0 : TLOG(TLVL_ERROR) << "Timeout (" << start_transition_timeout_ << " s) while waiting for Start after send_fragments thread started!";
469 0 : sender_thread_active_ = false;
470 : }
471 :
472 0 : while (sender_thread_active_)
473 : {
474 0 : startTime = artdaq::MonitoredQuantity::getCurrentTime();
475 :
476 0 : TLOG(TLVL_DEBUG + 35) << "send_fragments applyRequests start";
477 0 : sender_thread_active_ = fragment_buffer_ptr_->applyRequests(frags);
478 0 : TLOG(TLVL_DEBUG + 35) << "send_fragments applyRequests done (sender_thread_active_=" << sender_thread_active_ << ")";
479 : // 08-May-2015, KAB & JCF: if the generator getNext() method returns false
480 : // (which indicates that the data flow has stopped) *and* the reason that
481 : // it has stopped is because there was an exception that wasn't handled by
482 : // the experiment-specific FragmentGenerator class, we move to the
483 : // InRunError state so that external observers (e.g. RunControl or
484 : // DAQInterface) can see that there was a problem.
485 0 : if (!sender_thread_active_ && generator_ptr_ && generator_ptr_->exception())
486 : {
487 0 : parent_application_.in_run_failure();
488 : }
489 :
490 0 : delta_time = artdaq::MonitoredQuantity::getCurrentTime() - startTime;
491 :
492 0 : TLOG(TLVL_DEBUG + 34) << "send_fragments REQUEST_WAIT=" << delta_time;
493 0 : statsHelper_.addSample(REQUEST_WAIT_STAT_KEY, delta_time);
494 0 : metricMan->sendMetric("Avg Request Response Wait Time", delta_time, "seconds/fragment", 3, MetricMode::Average);
495 :
496 0 : if (!sender_thread_active_) { break; }
497 :
498 0 : size_t counter = 0;
499 0 : size_t total_bytes = 0;
500 :
501 0 : for (auto& fragPtr : frags)
502 : {
503 0 : if (fragPtr == nullptr)
504 : {
505 0 : TLOG(TLVL_WARNING) << "Encountered a bad fragment pointer in fragment " << fragment_count_ << ". "
506 0 : << "This is most likely caused by a problem with the Fragment Generator!";
507 0 : continue;
508 0 : }
509 0 : if (fragment_count_ == 0)
510 : {
511 0 : TLOG(TLVL_DEBUG + 32) << "Received first Fragment from Fragment Generator, sequence ID " << fragPtr->sequenceID() << ", size = " << fragPtr->sizeBytes() << " bytes.";
512 : }
513 0 : counter++;
514 0 : total_bytes += fragPtr->sizeBytes();
515 :
516 0 : if (artdaq::Fragment::isBroadcastFragmentType(fragPtr->type()))
517 : {
518 : // Just broadcast any system Fragments in the output
519 0 : artdaq::Fragment::sequence_id_t sequence_id = fragPtr->sequenceID();
520 0 : statsHelper_.addSample(FRAGMENTS_PROCESSED_STAT_KEY, fragPtr->sizeBytes());
521 0 : startTime = artdaq::MonitoredQuantity::getCurrentTime();
522 0 : TLOG(TLVL_DEBUG + 36) << "send_fragments seq=" << sequence_id << " sendFragment start";
523 0 : auto res = sender_ptr_->sendFragment(std::move(*fragPtr));
524 0 : TLOG(TLVL_DEBUG + 36) << "send_fragments seq=" << sequence_id << " sendFragment done (dest=" << res.first << ", sts=" << TransferInterface::CopyStatusToString(res.second) << ")";
525 0 : ++fragment_count_;
526 0 : auto output_wait = artdaq::MonitoredQuantity::getCurrentTime() - startTime;
527 0 : statsHelper_.addSample(OUTPUT_WAIT_STAT_KEY, output_wait);
528 0 : metricMan->sendMetric("Output Wait Time", output_wait, "seconds/fragment", 3, MetricMode::Average | MetricMode::Accumulate | MetricMode::Maximum | MetricMode::Minimum);
529 0 : continue;
530 0 : }
531 :
532 0 : artdaq::Fragment::sequence_id_t sequence_id = fragPtr->sequenceID();
533 0 : Globals::SetMFIteration("Sequence ID " + std::to_string(sequence_id));
534 0 : statsHelper_.addSample(FRAGMENTS_PROCESSED_STAT_KEY, fragPtr->sizeBytes());
535 :
536 : /*if ((fragment_count_ % 250) == 0)
537 : {
538 : TLOG(TLVL_DEBUG + 32)
539 : << "Sending fragment " << fragment_count_
540 : << " with sequence id " << sequence_id << ".";
541 : }*/
542 :
543 : // check for continous sequence IDs
544 0 : if (!skip_seqId_test_ && abs(static_cast<int64_t>(sequence_id) - static_cast<int64_t>(prev_seq_id_)) > 1)
545 : {
546 0 : TLOG(TLVL_WARNING)
547 0 : << "Missing sequence IDs: current sequence ID = "
548 0 : << sequence_id << ", previous sequence ID = "
549 0 : << prev_seq_id_ << ".";
550 : }
551 0 : prev_seq_id_ = sequence_id;
552 :
553 0 : startTime = artdaq::MonitoredQuantity::getCurrentTime();
554 0 : TLOG(TLVL_DEBUG + 36) << "send_fragments seq=" << sequence_id << " sendFragment start";
555 0 : auto res = sender_ptr_->sendFragment(std::move(*fragPtr));
556 0 : if (sender_ptr_->GetSentSequenceIDCount(sequence_id) == targetFragCount)
557 : {
558 0 : sender_ptr_->RemoveRoutingTableEntry(sequence_id);
559 : }
560 0 : TLOG(TLVL_DEBUG + 36) << "send_fragments seq=" << sequence_id << " sendFragment done (dest=" << res.first << ", sts=" << TransferInterface::CopyStatusToString(res.second) << ")";
561 0 : ++fragment_count_;
562 0 : auto output_wait = artdaq::MonitoredQuantity::getCurrentTime() - startTime;
563 0 : statsHelper_.addSample(OUTPUT_WAIT_STAT_KEY, output_wait);
564 0 : metricMan->sendMetric("Output Wait Time", output_wait, "seconds/fragment", 3, MetricMode::Average | MetricMode::Accumulate | MetricMode::Maximum | MetricMode::Minimum);
565 :
566 0 : bool readyToReport = statsHelper_.readyToReport();
567 0 : if (readyToReport)
568 : {
569 0 : TLOG(TLVL_INFO) << buildStatisticsString_();
570 : }
571 :
572 : // Turn on lvls (mem and/or slow) 3,13,14 to log every send.
573 0 : TLOG(((fragment_count_ == 1) ? TLVL_DEBUG + 32
574 : : (((fragment_count_ % 250) == 0 || readyToReport) ? TLVL_DEBUG + 36 : TLVL_DEBUG + 37)))
575 0 : << ((fragment_count_ == 1)
576 0 : ? "Sent first Fragment"
577 0 : : "Sending fragment " + std::to_string(fragment_count_))
578 0 : << " with SeqID " << sequence_id << ".";
579 : }
580 :
581 0 : if (counter > 0)
582 : {
583 0 : metricMan->sendMetric("Fragment Count", fragment_count_, "fragments", 1, MetricMode::LastPoint);
584 0 : metricMan->sendMetric("Fragment Rate", counter, "fragments", 1, MetricMode::Rate);
585 0 : metricMan->sendMetric("Fragment Size", total_bytes / counter, "bytes/fragment", 2, MetricMode::Average | MetricMode::Accumulate | MetricMode::Maximum | MetricMode::Minimum);
586 0 : metricMan->sendMetric("Data Rate", total_bytes, "bytes", 2, MetricMode::Rate);
587 : }
588 :
589 0 : frags.clear();
590 0 : std::this_thread::yield();
591 : }
592 :
593 0 : sender_ptr_.reset(nullptr);
594 :
595 : // 11-May-2015, KAB: call MetricManager::do_stop whenever we exit the
596 : // processing fragments loop so that metrics correctly go to zero when
597 : // there is no data flowing
598 0 : metricMan->do_stop();
599 :
600 0 : TLOG(TLVL_DEBUG + 32) << "send_fragments loop end";
601 0 : }
602 :
603 0 : std::string artdaq::BoardReaderCore::report(std::string const& which) const
604 : {
605 0 : std::string resultString;
606 :
607 : // pass the request to the FragmentGenerator instance, if it's available
608 0 : if (generator_ptr_ != nullptr && which != "core")
609 : {
610 0 : resultString = generator_ptr_->ReportCmd(which);
611 0 : if (resultString.length() > 0) { return resultString; }
612 : }
613 :
614 : // handle the request at this level, if we can
615 : // --> nothing here yet
616 :
617 : // if we haven't been able to come up with any report so far, say so
618 0 : std::string tmpString = app_name + " run number = ";
619 0 : tmpString.append(boost::lexical_cast<std::string>(run_id_.run()));
620 :
621 0 : tmpString.append(", Sent Fragment count = ");
622 0 : tmpString.append(boost::lexical_cast<std::string>(fragment_count_));
623 0 : if (which == "core")
624 : {
625 : // do nothing
626 : }
627 : //-----------------------------------------------------------------------------
628 : // P.Murat: add statistics report, the const/non const confusion to be cleaned up
629 : // by the maintainers
630 : //-----------------------------------------------------------------------------
631 0 : else if (which == "stats")
632 : {
633 0 : auto non_const_this = (artdaq::BoardReaderCore*)this;
634 0 : tmpString += ", " + non_const_this->buildStatisticsString_();
635 : }
636 : else
637 : {
638 0 : tmpString.append(". Command=\"" + which + "\" is not currently supported.");
639 : }
640 0 : return tmpString;
641 0 : }
642 :
643 0 : bool artdaq::BoardReaderCore::metaCommand(std::string const& command, std::string const& arg)
644 : {
645 0 : TLOG(TLVL_DEBUG + 32) << "metaCommand method called with "
646 0 : << "command = \"" << command << "\""
647 0 : << ", arg = \"" << arg << "\""
648 0 : << ".";
649 :
650 0 : if (generator_ptr_)
651 : {
652 0 : return generator_ptr_->metaCommand(command, arg);
653 : }
654 :
655 0 : return true;
656 : }
657 :
658 0 : std::string artdaq::BoardReaderCore::buildStatisticsString_()
659 : {
660 0 : std::ostringstream oss;
661 0 : double fragmentsGeneratedCount = 1.0;
662 0 : double fragmentsOutputCount = 1.0;
663 0 : oss << app_name << " statistics:" << std::endl;
664 :
665 0 : oss << " Fragments read: ";
666 0 : artdaq::MonitoredQuantityPtr mqPtr = artdaq::StatisticsCollection::getInstance().getMonitoredQuantity(FRAGMENTS_PER_READ_STAT_KEY);
667 0 : if (mqPtr.get() != nullptr)
668 : {
669 0 : artdaq::MonitoredQuantityStats stats;
670 0 : mqPtr->getStats(stats);
671 0 : oss << stats.recentValueSum << " fragments generated at "
672 0 : << stats.recentSampleRate << " getNext calls/sec, fragment rate = "
673 0 : << stats.recentValueRate << " fragments/sec, monitor window = "
674 0 : << stats.recentDuration << " sec, min::max read size = "
675 0 : << stats.recentValueMin
676 0 : << "::"
677 0 : << stats.recentValueMax
678 0 : << " fragments";
679 0 : fragmentsGeneratedCount = std::max(double(stats.recentSampleCount), 1.0);
680 0 : oss << " Average times per fragment: ";
681 0 : if (stats.recentSampleRate > 0.0)
682 : {
683 0 : oss << " elapsed time = "
684 0 : << (1.0 / stats.recentSampleRate) << " sec";
685 : }
686 0 : }
687 :
688 0 : oss << std::endl;
689 0 : mqPtr = artdaq::StatisticsCollection::getInstance().getMonitoredQuantity(FRAGMENTS_PROCESSED_STAT_KEY);
690 0 : if (mqPtr.get() != nullptr)
691 : {
692 0 : artdaq::MonitoredQuantityStats stats;
693 0 : mqPtr->getStats(stats);
694 0 : oss << " Fragment output statistics: "
695 0 : << stats.recentSampleCount << " fragments sent at "
696 0 : << stats.recentSampleRate << " fragments/sec, effective data rate = "
697 0 : << (stats.recentValueRate / 1024.0 / 1024.0) << " MB/sec, monitor window = "
698 0 : << stats.recentDuration << " sec, min::max event size = "
699 0 : << (stats.recentValueMin / 1024.0 / 1024.0)
700 0 : << "::"
701 0 : << (stats.recentValueMax / 1024.0 / 1024.0)
702 0 : << " MB" << std::endl;
703 0 : fragmentsOutputCount = std::max(double(stats.recentSampleCount), 1.0);
704 0 : }
705 :
706 : // 31-Dec-2014, KAB - Just a reminder that using "fragmentCount" in the
707 : // denominator of the calculations below is important because the way that
708 : // the accumulation of these statistics is done is not fragment-by-fragment
709 : // but read-by-read (where each read can contain multiple fragments).
710 : // 29-Aug-2016, KAB - BRSYNC_WAIT and OUTPUT_WAIT are now done fragment-by-
711 : // fragment, but we'll leave the calculation the same. (The alternative
712 : // would be to use recentValueAverage().)
713 :
714 0 : mqPtr = artdaq::StatisticsCollection::getInstance().getMonitoredQuantity(INPUT_WAIT_STAT_KEY);
715 0 : if (mqPtr.get() != nullptr)
716 : {
717 0 : oss << " Input wait time = "
718 0 : << (mqPtr->getRecentValueSum() / fragmentsGeneratedCount) << " s/fragment";
719 : }
720 0 : mqPtr = artdaq::StatisticsCollection::getInstance().getMonitoredQuantity(BUFFER_WAIT_STAT_KEY);
721 0 : if (mqPtr.get() != 0)
722 : {
723 0 : oss << ", buffer wait time = "
724 0 : << (mqPtr->getRecentValueSum() / fragmentsGeneratedCount) << " s/fragment";
725 : }
726 0 : mqPtr = artdaq::StatisticsCollection::getInstance().getMonitoredQuantity(REQUEST_WAIT_STAT_KEY);
727 0 : if (mqPtr.get() != 0)
728 : {
729 0 : oss << ", request wait time = "
730 0 : << (mqPtr->getRecentValueSum() / fragmentsOutputCount) << " s/fragment";
731 : }
732 :
733 0 : mqPtr = artdaq::StatisticsCollection::getInstance().getMonitoredQuantity(OUTPUT_WAIT_STAT_KEY);
734 0 : if (mqPtr.get() != nullptr)
735 : {
736 0 : oss << ", output wait time = "
737 0 : << (mqPtr->getRecentValueSum() / fragmentsOutputCount) << " s/fragment";
738 : }
739 : //-----------------------------------------------------------------------------
740 : // 2024-01-13 P.Murat: add SHM data
741 : //-----------------------------------------------------------------------------
742 0 : oss << fragment_buffer_ptr_->getStatReport();
743 :
744 0 : return oss.str();
745 0 : }
|
