Line data Source code
1 : #include "TRACE/tracemf.h" // Pre-empt TRACE/trace.h from Fragment.hh.
2 : #include "artdaq-core/Data/Fragment.hh"
3 :
4 : #define TRACE_NAME "TransferWrapper"
5 :
6 : #include "artdaq/ArtModules/detail/TransferWrapper.hh"
7 :
8 : #include "artdaq-core/Utilities/ExceptionHandler.hh"
9 : #include "artdaq-core/Utilities/TimeUtils.hh"
10 : #include "artdaq/DAQdata/NetMonHeader.hh"
11 : #include "artdaq/ExternalComms/MakeCommanderPlugin.hh"
12 : #include "artdaq/TransferPlugins/MakeTransferPlugin.hh"
13 :
14 : #include "cetlib_except/exception.h"
15 : #include "fhiclcpp/ParameterSet.h"
16 :
17 : #include <csignal>
18 : #include <iostream>
19 : #include <limits>
20 : #include <memory>
21 : #include <sstream>
22 : #include <string>
23 :
24 : namespace {
25 : volatile std::sig_atomic_t gSignalStatus = 0; ///< Stores singal from signal handler
26 : }
27 :
28 : /**
29 : * \brief Handle a Unix signal
30 : * \param signal Signal to handle
31 : */
32 0 : void signal_handler(int signal)
33 : {
34 0 : gSignalStatus = signal;
35 0 : }
36 :
37 0 : artdaq::TransferWrapper::TransferWrapper(const fhicl::ParameterSet& pset)
38 0 : : timeoutInUsecs_(pset.get<std::size_t>("timeoutInUsecs", 100000))
39 0 : , last_received_data_()
40 0 : , last_report_(std::chrono::steady_clock::now())
41 0 : , transfer_(nullptr)
42 0 : , commander_(nullptr)
43 0 : , pset_(pset)
44 0 : , dispatcherHost_(pset.get<std::string>("dispatcherHost", "localhost"))
45 0 : , dispatcherPort_(pset.get<std::string>("dispatcherPort", "5266"))
46 0 : , serverUrl_(pset.get<std::string>("server_url", "http://" + dispatcherHost_ + ":" + dispatcherPort_ + "/RPC2"))
47 0 : , maxEventsBeforeInit_(pset.get<std::size_t>("maxEventsBeforeInit", 5))
48 0 : , allowedFragmentTypes_(pset.get<std::vector<int>>("allowedFragmentTypes", {static_cast<int>(artdaq::Fragment::DataFragmentType),
49 : static_cast<int>(artdaq::Fragment::InitFragmentType),
50 : static_cast<int>(artdaq::Fragment::EndOfRunFragmentType),
51 : static_cast<int>(artdaq::Fragment::EndOfSubrunFragmentType),
52 : static_cast<int>(artdaq::Fragment::RunDataFragmentType),
53 : static_cast<int>(artdaq::Fragment::SubrunDataFragmentType)}))
54 0 : , runningStateTimeout_(pset.get<double>("dispatcherConnectTimeout", 0))
55 0 : , runningStateInterval_us_(pset.get<size_t>("dispatcherConnectRetryInterval_us", 1000000))
56 0 : , quitOnFragmentIntegrityProblem_(pset.get<bool>("quitOnFragmentIntegrityProblem", true))
57 0 : , multi_run_mode_(pset.get<bool>("multi_run_mode", false))
58 0 : , monitorRegistered_(false)
59 : {
60 0 : std::signal(SIGINT, signal_handler);
61 :
62 : try
63 : {
64 0 : if (metricMan)
65 : {
66 0 : metricMan->initialize(pset.get<fhicl::ParameterSet>("metrics", fhicl::ParameterSet()), "Online Monitor");
67 0 : metricMan->do_start();
68 : }
69 : }
70 0 : catch (...)
71 : {
72 0 : ExceptionHandler(ExceptionHandlerRethrow::no, "TransferWrapper: could not configure metrics");
73 0 : }
74 :
75 : // Clamp possible values
76 0 : if (runningStateInterval_us_ < 1000)
77 : {
78 0 : TLOG(TLVL_WARNING) << "Invalid value " << runningStateInterval_us_ << " us detected for dispatcherConnectRetryInterval_us. Setting to 1000 us";
79 0 : runningStateInterval_us_ = 1000;
80 : }
81 0 : if (runningStateInterval_us_ > 30000000)
82 : {
83 0 : TLOG(TLVL_WARNING) << "Invalid value " << runningStateInterval_us_ << " us detected for dispatcherConnectRetryInterval_us. Setting to 30,000,000 us";
84 0 : runningStateInterval_us_ = 30000000;
85 : }
86 :
87 0 : fhicl::ParameterSet new_pset(pset);
88 0 : if (!new_pset.has_key("server_url"))
89 : {
90 0 : new_pset.put<std::string>("server_url", serverUrl_);
91 : }
92 :
93 0 : artdaq::Commandable c;
94 0 : commander_ = MakeCommanderPlugin(new_pset, c);
95 0 : }
96 :
97 0 : artdaq::FragmentPtrs artdaq::TransferWrapper::receiveMessage()
98 : {
99 0 : artdaq::FragmentPtrs fragmentPtrs;
100 0 : bool receivedFragment = false;
101 : static bool initialized = false;
102 : static size_t fragments_received = 0;
103 :
104 0 : while (gSignalStatus == 0)
105 : {
106 0 : receivedFragment = false;
107 0 : auto fragmentPtr = std::make_unique<artdaq::Fragment>();
108 :
109 0 : while (!receivedFragment)
110 : {
111 0 : if (gSignalStatus != 0)
112 : {
113 0 : TLOG(TLVL_INFO) << "Ctrl-C appears to have been hit";
114 0 : return fragmentPtrs;
115 : }
116 0 : if (!monitorRegistered_)
117 : {
118 0 : registerMonitor();
119 0 : if (!monitorRegistered_)
120 : {
121 0 : return fragmentPtrs;
122 : }
123 : }
124 :
125 : try
126 : {
127 0 : auto result = transfer_->receiveFragment(*fragmentPtr, timeoutInUsecs_);
128 :
129 0 : if (result >= artdaq::TransferInterface::RECV_SUCCESS)
130 : {
131 0 : receivedFragment = true;
132 0 : fragments_received++;
133 :
134 : static size_t cntr = 0;
135 0 : ++cntr;
136 :
137 : // Report the first 10 events, then every 10th, then every 100th, and so
138 : // on, so a long run does not put one line per event into the log.
139 0 : size_t report_interval = 1;
140 0 : while (cntr / report_interval >= 10)
141 : {
142 0 : report_interval *= 10;
143 : }
144 :
145 0 : if (cntr % report_interval == 0)
146 : {
147 0 : auto mod = cntr % 10;
148 0 : auto suffix = "-th";
149 0 : if (mod == 1)
150 : {
151 0 : suffix = "-st";
152 : }
153 0 : if (mod == 2)
154 : {
155 0 : suffix = "-nd";
156 : }
157 0 : if (mod == 3)
158 : {
159 0 : suffix = "-rd";
160 : }
161 0 : TLOG(TLVL_INFO) << "Received " << cntr << suffix << " event, "
162 0 : << "seqID == " << fragmentPtr->sequenceID()
163 0 : << ", type == " << fragmentPtr->typeString();
164 : }
165 0 : last_received_data_ = std::chrono::steady_clock::now();
166 0 : continue;
167 0 : }
168 0 : if (result == artdaq::TransferInterface::DATA_END)
169 : {
170 0 : TLOG(TLVL_ERROR) << "Transfer Plugin disconnected or other unrecoverable error. Shutting down.";
171 0 : return fragmentPtrs;
172 : }
173 : else
174 : {
175 0 : auto tlvl = TLVL_DEBUG + 33;
176 0 : if (artdaq::TimeUtils::GetElapsedTime(last_report_) > 1.0 && artdaq::TimeUtils::GetElapsedTime(last_received_data_) > 1.0)
177 : {
178 0 : tlvl = TLVL_WARNING;
179 0 : last_report_ = std::chrono::steady_clock::now();
180 : }
181 :
182 0 : auto last_received_milliseconds = artdaq::TimeUtils::GetElapsedTimeMilliseconds(last_received_data_);
183 :
184 : // 02-Jun-2018, KAB: added status/result printout
185 : // to-do: add another else clause that explicitly checks for RECV_TIMEOUT
186 0 : TLOG(tlvl) << "Timeout occurred in call to transfer_->receiveFragmentFrom; will try again"
187 0 : << ", status = " << result << ", last received data " << last_received_milliseconds << " ms ago.";
188 : }
189 : }
190 0 : catch (...)
191 : {
192 0 : ExceptionHandler(ExceptionHandlerRethrow::yes,
193 : "Problem receiving data in TransferWrapper::receiveMessage");
194 0 : }
195 : }
196 :
197 0 : if (fragmentPtr->type() == artdaq::Fragment::EndOfDataFragmentType)
198 : {
199 0 : if (!multi_run_mode_)
200 : {
201 0 : TLOG(TLVL_DEBUG + 32) << "Received shutdown message, returning";
202 0 : fragmentPtrs.push_back(std::move(fragmentPtr));
203 : }
204 0 : return fragmentPtrs;
205 : }
206 :
207 0 : checkIntegrity(*fragmentPtr);
208 :
209 0 : if (initialized || fragmentPtr->type() == artdaq::Fragment::InitFragmentType)
210 : {
211 0 : initialized = true;
212 0 : fragmentPtrs.push_back(std::move(fragmentPtr));
213 0 : break;
214 : }
215 :
216 0 : if (fragments_received > maxEventsBeforeInit_)
217 : {
218 0 : TLOG(TLVL_ERROR) << "First " << maxEventsBeforeInit_ << " events received did not include the \"Init\" event containing necessary info for art; exiting...";
219 0 : throw cet::exception("TransferWrapper") << "First " << maxEventsBeforeInit_ << " events received did not include the \"Init\" event containing necessary info for art; exiting..."; // NOLINT(cert-err60-cpp)
220 : }
221 0 : }
222 :
223 0 : return fragmentPtrs;
224 0 : }
225 :
226 0 : std::shared_ptr<ArtdaqEvent> artdaq::TransferWrapper::receiveMessages()
227 : {
228 0 : std::shared_ptr<ArtdaqEvent> output = std::make_shared<ArtdaqEvent>();
229 :
230 0 : auto ptrs = receiveMessage();
231 0 : for (auto& ptr : ptrs)
232 : {
233 0 : auto fragType = ptr->type();
234 0 : auto fragPtr = ptr.release();
235 0 : ptr.reset(nullptr);
236 :
237 0 : if (output->fragments.count(fragType) == 0u)
238 : {
239 0 : output->fragments[fragType] = std::make_unique<artdaq::Fragments>();
240 : }
241 :
242 0 : output->fragments[fragType]->emplace_back(std::move(*fragPtr));
243 : }
244 :
245 0 : return output;
246 0 : }
247 :
248 0 : void artdaq::TransferWrapper::checkIntegrity(const artdaq::Fragment& fragment) const
249 : {
250 0 : const size_t artdaqheader = artdaq::detail::RawFragmentHeader::num_words() *
251 : sizeof(artdaq::detail::RawFragmentHeader::RawDataType);
252 0 : const auto payload = static_cast<size_t>(fragment.dataEndBytes() - fragment.dataBeginBytes());
253 0 : const size_t metadata = sizeof(artdaq::NetMonHeader);
254 0 : const size_t totalsize = fragment.sizeBytes();
255 :
256 0 : const auto type = static_cast<size_t>(fragment.type());
257 :
258 0 : if (totalsize != artdaqheader + metadata + payload)
259 : {
260 0 : std::stringstream errmsg;
261 0 : errmsg << "Error: artdaq fragment of type " << fragment.typeString() << ", sequence ID " << fragment.sequenceID() << " has internally inconsistent measures of its size, signalling data corruption: in bytes,"
262 0 : << " total size = " << totalsize << ", artdaq fragment header = " << artdaqheader << ", metadata = " << metadata << ", payload = " << payload;
263 :
264 0 : TLOG(TLVL_ERROR) << errmsg.str();
265 :
266 0 : if (quitOnFragmentIntegrityProblem_)
267 : {
268 0 : throw cet::exception("TransferWrapper") << errmsg.str(); // NOLINT(cert-err60-cpp)
269 : }
270 :
271 0 : return;
272 0 : }
273 :
274 0 : auto findloc = std::find(allowedFragmentTypes_.begin(), allowedFragmentTypes_.end(), static_cast<int>(type));
275 :
276 0 : if (findloc == allowedFragmentTypes_.end())
277 : {
278 0 : std::stringstream errmsg;
279 0 : errmsg << "Error: artdaq fragment appears to have type "
280 0 : << type << ", not found in the allowed fragment types list";
281 :
282 0 : TLOG(TLVL_ERROR) << errmsg.str();
283 0 : if (quitOnFragmentIntegrityProblem_)
284 : {
285 0 : throw cet::exception("TransferWrapper") << errmsg.str(); // NOLINT(cert-err60-cpp)
286 : }
287 :
288 0 : return;
289 0 : }
290 : }
291 :
292 0 : void artdaq::TransferWrapper::registerMonitor(bool setup_transfer)
293 : {
294 0 : if (setup_transfer)
295 : {
296 : try
297 : {
298 0 : transfer_.reset(nullptr);
299 0 : transfer_ = MakeTransferPlugin(pset_, "transfer_plugin", TransferInterface::Role::kReceive);
300 : }
301 0 : catch (...)
302 : {
303 0 : ExceptionHandler(ExceptionHandlerRethrow::yes,
304 : "TransferWrapper: failure in call to MakeTransferPlugin");
305 0 : }
306 : }
307 :
308 0 : auto start = std::chrono::steady_clock::now();
309 0 : auto sts = getDispatcherStatus();
310 0 : while (sts != "Running" && (runningStateTimeout_ == 0 || TimeUtils::GetElapsedTime(start) < runningStateTimeout_))
311 : {
312 0 : TLOG(TLVL_DEBUG + 32) << "Dispatcher state: " << sts;
313 0 : if (gSignalStatus != 0)
314 : {
315 0 : TLOG(TLVL_INFO) << "Ctrl-C appears to have been hit";
316 0 : return;
317 : }
318 0 : TLOG(TLVL_INFO) << "Waited " << std::fixed << std::setprecision(2) << TimeUtils::GetElapsedTime(start) << " s / " << runningStateTimeout_ << " s for Dispatcher to enter the Running state (state=" << sts << ")";
319 0 : usleep(runningStateInterval_us_);
320 0 : sts = getDispatcherStatus();
321 : }
322 0 : if (sts != "Running")
323 : {
324 0 : return;
325 : }
326 :
327 0 : auto dispatcherConfig = pset_.get<fhicl::ParameterSet>("dispatcher_config");
328 :
329 0 : int retry = 3;
330 :
331 0 : while (retry > 0)
332 : {
333 0 : label_ = dispatcherConfig.get<std::string>("unique_label") + "T" + std::to_string(time(0));
334 0 : dispatcherConfig.erase("unique_label");
335 0 : dispatcherConfig.put<std::string>("unique_label", label_);
336 0 : TLOG(TLVL_INFO) << "Attempting to register this monitor (\"" << label_
337 0 : << "\") with the dispatcher aggregator";
338 :
339 0 : auto status = commander_->send_register_monitor(dispatcherConfig.to_string());
340 :
341 0 : TLOG(TLVL_INFO) << "Response from dispatcher is \"" << status << "\"";
342 :
343 0 : if (status == "Success")
344 : {
345 0 : monitorRegistered_ = true;
346 0 : break;
347 : }
348 :
349 0 : TLOG(TLVL_WARNING) << "Error in TransferWrapper: attempt to register with dispatcher did not result in the \"Success\" response";
350 0 : usleep(100000);
351 :
352 0 : retry--;
353 0 : }
354 0 : }
355 :
356 0 : void artdaq::TransferWrapper::unregisterMonitor(bool force, std::string label)
357 : {
358 0 : if (!monitorRegistered_ && !force)
359 : {
360 0 : TLOG(TLVL_WARNING) << "The function to unregister the monitor was called, but the monitor doesn't appear to be registered";
361 0 : return;
362 : }
363 :
364 0 : if (label == "")
365 : {
366 0 : label = label_;
367 : }
368 :
369 0 : auto start_time = std::chrono::steady_clock::now();
370 0 : bool waiting = true;
371 0 : while (artdaq::TimeUtils::GetElapsedTime(start_time) < 5.0 && waiting)
372 : {
373 0 : std::string sts = getDispatcherStatus();
374 :
375 0 : if (sts.empty())
376 0 : return;
377 :
378 0 : if (sts == "busy")
379 : {
380 0 : TLOG(TLVL_INFO) << "The Dispatcher returned \"busy\", will wait 0.5s and retry";
381 0 : usleep(500000);
382 0 : continue;
383 0 : }
384 :
385 0 : if (sts != "Running" && sts != "Ready")
386 : {
387 0 : TLOG(TLVL_WARNING) << "The Dispatcher is not in the Running or Ready state, will not attempt to unregister (state: " << sts << ")";
388 0 : return;
389 : }
390 0 : waiting = false;
391 0 : }
392 0 : if (waiting)
393 : {
394 0 : TLOG(TLVL_WARNING) << "A timeout occurred waiting for the Dispatcher to leave the \"busy\" state, will not attempt to unregister";
395 0 : return;
396 : }
397 :
398 0 : int retry = 3;
399 0 : while (retry > 0)
400 : {
401 0 : TLOG(TLVL_INFO) << "Requesting that this monitor (" << label
402 0 : << ") be unregistered from the dispatcher aggregator";
403 :
404 0 : auto status = commander_->send_unregister_monitor(label);
405 :
406 0 : TLOG(TLVL_INFO) << "Response from dispatcher is \"" << status << "\"";
407 :
408 0 : if (status == "Success")
409 : {
410 0 : break;
411 : }
412 0 : else if (status == "busy")
413 : {
414 0 : TLOG(TLVL_DEBUG + 32) << "The Dispatcher returned \"busy\", will retry in 0.5s";
415 : }
416 : else
417 : {
418 0 : TLOG(TLVL_WARNING) << "The Dispatcher returned status " << status << " when attempting to unregister this monitor!";
419 : // throw cet::exception("TransferWrapper") << "Error in TransferWrapper: attempt to unregister with dispatcher did not result in the \"Success\" response";
420 : }
421 0 : retry--;
422 0 : usleep(500000);
423 0 : }
424 :
425 0 : TLOG(TLVL_INFO) << "Successfully unregistered the monitor from the Dispatcher";
426 0 : monitorRegistered_ = false;
427 : }
428 :
429 0 : std::string artdaq::TransferWrapper::getDispatcherStatus()
430 : {
431 : try
432 : {
433 0 : return commander_->send_status();
434 : }
435 0 : catch (std::exception const& ex)
436 : {
437 0 : TLOG(TLVL_WARNING) << "An exception was thrown trying to collect the Dispatcher's status. Most likely cause is the application is no longer running.";
438 0 : return "";
439 0 : }
440 : }
441 :
442 0 : artdaq::TransferWrapper::~TransferWrapper()
443 : {
444 0 : if (monitorRegistered_)
445 : {
446 : try
447 : {
448 0 : unregisterMonitor();
449 : }
450 0 : catch (...)
451 : {
452 0 : ExceptionHandler(ExceptionHandlerRethrow::no,
453 : "An exception occurred when trying to unregister monitor during TransferWrapper's destruction");
454 0 : }
455 : }
456 0 : artdaq::Globals::CleanUpGlobals();
457 0 : }
|
