Line data Source code
1 : #include "artdaq/DAQdata/Globals.hh"
2 : #define TRACE_NAME (app_name + "_BoardReaderApp").c_str() // NOLINT
3 :
4 : #include "artdaq-core/Utilities/ExceptionHandler.hh"
5 : #include "artdaq/Application/BoardReaderApp.hh"
6 :
7 : #include <boost/lexical_cast.hpp>
8 :
9 : #include <memory>
10 : #include <string>
11 :
12 0 : artdaq::BoardReaderApp::BoardReaderApp()
13 0 : : fragment_receiver_ptr_(nullptr)
14 : {
15 0 : }
16 :
17 : // *******************************************************************
18 : // *** The following methods implement the state machine operations.
19 : // *******************************************************************
20 :
21 0 : bool artdaq::BoardReaderApp::do_initialize(fhicl::ParameterSet const& pset, uint64_t timeout, uint64_t timestamp)
22 : {
23 0 : report_string_ = "";
24 0 : external_request_status_ = true;
25 :
26 : // in the following block, we first destroy the existing BoardReader
27 : // instance, then create a new one. Doing it in one step does not
28 : // produce the desired result since that creates a new instance and
29 : // then deletes the old one, and we need the opposite order.
30 0 : TLOG(TLVL_DEBUG + 32) << "Initializing first deleting old instance " << static_cast<void*>(fragment_receiver_ptr_.get());
31 0 : fragment_receiver_ptr_.reset(nullptr);
32 0 : fragment_receiver_ptr_ = std::make_unique<BoardReaderCore>(*this);
33 0 : TLOG(TLVL_DEBUG + 32) << "Initializing new BoardReaderCore at " << static_cast<void*>(fragment_receiver_ptr_.get()) << " with pset " << pset.to_string();
34 : try
35 : {
36 0 : external_request_status_ = fragment_receiver_ptr_->initialize(pset, timeout, timestamp);
37 : }
38 0 : catch (const std::exception& ex)
39 : {
40 0 : external_request_status_ = false;
41 0 : report_string_ = "Error initializing " + app_name + ": " + ex.what();
42 0 : return false;
43 0 : }
44 0 : catch (...)
45 : {
46 0 : external_request_status_ = false;
47 0 : report_string_ = "Error initializing " + app_name + " (unknown exception)";
48 0 : return false;
49 0 : }
50 0 : if (!external_request_status_)
51 : {
52 0 : auto core_error = fragment_receiver_ptr_->GetLastInitError();
53 0 : if (!core_error.empty())
54 : {
55 0 : report_string_ = "Error initializing " + app_name + ": " + core_error;
56 : }
57 : else
58 : {
59 0 : report_string_ = "Error initializing ";
60 0 : report_string_.append(app_name + " ");
61 0 : report_string_.append("with ParameterSet = \"" + pset.to_string() + "\".");
62 : }
63 0 : }
64 :
65 0 : TLOG(TLVL_DEBUG + 32) << "do_initialize(fhicl::ParameterSet, uint64_t, uint64_t): "
66 0 : << "Done initializing.";
67 0 : return external_request_status_;
68 : }
69 :
70 0 : bool artdaq::BoardReaderApp::do_start(art::RunID id, uint64_t timeout, uint64_t timestamp)
71 : {
72 0 : report_string_ = "";
73 0 : if (timeout == 0)
74 : {
75 0 : timeout = 3600; // seconds
76 : }
77 0 : fragment_receiver_ptr_->SetStartTransitionTimeout(timeout);
78 0 : external_request_status_ = true;
79 :
80 0 : boost::thread::attributes attrs;
81 0 : attrs.set_stack_size(4096 * 2000); // 8 MB
82 : try
83 : {
84 0 : fragment_output_thread_ = boost::thread(attrs, boost::bind(&BoardReaderCore::send_fragments, fragment_receiver_ptr_.get()));
85 : char tname[16]; // Size 16 - see man page pthread_setname_np(3) and/or prctl(2)
86 0 : snprintf(tname, sizeof(tname) - 1, "%d-FragOutput", my_rank); // NOLINT
87 0 : tname[sizeof(tname) - 1] = '\0'; // assure term. snprintf is not too evil :)
88 0 : auto handle = fragment_output_thread_.native_handle();
89 0 : pthread_setname_np(handle, tname);
90 0 : fragment_input_thread_ = boost::thread(attrs, boost::bind(&BoardReaderCore::receive_fragments, fragment_receiver_ptr_.get()));
91 0 : snprintf(tname, sizeof(tname) - 1, "%d-FragInput", my_rank); // NOLINT
92 0 : tname[sizeof(tname) - 1] = '\0'; // assure term. snprintf is not too evil :)
93 0 : handle = fragment_input_thread_.native_handle();
94 0 : pthread_setname_np(handle, tname);
95 : }
96 0 : catch (const boost::exception& e)
97 : {
98 0 : std::stringstream exception_string;
99 0 : exception_string << "Caught boost::exception starting Fragment Processing threads: " << boost::diagnostic_information(e) << ", errno=" << errno;
100 :
101 0 : ExceptionHandler(ExceptionHandlerRethrow::yes, exception_string.str());
102 0 : }
103 :
104 0 : auto start_wait = std::chrono::steady_clock::now();
105 0 : while (!fragment_receiver_ptr_->GetSenderThreadActive() || !fragment_receiver_ptr_->GetReceiverThreadActive())
106 : {
107 0 : if (TimeUtils::GetElapsedTime(start_wait) > timeout)
108 : {
109 0 : TLOG(TLVL_ERROR) << "Timeout occurred waiting for BoardReaderCore threads to start. Timeout = " << timeout << " s, Time waited = " << TimeUtils::GetElapsedTime(start_wait) << " s,"
110 0 : << " Receiver ready: " << std::boolalpha << fragment_receiver_ptr_->GetReceiverThreadActive() << ", Sender ready: " << fragment_receiver_ptr_->GetSenderThreadActive();
111 0 : external_request_status_ = false;
112 0 : break;
113 : }
114 0 : usleep(10000);
115 : }
116 :
117 : // Only if starting threads successful
118 0 : if (external_request_status_)
119 : {
120 0 : external_request_status_ = fragment_receiver_ptr_->start(id, timeout, timestamp);
121 : }
122 :
123 0 : if (!external_request_status_)
124 : {
125 0 : report_string_ = "Error starting ";
126 0 : report_string_.append(app_name + " ");
127 0 : report_string_.append("for run number ");
128 0 : report_string_.append(boost::lexical_cast<std::string>(id.run()));
129 0 : report_string_.append(", timeout ");
130 0 : report_string_.append(boost::lexical_cast<std::string>(timeout));
131 0 : report_string_.append(", timestamp ");
132 0 : report_string_.append(boost::lexical_cast<std::string>(timestamp));
133 0 : report_string_.append(".");
134 : }
135 :
136 0 : return external_request_status_;
137 0 : }
138 :
139 0 : bool artdaq::BoardReaderApp::do_stop(uint64_t timeout, uint64_t timestamp)
140 : {
141 0 : report_string_ = "";
142 0 : external_request_status_ = fragment_receiver_ptr_->stop(timeout, timestamp);
143 0 : if (!external_request_status_)
144 : {
145 0 : report_string_ = "Error stopping ";
146 0 : report_string_.append(app_name + ".");
147 0 : return false;
148 : }
149 0 : if (fragment_input_thread_.joinable())
150 : {
151 0 : TLOG(TLVL_DEBUG + 32) << "Joining fragment input (Generator) thread";
152 0 : fragment_input_thread_.join();
153 : }
154 0 : if (fragment_output_thread_.joinable())
155 : {
156 0 : TLOG(TLVL_DEBUG + 32) << "Joining fragment output (Sender) thread";
157 0 : fragment_output_thread_.join();
158 : }
159 :
160 0 : TLOG(TLVL_DEBUG + 32) << "BoardReader Stopped. Getting run statistics";
161 0 : int number_of_fragments_sent = -1;
162 0 : if (fragment_receiver_ptr_)
163 : {
164 0 : number_of_fragments_sent = fragment_receiver_ptr_->GetFragmentsProcessed();
165 : }
166 0 : TLOG(TLVL_DEBUG + 32) << "do_stop(uint64_t, uint64_t): "
167 0 : << "Number of fragments sent = " << number_of_fragments_sent
168 0 : << ".";
169 :
170 0 : return external_request_status_;
171 : }
172 :
173 0 : bool artdaq::BoardReaderApp::do_pause(uint64_t timeout, uint64_t timestamp)
174 : {
175 0 : report_string_ = "";
176 0 : external_request_status_ = fragment_receiver_ptr_->pause(timeout, timestamp);
177 0 : if (!external_request_status_)
178 : {
179 0 : report_string_ = "Error pausing ";
180 0 : report_string_.append(app_name + ".");
181 : }
182 :
183 0 : if (fragment_input_thread_.joinable()) fragment_input_thread_.join();
184 0 : if (fragment_output_thread_.joinable()) fragment_output_thread_.join();
185 0 : int number_of_fragments_sent = fragment_receiver_ptr_->GetFragmentsProcessed();
186 0 : TLOG(TLVL_DEBUG + 32) << "do_pause(uint64_t, uint64_t): "
187 0 : << "Number of fragments sent = " << number_of_fragments_sent
188 0 : << ".";
189 :
190 0 : return external_request_status_;
191 : }
192 :
193 0 : bool artdaq::BoardReaderApp::do_resume(uint64_t timeout, uint64_t timestamp)
194 : {
195 0 : report_string_ = "";
196 0 : if (timeout == 0)
197 : {
198 0 : timeout = 3600; // seconds
199 : }
200 0 : external_request_status_ = true;
201 :
202 0 : boost::thread::attributes attrs;
203 0 : attrs.set_stack_size(4096 * 2000); // 8 MB
204 : try
205 : {
206 0 : fragment_output_thread_ = boost::thread(attrs, boost::bind(&BoardReaderCore::send_fragments, fragment_receiver_ptr_.get()));
207 : char tname[16]; // Size 16 - see man page pthread_setname_np(3) and/or prctl(2)
208 0 : snprintf(tname, sizeof(tname) - 1, "%d-FragOutput", my_rank); // NOLINT
209 0 : tname[sizeof(tname) - 1] = '\0'; // assure term. snprintf is not too evil :)
210 0 : auto handle = fragment_output_thread_.native_handle();
211 0 : pthread_setname_np(handle, tname);
212 0 : fragment_input_thread_ = boost::thread(attrs, boost::bind(&BoardReaderCore::receive_fragments, fragment_receiver_ptr_.get()));
213 :
214 0 : snprintf(tname, sizeof(tname) - 1, "%d-FragInput", my_rank); // NOLINT
215 0 : tname[sizeof(tname) - 1] = '\0'; // assure term. snprintf is not too evil :)
216 0 : handle = fragment_input_thread_.native_handle();
217 0 : pthread_setname_np(handle, tname);
218 : }
219 0 : catch (const boost::exception& e)
220 : {
221 0 : std::stringstream exception_string;
222 0 : exception_string << "Caught boost::exception starting Fragment Processing threads: " << boost::diagnostic_information(e) << ", errno=" << errno;
223 :
224 0 : ExceptionHandler(ExceptionHandlerRethrow::yes, exception_string.str());
225 0 : }
226 :
227 0 : auto start_wait = std::chrono::steady_clock::now();
228 0 : while (!fragment_receiver_ptr_->GetSenderThreadActive() || !fragment_receiver_ptr_->GetReceiverThreadActive())
229 : {
230 0 : if (TimeUtils::GetElapsedTimeMicroseconds(start_wait) > timeout * 1000000)
231 : {
232 0 : TLOG(TLVL_ERROR) << "Timeout occurred waiting for BoardReaderCore threads to start. Timeout = " << timeout << " s, Time waited = " << TimeUtils::GetElapsedTime(start_wait) << " s,"
233 0 : << " Receiver ready: " << std::boolalpha << fragment_receiver_ptr_->GetReceiverThreadActive() << ", Sender ready: " << fragment_receiver_ptr_->GetSenderThreadActive();
234 0 : external_request_status_ = false;
235 0 : break;
236 : }
237 0 : usleep(10000);
238 : }
239 :
240 : // Only if starting threads successful
241 0 : if (external_request_status_)
242 : {
243 0 : external_request_status_ = fragment_receiver_ptr_->resume(timeout, timestamp);
244 : }
245 0 : if (!external_request_status_)
246 : {
247 0 : report_string_ = "Error resuming ";
248 0 : report_string_.append(app_name + ".");
249 : }
250 0 : return external_request_status_;
251 0 : }
252 :
253 0 : bool artdaq::BoardReaderApp::do_shutdown(uint64_t timeout)
254 : {
255 0 : report_string_ = "";
256 0 : external_request_status_ = fragment_receiver_ptr_->shutdown(timeout);
257 : // 02-Jun-2018, ELF & KAB: it's very, very unlikely that the following call is needed,
258 : // but just in case...
259 0 : if (fragment_input_thread_.joinable()) fragment_input_thread_.join();
260 0 : if (fragment_output_thread_.joinable()) fragment_output_thread_.join();
261 0 : if (!external_request_status_)
262 : {
263 0 : report_string_ = "Error shutting down ";
264 0 : report_string_.append(app_name + ".");
265 : }
266 0 : return external_request_status_;
267 : }
268 :
269 0 : bool artdaq::BoardReaderApp::do_soft_initialize(fhicl::ParameterSet const& pset, uint64_t timeout, uint64_t timestamp)
270 : {
271 0 : report_string_ = "";
272 0 : external_request_status_ = fragment_receiver_ptr_->soft_initialize(pset, timeout, timestamp);
273 0 : if (!external_request_status_)
274 : {
275 0 : report_string_ = "Error soft-initializing ";
276 0 : report_string_.append(app_name + " ");
277 0 : report_string_.append("with ParameterSet = \"" + pset.to_string() + "\".");
278 : }
279 0 : return external_request_status_;
280 : }
281 :
282 0 : bool artdaq::BoardReaderApp::do_reinitialize(fhicl::ParameterSet const& pset, uint64_t timeout, uint64_t timestamp)
283 : {
284 0 : external_request_status_ = fragment_receiver_ptr_->reinitialize(pset, timeout, timestamp);
285 0 : if (!external_request_status_)
286 : {
287 0 : report_string_ = "Error reinitializing ";
288 0 : report_string_.append(app_name + " ");
289 0 : report_string_.append("with ParameterSet = \"" + pset.to_string() + "\".");
290 : }
291 0 : return external_request_status_;
292 : }
293 :
294 0 : void artdaq::BoardReaderApp::BootedEnter()
295 : {
296 0 : TLOG(TLVL_DEBUG + 32) << "Booted state entry action called.";
297 :
298 : // the destruction of any existing BoardReaderCore has to happen in the
299 : // Booted Entry action rather than the Initialized Exit action because the
300 : // Initialized Exit action is only called after the "init" transition guard
301 : // condition is executed.
302 0 : fragment_receiver_ptr_.reset(nullptr);
303 0 : }
304 :
305 0 : bool artdaq::BoardReaderApp::do_meta_command(std::string const& command, std::string const& arg)
306 : {
307 0 : external_request_status_ = fragment_receiver_ptr_->metaCommand(command, arg);
308 0 : if (!external_request_status_)
309 : {
310 0 : report_string_ = "Error running meta-command on ";
311 0 : report_string_.append(app_name + " ");
312 0 : report_string_.append("with command = \"" + command + "\", arg = \"" + arg + "\".");
313 : }
314 0 : return external_request_status_;
315 : }
316 :
317 0 : std::string artdaq::BoardReaderApp::report(std::string const& which) const
318 : {
319 0 : std::string resultString;
320 :
321 : // if all that is requested is the latest state change result, return it
322 0 : if (which == "transition_status")
323 : {
324 0 : if (report_string_.length() > 0) { return report_string_; }
325 :
326 0 : return "Success";
327 : }
328 :
329 : //// if there is an outstanding report/message at the Commandable/Application
330 : //// level, prepend that
331 : // if (report_string_.length() > 0) {
332 : // resultString.append("*** Overall status message:\r\n");
333 : // resultString.append(report_string_ + "\r\n");
334 : // resultString.append("*** Requested report response:\r\n");
335 : // }
336 :
337 : // pass the request to the BoardReaderCore instance, if it's available
338 0 : if (fragment_receiver_ptr_ != nullptr)
339 : {
340 0 : resultString.append(fragment_receiver_ptr_->report(which));
341 : }
342 : else
343 : {
344 0 : resultString.append("This BoardReader has not yet been initialized and ");
345 0 : resultString.append("therefore can not provide reporting.");
346 : }
347 :
348 0 : return resultString;
349 0 : }
|
