Line data Source code
1 :
2 : #include "artdaq/DAQrate/SharedMemoryEventManager.hh"
3 : #include <sys/wait.h>
4 :
5 : #include <poll.h>
6 : #include <memory>
7 : #include <numeric>
8 :
9 : #include "artdaq-core/Core/StatisticsCollection.hh"
10 : #include "artdaq-core/Data/MetadataFragment.hh"
11 : #include "artdaq-core/Utilities/TraceLock.hh"
12 :
13 : #define TRACE_NAME (app_name + "_SharedMemoryEventManager").c_str()
14 :
15 : // clang-format off
16 : #define TLVL_ADDFRAGMENT 32
17 : #define TLVL_ADDINITFRAGMENT 33
18 : #define TLVL_BROADCASTFRAGMENT 34
19 : #define TLVL_BROADCASTFRAGMENTS 35
20 : #define TLVL_BROADCASTFRAGMENTS_2 36
21 : #define TLVL_CHECKPENDINGBROADCASTS 37
22 : #define TLVL_CHECKPENDINGBUFFERS 38
23 : #define TLVL_CHECKPENDINGBUFFERS_2 39
24 : #define TLVL_CHECKPENDINGBUFFERS_3 40
25 : #define TLVL_CHECKPENDINGBUFFERS_4 41
26 : #define TLVL_COMPLETEBUFFER 42
27 : #define TLVL_CONSTRUCTOR 43
28 : #define TLVL_DESTRUCTOR 43
29 : #define TLVL_DONEWRITINGFRAGMENT 44
30 : #define TLVL_ENDOFDATA 45
31 : #define TLVL_ENDOFDATA_2 46
32 : #define TLVL_ENDRUN 47
33 : #define TLVL_GETBUFFER 48
34 : #define TLVL_GETFRAGMENTCOUNT 49
35 : #define TLVL_GETSUBRUN 50
36 : #define TLVL_GETEVENTID 51
37 : #define TLVL_PARSEARTCOMMANDLINE 52
38 : #define TLVL_RECONFIGUREART 53
39 : #define TLVL_RUNART 54
40 : #define TLVL_RUNART_2 55
41 : #define TLVL_SENDINIT 56
42 : #define TLVL_SENDMETRICS 57
43 : #define TLVL_SHUTDOWN 58
44 : #define TLVL_STARTRUN 58
45 : #define TLVL_UPDATEARTCONFIG 59
46 : #define TLVL_WRITEFRAGMENTHEADER 60
47 : #define TLVL_BUFFER 61
48 : #define TLVL_BUFLCK 62
49 : // clang-format on
50 :
51 : std::mutex artdaq::SharedMemoryEventManager::sequence_id_mutex_;
52 : std::mutex artdaq::SharedMemoryEventManager::subrun_event_map_mutex_;
53 : const std::string artdaq::SharedMemoryEventManager::
54 : FRAGMENTS_RECEIVED_STAT_KEY("SharedMemoryEventManagerFragmentsReceived");
55 : const std::string artdaq::SharedMemoryEventManager::
56 : EVENTS_RELEASED_STAT_KEY("SharedMemoryEventManagerEventsReleased");
57 :
58 15 : artdaq::SharedMemoryEventManager::SharedMemoryEventManager(const fhicl::ParameterSet& pset, fhicl::ParameterSet art_pset)
59 30 : : SharedMemoryManager(pset.get<uint32_t>("shared_memory_key", Globals::SharedMemoryKey(0xEE000000)),
60 15 : pset.get<size_t>("buffer_count"),
61 75 : pset.has_key("max_event_size_bytes") ? pset.get<size_t>("max_event_size_bytes") : pset.get<size_t>("expected_fragments_per_event") * pset.get<size_t>("max_fragment_size_bytes"),
62 30 : pset.get<size_t>("stale_buffer_timeout_usec", pset.get<size_t>("event_queue_wait_time", 5) * 1000000),
63 30 : !pset.get<bool>("broadcast_mode", false))
64 30 : , num_art_processes_(pset.get<size_t>("art_analyzer_count", 1))
65 30 : , num_fragments_per_event_(pset.get<size_t>("expected_fragments_per_event"))
66 15 : , queue_size_(pset.get<size_t>("buffer_count"))
67 15 : , run_id_(0)
68 15 : , subrun_id_(0)
69 30 : , max_subrun_event_map_length_(pset.get<size_t>("max_subrun_lookup_table_size", 100))
70 30 : , subrun_transition_hold_time_s_(pset.get<double>("subrun_transition_hold_time_s", 0.001))
71 30 : , max_event_list_length_(pset.get<size_t>("max_event_list_length", 100))
72 30 : , use_sequence_id_for_event_number_(pset.get<bool>("use_sequence_id_for_event_number", true))
73 30 : , reset_event_number_for_subruns_(pset.get<bool>("reset_event_number_for_subruns", false))
74 50 : , overwrite_mode_(!pset.get<bool>("use_art", true) || pset.get<bool>("overwrite_mode", false) || pset.get<bool>("broadcast_mode", false))
75 30 : , shared_memory_ordering_(pset.get<bool>("shared_memory_ordering", false))
76 60 : , init_fragment_count_(pset.get<size_t>("init_fragment_count", pset.get<bool>("send_init_fragments", true) ? 1 : 0))
77 15 : , running_(false)
78 15 : , buffer_writes_pending_()
79 60 : , open_event_report_interval_ms_(pset.get<int>("open_event_report_interval_ms", pset.get<int>("incomplete_event_report_interval_ms", -1)))
80 15 : , last_open_event_report_time_(std::chrono::steady_clock::now())
81 15 : , last_backpressure_report_time_(std::chrono::steady_clock::now())
82 15 : , last_fragment_header_write_time_(std::chrono::steady_clock::now())
83 45 : , event_timing_(pset.get<size_t>("buffer_count"))
84 30 : , broadcast_timeout_ms_(pset.get<int>("fragment_broadcast_timeout_ms", 3000))
85 15 : , run_event_count_(0)
86 15 : , run_incomplete_event_count_(0)
87 15 : , subrun_event_count_(0)
88 15 : , subrun_incomplete_event_count_(0)
89 15 : , oversize_fragment_count_(0)
90 30 : , maximum_oversize_fragment_count_(pset.get<int>("maximum_oversize_fragment_count", 1))
91 15 : , restart_art_(false)
92 30 : , always_restart_art_(pset.get<bool>("restart_crashed_art_processes", true))
93 30 : , manual_art_(pset.get<bool>("manual_art", false))
94 15 : , current_art_pset_(art_pset)
95 45 : , art_cmdline_(pset.get<std::string>("art_command_line", "art -c #CONFIG_FILE#"))
96 30 : , art_process_index_offset_(pset.get<size_t>("art_index_offset", 0))
97 30 : , minimum_art_lifetime_s_(pset.get<double>("minimum_art_lifetime_s", 2.0))
98 30 : , art_event_processing_time_us_(pset.get<size_t>("expected_art_event_processing_time_us", 1000000))
99 30 : , capture_art_stdout_(pset.get<bool>("capture_art_stdout", true))
100 30 : , capture_art_stderr_(pset.get<bool>("capture_art_stderr", true))
101 15 : , requests_(nullptr)
102 15 : , tokens_(nullptr)
103 15 : , data_pset_(pset)
104 60 : , broadcasts_(pset.get<uint32_t>("broadcast_shared_memory_key", Globals::SharedMemoryKey(0xBB000000)),
105 45 : pset.get<size_t>("broadcast_buffer_count", 10),
106 15 : pset.get<size_t>("broadcast_buffer_size", 0x100000),
107 345 : pset.get<int>("expected_art_event_processing_time_us", 100000) * pset.get<size_t>("buffer_count"), false)
108 : {
109 15 : subrun_event_map_[0] = 1;
110 15 : buffer_sequence_ids_.resize(size(), Fragment::InvalidSequenceID);
111 15 : SetMinWriteSize(sizeof(detail::RawEventHeader) + sizeof(detail::RawFragmentHeader));
112 15 : broadcasts_.SetMinWriteSize(sizeof(detail::RawEventHeader) + sizeof(detail::RawFragmentHeader));
113 :
114 15 : RegisterWriter();
115 15 : broadcasts_.RegisterWriter();
116 :
117 45 : if (!pset.get<bool>("use_art", true))
118 : {
119 30 : TLOG(TLVL_INFO) << "BEGIN SharedMemoryEventManager CONSTRUCTOR with use_art:false";
120 10 : num_art_processes_ = 0;
121 : }
122 : else
123 : {
124 15 : TLOG(TLVL_INFO) << "BEGIN SharedMemoryEventManager CONSTRUCTOR with use_art:true";
125 10 : TLOG(TLVL_CONSTRUCTOR) << "art_pset is " << art_pset.to_string();
126 : }
127 :
128 15 : if (manual_art_)
129 0 : current_art_config_file_ = std::make_shared<art_config_file>(art_pset, GetKey(), GetBroadcastKey());
130 : else
131 15 : current_art_config_file_ = std::make_shared<art_config_file>(art_pset);
132 :
133 15 : if (overwrite_mode_ && num_art_processes_ > 0)
134 : {
135 0 : TLOG(TLVL_WARNING) << "Art is configured to run, but overwrite mode is enabled! Check your configuration if this in unintentional!";
136 0 : }
137 15 : else if (overwrite_mode_)
138 : {
139 30 : TLOG(TLVL_INFO) << "Overwrite Mode enabled, no configured art processes at startup";
140 : }
141 :
142 103 : for (size_t ii = 0; ii < size(); ++ii)
143 : {
144 88 : buffer_writes_pending_[ii] = 0;
145 : // Make sure the mutexes are created once
146 88 : std::lock_guard<std::mutex> lk(buffer_mutexes_[ii]);
147 88 : }
148 :
149 15 : if (!IsValid())
150 : {
151 0 : TLOG(TLVL_ERROR) << "Unable to attach to Shared Memory!";
152 0 : throw cet::exception(app_name + "_SharedMemoryEventManager") << "Unable to attach to Shared Memory!"; // NOLINT(cert-err60-cpp)
153 : }
154 :
155 30 : TLOG(TLVL_CONSTRUCTOR) << "Setting Writer rank to " << my_rank;
156 15 : SetRank(my_rank);
157 30 : TLOG(TLVL_CONSTRUCTOR) << "Writer Rank is " << GetRank();
158 :
159 15 : statsHelper_.addMonitoredQuantityName(FRAGMENTS_RECEIVED_STAT_KEY);
160 15 : statsHelper_.addMonitoredQuantityName(EVENTS_RELEASED_STAT_KEY);
161 :
162 : // fetch the monitoring parameters and create the MonitoredQuantity instances
163 15 : statsHelper_.createCollectors(pset, 100, 30.0, 60.0, EVENTS_RELEASED_STAT_KEY);
164 :
165 30 : TLOG(TLVL_CONSTRUCTOR) << "END CONSTRUCTOR";
166 15 : }
167 :
168 16 : artdaq::SharedMemoryEventManager::~SharedMemoryEventManager() noexcept
169 : {
170 30 : TLOG(TLVL_DESTRUCTOR) << "DESTRUCTOR";
171 15 : if (running_)
172 : {
173 : try
174 : {
175 8 : endOfData();
176 : }
177 0 : catch (...)
178 : {
179 : // IGNORED
180 0 : }
181 : }
182 :
183 15 : UnregisterWriter();
184 15 : broadcasts_.UnregisterWriter();
185 30 : TLOG(TLVL_DESTRUCTOR) << "Destructor END";
186 16 : }
187 :
188 7138 : bool artdaq::SharedMemoryEventManager::AddFragment(detail::RawFragmentHeader frag, void* dataPtr)
189 : {
190 7138 : if (!running_) return true;
191 :
192 14276 : TLOG(TLVL_ADDFRAGMENT) << "AddFragment(Header, ptr) BEGIN frag.word_count=" << frag.word_count
193 7138 : << ", sequence_id=" << frag.sequence_id;
194 7138 : auto buffer = getBufferForSequenceID_(frag.sequence_id, true, frag.timestamp);
195 14276 : TLOG(TLVL_ADDFRAGMENT) << "Using buffer " << buffer << " for seqid=" << frag.sequence_id;
196 7138 : if (buffer == -1)
197 : {
198 5067 : return false;
199 : }
200 2071 : if (buffer == -2)
201 : {
202 0 : TLOG(TLVL_ERROR) << "Dropping event because data taking has already passed this event number: " << frag.sequence_id;
203 0 : return true;
204 : }
205 :
206 2071 : auto hdr = getEventHeader_(buffer);
207 2071 : hdr->run_id = run_id_;
208 2071 : hdr->subrun_id = GetSubrunForSequenceID(frag.sequence_id);
209 2071 : hdr->event_id = GetEventIDForFragment(frag.sequence_id, frag.timestamp);
210 :
211 4142 : TLOG(TLVL_ADDFRAGMENT) << "AddFragment before Write calls";
212 2071 : Write(buffer, dataPtr, frag.word_count * sizeof(RawDataType));
213 :
214 4142 : TLOG(TLVL_ADDFRAGMENT) << "Checking for complete event";
215 2071 : auto fragmentCount = GetFragmentCount(frag.sequence_id);
216 2071 : hdr->is_complete = fragmentCount == num_fragments_per_event_ && buffer_writes_pending_[buffer] == 0;
217 4142 : TLOG(TLVL_ADDFRAGMENT) << "hdr->is_complete=" << std::boolalpha << hdr->is_complete
218 0 : << ", fragmentCount=" << fragmentCount
219 0 : << ", num_fragments_per_event=" << num_fragments_per_event_
220 2071 : << ", buffer_writes_pending_[buffer]=" << buffer_writes_pending_[buffer];
221 :
222 2071 : complete_buffer_(buffer);
223 :
224 4142 : TLOG(TLVL_ADDFRAGMENT) << "AddFragment END";
225 2071 : statsHelper_.addSample(FRAGMENTS_RECEIVED_STAT_KEY, frag.word_count * sizeof(RawDataType));
226 2071 : return true;
227 : }
228 :
229 2075 : bool artdaq::SharedMemoryEventManager::AddFragment(FragmentPtr frag, size_t timeout_usec, FragmentPtr& outfrag)
230 : {
231 4150 : TLOG(TLVL_ADDFRAGMENT) << "AddFragment(FragmentPtr) BEGIN";
232 2075 : auto hdr = *reinterpret_cast<detail::RawFragmentHeader*>(frag->headerAddress()); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
233 2075 : auto data = frag->headerAddress();
234 2075 : auto start = std::chrono::steady_clock::now();
235 2075 : bool sts = false;
236 11288 : while (!sts && TimeUtils::GetElapsedTimeMicroseconds(start) < timeout_usec)
237 : {
238 7138 : sts = AddFragment(hdr, data);
239 7138 : if (!sts)
240 : {
241 5067 : usleep(1000);
242 : }
243 : }
244 2075 : if (!sts)
245 : {
246 4 : outfrag = std::move(frag);
247 : }
248 4150 : TLOG(TLVL_ADDFRAGMENT) << "AddFragment(FragmentPtr) RETURN " << std::boolalpha << sts;
249 2075 : return sts;
250 : }
251 :
252 26 : artdaq::RawDataType* artdaq::SharedMemoryEventManager::WriteFragmentHeader(detail::RawFragmentHeader frag, bool dropIfNoBuffersAvailable)
253 : {
254 26 : if (!running_) return nullptr;
255 50 : TLOG(TLVL_WRITEFRAGMENTHEADER) << "WriteFragmentHeader BEGIN, seqID=" << frag.sequence_id;
256 25 : auto buffer = getBufferForSequenceID_(frag.sequence_id, true, frag.timestamp);
257 :
258 25 : if (buffer < 0)
259 : {
260 2 : if (buffer == -1 && !dropIfNoBuffersAvailable)
261 : {
262 0 : std::unique_lock<std::mutex> bp_lk(sequence_id_mutex_);
263 0 : if (TimeUtils::GetElapsedTime(last_backpressure_report_time_) > 1.0)
264 : {
265 0 : TLOG(TLVL_WARNING) << app_name << ": Back-pressure condition: All Shared Memory buffers have been full for " << TimeUtils::GetElapsedTime(last_fragment_header_write_time_) << " s! There are " << (GetAttachedCount() - 1) << " art processes connected.";
266 0 : last_backpressure_report_time_ = std::chrono::steady_clock::now();
267 0 : if (GetAttachedCount() == 1 && !restart_art_)
268 : {
269 0 : TLOG(TLVL_ERROR) << "All art processes have died, and restarting was unsuccessful. Check PMT log file for error messages";
270 0 : throw cet::exception(app_name + "_SharedMemoryEventManager") << "All art processes have died, and restarting was unsuccessful. Check PMT log file for error messages"; // NOLINT(cert-err60-cpp)
271 : }
272 : }
273 0 : if (metricMan)
274 : {
275 0 : metricMan->sendMetric("Back-pressure wait time", TimeUtils::GetElapsedTime(last_fragment_header_write_time_), "s", 1, MetricMode::LastPoint);
276 : }
277 0 : TLOG(TLVL_WRITEFRAGMENTHEADER) << "No shared memory buffers available, seqID=" << frag.sequence_id;
278 0 : return nullptr;
279 0 : }
280 2 : if (buffer == -2)
281 : {
282 6 : TLOG(TLVL_ERROR) << "Dropping fragment with sequence id " << frag.sequence_id << " and fragment id " << frag.fragment_id << " because data taking has already passed this event.";
283 : }
284 : else
285 : {
286 0 : TLOG(TLVL_INFO) << "Dropping fragment with sequence id " << frag.sequence_id << " and fragment id " << frag.fragment_id << " because there is no room in the queue and reliable mode is off.";
287 : }
288 2 : dropped_data_.emplace_back(frag, std::make_unique<Fragment>(frag.word_count - frag.num_words()));
289 2 : auto it = dropped_data_.rbegin();
290 :
291 4 : TLOG(TLVL_WRITEFRAGMENTHEADER) << "Dropping fragment with sequence id " << frag.sequence_id << " and fragment id " << frag.fragment_id << " into "
292 2 : << static_cast<void*>(it->second->dataBegin()) << " sz=" << it->second->dataSizeBytes();
293 :
294 2 : return it->second->dataBegin();
295 : }
296 :
297 23 : last_backpressure_report_time_ = std::chrono::steady_clock::now();
298 23 : last_fragment_header_write_time_ = std::chrono::steady_clock::now();
299 : // Increment this as soon as we know we want to use the buffer
300 23 : buffer_writes_pending_[buffer]++;
301 :
302 23 : if (metricMan)
303 : {
304 161 : metricMan->sendMetric("Input Fragment Rate", 1, "Fragments/s", 1, MetricMode::Rate);
305 : }
306 :
307 46 : TLOG(TLVL_BUFLCK) << "WriteFragmentHeader: obtaining buffer_mutexes lock for buffer " << buffer << ", seqID=" << frag.sequence_id;
308 : ;
309 :
310 23 : std::unique_lock<std::mutex> lk(buffer_mutexes_.at(buffer));
311 :
312 46 : TLOG(TLVL_BUFLCK) << "WriteFragmentHeader: obtained buffer_mutexes lock for buffer " << buffer << ", seqID=" << frag.sequence_id;
313 :
314 23 : auto hdrpos = reinterpret_cast<RawDataType*>(GetWritePos(buffer)); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
315 23 : Write(buffer, &frag, frag.num_words() * sizeof(RawDataType));
316 :
317 23 : auto pos = reinterpret_cast<RawDataType*>(GetWritePos(buffer)); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
318 23 : if (frag.word_count - frag.num_words() > 0)
319 : {
320 23 : auto sts = IncrementWritePos(buffer, (frag.word_count - frag.num_words()) * sizeof(RawDataType));
321 :
322 23 : if (!sts)
323 : {
324 0 : reinterpret_cast<detail::RawFragmentHeader*>(hdrpos)->word_count = frag.num_words(); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
325 0 : reinterpret_cast<detail::RawFragmentHeader*>(hdrpos)->type = Fragment::ErrorFragmentType; // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
326 0 : TLOG(TLVL_ERROR) << "Dropping over-size fragment with sequence id " << frag.sequence_id << " and fragment id " << frag.fragment_id << " because there is no room in the current buffer for this Fragment! (Keeping header)";
327 0 : dropped_data_.emplace_back(frag, std::make_unique<Fragment>(frag.word_count - frag.num_words()));
328 0 : auto it = dropped_data_.rbegin();
329 :
330 0 : oversize_fragment_count_++;
331 :
332 0 : if (maximum_oversize_fragment_count_ > 0 && oversize_fragment_count_ >= maximum_oversize_fragment_count_)
333 : {
334 0 : lk.unlock();
335 0 : TLOG(TLVL_ERROR) << "Too many over-size Fragments received! Please adjust max_event_size_bytes or max_fragment_size_bytes!";
336 0 : throw cet::exception("Too many over-size Fragments received! Please adjust max_event_size_bytes or max_fragment_size_bytes!");
337 : }
338 :
339 0 : TLOG(TLVL_WRITEFRAGMENTHEADER) << "Dropping over-size fragment with sequence id " << frag.sequence_id << " and fragment id " << frag.fragment_id
340 0 : << " into " << static_cast<void*>(it->second->dataBegin());
341 0 : return it->second->dataBegin();
342 : }
343 : }
344 46 : TLOG(TLVL_WRITEFRAGMENTHEADER) << "WriteFragmentHeader END, seqID=" << frag.sequence_id;
345 23 : return pos;
346 23 : }
347 :
348 25 : void artdaq::SharedMemoryEventManager::DoneWritingFragment(detail::RawFragmentHeader frag)
349 : {
350 50 : TLOG(TLVL_DONEWRITINGFRAGMENT) << "DoneWritingFragment BEGIN";
351 :
352 25 : auto buffer = getBufferForSequenceID_(frag.sequence_id, false, frag.timestamp);
353 25 : if (buffer < 0)
354 : {
355 2 : for (auto it = dropped_data_.begin(); it != dropped_data_.end(); ++it)
356 : {
357 2 : if (frag.operator==(it->first)) // TODO, ELF 5/26/2023: Workaround until artdaq_core can be fixed for C++20
358 : {
359 2 : dropped_data_.erase(it);
360 2 : return;
361 : }
362 : }
363 0 : if (buffer == -1)
364 : {
365 0 : Detach(true, app_name + "SharedMemoryEventManager",
366 : "getBufferForSequenceID_ returned -1 in DoneWritingFragment. This indicates a possible mismatch between expected Fragment count and the actual number of Fragments received.");
367 : }
368 0 : return;
369 : }
370 :
371 23 : if (!frag.valid)
372 : {
373 0 : UpdateFragmentHeader(buffer, frag);
374 : }
375 :
376 23 : statsHelper_.addSample(FRAGMENTS_RECEIVED_STAT_KEY, frag.word_count * sizeof(RawDataType));
377 : {
378 46 : TLOG(TLVL_BUFLCK) << "DoneWritingFragment: obtaining buffer_mutexes lock for buffer " << buffer;
379 :
380 23 : std::unique_lock<std::mutex> lk(buffer_mutexes_.at(buffer));
381 :
382 46 : TLOG(TLVL_BUFLCK) << "DoneWritingFragment: obtained buffer_mutexes lock for buffer " << buffer;
383 :
384 46 : TLOG(TLVL_DONEWRITINGFRAGMENT) << "DoneWritingFragment: Received Fragment with sequence ID " << frag.sequence_id << " and fragment id " << frag.fragment_id << " (type " << static_cast<int>(frag.type) << ")";
385 23 : auto hdr = getEventHeader_(buffer);
386 23 : hdr->run_id = run_id_;
387 23 : hdr->subrun_id = GetSubrunForSequenceID(frag.sequence_id);
388 23 : hdr->event_id = GetEventIDForFragment(frag.sequence_id, frag.timestamp);
389 :
390 46 : TLOG(TLVL_DONEWRITINGFRAGMENT) << "DoneWritingFragment: Updating buffer touch time";
391 23 : TouchBuffer(buffer);
392 :
393 23 : if (buffer_writes_pending_[buffer] > 1)
394 : {
395 6 : TLOG(TLVL_DONEWRITINGFRAGMENT) << "Done writing fragment, but there's another writer. Not doing bookkeeping steps.";
396 3 : buffer_writes_pending_[buffer]--;
397 3 : return;
398 : }
399 40 : TLOG(TLVL_DONEWRITINGFRAGMENT) << "Done writing fragment, and no other writer. Doing bookkeeping steps.";
400 20 : auto frag_count = GetFragmentCount(frag.sequence_id);
401 20 : hdr->is_complete = frag_count >= num_fragments_per_event_;
402 :
403 20 : if (frag_count > num_fragments_per_event_)
404 : {
405 3 : TLOG(TLVL_WARNING) << "DoneWritingFragment: This Event has more Fragments ( " << frag_count << " ) than specified in configuration ( " << num_fragments_per_event_ << " )!"
406 2 : << " This is probably due to a misconfiguration and is *not* a reliable mode!";
407 : }
408 :
409 40 : TLOG(TLVL_DONEWRITINGFRAGMENT) << "DoneWritingFragment: Received Fragment with sequence ID " << frag.sequence_id << " and fragment id " << frag.fragment_id << ", count/expected = " << frag_count << "/" << num_fragments_per_event_;
410 : #if ART_SUPPORTS_DUPLICATE_EVENTS
411 : if (!hdr->is_complete && released_incomplete_events_.count(frag.sequence_id))
412 : {
413 : hdr->is_complete = frag_count >= released_incomplete_events_[frag.sequence_id] && buffer_writes_pending_[buffer] == 0;
414 : }
415 : #endif
416 :
417 20 : complete_buffer_(buffer);
418 :
419 : // Move this down here to avoid race condition
420 20 : buffer_writes_pending_[buffer]--;
421 23 : }
422 40 : TLOG(TLVL_DONEWRITINGFRAGMENT) << "DoneWritingFragment END";
423 : }
424 :
425 2101 : size_t artdaq::SharedMemoryEventManager::GetFragmentCount(Fragment::sequence_id_t seqID, Fragment::type_t type)
426 : {
427 2101 : return GetFragmentCountInBuffer(getBufferForSequenceID_(seqID, false), type);
428 : }
429 :
430 2106 : size_t artdaq::SharedMemoryEventManager::GetFragmentCountInBuffer(int buffer, Fragment::type_t type)
431 : {
432 2106 : if (buffer < 0)
433 : {
434 0 : return 0;
435 : }
436 2106 : ResetReadPos(buffer);
437 2106 : IncrementReadPos(buffer, sizeof(detail::RawEventHeader));
438 :
439 2106 : size_t count = 0;
440 :
441 8326 : while (MoreDataInBuffer(buffer))
442 : {
443 6220 : auto fragHdr = reinterpret_cast<artdaq::detail::RawFragmentHeader*>(GetReadPos(buffer)); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
444 6220 : IncrementReadPos(buffer, fragHdr->word_count * sizeof(RawDataType));
445 6220 : if (type != Fragment::InvalidFragmentType && fragHdr->type != type)
446 : {
447 : // Skip fragments with the wrong type, as they were over-size and truncated to the header
448 0 : continue;
449 : }
450 12440 : TLOG(TLVL_GETFRAGMENTCOUNT) << "Adding Fragment with size=" << fragHdr->word_count << " to Fragment count";
451 6220 : ++count;
452 : }
453 :
454 2106 : return count;
455 : }
456 :
457 0 : void artdaq::SharedMemoryEventManager::UpdateFragmentHeader(int buffer, artdaq::detail::RawFragmentHeader hdr)
458 : {
459 0 : if (buffer < 0)
460 : {
461 0 : return;
462 : }
463 0 : ResetReadPos(buffer);
464 0 : IncrementReadPos(buffer, sizeof(detail::RawEventHeader));
465 :
466 0 : while (MoreDataInBuffer(buffer))
467 : {
468 0 : auto fragHdr = reinterpret_cast<artdaq::detail::RawFragmentHeader*>(GetReadPos(buffer)); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
469 0 : if (hdr.fragment_id == fragHdr->fragment_id)
470 : {
471 0 : *fragHdr = hdr;
472 0 : break;
473 : }
474 : }
475 :
476 0 : return;
477 : }
478 :
479 6 : void artdaq::SharedMemoryEventManager::RunArt(size_t process_index, const std::shared_ptr<std::atomic<pid_t>>& pid_out)
480 : {
481 : do
482 : {
483 6 : auto start_time = std::chrono::steady_clock::now();
484 6 : send_init_frags_();
485 18 : TLOG(TLVL_INFO) << "Starting art process with config file " << current_art_config_file_->getFileName();
486 :
487 6 : pid_t pid = 0;
488 6 : bool piped_output = true;
489 : int stdoutpipefd[2];
490 : int stderrpipefd[2];
491 :
492 6 : if (!manual_art_)
493 : {
494 6 : if (pipe(stdoutpipefd) == -1)
495 : {
496 0 : TLOG(TLVL_ERROR) << "Error creating pipe for art process stdout: " << errno << " (" << strerror(errno) << ").";
497 0 : piped_output = false;
498 : }
499 6 : if (piped_output && pipe(stderrpipefd) == -1)
500 : {
501 0 : TLOG(TLVL_ERROR) << "Error creating pipe for art process stderr: " << errno << " (" << strerror(errno) << ").";
502 0 : close(stdoutpipefd[0]);
503 0 : close(stdoutpipefd[1]);
504 0 : piped_output = false;
505 : }
506 6 : pid = fork();
507 6 : if (pid == 0)
508 : { /* child */
509 0 : if (piped_output)
510 : {
511 0 : close(stdoutpipefd[0]); // Close read end of stdout pipe
512 0 : close(stderrpipefd[0]); // Close read end of stderr pipe
513 0 : dup2(stdoutpipefd[1], STDOUT_FILENO); // Redirect stdout to pipe
514 0 : dup2(stderrpipefd[1], STDERR_FILENO); // Redirect stderr to pipe
515 : }
516 :
517 : // 23-May-2018, KAB: added the setting of the partition number env var
518 : // in the environment of the child art process so that Globals.hh
519 : // will pick it up there and provide it to the artdaq classes that
520 : // are used in data transfers, etc. within the art process.
521 0 : std::string envVarKey = "ARTDAQ_PARTITION_NUMBER";
522 0 : std::string envVarValue = std::to_string(Globals::GetPartitionNumber());
523 0 : if (setenv(envVarKey.c_str(), envVarValue.c_str(), 1) != 0)
524 : {
525 0 : TLOG(TLVL_ERROR) << "Error setting environment variable \"" << envVarKey
526 0 : << "\" in the environment of a child art process. "
527 0 : << "This may result in incorrect TCP port number "
528 0 : << "assignments or other issues, and data may "
529 0 : << "not flow through the system correctly.";
530 : }
531 0 : envVarKey = "ARTDAQ_APPLICATION_NAME";
532 0 : envVarValue = app_name;
533 0 : if (setenv(envVarKey.c_str(), envVarValue.c_str(), 1) != 0)
534 : {
535 0 : TLOG(TLVL_RUNART) << "Error setting environment variable \"" << envVarKey
536 0 : << "\" in the environment of a child art process. ";
537 : }
538 0 : envVarKey = "ARTDAQ_RANK";
539 0 : envVarValue = std::to_string(my_rank);
540 0 : if (setenv(envVarKey.c_str(), envVarValue.c_str(), 1) != 0)
541 : {
542 0 : TLOG(TLVL_RUNART) << "Error setting environment variable \"" << envVarKey
543 0 : << "\" in the environment of a child art process. ";
544 : }
545 :
546 0 : TLOG(TLVL_RUNART_2) << "Parsing art command line";
547 0 : auto args = parse_art_command_line_(current_art_config_file_, process_index);
548 :
549 0 : TLOG(TLVL_RUNART_2) << "Calling execvp with application name " << args[0];
550 0 : execvp(args[0], &args[0]);
551 :
552 0 : TLOG(TLVL_RUNART_2) << "Application exited, cleaning up";
553 0 : for (auto& arg : args)
554 : {
555 0 : delete[] arg;
556 : }
557 :
558 0 : exit(1);
559 0 : }
560 : }
561 : else
562 : {
563 : // Using cin/cout here to ensure console is active (artdaqDriver)
564 0 : std::cout << "Please run the following command in a separate terminal:" << std::endl
565 0 : << "art -c " << current_art_config_file_->getFileName() << std::endl
566 0 : << "Then, in a third terminal, execute: \"ps aux|grep [a]rt -c " << current_art_config_file_->getFileName() << "\" and note the PID of the art process." << std::endl
567 0 : << "Finally, return to this window and enter the pid: " << std::endl;
568 0 : std::cin >> pid;
569 : }
570 6 : *pid_out = pid;
571 :
572 18 : TLOG(TLVL_INFO) << "PID of new art process is " << pid;
573 : {
574 6 : std::unique_lock<std::mutex> lk(art_process_mutex_);
575 6 : art_processes_.insert(pid);
576 6 : }
577 : siginfo_t status;
578 6 : auto sts = 0;
579 6 : if (!manual_art_)
580 : {
581 : // Read output from art process and report it to TRACE
582 6 : if (piped_output)
583 : {
584 6 : close(stdoutpipefd[1]); // Close write end of stdout pipe
585 6 : close(stderrpipefd[1]); // Close write end of stderr pipe
586 :
587 6 : std::string art_tname = app_name + "_art_stdout";
588 : char buf[PIPE_BUF];
589 : struct pollfd fds[2];
590 6 : fds[0].fd = stdoutpipefd[0];
591 6 : fds[0].events = POLLIN;
592 6 : fds[1].fd = stderrpipefd[0];
593 6 : fds[1].events = POLLIN;
594 :
595 : do
596 : {
597 2623 : sts = waitid(P_PID, pid, &status, WEXITED | WNOHANG);
598 2623 : poll(fds, 2, 1000);
599 2623 : if (fds[0].revents & POLLIN)
600 : {
601 105 : ssize_t count = read(stdoutpipefd[0], buf, sizeof(buf) - 1);
602 105 : if (count > 0 && capture_art_stdout_)
603 : {
604 420 : TLOG(TLVL_INFO, art_tname) << "art[" << pid << "]" << std::string(buf, count);
605 : }
606 : }
607 2623 : if (fds[1].revents & POLLIN)
608 : {
609 0 : ssize_t count = read(stderrpipefd[0], buf, sizeof(buf) - 1);
610 0 : if (count > 0 && capture_art_stderr_)
611 : {
612 0 : TLOG(TLVL_ERROR, art_tname) << "art[" << pid << "]" << std::string(buf, count);
613 : }
614 : }
615 2623 : } while (status.si_code != CLD_DUMPED && status.si_code != CLD_KILLED && status.si_code != CLD_EXITED && sts == 0);
616 6 : }
617 : else
618 : {
619 0 : sts = waitid(P_PID, pid, &status, WEXITED);
620 : }
621 : }
622 : else
623 : {
624 0 : while (kill(pid, 0) >= 0) usleep(10000);
625 :
626 0 : TLOG(TLVL_INFO) << "Faking good exit status, please see art process for actual exit status!";
627 0 : status.si_code = CLD_EXITED;
628 0 : status.si_status = 0;
629 : }
630 18 : TLOG(TLVL_INFO) << "Removing PID " << pid << " from process list";
631 : {
632 6 : std::unique_lock<std::mutex> lk(art_process_mutex_);
633 6 : art_processes_.erase(pid);
634 6 : }
635 6 : if (sts < 0)
636 : {
637 0 : TLOG(TLVL_WARNING) << "Error occurred in waitid for art process " << pid << ": " << errno << " (" << strerror(errno) << ").";
638 : }
639 6 : else if (status.si_code == CLD_EXITED && status.si_status == 0)
640 : {
641 18 : TLOG(TLVL_INFO) << "art process " << pid << " exited normally, " << (restart_art_ ? "restarting" : "not restarting");
642 6 : }
643 : else
644 : {
645 0 : auto art_lifetime = TimeUtils::GetElapsedTime(start_time);
646 0 : if (art_lifetime < minimum_art_lifetime_s_)
647 : {
648 0 : restart_art_ = false;
649 : }
650 :
651 0 : auto exit_type = "exited with status code";
652 0 : switch (status.si_code)
653 : {
654 0 : case CLD_DUMPED:
655 : case CLD_KILLED:
656 0 : exit_type = "was killed with signal";
657 0 : break;
658 0 : case CLD_EXITED:
659 : default:
660 0 : break;
661 : }
662 :
663 0 : TLOG((restart_art_ ? TLVL_WARNING : TLVL_ERROR))
664 0 : << "art process " << pid << " " << exit_type << " " << status.si_status
665 0 : << (status.si_code == CLD_DUMPED ? " (core dumped)" : "")
666 0 : << " after running for " << std::setprecision(2) << std::fixed << art_lifetime << " seconds, "
667 0 : << (restart_art_ ? "restarting" : "not restarting");
668 : }
669 6 : } while (restart_art_);
670 6 : }
671 :
672 17 : bool artdaq::SharedMemoryEventManager::StartArt()
673 : {
674 17 : size_t initialCount = GetAttachedCount();
675 17 : restart_art_ = always_restart_art_;
676 17 : if (num_art_processes_ == 0)
677 : {
678 11 : return true;
679 : }
680 12 : for (size_t ii = 0; ii < num_art_processes_; ++ii)
681 : {
682 6 : StartArtProcess(current_art_pset_, ii);
683 : }
684 6 : auto startTime = std::chrono::steady_clock::now();
685 6 : while (GetAttachedCount() - initialCount != num_art_processes_)
686 : {
687 0 : TLOG(TLVL_INFO) << "Waiting for all art processes to connect to shared memory, " << TimeUtils::GetElapsedTime(startTime) << " s elapsed.";
688 0 : std::this_thread::sleep_for(std::chrono::seconds(1));
689 0 : if (!restart_art_)
690 : {
691 0 : TLOG(TLVL_ERROR) << "Error occurred while starting art processes, aborting. Check PMT log for error messages.";
692 0 : return false;
693 : }
694 : }
695 6 : return true;
696 : }
697 :
698 6 : pid_t artdaq::SharedMemoryEventManager::StartArtProcess(fhicl::ParameterSet pset, size_t process_index)
699 : {
700 : static std::mutex start_art_mutex;
701 6 : std::unique_lock<std::mutex> lk(start_art_mutex);
702 : // TraceLock lk(start_art_mutex, 15, "StartArtLock");
703 6 : restart_art_ = always_restart_art_;
704 6 : auto initialCount = GetAttachedCount();
705 6 : auto startTime = std::chrono::steady_clock::now();
706 :
707 6 : if (pset != current_art_pset_ || !current_art_config_file_)
708 : {
709 0 : current_art_pset_ = pset;
710 0 : if (manual_art_)
711 0 : current_art_config_file_ = std::make_shared<art_config_file>(pset, GetKey(), GetBroadcastKey());
712 : else
713 0 : current_art_config_file_ = std::make_shared<art_config_file>(pset);
714 : }
715 6 : std::shared_ptr<std::atomic<pid_t>> pid(new std::atomic<pid_t>(-1));
716 12 : boost::thread thread([this, process_index, pid] { RunArt(process_index, pid); });
717 6 : thread.detach();
718 :
719 6 : auto currentCount = GetAttachedCount() - initialCount;
720 31 : while ((currentCount < 1 || *pid <= 0) && (TimeUtils::GetElapsedTime(startTime) < 5 || manual_art_))
721 : {
722 25 : usleep(10000);
723 25 : currentCount = GetAttachedCount() - initialCount;
724 : }
725 6 : if ((currentCount < 1 || *pid <= 0) && manual_art_)
726 : {
727 0 : TLOG(TLVL_WARNING) << "Manually-started art process has not connected to shared memory or has bad PID: connected:" << currentCount << ", PID:" << pid;
728 0 : return 0;
729 : }
730 6 : if (currentCount < 1 || *pid <= 0)
731 : {
732 0 : TLOG(TLVL_WARNING) << "art process has not started after 5s. Check art configuration!"
733 0 : << " (pid=" << *pid << ", attachedCount=" << currentCount << ")";
734 0 : return 0;
735 : }
736 :
737 18 : TLOG(TLVL_INFO) << std::setw(4) << std::fixed << "art initialization took "
738 12 : << TimeUtils::GetElapsedTime(startTime) << " seconds.";
739 :
740 6 : return *pid;
741 6 : }
742 :
743 6 : void artdaq::SharedMemoryEventManager::ShutdownArtProcesses(std::set<pid_t>& pids)
744 : {
745 6 : restart_art_ = false;
746 : // current_art_config_file_ = nullptr;
747 : // current_art_pset_ = fhicl::ParameterSet();
748 :
749 370 : auto check_pids = [&](bool print) {
750 370 : std::unique_lock<std::mutex> lk(art_process_mutex_);
751 735 : for (auto pid = pids.begin(); pid != pids.end();)
752 : {
753 : // 08-May-2018, KAB: protect against killing invalid PIDS
754 :
755 365 : if (*pid <= 0)
756 : {
757 0 : TLOG(TLVL_WARNING) << "Removing an invalid PID (" << *pid
758 0 : << ") from the shutdown list.";
759 0 : pid = pids.erase(pid);
760 : }
761 365 : else if (kill(*pid, 0) < 0)
762 : {
763 1 : pid = pids.erase(pid);
764 : }
765 : else
766 : {
767 364 : if (print)
768 : {
769 0 : std::cout << *pid << " ";
770 : }
771 364 : ++pid;
772 : }
773 : }
774 370 : };
775 370 : auto count_pids = [&]() {
776 370 : std::unique_lock<std::mutex> lk(art_process_mutex_);
777 740 : return pids.size();
778 370 : };
779 6 : check_pids(false);
780 6 : if (count_pids() == 0)
781 : {
782 0 : TLOG(TLVL_SHUTDOWN) << "All art processes already exited, nothing to do.";
783 0 : usleep(1000);
784 6 : return;
785 : }
786 :
787 6 : if (!manual_art_)
788 : {
789 6 : int graceful_wait_ms = art_event_processing_time_us_ * size() * 10 / 1000;
790 6 : int gentle_wait_ms = art_event_processing_time_us_ * size() * 2 / 1000;
791 6 : int int_wait_ms = art_event_processing_time_us_ * size() / 1000;
792 6 : auto shutdown_start = std::chrono::steady_clock::now();
793 :
794 : // if (!overwrite_mode_)
795 : {
796 12 : TLOG(TLVL_SHUTDOWN) << "Waiting up to " << graceful_wait_ms << " ms for all art processes to exit gracefully";
797 364 : for (int ii = 0; ii < graceful_wait_ms; ++ii)
798 : {
799 364 : usleep(1000);
800 :
801 364 : check_pids(false);
802 364 : if (count_pids() == 0)
803 : {
804 18 : TLOG(TLVL_INFO) << "All art processes exited after " << TimeUtils::GetElapsedTimeMilliseconds(shutdown_start) << " ms.";
805 6 : return;
806 : }
807 : }
808 : }
809 :
810 : {
811 0 : TLOG(TLVL_SHUTDOWN) << "Gently informing art processes that it is time to shut down";
812 0 : std::unique_lock<std::mutex> lk(art_process_mutex_);
813 0 : for (auto pid : pids)
814 : {
815 0 : TLOG(TLVL_SHUTDOWN) << "Sending SIGQUIT to pid " << pid;
816 0 : kill(pid, SIGQUIT);
817 : }
818 0 : }
819 :
820 0 : TLOG(TLVL_SHUTDOWN) << "Waiting up to " << gentle_wait_ms << " ms for all art processes to exit from SIGQUIT";
821 0 : for (int ii = 0; ii < gentle_wait_ms; ++ii)
822 : {
823 0 : usleep(1000);
824 :
825 0 : check_pids(false);
826 0 : if (count_pids() == 0)
827 : {
828 0 : TLOG(TLVL_INFO) << "All art processes exited after " << TimeUtils::GetElapsedTimeMilliseconds(shutdown_start) << " ms (SIGQUIT).";
829 0 : return;
830 : }
831 : }
832 :
833 : {
834 0 : TLOG(TLVL_SHUTDOWN) << "Insisting that the art processes shut down";
835 0 : std::unique_lock<std::mutex> lk(art_process_mutex_);
836 0 : for (auto pid : pids)
837 : {
838 0 : kill(pid, SIGINT);
839 : }
840 0 : }
841 :
842 0 : TLOG(TLVL_SHUTDOWN) << "Waiting up to " << int_wait_ms << " ms for all art processes to exit from SIGINT";
843 0 : for (int ii = 0; ii < int_wait_ms; ++ii)
844 : {
845 0 : usleep(1000);
846 :
847 0 : check_pids(false);
848 :
849 0 : if (count_pids() == 0)
850 : {
851 0 : TLOG(TLVL_INFO) << "All art processes exited after " << TimeUtils::GetElapsedTimeMilliseconds(shutdown_start) << " ms (SIGINT).";
852 0 : return;
853 : }
854 : }
855 :
856 0 : TLOG(TLVL_SHUTDOWN) << "Killing remaning art processes with extreme prejudice";
857 0 : while (count_pids() > 0)
858 : {
859 : {
860 0 : std::unique_lock<std::mutex> lk(art_process_mutex_);
861 0 : kill(*pids.begin(), SIGKILL);
862 0 : usleep(1000);
863 0 : }
864 0 : check_pids(false);
865 : }
866 0 : TLOG(TLVL_INFO) << "All art processes exited after " << TimeUtils::GetElapsedTimeMilliseconds(shutdown_start) << " ms (SIGKILL).";
867 : }
868 : else
869 : {
870 0 : std::cout << "Please shut down all art processes, then hit return/enter" << std::endl;
871 0 : while (count_pids() > 0)
872 : {
873 0 : std::cout << "The following PIDs are running: ";
874 0 : check_pids(true);
875 0 : std::cout << std::endl;
876 0 : usleep(500000);
877 : }
878 : }
879 : }
880 :
881 1 : void artdaq::SharedMemoryEventManager::ReconfigureArt(fhicl::ParameterSet art_pset, run_id_t newRun, int n_art_processes)
882 : {
883 2 : TLOG(TLVL_RECONFIGUREART) << "ReconfigureArt BEGIN";
884 1 : if (restart_art_ || !always_restart_art_) // Art is running
885 : {
886 0 : endOfData();
887 : }
888 11 : for (size_t ii = 0; ii < broadcasts_.size(); ++ii)
889 : {
890 10 : broadcasts_.MarkBufferEmpty(ii, true);
891 : }
892 1 : if (newRun == 0)
893 : {
894 1 : newRun = run_id_ + 1;
895 : }
896 :
897 1 : if (art_pset != current_art_pset_ || !current_art_config_file_)
898 : {
899 1 : current_art_pset_ = art_pset;
900 1 : if (manual_art_)
901 0 : current_art_config_file_ = std::make_shared<art_config_file>(art_pset, GetKey(), GetBroadcastKey());
902 : else
903 1 : current_art_config_file_ = std::make_shared<art_config_file>(art_pset);
904 : }
905 :
906 1 : if (n_art_processes != -1)
907 : {
908 0 : TLOG(TLVL_INFO) << "Setting number of art processes to " << n_art_processes;
909 0 : num_art_processes_ = n_art_processes;
910 : }
911 1 : startRun(newRun);
912 2 : TLOG(TLVL_RECONFIGUREART) << "ReconfigureArt END";
913 1 : }
914 :
915 16 : bool artdaq::SharedMemoryEventManager::endOfData()
916 : {
917 16 : running_ = false;
918 : {
919 16 : std::lock_guard<std::mutex> lk(init_fragments_mutex_);
920 16 : init_fragment_map_.clear();
921 16 : }
922 16 : init_frags_sent_ = false;
923 32 : TLOG(TLVL_ENDOFDATA) << "SharedMemoryEventManager::endOfData";
924 16 : restart_art_ = false;
925 :
926 16 : auto start = std::chrono::steady_clock::now();
927 212 : auto pendingWriteCount = std::accumulate(buffer_writes_pending_.begin(), buffer_writes_pending_.end(), 0, [](int a, auto& b) { return a + b.second.load(); });
928 32 : TLOG(TLVL_ENDOFDATA) << "endOfData: Waiting for " << pendingWriteCount << " pending writes to complete";
929 26 : while (pendingWriteCount > 0 && TimeUtils::GetElapsedTimeMicroseconds(start) < 1000000)
930 : {
931 10 : usleep(10000);
932 50 : pendingWriteCount = std::accumulate(buffer_writes_pending_.begin(), buffer_writes_pending_.end(), 0, [](int a, auto& b) { return a + b.second.load(); });
933 : }
934 :
935 16 : size_t initialStoreSize = GetOpenEventCount();
936 32 : TLOG(TLVL_ENDOFDATA) << "endOfData: Flushing " << initialStoreSize
937 16 : << " stale events from the SharedMemoryEventManager.";
938 16 : int counter = initialStoreSize;
939 20 : while (!active_buffers_.empty() && counter > 0)
940 : {
941 4 : complete_buffer_(*active_buffers_.begin());
942 4 : counter--;
943 : }
944 32 : TLOG(TLVL_ENDOFDATA) << "endOfData: Done flushing, there are now " << GetOpenEventCount()
945 16 : << " stale events in the SharedMemoryEventManager.";
946 :
947 16 : size_t incomplete_at_shutdown = active_buffers_.size();
948 16 : size_t force_flushed_count = 0;
949 :
950 : // Grace period: wait up to one stale_buffer_timeout for late fragments to arrive and complete events
951 16 : if (!active_buffers_.empty())
952 : {
953 4 : auto grace_timeout_us = GetBufferTimeout();
954 8 : TLOG(TLVL_ENDOFDATA) << "endOfData: Waiting up to " << (grace_timeout_us / 1000000.0)
955 4 : << " s for " << active_buffers_.size() << " incomplete events to finish";
956 4 : start = std::chrono::steady_clock::now();
957 1016 : while (!active_buffers_.empty() && TimeUtils::GetElapsedTimeMicroseconds(start) < grace_timeout_us)
958 : {
959 1012 : usleep(10000);
960 1012 : counter = active_buffers_.size();
961 2024 : while (!active_buffers_.empty() && counter > 0)
962 : {
963 1012 : complete_buffer_(*active_buffers_.begin());
964 1012 : counter--;
965 : }
966 : }
967 :
968 : // Force-flush any remaining incomplete events
969 4 : if (!active_buffers_.empty())
970 : {
971 8 : TLOG(TLVL_ENDOFDATA) << "endOfData: Force-flushing " << active_buffers_.size() << " incomplete events after grace period";
972 4 : std::unique_lock<std::mutex> lk(sequence_id_mutex_);
973 4 : auto buffers_to_flush = active_buffers_;
974 8 : for (auto buf : buffers_to_flush)
975 : {
976 8 : if (buffer_writes_pending_[buf].load() != 0)
977 : {
978 0 : continue;
979 : }
980 4 : auto hdr = getEventHeader_(buf);
981 4 : if (requests_)
982 : {
983 4 : requests_->RemoveRequest(hdr->sequence_id);
984 : }
985 4 : active_buffers_.erase(buf);
986 4 : pending_buffers_.insert(buf);
987 4 : run_incomplete_event_count_++;
988 4 : force_flushed_count++;
989 4 : if (released_incomplete_events_.count(hdr->sequence_id) == 0u)
990 : {
991 4 : released_incomplete_events_[hdr->sequence_id] = num_fragments_per_event_ - GetFragmentCountInBuffer(buf);
992 : }
993 : else
994 : {
995 0 : released_incomplete_events_[hdr->sequence_id] -= GetFragmentCountInBuffer(buf);
996 : }
997 12 : TLOG(TLVL_WARNING) << "endOfData: Event " << hdr->sequence_id
998 4 : << " is incomplete (missing " << released_incomplete_events_[hdr->sequence_id]
999 8 : << " Fragments). Force-releasing to art.";
1000 : }
1001 4 : check_pending_buffers_(lk);
1002 4 : }
1003 : }
1004 :
1005 16 : if (incomplete_at_shutdown == 0)
1006 : {
1007 36 : TLOG(TLVL_INFO) << "endOfData summary: All events were complete at shutdown.";
1008 : }
1009 4 : else if (force_flushed_count == 0)
1010 : {
1011 0 : TLOG(TLVL_INFO) << "endOfData summary: " << incomplete_at_shutdown << " incomplete events all completed during grace period.";
1012 : }
1013 : else
1014 : {
1015 12 : TLOG(TLVL_INFO) << "endOfData summary: " << force_flushed_count << " of " << incomplete_at_shutdown
1016 8 : << " incomplete events were force-flushed after grace period.";
1017 : }
1018 :
1019 32 : TLOG(TLVL_ENDOFDATA) << "Waiting for " << (ReadReadyCount() + (size() - WriteReadyCount(overwrite_mode_))) << " outstanding buffers...";
1020 16 : start = std::chrono::steady_clock::now();
1021 16 : auto lastReadCount = ReadReadyCount() + (size() - WriteReadyCount(overwrite_mode_));
1022 16 : auto end_of_data_wait_us = art_event_processing_time_us_ * (lastReadCount > 0 ? lastReadCount : 1); // size();
1023 :
1024 16 : auto outstanding_buffer_wait_time = art_event_processing_time_us_ > 100000 ? 100000 : art_event_processing_time_us_;
1025 :
1026 : // We will wait until no buffer has been read for the end of data wait seconds, or no art processes are left.
1027 32 : while (lastReadCount > 0 && (end_of_data_wait_us == 0 || TimeUtils::GetElapsedTimeMicroseconds(start) < end_of_data_wait_us) && get_art_process_count_() > 0)
1028 : {
1029 16 : auto temp = ReadReadyCount() + (size() - WriteReadyCount(overwrite_mode_));
1030 16 : if (temp != lastReadCount)
1031 : {
1032 12 : TLOG(TLVL_ENDOFDATA_2) << "Waiting for " << temp << " outstanding buffers...";
1033 6 : lastReadCount = temp;
1034 6 : start = std::chrono::steady_clock::now();
1035 : }
1036 16 : if (lastReadCount > 0)
1037 : {
1038 20 : TLOG(TLVL_ENDOFDATA_2) << "About to sleep " << outstanding_buffer_wait_time << " us - lastReadCount=" << lastReadCount << " size=" << size() << " end_of_data_wait_us=" << end_of_data_wait_us;
1039 10 : usleep(outstanding_buffer_wait_time);
1040 : }
1041 : }
1042 :
1043 32 : TLOG(TLVL_ENDOFDATA) << "endOfData: After wait for outstanding buffers. Still outstanding: " << lastReadCount << ", time waited: "
1044 16 : << TimeUtils::GetElapsedTime(start) << " s / " << (end_of_data_wait_us / 1000000.0) << " s, art process count: " << get_art_process_count_();
1045 :
1046 32 : TLOG(TLVL_ENDOFDATA) << "endOfData: Broadcasting EndOfData Fragment";
1047 16 : FragmentPtrs broadcast;
1048 16 : broadcast.emplace_back(Fragment::eodFrag(GetBufferCount()));
1049 16 : bool success = broadcastFragments_(broadcast);
1050 16 : if (!success)
1051 : {
1052 0 : TLOG(TLVL_ENDOFDATA) << "endOfData: Clearing buffers to make room for EndOfData Fragment";
1053 0 : for (size_t ii = 0; ii < broadcasts_.size(); ++ii)
1054 : {
1055 0 : broadcasts_.MarkBufferEmpty(ii, true);
1056 : }
1057 0 : broadcastFragments_(broadcast);
1058 : }
1059 16 : auto endOfDataProcessingStart = std::chrono::steady_clock::now();
1060 22 : while (get_art_process_count_() > 0)
1061 : {
1062 12 : TLOG(TLVL_ENDOFDATA) << "There are " << get_art_process_count_() << " art processes remaining. Proceeding to shutdown.";
1063 :
1064 6 : ShutdownArtProcesses(art_processes_);
1065 : }
1066 32 : TLOG(TLVL_ENDOFDATA) << "It took " << TimeUtils::GetElapsedTime(endOfDataProcessingStart) << " s for all art processes to close after sending EndOfData Fragment";
1067 :
1068 : // Reset the manager ID counters on both shared memory segments. Manager IDs are only
1069 : // valid in the range 0-63 (readers/writers are tracked in 64-bit masks), and IDs are
1070 : // never recycled on Detach. Without resetting the broadcast segment as well, its
1071 : // counter grows by one for every art process started during the lifetime of this
1072 : // process, and after 63 art process launches (e.g. ~13 runs with 5 art processes each)
1073 : // every new art process logs "Too many processes attached to shared memory" and
1074 : // reader/writer tracking on the broadcast segment becomes unreliable.
1075 16 : ResetAttachedCount();
1076 16 : broadcasts_.ResetAttachedCount();
1077 :
1078 32 : TLOG(TLVL_ENDOFDATA) << "endOfData: Clearing buffers";
1079 114 : for (size_t ii = 0; ii < size(); ++ii)
1080 : {
1081 98 : MarkBufferEmpty(ii, true);
1082 : }
1083 :
1084 16 : released_events_.clear();
1085 16 : released_incomplete_events_.clear();
1086 16 : std::fill(buffer_sequence_ids_.begin(), buffer_sequence_ids_.end(), Fragment::InvalidSequenceID);
1087 :
1088 32 : TLOG(TLVL_ENDOFDATA) << "endOfData END";
1089 48 : TLOG(TLVL_INFO) << "EndOfData Complete. Final Buffer ID is " << GetLastSeenBufferID() << ".";
1090 16 : return true;
1091 16 : }
1092 :
1093 17 : bool artdaq::SharedMemoryEventManager::startRun(run_id_t runID)
1094 : {
1095 17 : running_ = true;
1096 : {
1097 17 : std::lock_guard<std::mutex> lk(init_fragments_mutex_);
1098 17 : init_fragment_map_.clear();
1099 17 : }
1100 17 : init_frags_sent_ = false;
1101 17 : statsHelper_.resetStatistics();
1102 34 : TLOG(TLVL_STARTRUN) << "startRun: Clearing broadcast buffers";
1103 187 : for (size_t ii = 0; ii < broadcasts_.size(); ++ii)
1104 : {
1105 170 : broadcasts_.MarkBufferEmpty(ii, true);
1106 : }
1107 17 : released_events_.clear();
1108 17 : released_incomplete_events_.clear();
1109 17 : std::fill(buffer_sequence_ids_.begin(), buffer_sequence_ids_.end(), Fragment::InvalidSequenceID);
1110 : // If we fail to start the art processes, abort Start
1111 17 : if (!StartArt())
1112 : {
1113 0 : return false;
1114 : }
1115 17 : run_id_ = runID;
1116 17 : subrun_id_ = 1;
1117 : {
1118 17 : std::unique_lock<std::mutex> lk(subrun_event_map_mutex_);
1119 17 : subrun_event_map_.clear();
1120 17 : subrun_event_map_[0] = 1;
1121 17 : }
1122 17 : run_event_count_ = 0;
1123 17 : run_incomplete_event_count_ = 0;
1124 17 : requests_ = std::make_unique<RequestSender>(data_pset_);
1125 17 : if (requests_)
1126 : {
1127 17 : requests_->SetRunNumber(static_cast<uint32_t>(run_id_));
1128 : }
1129 51 : if (data_pset_.has_key("routing_token_config"))
1130 : {
1131 0 : auto rmPset = data_pset_.get<fhicl::ParameterSet>("routing_token_config");
1132 0 : if (rmPset.get<bool>("use_routing_manager", false))
1133 : {
1134 0 : tokens_ = std::make_unique<TokenSender>(rmPset);
1135 0 : tokens_->SetRunNumber(static_cast<uint32_t>(run_id_));
1136 0 : tokens_->SendRoutingToken(queue_size_, run_id_);
1137 : }
1138 0 : }
1139 34 : TLOG(TLVL_STARTRUN) << "Starting run " << run_id_
1140 0 : << ", max queue size = "
1141 0 : << queue_size_
1142 0 : << ", queue size = "
1143 17 : << GetLockedBufferCount();
1144 17 : if (metricMan)
1145 : {
1146 119 : metricMan->sendMetric("Run Number", static_cast<uint64_t>(run_id_), "Run", 1, MetricMode::LastPoint | MetricMode::Persist);
1147 : }
1148 17 : return true;
1149 : }
1150 :
1151 1 : bool artdaq::SharedMemoryEventManager::endRun()
1152 : {
1153 3 : TLOG(TLVL_INFO) << "Ending run " << run_id_;
1154 2 : TLOG(TLVL_ENDRUN) << "Shutting down RequestSender";
1155 1 : requests_.reset(nullptr);
1156 2 : TLOG(TLVL_ENDRUN) << "Shutting down TokenSender";
1157 1 : tokens_.reset(nullptr);
1158 :
1159 3 : TLOG(TLVL_INFO) << "Run " << run_id_ << " has ended. There were " << run_event_count_ << " events in this run.";
1160 1 : run_event_count_ = 0;
1161 1 : run_incomplete_event_count_ = 0;
1162 1 : oversize_fragment_count_ = 0;
1163 : {
1164 1 : std::unique_lock<std::mutex> lk(subrun_event_map_mutex_);
1165 1 : subrun_event_map_.clear();
1166 1 : subrun_event_map_[0] = 1;
1167 1 : }
1168 1 : return true;
1169 : }
1170 :
1171 5 : void artdaq::SharedMemoryEventManager::rolloverSubrun(sequence_id_t boundary, subrun_id_t subrun, bool sendFragment)
1172 : {
1173 : // Generated EndOfSubrun Fragments have Sequence ID 0 and should be ignored
1174 5 : if (boundary == 0 || boundary == Fragment::InvalidSequenceID)
1175 : {
1176 2 : return;
1177 : }
1178 :
1179 5 : std::unique_lock<std::mutex> lk(subrun_event_map_mutex_);
1180 :
1181 : // Don't re-rollover to an already-defined subrun
1182 5 : if (!subrun_event_map_.empty() && subrun_event_map_.rbegin()->second >= subrun)
1183 : {
1184 2 : return;
1185 : }
1186 9 : TLOG(TLVL_INFO) << "Will roll over to subrun " << subrun << " when I reach Sequence ID " << (boundary + 1);
1187 3 : subrun_event_map_[boundary + 1] = subrun;
1188 3 : while (subrun_event_map_.size() > max_subrun_event_map_length_)
1189 : {
1190 0 : subrun_event_map_.erase(subrun_event_map_.begin());
1191 : }
1192 :
1193 3 : if (sendFragment)
1194 : {
1195 0 : auto endOfSubrunFrag = artdaq::MetadataFragment::CreateEndOfSubrunFragment(my_rank, boundary, subrun, 0);
1196 0 : BroadcastFragment(endOfSubrunFrag);
1197 0 : }
1198 5 : }
1199 :
1200 2 : void artdaq::SharedMemoryEventManager::rolloverSubrun(bool sendFragment)
1201 : {
1202 2 : Fragment::sequence_id_t seqID = 0;
1203 2 : subrun_id_t subrun = 0;
1204 : {
1205 2 : std::unique_lock<std::mutex> lk(subrun_event_map_mutex_);
1206 5 : for (auto& it : subrun_event_map_)
1207 : {
1208 3 : if (it.first >= seqID)
1209 : {
1210 3 : seqID = it.first + 1;
1211 : }
1212 3 : if (it.second >= subrun)
1213 : {
1214 3 : subrun = it.second + 1;
1215 : }
1216 : }
1217 2 : }
1218 2 : rolloverSubrun(seqID, subrun, sendFragment);
1219 2 : }
1220 :
1221 0 : void artdaq::SharedMemoryEventManager::sendMetrics()
1222 : {
1223 0 : if (metricMan)
1224 : {
1225 0 : metricMan->sendMetric("Open Event Count", GetOpenEventCount(), "events", 1, MetricMode::LastPoint);
1226 0 : metricMan->sendMetric("Pending Event Count", GetPendingEventCount(), "events", 1, MetricMode::LastPoint);
1227 : }
1228 :
1229 0 : if (open_event_report_interval_ms_ > 0 && GetLockedBufferCount() != 0u)
1230 : {
1231 0 : if (TimeUtils::GetElapsedTimeMilliseconds(last_open_event_report_time_) < static_cast<size_t>(open_event_report_interval_ms_))
1232 : {
1233 0 : return;
1234 : }
1235 :
1236 0 : last_open_event_report_time_ = std::chrono::steady_clock::now();
1237 0 : std::ostringstream oss;
1238 0 : oss << "Open Events (expecting " << num_fragments_per_event_ << " Fragments): ";
1239 0 : for (auto& ev : active_buffers_)
1240 : {
1241 0 : auto hdr = getEventHeader_(ev);
1242 0 : oss << hdr->sequence_id << " (has " << GetFragmentCount(hdr->sequence_id) << " Fragments), ";
1243 : }
1244 0 : TLOG(TLVL_SENDMETRICS) << oss.str();
1245 0 : }
1246 : }
1247 :
1248 6095 : artdaq::detail::RawEventHeader* artdaq::SharedMemoryEventManager::getEventHeader_(int buffer)
1249 : {
1250 6095 : return reinterpret_cast<detail::RawEventHeader*>(GetBufferStart(buffer)); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
1251 : }
1252 :
1253 2565 : artdaq::SharedMemoryEventManager::subrun_id_t artdaq::SharedMemoryEventManager::GetSubrunForSequenceID(Fragment::sequence_id_t seqID)
1254 : {
1255 2565 : subrun_id_t subrun = 1;
1256 2565 : if (init_fragment_count_ > 0)
1257 : {
1258 6 : TLOG(TLVL_GETSUBRUN) << "init_fragment_count > 0 (processing art events): Decoding subrun from sequenceID " << seqID;
1259 3 : subrun = seqID >> 32;
1260 : }
1261 : else
1262 : {
1263 2562 : std::unique_lock<std::mutex> lk(subrun_event_map_mutex_);
1264 :
1265 5124 : TLOG(TLVL_GETSUBRUN) << "GetSubrunForSequenceID BEGIN map size = " << subrun_event_map_.size();
1266 2562 : auto it = subrun_event_map_.begin();
1267 :
1268 5127 : while (it->first <= seqID && it != subrun_event_map_.end())
1269 : {
1270 5130 : TLOG(TLVL_GETSUBRUN) << "Map has sequence ID " << it->first << ", subrun " << it->second << " (looking for <= " << seqID << ")";
1271 2565 : subrun = it->second;
1272 2565 : ++it;
1273 : }
1274 2562 : }
1275 :
1276 5130 : TLOG(TLVL_GETSUBRUN) << "GetSubrunForSequenceID returning subrun " << subrun << " for sequence ID " << seqID;
1277 2565 : return subrun;
1278 : }
1279 :
1280 2537 : artdaq::SharedMemoryEventManager::event_id_t artdaq::SharedMemoryEventManager::GetEventIDForFragment(Fragment::sequence_id_t seqID, Fragment::timestamp_t timestamp)
1281 : {
1282 2537 : event_id_t event = 1;
1283 2537 : if (init_fragment_count_ > 0)
1284 : {
1285 4 : TLOG(TLVL_GETEVENTID) << "init_fragment_count > 0 (processing art events): Decoding event ID from sequenceID " << seqID;
1286 2 : event = seqID & 0xFFFFFFFF;
1287 : }
1288 : else
1289 : {
1290 2535 : sequence_id_t subrun_start = 0;
1291 2535 : if (reset_event_number_for_subruns_)
1292 : {
1293 0 : std::unique_lock<std::mutex> lk(subrun_event_map_mutex_);
1294 0 : TLOG(TLVL_GETEVENTID) << "GetEventIDForFragment BEGIN map size = " << subrun_event_map_.size();
1295 0 : auto it = subrun_event_map_.begin();
1296 0 : while (it->first < seqID && it != subrun_event_map_.end())
1297 : {
1298 0 : TLOG(TLVL_GETEVENTID) << "Map has sequence ID " << it->first << ", event " << it->second << " (looking for <= " << seqID << ")";
1299 0 : subrun_start = it->first;
1300 0 : ++it;
1301 : }
1302 0 : }
1303 :
1304 2535 : event = use_sequence_id_for_event_number_ ? (seqID - subrun_start) : (timestamp - subrun_start);
1305 : }
1306 5074 : TLOG(TLVL_GETEVENTID) << "GetEventIDForFragment returning event ID " << event << " for sequence ID " << seqID;
1307 2537 : return event;
1308 : }
1309 :
1310 9289 : int artdaq::SharedMemoryEventManager::getBufferForSequenceID_(Fragment::sequence_id_t seqID, bool create_new, Fragment::timestamp_t timestamp)
1311 : {
1312 18578 : TLOG(TLVL_GETBUFFER) << "getBufferForSequenceID " << seqID << " BEGIN";
1313 :
1314 9289 : std::unique_lock<std::mutex> lk(sequence_id_mutex_);
1315 :
1316 18578 : TLOG(TLVL_GETBUFFER) << "getBufferForSequenceID obtained sequence_id_mutex for seqid=" << seqID;
1317 :
1318 : #if !ART_SUPPORTS_DUPLICATE_EVENTS
1319 9289 : if (released_incomplete_events_.count(seqID) != 0u)
1320 : {
1321 6 : TLOG(TLVL_ERROR) << "Event " << seqID << " has already been marked \"Incomplete\" and sent to art!";
1322 2 : return -2;
1323 : }
1324 9287 : if (released_events_.count(seqID) != 0u)
1325 : {
1326 6 : TLOG(TLVL_ERROR) << "Event " << seqID << " has already been completed and released to art! Check configuration for inconsistent Fragment count per event!";
1327 2 : return -2;
1328 : }
1329 : #endif
1330 80864 : for (size_t ii = 0; ii < size(); ++ii)
1331 : {
1332 75354 : if (buffer_sequence_ids_[ii] == seqID)
1333 : {
1334 7550 : TLOG(TLVL_GETBUFFER) << "getBufferForSequenceID " << seqID << " returning buffer " << ii << " (already assigned)";
1335 3775 : return ii;
1336 : }
1337 : }
1338 :
1339 5510 : auto buffers = GetBuffersOwnedByManager();
1340 5513 : for (auto& buf : buffers)
1341 : {
1342 3 : auto hdr = getEventHeader_(buf);
1343 3 : buffer_sequence_ids_[buf] = hdr->sequence_id; // Update sequence ID cache
1344 3 : if (hdr->sequence_id == seqID)
1345 : {
1346 0 : TLOG(TLVL_GETBUFFER) << "getBufferForSequenceID " << seqID << " returning " << buf;
1347 0 : return buf;
1348 : }
1349 : }
1350 :
1351 5510 : if (!create_new)
1352 : {
1353 0 : return -1;
1354 : }
1355 :
1356 5510 : check_pending_buffers_(lk);
1357 5510 : int new_buffer = GetBufferForWriting(false, shared_memory_ordering_ ? static_cast<size_t>(seqID) : 0);
1358 :
1359 5510 : if (new_buffer == -1)
1360 : {
1361 5067 : new_buffer = GetBufferForWriting(overwrite_mode_, shared_memory_ordering_ ? static_cast<size_t>(seqID) : 0);
1362 : }
1363 :
1364 5510 : if (new_buffer == -1)
1365 : {
1366 5067 : return -1;
1367 : }
1368 886 : TLOG(TLVL_BUFLCK) << "getBufferForSequenceID_: obtaining buffer_mutexes lock for buffer " << new_buffer;
1369 443 : std::unique_lock<std::mutex> buffer_lk(buffer_mutexes_.at(new_buffer));
1370 886 : TLOG(TLVL_BUFLCK) << "getBufferForSequenceID_: obtained buffer_mutexes lock for buffer " << new_buffer;
1371 :
1372 443 : event_timing_[new_buffer] = std::chrono::steady_clock::now();
1373 :
1374 443 : auto hdr = getEventHeader_(new_buffer);
1375 443 : hdr->is_complete = false;
1376 443 : hdr->run_id = run_id_;
1377 443 : hdr->subrun_id = GetSubrunForSequenceID(seqID);
1378 443 : hdr->event_id = GetEventIDForFragment(seqID, timestamp);
1379 443 : hdr->sequence_id = seqID;
1380 443 : hdr->timestamp = timestamp;
1381 443 : buffer_writes_pending_[new_buffer] = 0;
1382 443 : buffer_sequence_ids_[new_buffer] = seqID;
1383 443 : IncrementWritePos(new_buffer, sizeof(detail::RawEventHeader));
1384 443 : Globals::SetMFIteration("Sequence ID " + std::to_string(seqID));
1385 :
1386 886 : TLOG(TLVL_BUFFER) << "getBufferForSequenceID placing " << new_buffer << " to active.";
1387 443 : active_buffers_.insert(new_buffer);
1388 886 : TLOG(TLVL_BUFFER) << "Buffer occupancy now (total,full,reading,empty,pending,active)=("
1389 0 : << size() << ","
1390 0 : << ReadReadyCount() << ","
1391 0 : << WriteReadyCount(true) - WriteReadyCount(false) - ReadReadyCount() << ","
1392 0 : << WriteReadyCount(false) << ","
1393 0 : << pending_buffers_.size() << ","
1394 443 : << active_buffers_.size() << ")";
1395 :
1396 443 : if (requests_)
1397 : {
1398 443 : requests_->AddRequest(seqID, timestamp);
1399 : }
1400 886 : TLOG(TLVL_GETBUFFER) << "getBufferForSequenceID " << seqID << " returning newly initialized buffer " << new_buffer;
1401 443 : return new_buffer;
1402 9289 : }
1403 :
1404 0 : bool artdaq::SharedMemoryEventManager::hasFragments_(int buffer)
1405 : {
1406 0 : if (buffer == -1)
1407 : {
1408 0 : return true;
1409 : }
1410 0 : if (!CheckBuffer(buffer, BufferSemaphoreFlags::Writing))
1411 : {
1412 0 : return true;
1413 : }
1414 0 : ResetReadPos(buffer);
1415 0 : IncrementReadPos(buffer, sizeof(detail::RawEventHeader));
1416 0 : return MoreDataInBuffer(buffer);
1417 : }
1418 :
1419 3107 : void artdaq::SharedMemoryEventManager::complete_buffer_(int buffer)
1420 : {
1421 3107 : auto hdr = getEventHeader_(buffer);
1422 3107 : if (hdr != nullptr && hdr->is_complete)
1423 : {
1424 876 : TLOG(TLVL_COMPLETEBUFFER) << "complete_buffer_: This fragment completes event " << hdr->sequence_id << ".";
1425 :
1426 : {
1427 876 : TLOG(TLVL_BUFFER) << "complete_buffer_ moving " << buffer << " from active to pending.";
1428 :
1429 876 : TLOG(TLVL_BUFLCK) << "complete_buffer_: obtaining sequence_id_mutex lock for seqid=" << hdr->sequence_id;
1430 438 : std::unique_lock<std::mutex> lk(sequence_id_mutex_);
1431 876 : TLOG(TLVL_BUFLCK) << "complete_buffer_: obtained sequence_id_mutex lock for seqid=" << hdr->sequence_id;
1432 438 : active_buffers_.erase(buffer);
1433 438 : pending_buffers_.insert(buffer);
1434 438 : released_events_.insert(hdr->sequence_id);
1435 538 : while (released_events_.size() > max_event_list_length_)
1436 : {
1437 100 : released_events_.erase(released_events_.begin());
1438 : }
1439 :
1440 876 : TLOG(TLVL_BUFFER) << "Buffer occupancy now (total,full,reading,empty,pending,active)=("
1441 0 : << size() << ","
1442 0 : << ReadReadyCount() << ","
1443 0 : << WriteReadyCount(true) - WriteReadyCount(false) - ReadReadyCount() << ","
1444 0 : << WriteReadyCount(false) << ","
1445 0 : << pending_buffers_.size() << ","
1446 438 : << active_buffers_.size() << ")";
1447 438 : check_pending_buffers_(lk);
1448 438 : }
1449 438 : if (requests_)
1450 : {
1451 438 : requests_->RemoveRequest(hdr->sequence_id);
1452 : }
1453 : }
1454 3107 : check_pending_broadcasts_();
1455 3107 : }
1456 :
1457 0 : bool artdaq::SharedMemoryEventManager::bufferComparator(int bufA, int bufB)
1458 : {
1459 0 : return getEventHeader_(bufA) < getEventHeader_(bufB);
1460 : }
1461 :
1462 10 : void artdaq::SharedMemoryEventManager::CheckPendingBuffers()
1463 : {
1464 : {
1465 20 : TLOG(TLVL_BUFLCK) << "Obtaining sequence_id_mutex_";
1466 10 : std::unique_lock<std::mutex> lk(sequence_id_mutex_);
1467 20 : TLOG(TLVL_BUFLCK) << "Obtained sequence_id_mutex_";
1468 :
1469 10 : check_pending_buffers_(lk);
1470 10 : }
1471 10 : check_pending_broadcasts_();
1472 10 : }
1473 :
1474 5962 : void artdaq::SharedMemoryEventManager::check_pending_buffers_(std::unique_lock<std::mutex> const& lock)
1475 : {
1476 11924 : TLOG(TLVL_CHECKPENDINGBUFFERS) << "check_pending_buffers_ BEGIN Locked=" << std::boolalpha << lock.owns_lock();
1477 :
1478 5962 : auto buffers = GetBuffersOwnedByManager();
1479 6419 : for (auto buf : buffers)
1480 : {
1481 457 : if (ResetBuffer(buf) && (pending_buffers_.count(buf) == 0u))
1482 : {
1483 2 : TLOG(TLVL_CHECKPENDINGBUFFERS) << "check_pending_buffers_ Incomplete buffer detected, buf=" << buf << " active_bufers_.count(buf)=" << active_buffers_.count(buf) << " buffer_writes_pending_[buf]=" << buffer_writes_pending_[buf].load();
1484 1 : auto hdr = getEventHeader_(buf);
1485 2 : if ((active_buffers_.count(buf) != 0u) && buffer_writes_pending_[buf].load() == 0)
1486 : {
1487 1 : if (requests_)
1488 : {
1489 1 : requests_->RemoveRequest(hdr->sequence_id);
1490 : }
1491 2 : TLOG(TLVL_BUFFER) << "check_pending_buffers_ moving buffer " << buf << " from active to pending";
1492 1 : active_buffers_.erase(buf);
1493 1 : pending_buffers_.insert(buf);
1494 2 : TLOG(TLVL_BUFFER) << "Buffer occupancy now (total,full,reading,empty,pending,active)=("
1495 0 : << size() << ","
1496 0 : << ReadReadyCount() << ","
1497 0 : << WriteReadyCount(true) - WriteReadyCount(false) - ReadReadyCount() << ","
1498 0 : << WriteReadyCount(false) << ","
1499 0 : << pending_buffers_.size() << ","
1500 1 : << active_buffers_.size() << ")";
1501 :
1502 1 : run_incomplete_event_count_++;
1503 1 : if (metricMan)
1504 : {
1505 7 : metricMan->sendMetric("Incomplete Event Rate", 1, "events/s", 3, MetricMode::Rate);
1506 : }
1507 1 : if (released_incomplete_events_.count(hdr->sequence_id) == 0u)
1508 : {
1509 1 : released_incomplete_events_[hdr->sequence_id] = num_fragments_per_event_ - GetFragmentCountInBuffer(buf);
1510 : }
1511 : else
1512 : {
1513 0 : released_incomplete_events_[hdr->sequence_id] -= GetFragmentCountInBuffer(buf);
1514 : }
1515 :
1516 3 : TLOG(TLVL_WARNING) << "Event " << hdr->sequence_id
1517 2 : << " was opened " << TimeUtils::GetElapsedTime(event_timing_[buf]) << " s ago"
1518 1 : << " and has timed out (missing " << released_incomplete_events_[hdr->sequence_id] << " Fragments)."
1519 2 : << "Scheduling release to art.";
1520 : }
1521 : }
1522 : }
1523 :
1524 5962 : std::list<int> sorted_buffers(pending_buffers_.begin(), pending_buffers_.end());
1525 5962 : sorted_buffers.sort([this](int a, int b) { return bufferComparator(a, b); });
1526 :
1527 5962 : auto available_buffers = WriteReadyCount(overwrite_mode_);
1528 5962 : auto counter = 0;
1529 5962 : double eventSize = 0;
1530 5962 : double eventTime = 0;
1531 6405 : for (auto buf : sorted_buffers)
1532 : {
1533 443 : auto hdr = getEventHeader_(buf);
1534 443 : auto thisEventSize = BufferDataSize(buf);
1535 :
1536 443 : bool currentSubrun = hdr->subrun_id == subrun_id_;
1537 :
1538 443 : if (hdr->subrun_id > subrun_id_ && available_buffers > 0 && TimeUtils::GetElapsedTime(last_event_time_) < subrun_transition_hold_time_s_)
1539 : {
1540 0 : TLOG(TLVL_CHECKPENDINGBUFFERS_4) << "Holding event " << std::to_string(hdr->sequence_id) << " (sr=" << hdr->subrun_id << ") in buffer " << buf << ", "
1541 0 : << "event_size=" << thisEventSize << ", buffer_size=" << BufferSize();
1542 0 : continue;
1543 0 : }
1544 :
1545 886 : TLOG(TLVL_CHECKPENDINGBUFFERS_4) << "Releasing event " << std::to_string(hdr->sequence_id) << " (sr=" << hdr->subrun_id << ") in buffer " << buf << " to art, "
1546 443 : << "event_size=" << thisEventSize << ", buffer_size=" << BufferSize();
1547 443 : statsHelper_.addSample(EVENTS_RELEASED_STAT_KEY, thisEventSize);
1548 :
1549 886 : TLOG(TLVL_BUFFER) << "check_pending_buffers_ removing buffer " << buf << " moving from pending to full";
1550 443 : MarkBufferFull(buf);
1551 443 : run_event_count_++;
1552 443 : counter++;
1553 443 : eventSize += thisEventSize;
1554 443 : eventTime += TimeUtils::GetElapsedTime(event_timing_[buf]);
1555 443 : pending_buffers_.erase(buf);
1556 443 : if (currentSubrun)
1557 : {
1558 442 : last_event_time_ = std::chrono::steady_clock::now();
1559 : }
1560 : }
1561 11924 : TLOG(TLVL_BUFFER) << "Buffer occupancy now (total,full,reading,empty,pending,active)=("
1562 0 : << size() << ","
1563 0 : << ReadReadyCount() << ","
1564 0 : << WriteReadyCount(true) - WriteReadyCount(false) - ReadReadyCount() << ","
1565 0 : << WriteReadyCount(false) << ","
1566 0 : << pending_buffers_.size() << ","
1567 5962 : << active_buffers_.size() << ")";
1568 :
1569 5962 : if (tokens_ && tokens_->RoutingTokenSendsEnabled())
1570 : {
1571 0 : TLOG(TLVL_CHECKPENDINGBUFFERS_3) << "Sent tokens: " << tokens_->GetSentTokenCount() << ", Event count: " << run_event_count_;
1572 0 : auto outstanding_tokens = tokens_->GetSentTokenCount() - run_event_count_;
1573 :
1574 0 : TLOG(TLVL_CHECKPENDINGBUFFERS_3) << "check_pending_buffers_: outstanding_tokens: " << outstanding_tokens << ", available_buffers: " << available_buffers
1575 0 : << ", tokens_to_send: " << available_buffers - outstanding_tokens;
1576 :
1577 0 : if (available_buffers > outstanding_tokens)
1578 : {
1579 0 : auto tokens_to_send = available_buffers - outstanding_tokens;
1580 :
1581 0 : while (tokens_to_send > 0)
1582 : {
1583 0 : TLOG(TLVL_CHECKPENDINGBUFFERS_3) << "check_pending_buffers_: Sending a Routing Token";
1584 0 : tokens_->SendRoutingToken(1, run_id_);
1585 0 : tokens_to_send--;
1586 : }
1587 : }
1588 : }
1589 :
1590 5962 : if (statsHelper_.readyToReport())
1591 : {
1592 0 : std::string statString = buildStatisticsString_();
1593 0 : TLOG(TLVL_INFO) << statString;
1594 0 : }
1595 :
1596 5962 : if (metricMan)
1597 : {
1598 11924 : TLOG(TLVL_CHECKPENDINGBUFFERS_2) << "check_pending_buffers_: Sending Metrics";
1599 35772 : metricMan->sendMetric("Event Rate", counter, "Events", 1, MetricMode::Rate);
1600 35772 : metricMan->sendMetric("Data Rate", eventSize, "Bytes", 1, MetricMode::Rate);
1601 5962 : if (counter > 0)
1602 : {
1603 2658 : metricMan->sendMetric("Average Event Size", eventSize / counter, "Bytes", 1, MetricMode::Average);
1604 3101 : metricMan->sendMetric("Average Event Building Time", eventTime / counter, "s", 1, MetricMode::Average);
1605 : }
1606 :
1607 35772 : metricMan->sendMetric("Events Released to art this run", run_event_count_, "Events", 1, MetricMode::LastPoint);
1608 35772 : metricMan->sendMetric("Incomplete Events Released to art this run", run_incomplete_event_count_, "Events", 1, MetricMode::LastPoint);
1609 5962 : if (tokens_ && tokens_->RoutingTokenSendsEnabled())
1610 : {
1611 0 : metricMan->sendMetric("Tokens sent", tokens_->GetSentTokenCount(), "Tokens", 2, MetricMode::LastPoint);
1612 : }
1613 :
1614 5962 : auto bufferReport = GetBufferReport();
1615 5962 : int full = 0, empty = 0, writing = 0, reading = 0;
1616 65428 : for (auto& buf : bufferReport)
1617 : {
1618 59466 : switch (buf.second)
1619 : {
1620 54817 : case BufferSemaphoreFlags::Full:
1621 54817 : full++;
1622 54817 : break;
1623 4548 : case BufferSemaphoreFlags::Empty:
1624 4548 : empty++;
1625 4548 : break;
1626 14 : case BufferSemaphoreFlags::Writing:
1627 14 : writing++;
1628 14 : break;
1629 87 : case BufferSemaphoreFlags::Reading:
1630 87 : reading++;
1631 87 : break;
1632 : }
1633 : }
1634 5962 : auto total = size();
1635 11924 : TLOG(TLVL_CHECKPENDINGBUFFERS_2) << "Buffer usage: full=" << full << ", empty=" << empty << ", writing=" << writing << ", reading=" << reading << ", total=" << total;
1636 :
1637 35772 : metricMan->sendMetric("Shared Memory Full Buffers", full, "buffers", 2, MetricMode::LastPoint);
1638 35772 : metricMan->sendMetric("Shared Memory Available Buffers", empty, "buffers", 2, MetricMode::LastPoint);
1639 35772 : metricMan->sendMetric("Shared Memory Pending Buffers", writing, "buffers", 2, MetricMode::LastPoint);
1640 35772 : metricMan->sendMetric("Shared Memory Reading Buffers", reading, "buffers", 2, MetricMode::LastPoint);
1641 5962 : if (total > 0)
1642 : {
1643 35772 : metricMan->sendMetric("Shared Memory Full %", full * 100 / static_cast<double>(total), "%", 2, MetricMode::LastPoint);
1644 41734 : metricMan->sendMetric("Shared Memory Available %", empty * 100 / static_cast<double>(total), "%", 2, MetricMode::LastPoint);
1645 : }
1646 5962 : }
1647 11924 : TLOG(TLVL_CHECKPENDINGBUFFERS) << "check_pending_buffers_ END";
1648 5962 : }
1649 :
1650 0 : void artdaq::SharedMemoryEventManager::BroadcastFragment(FragmentPtr& frag)
1651 : {
1652 : {
1653 0 : std::lock_guard<std::mutex> lk(broadcast_mutex_);
1654 :
1655 0 : bool entry_found = false;
1656 0 : for (auto& entry : pending_broadcasts_)
1657 : {
1658 0 : if (entry.type == frag->type() && entry.sequence_id == frag->sequenceID())
1659 : {
1660 0 : TLOG(TLVL_BROADCASTFRAGMENT) << "Received BroadcastFragment of type " << static_cast<int>(frag->type()) << ", seqID " << frag->sequenceID() << " matching current pending_broadcasts_ entry. frags=" << entry.fragments.size() + 1 << "/" << init_fragment_count_;
1661 0 : entry.fragments.push_back(std::move(frag));
1662 0 : entry_found = true;
1663 0 : break;
1664 : }
1665 : }
1666 0 : if (!entry_found)
1667 : {
1668 0 : TLOG(TLVL_BROADCASTFRAGMENT) << "Received BroadcastFragment of type " << static_cast<int>(frag->type()) << ", seqID " << frag->sequenceID() << ", creating new pending_broadcasts_ entry";
1669 0 : pending_broadcasts_.emplace_back();
1670 0 : pending_broadcasts_.back().deadline = std::chrono::steady_clock::now() + std::chrono::microseconds(GetBufferTimeout());
1671 0 : pending_broadcasts_.back().type = frag->type();
1672 0 : pending_broadcasts_.back().sequence_id = frag->sequenceID();
1673 0 : pending_broadcasts_.back().subrun_id = GetSubrunForSequenceID(frag->sequenceID());
1674 0 : pending_broadcasts_.back().fragments.push_back(std::move(frag));
1675 : }
1676 0 : }
1677 0 : check_pending_broadcasts_();
1678 0 : }
1679 :
1680 3117 : void artdaq::SharedMemoryEventManager::check_pending_broadcasts_()
1681 : {
1682 3117 : std::lock_guard<std::mutex> lk(broadcast_mutex_);
1683 :
1684 3117 : auto entry = pending_broadcasts_.begin();
1685 3117 : auto now = std::chrono::steady_clock::now();
1686 3117 : while (entry != pending_broadcasts_.end())
1687 : {
1688 0 : if (running_)
1689 : {
1690 0 : if ((!init_frags_sent_ || entry->fragments.size() == 0 || (entry->fragments.size() < init_fragment_count_ && now < entry->deadline)))
1691 : {
1692 0 : entry++;
1693 0 : continue;
1694 : }
1695 0 : if (entry->fragments.front()->type() == Fragment::EndOfSubrunFragmentType || entry->fragments.front()->type() == artdaq::Fragment::SubrunDataFragmentType)
1696 : {
1697 0 : if (entry->subrun_id == subrun_id_ && TimeUtils::GetElapsedTime(last_event_time_) < subrun_transition_hold_time_s_)
1698 : {
1699 0 : TLOG(TLVL_CHECKPENDINGBROADCASTS) << "Holding entry size = " << entry->fragments.size() << " / " << init_fragment_count_ << ", lead SeqID = " << entry->fragments.front()->sequenceID() << ", subrun=" << entry->subrun_id << " because it is EndOfSubrun and hold time has not expired";
1700 0 : entry++;
1701 0 : continue;
1702 0 : }
1703 : }
1704 : }
1705 :
1706 0 : TLOG(TLVL_CHECKPENDINGBROADCASTS) << "Broadcasting entry init_frags_sent_=" << init_frags_sent_ << ", size=" << entry->fragments.size() << "/" << init_fragment_count_ << ", subrun=" << entry->subrun_id << ", lead SeqID=" << entry->fragments.front()->sequenceID() << " deadline delta=" << std::chrono::duration_cast<std::chrono::microseconds>(now - entry->deadline).count() << " us";
1707 0 : broadcastFragments_(entry->fragments);
1708 0 : entry = pending_broadcasts_.erase(entry);
1709 : }
1710 3117 : }
1711 :
1712 22 : bool artdaq::SharedMemoryEventManager::broadcastFragments_(FragmentPtrs& frags)
1713 : {
1714 22 : if (frags.empty())
1715 : {
1716 0 : TLOG(TLVL_ERROR) << "Requested broadcast but no Fragments given!";
1717 0 : return false;
1718 : }
1719 22 : if (!broadcasts_.IsValid())
1720 : {
1721 0 : TLOG(TLVL_ERROR) << "Broadcast attempted but broadcast shared memory is unavailable!";
1722 0 : return false;
1723 : }
1724 44 : TLOG(TLVL_BROADCASTFRAGMENTS) << "Broadcasting " << frags.size() << " Fragments with lead seqID=" << frags.front()->sequenceID()
1725 0 : << ", type " << detail::RawFragmentHeader::SystemTypeToString(frags.front()->type())
1726 22 : << ", size=" << frags.front()->sizeBytes() << "B.";
1727 22 : auto buffer = broadcasts_.GetBufferForWriting(false);
1728 44 : TLOG(TLVL_BROADCASTFRAGMENTS) << "broadcastFragments_: after getting buffer 1st buffer=" << buffer;
1729 22 : auto start_time = std::chrono::steady_clock::now();
1730 22 : while (buffer == -1 && TimeUtils::GetElapsedTimeMilliseconds(start_time) < static_cast<size_t>(broadcast_timeout_ms_))
1731 : {
1732 0 : usleep(10000);
1733 0 : buffer = broadcasts_.GetBufferForWriting(true); // Go into overwrite mode
1734 : }
1735 44 : TLOG(TLVL_BROADCASTFRAGMENTS) << "broadcastFragments_: after getting buffer w/timeout, buffer=" << buffer << ", elapsed time=" << TimeUtils::GetElapsedTime(start_time) << " s.";
1736 22 : if (buffer == -1)
1737 : {
1738 0 : TLOG(TLVL_ERROR) << "Broadcast of fragment type " << frags.front()->typeString() << " failed due to timeout waiting for buffer!";
1739 0 : return false;
1740 : }
1741 :
1742 44 : TLOG(TLVL_BROADCASTFRAGMENTS) << "broadcastFragments_: Filling in RawEventHeader";
1743 22 : auto hdr = reinterpret_cast<detail::RawEventHeader*>(broadcasts_.GetBufferStart(buffer)); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
1744 22 : hdr->run_id = run_id_;
1745 22 : hdr->subrun_id = GetSubrunForSequenceID(frags.front()->sequenceID());
1746 22 : hdr->sequence_id = frags.front()->sequenceID();
1747 22 : hdr->is_complete = true;
1748 22 : broadcasts_.IncrementWritePos(buffer, sizeof(detail::RawEventHeader));
1749 :
1750 22 : if (frags.front()->type() == artdaq::Fragment::EndOfSubrunFragmentType || frags.front()->type() == artdaq::Fragment::SubrunDataFragmentType)
1751 : {
1752 0 : subrun_id_ = hdr->subrun_id + 1;
1753 : }
1754 :
1755 44 : for (auto& frag : frags)
1756 : {
1757 22 : if (frag->sequenceID() != hdr->sequence_id || frag->type() != frags.front()->type())
1758 : {
1759 0 : TLOG(TLVL_WARNING) << "Skipping fragment due to Type/SeqID mismatch! seqID=" << frag->sequenceID() << " (expected " << hdr->sequence_id << "), type=" << static_cast<int>(frag->type()) << " (" << static_cast<int>(frags.front()->type()) << ")";
1760 0 : continue;
1761 0 : }
1762 44 : TLOG(TLVL_BROADCASTFRAGMENTS_2) << "broadcastFragments_ before Fragment Write call seqID=" << frag->sequenceID() << ", fragID=" << frag->fragmentID() << ", type=" << static_cast<int>(frag->type());
1763 22 : broadcasts_.Write(buffer, frag->headerAddress(), frag->size() * sizeof(RawDataType));
1764 : }
1765 :
1766 44 : TLOG(TLVL_BROADCASTFRAGMENTS) << "broadcastFragments_ Marking buffer full";
1767 22 : broadcasts_.MarkBufferFull(buffer, -1);
1768 44 : TLOG(TLVL_BROADCASTFRAGMENTS) << "broadcastFragments_ Complete";
1769 22 : return true;
1770 : }
1771 :
1772 0 : std::vector<char*> artdaq::SharedMemoryEventManager::parse_art_command_line_(const std::shared_ptr<art_config_file>& config_file, size_t process_index)
1773 : {
1774 0 : auto offset_index = process_index + art_process_index_offset_;
1775 0 : TLOG(TLVL_PARSEARTCOMMANDLINE) << "parse_art_command_line_: Parsing command line " << art_cmdline_ << ", config_file: " << config_file->getFileName() << ", index: " << process_index << " (w/offset: " << offset_index << ")";
1776 0 : std::string art_cmdline_tmp = art_cmdline_;
1777 0 : auto filenameit = art_cmdline_tmp.find("#CONFIG_FILE#");
1778 0 : if (filenameit != std::string::npos)
1779 : {
1780 0 : art_cmdline_tmp.replace(filenameit, 13, config_file->getFileName());
1781 : }
1782 0 : auto indexit = art_cmdline_tmp.find("#PROCESS_INDEX#");
1783 0 : if (indexit != std::string::npos)
1784 : {
1785 0 : art_cmdline_tmp.replace(indexit, 15, std::to_string(offset_index));
1786 : }
1787 0 : TLOG(TLVL_PARSEARTCOMMANDLINE) << "parse_art_command_line_: After replacing index and config parameters, command line is " << art_cmdline_tmp;
1788 :
1789 0 : std::istringstream iss(art_cmdline_tmp);
1790 0 : auto tokens = std::vector<std::string>{std::istream_iterator<std::string>{iss}, std::istream_iterator<std::string>{}};
1791 0 : std::vector<char*> output;
1792 :
1793 0 : for (auto& token : tokens)
1794 : {
1795 0 : TLOG(TLVL_PARSEARTCOMMANDLINE) << "parse_art_command_line_: Adding cmdline token " << token << " to output list";
1796 0 : output.emplace_back(new char[token.length() + 1]);
1797 0 : memcpy(output.back(), token.c_str(), token.length());
1798 0 : output.back()[token.length()] = '\0'; // NOLINT(cppcoreguidelines-pro-bounds-pointer-arithmetic)
1799 : }
1800 0 : output.emplace_back(nullptr);
1801 :
1802 0 : return output;
1803 0 : }
1804 :
1805 6 : void artdaq::SharedMemoryEventManager::send_init_frags_()
1806 : {
1807 6 : std::lock_guard<std::mutex> lk(init_fragments_mutex_);
1808 6 : if (init_fragment_map_size_() >= init_fragment_count_ && init_fragment_count_ > 0)
1809 : {
1810 0 : TLOG(TLVL_INFO) << "Broadcasting " << init_fragment_map_size_() << " Init Fragment(s) to all art subprocesses...";
1811 :
1812 0 : FragmentPtrs init_fragments;
1813 0 : for (auto& fragment_id_pair : init_fragment_map_)
1814 : {
1815 0 : for (auto& ts_pair : fragment_id_pair.second)
1816 : {
1817 0 : init_fragments.emplace_back(std::make_unique<Fragment>(*ts_pair.second));
1818 : }
1819 : }
1820 :
1821 0 : broadcastFragments_(init_fragments);
1822 0 : TLOG(TLVL_SENDINIT) << "Init Fragment sent";
1823 0 : init_frags_sent_ = true;
1824 0 : }
1825 6 : else if (init_fragment_count_ > 0 && init_fragment_map_size_() == 0)
1826 : {
1827 0 : TLOG(TLVL_INFO) << "Cannot send Init Fragment(s) because I haven't yet received any! Set send_init_fragments to false or init_fragment_count to 0 if this process does not receive serialized art events to avoid potentially lengthy timeouts!";
1828 : }
1829 6 : else if (init_fragment_count_ > 0)
1830 : {
1831 0 : TLOG(TLVL_INFO) << "Cannot send Init Fragment(s) because I haven't yet received them (have " << init_fragment_map_size_() << " of " << init_fragment_count_ << ")!";
1832 : }
1833 : else
1834 : {
1835 : // Send an empty Init Fragment so that ArtdaqInput knows that this is a pure-Fragment input
1836 6 : artdaq::FragmentPtrs begin_run_fragments_;
1837 6 : begin_run_fragments_.emplace_back(new artdaq::Fragment());
1838 6 : begin_run_fragments_.back()->setSystemType(artdaq::Fragment::InitFragmentType);
1839 6 : broadcastFragments_(begin_run_fragments_);
1840 6 : init_frags_sent_ = true;
1841 6 : }
1842 6 : }
1843 :
1844 0 : void artdaq::SharedMemoryEventManager::AddInitFragment(FragmentPtr& frag)
1845 : {
1846 0 : std::unique_lock<std::mutex> lk(init_fragments_mutex_);
1847 :
1848 0 : auto fragId = frag->fragmentID();
1849 0 : auto ts = frag->timestamp();
1850 :
1851 0 : init_fragment_map_[fragId][ts] = std::move(frag);
1852 0 : TLOG(TLVL_ADDINITFRAGMENT) << "Received Init Fragment from rank " << fragId << ", art process id " << ts << ". Now have " << init_fragment_map_size_() << " of " << init_fragment_count_;
1853 :
1854 : // Don't send until all init fragments have been received
1855 0 : if (init_fragment_map_size_() >= init_fragment_count_)
1856 : {
1857 0 : lk.unlock();
1858 0 : send_init_frags_();
1859 : }
1860 0 : }
1861 :
1862 6 : size_t artdaq::SharedMemoryEventManager::init_fragment_map_size_() const
1863 : {
1864 6 : size_t size = 0;
1865 :
1866 6 : for (auto& frag_id_pair : init_fragment_map_)
1867 : {
1868 0 : size += frag_id_pair.second.size();
1869 : }
1870 :
1871 6 : return size;
1872 : }
1873 :
1874 0 : void artdaq::SharedMemoryEventManager::UpdateArtConfiguration(fhicl::ParameterSet art_pset)
1875 : {
1876 0 : TLOG(TLVL_UPDATEARTCONFIG) << "UpdateArtConfiguration BEGIN";
1877 0 : if (art_pset != current_art_pset_ || !current_art_config_file_)
1878 : {
1879 0 : current_art_pset_ = art_pset;
1880 0 : if (manual_art_)
1881 0 : current_art_config_file_ = std::make_shared<art_config_file>(art_pset, GetKey(), GetBroadcastKey());
1882 : else
1883 0 : current_art_config_file_ = std::make_shared<art_config_file>(art_pset);
1884 : }
1885 0 : TLOG(TLVL_UPDATEARTCONFIG) << "UpdateArtConfiguration END";
1886 0 : }
1887 :
1888 0 : std::string artdaq::SharedMemoryEventManager::buildStatisticsString_() const
1889 : {
1890 0 : std::ostringstream oss;
1891 0 : oss << app_name << " statistics:" << std::endl;
1892 :
1893 : artdaq::MonitoredQuantityPtr mqPtr =
1894 0 : artdaq::StatisticsCollection::getInstance().getMonitoredQuantity(EVENTS_RELEASED_STAT_KEY);
1895 0 : if (mqPtr.get() != nullptr)
1896 : {
1897 0 : artdaq::MonitoredQuantityStats stats;
1898 0 : mqPtr->getStats(stats);
1899 0 : oss << " Event statistics: " << stats.recentSampleCount << " events released at " << stats.recentSampleRate
1900 0 : << " events/sec, effective data rate = "
1901 0 : << (stats.recentValueRate / 1024.0 / 1024.0)
1902 0 : << " MB/sec, monitor window = " << stats.recentDuration
1903 0 : << " sec, min::max event size = " << (stats.recentValueMin / 1024.0 / 1024.0)
1904 0 : << "::" << (stats.recentValueMax / 1024.0 / 1024.0) << " MB" << std::endl;
1905 0 : if (stats.recentSampleRate > 0.0)
1906 : {
1907 0 : oss << " Average time per event: ";
1908 0 : oss << " elapsed time = " << (1.0 / stats.recentSampleRate) << " sec" << std::endl;
1909 : }
1910 0 : }
1911 :
1912 0 : mqPtr = artdaq::StatisticsCollection::getInstance().getMonitoredQuantity(FRAGMENTS_RECEIVED_STAT_KEY);
1913 0 : if (mqPtr.get() != nullptr)
1914 : {
1915 0 : artdaq::MonitoredQuantityStats stats;
1916 0 : mqPtr->getStats(stats);
1917 0 : oss << " Fragment statistics: " << stats.recentSampleCount << " fragments received at " << stats.recentSampleRate
1918 0 : << " fragments/sec, effective data rate = "
1919 0 : << (stats.recentValueRate / 1024.0 / 1024.0)
1920 0 : << " MB/sec, monitor window = " << stats.recentDuration
1921 0 : << " sec, min::max fragment size = " << (stats.recentValueMin / 1024.0 / 1024.0)
1922 0 : << "::" << (stats.recentValueMax / 1024.0 / 1024.0) << " MB" << std::endl;
1923 0 : }
1924 :
1925 0 : oss << " Event counts: Run -- " << run_event_count_ << " Total, " << run_incomplete_event_count_ << " Incomplete."
1926 0 : << " Subrun -- " << subrun_event_count_ << " Total, " << subrun_incomplete_event_count_ << " Incomplete. "
1927 0 : << std::endl;
1928 : //-----------------------------------------------------------------------------
1929 : // P.Murat: add statistics on the SHM buffers
1930 : // there are 4 different flags: 0:empty, 1:writing; 2:full 3:reading
1931 : // want statistics on all of them
1932 : //-----------------------------------------------------------------------------
1933 : // auto = std::vector<std::pair<int, artdaq::SharedMemoryManager::BufferSemaphoreFlags>>
1934 0 : artdaq::SharedMemoryEventManager* nc_this = (artdaq::SharedMemoryEventManager*)this;
1935 :
1936 0 : int bsize = nc_this->BufferSize();
1937 :
1938 0 : auto v = nc_this->GetBufferReport();
1939 :
1940 0 : int nbb[4] = {0, 0, 0, 0};
1941 :
1942 0 : int nbuff = v.size();
1943 0 : for (int i = 0; i < nbuff; i++)
1944 : {
1945 0 : auto x = v[i];
1946 0 : int flag = (int)x.second;
1947 0 : nbb[flag]++;
1948 : }
1949 :
1950 0 : oss << " Shared Memory: "
1951 0 : << nbuff << " buffers of " << bsize << " B, "
1952 0 : << nbb[0] << " Empty, " << nbb[1] << " Writing, " << nbb[2] << " Full, " << nbb[3] << " reading"
1953 0 : << std::endl;
1954 :
1955 0 : return oss.str();
1956 0 : }
|
