Line data Source code
1 : #include "TRACE/tracemf.h"
2 :
3 : #include "artdaq/ArtModules/ArtdaqSharedMemoryServiceInterface.h"
4 : #include "artdaq/DAQdata/Globals.hh"
5 :
6 : #include "artdaq-core/Core/SharedMemoryEventReceiver.hh"
7 : #include "artdaq-core/Utilities/ExceptionHandler.hh"
8 :
9 : #include "art/Framework/Services/Registry/ServiceDefinitionMacros.h"
10 : #include "art/Framework/Services/Registry/ServiceHandle.h"
11 : #include "fhiclcpp/types/Atom.h"
12 : #include "fhiclcpp/types/Comment.h"
13 : #include "fhiclcpp/types/ConfigurationTable.h"
14 : #include "fhiclcpp/types/Name.h"
15 :
16 : #include <cstdint>
17 : #include <memory>
18 :
19 : #define TRACE_NAME "ArtdaqSharedMemoryService"
20 : // ----------------------------------------------------------------------
21 :
22 : /**
23 : * \brief ArtdaqSharedMemoryService extends ArtdaqSharedMemoryServiceInterface.
24 : * It receives events from shared memory using SharedMemoryEventReceiver. It also manages the artdaq Global varaibles my_rank and app_name.
25 : * Users should retrieve a ServiceHandle to this class before using artdaq Globals to ensure the correct values are used.
26 : */
27 : class ArtdaqSharedMemoryService : public ArtdaqSharedMemoryServiceInterface
28 : {
29 : public:
30 : /// <summary>
31 : /// Allowed Configuration parameters of NetMonTransportService. May be used for configuration validation
32 : /// </summary>
33 : struct Config
34 : {
35 : /// "shared_memory_key" (Default: 0xBEE70000 + pid): Key to use when connecting to shared memory. Will default to 0xBEE70000 + getppid().
36 : fhicl::Atom<uint32_t> shared_memory_key{fhicl::Name{"shared_memory_key"}, fhicl::Comment{"Key to use when connecting to shared memory. Will default to 0xBEE70000 + getppid()."}, 0xBEE70000};
37 : /// "shared_memory_key" (Default: 0xCEE70000 + pid): Key to use when connecting to broadcast shared memory. Will default to 0xCEE70000 + getppid().
38 : fhicl::Atom<uint32_t> broadcast_shared_memory_key{fhicl::Name{"broadcast_shared_memory_key"}, fhicl::Comment{"Key to use when connecting to broadcast shared memory. Will default to 0xCEE70000 + getppid()."}, 0xCEE70000};
39 : /// "rank" (OPTIONAL) : The rank of this applicaiton, for use by non - artdaq applications running NetMonTransportService
40 : fhicl::Atom<int> rank{fhicl::Name{"rank"}, fhicl::Comment{"Rank of this artdaq application. Used for data transfers"}};
41 : /// "subrun_closure_threshold" (Default: 5) Minimum number of events in event ordering list before releasing a subrun/run change event. -1 to always require safety valve timeout to release subrun/run change events. 0 to always release subrun/run change events immediately.
42 : fhicl::Atom<int> subrun_closure_threshold{fhicl::Name{"subrun_closure_threshold"}, fhicl::Comment{"Minimum number of events in event ordering list before releasing a subrun/run change event"}, 1};
43 : /// "safety_valve_timeout_s" (Default: 10.0): Maximum time (in s) to wait before releasing the front of the event ordering list
44 : fhicl::Atom<double> safety_valve_timeout_s{fhicl::Name{"safety_valve_timeout_s"}, fhicl::Comment{"Maximum time (in s) to wait before releasing the front of the event ordering list"}, 10.0};
45 : };
46 : /// Used for ParameterSet validation (if desired)
47 : using Parameters = fhicl::WrappedTable<Config>;
48 :
49 : /**
50 : * \brief NetMonTransportService Destructor. Calls disconnect().
51 : */
52 : virtual ~ArtdaqSharedMemoryService();
53 :
54 : /**
55 : * \brief NetMonTransportService Constructor
56 : * \param pset ParameterSet used to configure NetMonTransportService and DataSenderManager. See NetMonTransportService::Config
57 : */
58 : ArtdaqSharedMemoryService(fhicl::ParameterSet const& pset, art::ActivityRegistry&);
59 :
60 : /**
61 : * \brief Receive event(s) from the shared memory
62 : * \param broadcast Whether to only attempt to receive a broadcast (broadcasts are always preferentially received over data)
63 : * \return Map of Fragment types retrieved from shared memory
64 : */
65 : std::shared_ptr<ArtdaqEvent> ReceiveEvent(bool broadcast) override;
66 :
67 : /**
68 : * \brief Get the number of events which are ready to be read
69 : * \return The number of events which can be read
70 : */
71 0 : size_t GetQueueSize() override { return incoming_events_->ReadReadyCount(); }
72 :
73 : /**
74 : * \brief Get the maximum number of events which can be stored in the shared memory
75 : * \return The maximum number of events which can be stored in the shared memory
76 : */
77 0 : size_t GetQueueCapacity() override { return incoming_events_->size(); }
78 :
79 : /**
80 : * \brief Get the ID of this art process
81 : * \return The ID of this art process from the shared memory segment
82 : */
83 0 : size_t GetMyId() override { return incoming_events_->GetMyId(); }
84 :
85 : private:
86 : ArtdaqSharedMemoryService(ArtdaqSharedMemoryService const&) = delete;
87 : ArtdaqSharedMemoryService(ArtdaqSharedMemoryService&&) = delete;
88 : ArtdaqSharedMemoryService& operator=(ArtdaqSharedMemoryService const&) = delete;
89 : ArtdaqSharedMemoryService& operator=(ArtdaqSharedMemoryService&&) = delete;
90 :
91 : std::shared_ptr<ArtdaqEvent> ReadEventFromSharedMemory(bool broadcast);
92 :
93 : private:
94 : std::unique_ptr<artdaq::SharedMemoryEventReceiver> incoming_events_;
95 : std::list<std::shared_ptr<ArtdaqEvent>> event_ordering_;
96 : std::set<artdaq::Fragment::sequence_id_t> released_broadcast_sequence_ids_;
97 : size_t read_timeout_;
98 : int subrun_closure_threshold_{1};
99 : double safety_valve_timeout_s_{10.0};
100 : bool last_read_timeout_{false};
101 : bool resume_after_timeout_;
102 : bool printed_exit_message_{false};
103 : bool end_of_data_received_{false};
104 : bool subrun_has_events_{false};
105 : uint32_t current_subrun_{0};
106 : };
107 :
108 0 : DECLARE_ART_SERVICE_INTERFACE_IMPL(ArtdaqSharedMemoryService, ArtdaqSharedMemoryServiceInterface, LEGACY)
109 :
110 : static fhicl::ParameterSet empty_pset;
111 :
112 : // clang-format off
113 : #define TLVL_CONSTRUCTOR TLVL_DEBUG + 32
114 : #define TLVL_READEVENT TLVL_DEBUG + 33
115 : #define TLVL_READEVENT_2 TLVL_DEBUG + 34
116 : #define TLVL_READEVENT_3 TLVL_DEBUG + 35
117 : #define TLVL_RECEIVEEVENT TLVL_DEBUG + 36
118 : #define TLVL_RECEIVEEVENT_2 TLVL_DEBUG + 37
119 : #define TLVL_RECEIVEEVENT_3 TLVL_DEBUG + 38
120 : #define TLVL_RECEIVEEVENT_4 TLVL_DEBUG + 39
121 : // clang-format on
122 :
123 0 : ArtdaqSharedMemoryService::ArtdaqSharedMemoryService(fhicl::ParameterSet const& pset, art::ActivityRegistry& /*unused*/)
124 0 : : incoming_events_(nullptr)
125 0 : , event_ordering_()
126 0 : , read_timeout_(pset.get<size_t>("read_timeout_us", static_cast<size_t>(pset.get<double>("waiting_time", 600.0) * 1000000)))
127 0 : , subrun_closure_threshold_(pset.get<int>("subrun_closure_threshold", artdaq::SharedMemoryManager::GetCatchUpFactor() + 2)) // +2, one for ESRF itself, one for extra padding to ensure that no catch-up is being performed
128 0 : , safety_valve_timeout_s_(pset.get<double>("safety_valve_timeout_s", 10.0))
129 0 : , resume_after_timeout_(pset.get<bool>("resume_after_timeout", true))
130 : {
131 0 : TLOG(TLVL_CONSTRUCTOR) << "ArtdaqSharedMemoryService CONSTRUCTOR";
132 :
133 0 : incoming_events_ = std::make_unique<artdaq::SharedMemoryEventReceiver>(
134 0 : pset.get<int>("shared_memory_key", artdaq::Globals::SharedMemoryKey(0xEE000000, true)),
135 0 : pset.get<int>("broadcast_shared_memory_key", artdaq::Globals::SharedMemoryKey(0xBB000000, true)));
136 :
137 0 : char const* artapp_env = getenv("ARTDAQ_APPLICATION_NAME");
138 0 : std::string artapp_str;
139 0 : if (artapp_env != nullptr)
140 : {
141 0 : artapp_str = std::string(artapp_env) + "_";
142 : }
143 :
144 0 : TLOG(TLVL_CONSTRUCTOR) << "Setting app_name";
145 0 : artdaq::Globals::my_art_id_ = incoming_events_->GetMyId();
146 0 : app_name = artapp_str + "art" + std::to_string(artdaq::Globals::my_art_id_);
147 : // artdaq::configureMessageFacility(app_name.c_str()); // ELF 11/20/2020: MessageFacility already configured by initialization pset
148 :
149 0 : artapp_env = getenv("ARTDAQ_RANK");
150 0 : if (artapp_env != nullptr && my_rank < 0)
151 : {
152 0 : TLOG(TLVL_CONSTRUCTOR) << "Setting rank from envrionment";
153 0 : my_rank = strtol(artapp_env, nullptr, 10);
154 0 : }
155 : else
156 : {
157 0 : TLOG(TLVL_CONSTRUCTOR) << "Setting my_rank from shared memory";
158 0 : my_rank = incoming_events_->GetRank();
159 : }
160 :
161 : try
162 : {
163 0 : if (metricMan)
164 : {
165 0 : metricMan->initialize(pset.get<fhicl::ParameterSet>("metrics", fhicl::ParameterSet()), app_name);
166 0 : metricMan->do_start();
167 : }
168 : }
169 0 : catch (...)
170 : {
171 0 : artdaq::ExceptionHandler(artdaq::ExceptionHandlerRethrow::no, "Error loading metrics in ArtdaqSharedMemoryService()");
172 0 : }
173 :
174 0 : TLOG(TLVL_INFO) << "app_name is " << app_name << ", rank " << my_rank;
175 0 : }
176 :
177 0 : ArtdaqSharedMemoryService::~ArtdaqSharedMemoryService()
178 : {
179 0 : artdaq::Globals::CleanUpGlobals();
180 0 : }
181 :
182 0 : std::shared_ptr<ArtdaqEvent> ArtdaqSharedMemoryService::ReadEventFromSharedMemory(bool broadcast)
183 : {
184 0 : TLOG(TLVL_READEVENT) << "ReadEventFromSharedMemory BEGIN";
185 0 : std::shared_ptr<ArtdaqEvent> output_event;
186 :
187 0 : while (output_event == nullptr)
188 : {
189 0 : TLOG(TLVL_READEVENT_2) << "ReadEventFromSharedMemory: Waiting for available buffer";
190 0 : bool got_event = false;
191 0 : auto start_time = std::chrono::steady_clock::now();
192 0 : auto read_timeout_to_use = read_timeout_ > 100000 ? 100000 : read_timeout_;
193 0 : if (!resume_after_timeout_ || broadcast) read_timeout_to_use = read_timeout_;
194 0 : while (!incoming_events_->IsEndOfData() && !got_event)
195 : {
196 0 : got_event = incoming_events_->ReadyForRead(broadcast, read_timeout_to_use);
197 0 : if (!got_event && (!resume_after_timeout_ || broadcast)) // Only try broadcasts once!
198 : {
199 0 : TLOG(TLVL_ERROR) << "Timeout occurred! No data received after " << read_timeout_to_use << " us. Returning empty Fragment list!";
200 0 : last_read_timeout_ = true;
201 0 : return nullptr;
202 : }
203 0 : if (!got_event && artdaq::TimeUtils::GetElapsedTimeMicroseconds(start_time) > read_timeout_)
204 : {
205 0 : TLOG(TLVL_READEVENT_2) << "Timeout occurred! No data received after " << artdaq::TimeUtils::GetElapsedTimeMicroseconds(start_time) << " us. Retrying.";
206 0 : last_read_timeout_ = true;
207 : }
208 : }
209 :
210 0 : if (incoming_events_->IsEndOfData())
211 : {
212 0 : if (!printed_exit_message_)
213 : {
214 0 : TLOG(TLVL_INFO) << "End of Data signal received, exiting";
215 0 : printed_exit_message_ = true;
216 : }
217 0 : return nullptr;
218 : }
219 :
220 0 : TLOG(TLVL_READEVENT) << "ReadEventFromSharedMemory: Reading buffer header";
221 0 : last_read_timeout_ = false;
222 0 : output_event = std::make_shared<ArtdaqEvent>();
223 0 : auto errflag = false;
224 0 : auto hdrPtr = incoming_events_->ReadHeader(errflag);
225 0 : if (errflag || hdrPtr == nullptr)
226 : { // Buffer was changed out from under reader!
227 0 : incoming_events_->ReleaseBuffer();
228 0 : continue; // retry
229 : }
230 0 : output_event->header = std::make_shared<artdaq::detail::RawEventHeader>(*hdrPtr);
231 0 : TLOG(TLVL_READEVENT) << "ReadEventFromSharedMemory: Getting Fragment types";
232 0 : auto fragmentTypes = incoming_events_->GetFragmentTypes(errflag);
233 0 : if (errflag)
234 : { // Buffer was changed out from under reader!
235 0 : incoming_events_->ReleaseBuffer();
236 0 : continue; // retry
237 : }
238 0 : if (fragmentTypes.empty())
239 : {
240 0 : TLOG(TLVL_ERROR) << "Event has no Fragments! Aborting!";
241 0 : incoming_events_->ReleaseBuffer();
242 0 : return nullptr;
243 : }
244 :
245 0 : for (auto const& type : fragmentTypes)
246 : {
247 0 : TLOG(TLVL_READEVENT_3) << "ReadEventFromSharedMemory: Getting all Fragments of type " << static_cast<int>(type);
248 0 : output_event->fragments[type] = incoming_events_->GetFragmentsByType(errflag, type);
249 0 : if (!output_event->fragments[type])
250 : {
251 0 : TLOG(TLVL_WARNING) << "Error retrieving Fragments from shared memory! (Most likely due to a buffer overwrite) Retrying...";
252 0 : incoming_events_->ReleaseBuffer();
253 0 : output_event->fragments.clear();
254 0 : continue;
255 0 : }
256 : // Events coming out of the EventStore are not sorted but need to be sorted by sequence ID before they can be passed to art.
257 0 : std::sort(output_event->fragments[type]->begin(), output_event->fragments[type]->end(), artdaq::fragmentSequenceIDCompare);
258 : }
259 0 : TLOG(TLVL_READEVENT) << "ReadEventFromSharedMemory: Releasing buffer";
260 0 : incoming_events_->ReleaseBuffer();
261 0 : }
262 :
263 0 : TLOG(TLVL_READEVENT) << "ReadEventFromSharedMemory END";
264 0 : return output_event;
265 0 : }
266 :
267 0 : std::shared_ptr<ArtdaqEvent> ArtdaqSharedMemoryService::ReceiveEvent(bool broadcast)
268 : {
269 0 : TLOG(TLVL_RECEIVEEVENT) << "ReceiveEvent BEGIN";
270 0 : std::shared_ptr<ArtdaqEvent> output_event;
271 0 : auto start_time = std::chrono::steady_clock::now();
272 :
273 0 : while (output_event == nullptr)
274 : {
275 : // If we experienced a timeout, or have an EndOfData event, drain any held Start/End Run/SubRun events
276 0 : if (last_read_timeout_ || end_of_data_received_)
277 : {
278 0 : if (event_ordering_.size() > 0)
279 : {
280 0 : output_event = event_ordering_.front();
281 0 : event_ordering_.pop_front();
282 0 : break; // while(output_event == nullptr)
283 : }
284 : // Don't try to get more data if we have an EndOfData event
285 0 : if (end_of_data_received_) { break; }
286 : }
287 :
288 0 : if (event_ordering_.size() > 0)
289 : {
290 0 : auto first_type = event_ordering_.front()->FirstFragmentType();
291 0 : auto first_sr = event_ordering_.front()->header->subrun_id;
292 : // If there is an Init Fragment, return it
293 0 : if (first_type == artdaq::Fragment::InitFragmentType)
294 : {
295 0 : TLOG(TLVL_RECEIVEEVENT) << "Returning Init Fragment";
296 0 : output_event = event_ordering_.front();
297 0 : event_ordering_.pop_front();
298 0 : break; // while(output_event == nullptr)
299 : }
300 0 : if (current_subrun_ != 0 && first_sr <= current_subrun_)
301 : {
302 0 : if (artdaq::Fragment::isUserFragmentType(first_type) || first_type == artdaq::Fragment::ContainerFragmentType || first_type == artdaq::Fragment::EmptyFragmentType || first_type == artdaq::Fragment::DataFragmentType)
303 : {
304 0 : TLOG(TLVL_RECEIVEEVENT) << "Returning Fragment due to subrun match";
305 0 : output_event = event_ordering_.front();
306 0 : event_ordering_.pop_front();
307 0 : break; // while(output_event == nullptr)
308 : }
309 0 : else if (subrun_closure_threshold_ >= 0 && event_ordering_.size() > static_cast<size_t>(subrun_closure_threshold_))
310 : {
311 : // First Fragment is broadcast (begin/end run/subrun), but there's more in event ordering!
312 0 : TLOG(TLVL_RECEIVEEVENT) << "Returning Broadcast Fragment due to subrun closure";
313 :
314 0 : if (released_broadcast_sequence_ids_.count(event_ordering_.front()->header->sequence_id) == 0)
315 : {
316 0 : output_event = event_ordering_.front();
317 0 : released_broadcast_sequence_ids_.insert(event_ordering_.front()->header->sequence_id);
318 0 : while (released_broadcast_sequence_ids_.size() > 1000) { released_broadcast_sequence_ids_.erase(released_broadcast_sequence_ids_.begin()); }
319 : }
320 : else
321 : {
322 0 : TLOG(TLVL_WARNING) << "Discarding duplicate Broadcast with sequence ID " << event_ordering_.front()->header->sequence_id << " and count " << event_ordering_.front()->fragments.size();
323 : }
324 0 : event_ordering_.pop_front();
325 0 : continue; // while(output_event == nullptr)
326 0 : }
327 0 : }
328 0 : else if (current_subrun_ != 0 && first_sr > current_subrun_ + 1)
329 : {
330 0 : TLOG(TLVL_RECEIVEEVENT) << "Returning Fragment due to stale current_subrun_ (first_sr=" << first_sr << ", current_subrun_=" << current_subrun_ << ")";
331 0 : output_event = event_ordering_.front();
332 0 : event_ordering_.pop_front();
333 0 : break; // while(output_event == nullptr)
334 : }
335 :
336 0 : if (event_ordering_.size() == 1 && (first_type == artdaq::Fragment::EndOfRunFragmentType || first_type == artdaq::Fragment::RunDataFragmentType))
337 : {
338 0 : TLOG(TLVL_RECEIVEEVENT) << "Returning Broadcast Fragment due to end-of-run";
339 0 : output_event = event_ordering_.front();
340 0 : event_ordering_.pop_front();
341 0 : break; // while(output_event == nullptr)
342 : }
343 :
344 0 : if (current_subrun_ == 0)
345 : {
346 0 : if (artdaq::Fragment::isUserFragmentType(first_type) || first_type == artdaq::Fragment::ContainerFragmentType || first_type == artdaq::Fragment::EmptyFragmentType || first_type == artdaq::Fragment::DataFragmentType)
347 : {
348 0 : TLOG(TLVL_RECEIVEEVENT) << "Returning Fragment due to unset subrun";
349 0 : output_event = event_ordering_.front();
350 0 : event_ordering_.pop_front();
351 0 : break; // while(output_event == nullptr)
352 : }
353 : // We cannot close a subrun that has not yet been opened
354 0 : if (first_type == artdaq::Fragment::EndOfSubrunFragmentType || first_type == artdaq::Fragment::SubrunDataFragmentType)
355 : {
356 0 : TLOG(TLVL_WARNING) << "Subrun is unset, discarding EndOfSubrun Fragment(s) for subrun " << first_sr;
357 0 : event_ordering_.pop_front();
358 : }
359 : // Likewise, we cannot close a run that is not open
360 0 : if (first_type == artdaq::Fragment::EndOfRunFragmentType || first_type == artdaq::Fragment::RunDataFragmentType)
361 : {
362 0 : TLOG(TLVL_WARNING) << "Subrun is unset, discarding EndOfRun Fragment(s) for run " << event_ordering_.front()->header->run_id;
363 0 : event_ordering_.pop_front();
364 : }
365 : }
366 :
367 0 : if (artdaq::TimeUtils::GetElapsedTime(start_time) > safety_valve_timeout_s_)
368 : {
369 0 : TLOG(subrun_closure_threshold_ < 0 ? TLVL_INFO : TLVL_WARNING) << "Returning Fragment due to safety valve timeout (" << safety_valve_timeout_s_ << " s). event_ordering_ size=" << event_ordering_.size()
370 0 : << " (th=" << subrun_closure_threshold_ << "), first event type=" << static_cast<int>(first_type) << " sr="
371 0 : << first_sr << " (c=" << current_subrun_ << ")";
372 0 : output_event = event_ordering_.front();
373 0 : event_ordering_.pop_front();
374 0 : break; // while(output_event == nullptr)
375 : }
376 : }
377 :
378 0 : auto next_event = ReadEventFromSharedMemory(broadcast);
379 0 : if (next_event == nullptr)
380 : {
381 0 : if (event_ordering_.size() > 0)
382 : {
383 0 : output_event = event_ordering_.front();
384 0 : event_ordering_.pop_front();
385 : }
386 : else
387 : {
388 : // Will return nullptr
389 0 : break; // while(output_event == nullptr)
390 : }
391 : }
392 : else
393 : {
394 : // Reset start time when new event arrives
395 0 : start_time = std::chrono::steady_clock::now();
396 0 : TLOG(TLVL_RECEIVEEVENT_2) << "Adding ArtdaqEvent with run=" << next_event->header->run_id << ", subrun=" << next_event->header->subrun_id << ", seq=" << next_event->header->sequence_id << ", and type " << static_cast<int>(next_event->FirstFragmentType()) << " to event ordering list";
397 0 : if (next_event->FirstFragmentType() == artdaq::Fragment::EndOfDataFragmentType) { end_of_data_received_ = true; }
398 0 : else if (next_event->header->subrun_id < current_subrun_ && next_event->FirstFragmentType() != artdaq::Fragment::RunDataFragmentType && next_event->FirstFragmentType() == artdaq::Fragment::EndOfRunFragmentType)
399 : {
400 0 : auto seq_mask = 0xFFFFFFFF & next_event->header->sequence_id;
401 0 : TLOG(TLVL_WARNING) << "ArtdaqEvent with run = " << next_event->header->run_id << ", subrun = " << next_event->header->subrun_id << ", seq = " << next_event->header->sequence_id << " (32b mask " << seq_mask << "), and type " << static_cast<int>(next_event->FirstFragmentType()) << " is from a previous subrun! (current=" << current_subrun_ << ")";
402 : }
403 0 : event_ordering_.push_back(next_event);
404 0 : if (event_ordering_.size() > 1)
405 : {
406 0 : TLOG(TLVL_RECEIVEEVENT_2) << "event_ordering_ size is now " << event_ordering_.size();
407 : }
408 0 : event_ordering_.sort();
409 : }
410 0 : }
411 :
412 0 : if (output_event != nullptr)
413 : {
414 0 : auto type = output_event->FirstFragmentType();
415 0 : TLOG(TLVL_RECEIVEEVENT_3) << "Returning ArtdaqEvent with run=" << output_event->header->run_id << ", subrun=" << output_event->header->subrun_id
416 0 : << ", seq=" << output_event->header->sequence_id << ", and type " << static_cast<int>(type);
417 0 : if (output_event->header->subrun_id > current_subrun_ && output_event->header->subrun_id != 65535) // EndOfRun Fragments have subrun -1
418 : {
419 0 : if (current_subrun_ != 0)
420 : {
421 0 : TLOG(TLVL_WARNING) << "Event subrun " << output_event->header->subrun_id << " is greater than current_subrun_ (" << current_subrun_ << "), incrementing";
422 : }
423 : else
424 : {
425 0 : TLOG(TLVL_DEBUG) << "Incrementing current_subrun_ from 0 to " << output_event->header->subrun_id << " due to unset subrun";
426 : }
427 0 : current_subrun_ = output_event->header->subrun_id;
428 : }
429 0 : if (type == artdaq::Fragment::EndOfSubrunFragmentType || type == artdaq::Fragment::SubrunDataFragmentType || type == artdaq::Fragment::EndOfRunFragmentType || type == artdaq::Fragment::RunDataFragmentType || type == artdaq::Fragment::InitFragmentType)
430 : {
431 0 : if (type == artdaq::Fragment::EndOfSubrunFragmentType || type == artdaq::Fragment::SubrunDataFragmentType)
432 : {
433 0 : TLOG(TLVL_RECEIVEEVENT_4) << "EndOfSubrun or SubrunData Fragment recieved, incrementing current_subrun from " << current_subrun_ << " to " << (current_subrun_ + 1);
434 0 : current_subrun_++;
435 0 : subrun_has_events_ = false;
436 0 : }
437 : else
438 : {
439 0 : TLOG(TLVL_DEBUG) << "Due to run/subrun/control conditions, setting current_subrun to 0";
440 0 : current_subrun_ = 0;
441 0 : subrun_has_events_ = false;
442 : }
443 0 : }
444 : else
445 : {
446 0 : subrun_has_events_ = true;
447 : }
448 : }
449 :
450 0 : TLOG(TLVL_RECEIVEEVENT) << "ReceiveEvent END";
451 :
452 0 : return output_event;
453 0 : }
454 :
455 0 : DEFINE_ART_SERVICE_INTERFACE_IMPL(ArtdaqSharedMemoryService, ArtdaqSharedMemoryServiceInterface)
|
