Line data Source code
1 : #include "TRACE/tracemf.h"
2 : #include "artdaq/DAQdata/Globals.hh" // include these 2 first -
3 : #define TRACE_NAME (app_name + "_DataReceiverCore").c_str()
4 :
5 : #include "artdaq/Application/DataReceiverCore.hh"
6 :
7 : #include "artdaq-core/Utilities/ExceptionHandler.hh"
8 :
9 : #include <memory>
10 : #include <string>
11 :
12 0 : artdaq::DataReceiverCore::DataReceiverCore()
13 0 : : stop_requested_(false)
14 0 : , pause_requested_(false)
15 0 : , run_is_paused_(false)
16 :
17 : {
18 0 : TLOG(TLVL_DEBUG + 32) << "Constructor";
19 0 : }
20 :
21 0 : artdaq::DataReceiverCore::~DataReceiverCore()
22 : {
23 0 : TLOG(TLVL_DEBUG + 32) << "Destructor";
24 0 : }
25 :
26 0 : bool artdaq::DataReceiverCore::initializeDataReceiver(fhicl::ParameterSet const& pset, fhicl::ParameterSet const& data_pset, fhicl::ParameterSet const& metric_pset)
27 : {
28 : // other parameters
29 0 : verbose_ = data_pset.get<bool>("verbose", true);
30 :
31 : // TRACE here so that mftrace_module and mftrace_iteration are ready by mftrace...should set it for all subsequent traces
32 0 : TLOG(TLVL_INFO) << "Initializing Data Receiver";
33 :
34 0 : if (metric_pset.is_empty())
35 : {
36 0 : TLOG(TLVL_INFO) << "No metric plugins appear to be defined";
37 : }
38 : try
39 : {
40 0 : metricMan->initialize(metric_pset, app_name);
41 : }
42 0 : catch (...)
43 : {
44 0 : ExceptionHandler(ExceptionHandlerRethrow::no,
45 : "Error loading metrics in DataReceiverCore::initialize()");
46 0 : }
47 :
48 0 : fhicl::ParameterSet art_pset = pset;
49 0 : if (art_pset.has_key("art"))
50 : {
51 0 : art_pset = art_pset.get<fhicl::ParameterSet>("art");
52 : }
53 : else
54 : {
55 0 : art_pset.erase("daq");
56 : }
57 :
58 : // Add the "metrics" block
59 0 : auto art_services_pset = art_pset.get<fhicl::ParameterSet>("services");
60 0 : auto art_services_ArtdaqSharedMemoryServiceInterface_pset = art_services_pset.get<fhicl::ParameterSet>("ArtdaqSharedMemoryServiceInterface");
61 0 : if (!art_services_ArtdaqSharedMemoryServiceInterface_pset.has_key("metrics"))
62 : {
63 0 : art_services_ArtdaqSharedMemoryServiceInterface_pset.put<fhicl::ParameterSet>("metrics", metric_pset);
64 0 : art_services_pset.erase("ArtdaqSharedMemoryServiceInterface");
65 0 : art_services_pset.put<fhicl::ParameterSet>("ArtdaqSharedMemoryServiceInterface", art_services_ArtdaqSharedMemoryServiceInterface_pset);
66 0 : art_pset.erase("services");
67 0 : art_pset.put<fhicl::ParameterSet>("services", art_services_pset);
68 : }
69 : else
70 : {
71 0 : TLOG(TLVL_INFO) << "Metrics block already exists in ArtdaqSharedMemoryServiceInterface, not adding metrics from DAQ ParameterSet";
72 : }
73 :
74 0 : fhicl::ParameterSet data_tmp = data_pset;
75 0 : if (data_pset.has_key("expected_events_per_bunch"))
76 : {
77 0 : data_tmp.put<int>("expected_fragments_per_event", data_pset.get<int>("expected_events_per_bunch"));
78 : }
79 :
80 0 : if (data_pset.has_key("rank"))
81 : {
82 0 : if (my_rank >= 0 && data_pset.get<int>("rank") != my_rank)
83 : {
84 0 : TLOG(TLVL_WARNING) << "Rank specified at startup is different than rank specified at configure! Using rank received at configure!";
85 : }
86 0 : my_rank = data_pset.get<int>("rank");
87 : }
88 0 : if (my_rank == -1)
89 : {
90 0 : TLOG(TLVL_ERROR) << "Rank not specified at startup or in configuration! Aborting";
91 0 : exit(1);
92 : }
93 :
94 0 : event_store_ptr_ = std::make_shared<SharedMemoryEventManager>(data_tmp, art_pset);
95 0 : art_pset_ = art_pset;
96 0 : TLOG(TLVL_DEBUG + 32) << "Resulting art_pset_: \"" << art_pset_.to_string() << "\".";
97 :
98 0 : receiver_ptr_ = std::make_unique<artdaq::DataReceiverManager>(data_tmp, event_store_ptr_);
99 :
100 0 : return true;
101 0 : }
102 :
103 0 : bool artdaq::DataReceiverCore::start(art::RunID id)
104 : {
105 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Starting run " << id.run();
106 :
107 : // 13-Jul-2018, KAB: added code to update the art_pset inside the event store
108 : // with configuration archive information
109 : // so that the config info will be stored in the output art/ROOT file.
110 : // (Note that we don't bother looping over the config_archive_entries if that
111 : // map is empty, but we *do* still update the art configuration with art_pset_
112 : // at each begin-run because the config archive may be non-empty one time through
113 : // and then empty the next time.)
114 0 : fhicl::ParameterSet temp_pset = art_pset_;
115 0 : if (!config_archive_entries_.empty())
116 : {
117 0 : fhicl::ParameterSet config_pset;
118 0 : for (auto& entry : config_archive_entries_)
119 : {
120 0 : config_pset.put(entry.first, entry.second);
121 : }
122 0 : temp_pset.put_or_replace("configuration_documents", config_pset);
123 0 : }
124 0 : event_store_ptr_->UpdateArtConfiguration(temp_pset);
125 :
126 0 : stop_requested_.store(false);
127 0 : pause_requested_.store(false);
128 0 : run_is_paused_.store(false);
129 0 : metricMan->do_start();
130 0 : auto res = event_store_ptr_->startRun(id.run());
131 0 : if (!res) { return false; }
132 0 : receiver_ptr_->start_threads();
133 :
134 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Completed the Start transition for run " << event_store_ptr_->runID();
135 0 : return true;
136 0 : }
137 :
138 0 : bool artdaq::DataReceiverCore::stop()
139 : {
140 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Stopping run " << event_store_ptr_->runID();
141 0 : bool endSucceeded = false;
142 : int attemptsToEnd;
143 0 : receiver_ptr_->stop_threads();
144 :
145 : // 21-Jun-2013, KAB - the stop_requested_ variable must be set
146 : // before the flush lock so that the processFragments loop will
147 : // exit (after the timeout), the lock will be released (in the
148 : // processFragments method), and this method can continue.
149 0 : stop_requested_.store(true);
150 :
151 0 : TLOG(TLVL_DEBUG + 32) << "Ending run " << event_store_ptr_->runID();
152 0 : attemptsToEnd = 1;
153 0 : endSucceeded = event_store_ptr_->endRun();
154 0 : while (!endSucceeded && attemptsToEnd < 3)
155 : {
156 0 : ++attemptsToEnd;
157 0 : TLOG(TLVL_DEBUG + 32) << "Retrying EventStore::endRun()";
158 0 : endSucceeded = event_store_ptr_->endRun();
159 : }
160 0 : if (!endSucceeded)
161 : {
162 0 : TLOG(TLVL_ERROR)
163 0 : << "EventStore::endRun in stop method failed after three tries.";
164 : }
165 0 : TLOG(TLVL_DEBUG + 32) << "Done Ending run " << event_store_ptr_->runID();
166 :
167 0 : attemptsToEnd = 1;
168 0 : TLOG(TLVL_DEBUG + 32) << "stop: Calling EventStore::endOfData";
169 0 : endSucceeded = event_store_ptr_->endOfData();
170 0 : while (!endSucceeded && attemptsToEnd < 3)
171 : {
172 0 : ++attemptsToEnd;
173 0 : TLOG(TLVL_DEBUG + 32) << "Retrying EventStore::endOfData()";
174 0 : endSucceeded = event_store_ptr_->endOfData();
175 : }
176 :
177 0 : run_is_paused_.store(false);
178 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Completed the Stop transition for run " << event_store_ptr_->runID();
179 0 : return true;
180 : }
181 :
182 0 : bool artdaq::DataReceiverCore::pause()
183 : {
184 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Pausing run " << event_store_ptr_->runID();
185 0 : pause_requested_.store(true);
186 0 : run_is_paused_.store(true);
187 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Completed the Pause transition for run " << event_store_ptr_->runID();
188 0 : return true;
189 : }
190 :
191 0 : bool artdaq::DataReceiverCore::resume()
192 : {
193 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Resuming run " << event_store_ptr_->runID();
194 0 : pause_requested_.store(false);
195 0 : metricMan->do_start();
196 0 : event_store_ptr_->rolloverSubrun(true);
197 0 : run_is_paused_.store(false);
198 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Completed the Resume transition for run " << event_store_ptr_->runID();
199 0 : return true;
200 : }
201 :
202 0 : bool artdaq::DataReceiverCore::shutdown()
203 : {
204 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Starting Shutdown transition";
205 :
206 : /* We don't care about flushing data here. The only way to transition to the
207 : shutdown state is from a state where there is no data taking. All we have
208 : to do is signal the art input module that we're done taking data so that
209 : it can wrap up whatever it needs to do. */
210 :
211 0 : TLOG(TLVL_DEBUG + 32) << "shutdown: Shutting down DataReceiverManager";
212 0 : receiver_ptr_.reset(nullptr);
213 :
214 0 : bool endSucceeded = false;
215 0 : int attemptsToEnd = 1;
216 0 : TLOG(TLVL_DEBUG + 32) << "shutdown: Calling EventStore::endOfData";
217 0 : endSucceeded = event_store_ptr_->endOfData();
218 0 : while (!endSucceeded && attemptsToEnd < 3)
219 : {
220 0 : ++attemptsToEnd;
221 0 : TLOG(TLVL_DEBUG + 32) << "Retrying EventStore::endOfData()";
222 0 : endSucceeded = event_store_ptr_->endOfData();
223 : }
224 :
225 0 : TLOG(TLVL_DEBUG + 32) << "shutdown: Shutting down SharedMemoryEventManager";
226 0 : event_store_ptr_.reset();
227 :
228 0 : TLOG(TLVL_DEBUG + 32) << "shutdown: Shutting down MetricManager";
229 0 : metricMan->shutdown();
230 :
231 0 : TLOG(TLVL_DEBUG + 32) << "shutdown: Complete";
232 0 : TLOG((verbose_ ? TLVL_INFO : TLVL_DEBUG + 32)) << "Completed Shutdown transition";
233 0 : return endSucceeded;
234 : }
235 :
236 0 : bool artdaq::DataReceiverCore::soft_initialize(fhicl::ParameterSet const& pset)
237 : {
238 0 : TLOG(TLVL_DEBUG + 32) << "soft_initialize method called with DAQ "
239 0 : << "ParameterSet = \"" << pset.to_string()
240 0 : << "\".";
241 0 : return true;
242 : }
243 :
244 0 : bool artdaq::DataReceiverCore::reinitialize(fhicl::ParameterSet const& pset)
245 : {
246 0 : TLOG(TLVL_DEBUG + 32) << "reinitialize method called with DAQ "
247 0 : << "ParameterSet = \"" << pset.to_string()
248 0 : << "\".";
249 0 : event_store_ptr_ = nullptr;
250 0 : return initialize(pset);
251 : }
252 :
253 0 : bool artdaq::DataReceiverCore::rollover_subrun(uint64_t boundary, uint32_t subrun)
254 : {
255 0 : if (event_store_ptr_)
256 : {
257 0 : event_store_ptr_->rolloverSubrun(boundary, subrun, true);
258 0 : return true;
259 : }
260 0 : return false;
261 : }
262 :
263 0 : std::string artdaq::DataReceiverCore::report(std::string const& which) const
264 : {
265 0 : if (which == "open_event_count")
266 : {
267 0 : if (event_store_ptr_ != nullptr)
268 : {
269 0 : return std::to_string(event_store_ptr_->GetOpenEventCount());
270 : }
271 :
272 0 : return "-1";
273 : }
274 0 : if (which == "event_count")
275 : {
276 0 : if (receiver_ptr_ != nullptr)
277 : {
278 0 : return std::to_string(receiver_ptr_->count());
279 : }
280 :
281 0 : return "0";
282 : }
283 0 : else if (which == "stats")
284 : {
285 0 : if (event_store_ptr_ != nullptr)
286 0 : return event_store_ptr_->BuildStatisticsString();
287 : else
288 0 : return "-1";
289 : }
290 :
291 : // lots of cool stuff that we can do here
292 : // - report on the number of fragments received and the number
293 : // of events built (in the current or previous run
294 : // - report on the number of incomplete events in the EventStore
295 : // (if running)
296 0 : std::string tmpString;
297 0 : if (event_store_ptr_ != nullptr)
298 : {
299 0 : tmpString.append(app_name + " run number = " + std::to_string(event_store_ptr_->runID()) + ".\n");
300 : }
301 0 : tmpString.append("Command \"" + which + "\" is not currently supported.");
302 0 : return tmpString;
303 0 : }
|
