Line data Source code
1 : #ifndef ARTDAQ_DAQRATE_SHAREDMEMORYEVENTMANAGER_HH
2 : #define ARTDAQ_DAQRATE_SHAREDMEMORYEVENTMANAGER_HH
3 :
4 : #include "TRACE/tracemf.h" // Pre-empt TRACE/trace.h from Fragment.hh.
5 : #include "artdaq-core/Data/Fragment.hh"
6 :
7 : #include "artdaq-core/Core/SharedMemoryManager.hh"
8 : #include "artdaq-core/Data/RawEvent.hh"
9 : #include "artdaq-core/Utilities/configureMessageFacility.hh"
10 : #include "artdaq/DAQrate/StatisticsHelper.hh"
11 : #include "artdaq/DAQrate/detail/RequestSender.hh"
12 : #include "artdaq/DAQrate/detail/TokenSender.hh"
13 :
14 : #include "fhiclcpp/types/Atom.h"
15 : #include "fhiclcpp/types/Comment.h"
16 : #include "fhiclcpp/types/ConfigurationTable.h"
17 : #include "fhiclcpp/types/OptionalTable.h"
18 : #include "fhiclcpp/types/TableFragment.h"
19 :
20 : #define ART_SUPPORTS_DUPLICATE_EVENTS 0
21 :
22 : #include <sys/stat.h>
23 : #include <deque>
24 : #include <fstream>
25 : #include <iomanip>
26 : #include <set>
27 :
28 : namespace artdaq {
29 :
30 : /**
31 : * \brief art_config_file wraps a temporary file used to configure art
32 : */
33 : class art_config_file
34 : {
35 : public:
36 : /**
37 : * \brief art_config_file Constructor
38 : * \param ps ParameterSet to write to temporary file
39 : * \param shm_key Shared Memory key to use (if 0, child program will use parent PID to generate)
40 : * \param broadcast_key Shared Memory key to use for broadcasts (if 0, child program will use parent PID to generate)
41 : */
42 16 : explicit art_config_file(fhicl::ParameterSet ps, uint32_t shm_key = 0, uint32_t broadcast_key = 0)
43 16 : : dir_name_("/tmp/partition_" + std::to_string(Globals::GetPartitionNumber()))
44 16 : , file_name_(dir_name_ + "/artConfig_" + std::to_string(my_rank) + "_" + std::to_string(artdaq::TimeUtils::gettimeofday_us()) + ".fcl")
45 : {
46 16 : mkdir(dir_name_.c_str(), 0777); // Allowed to fail if directory already exists
47 :
48 16 : std::ofstream of(file_name_, std::ofstream::trunc);
49 16 : if (of.fail())
50 : {
51 : // Probably a permissions error...
52 0 : dir_name_ = "/tmp/partition_" + std::to_string(Globals::GetPartitionNumber()) + "_" + std::to_string(getuid());
53 0 : mkdir(dir_name_.c_str(), 0777); // Allowed to fail if directory already exists
54 0 : file_name_ = dir_name_ + "/artConfig_" + std::to_string(my_rank) + "_" + std::to_string(artdaq::TimeUtils::gettimeofday_us()) + ".fcl";
55 :
56 0 : of.open(file_name_, std::ofstream::trunc);
57 0 : if (of.fail())
58 : {
59 0 : TLOG(TLVL_ERROR, "ArtConfigFile") << "Failed to open configuration file after two attemps! ABORTING!";
60 0 : exit(46);
61 : }
62 : }
63 16 : of << ps.to_string();
64 :
65 60 : if (ps.has_key("services") && ps.has_key("services.message"))
66 : {
67 0 : auto existing_message_config = ps.get<fhicl::ParameterSet>("services.message");
68 0 : auto existing_destinations = existing_message_config.get<fhicl::ParameterSet>("destinations");
69 0 : auto generated_message_config = generateMessageFacilityConfiguration(mf::GetApplicationName().c_str(), true, false, "-art");
70 0 : auto generated_message_pset = fhicl::ParameterSet::make(generated_message_config);
71 0 : auto generated_destinations = generated_message_pset.get<fhicl::ParameterSet>("destinations");
72 0 : for (auto& dest : generated_destinations.get_pset_names())
73 : {
74 0 : existing_destinations.put(dest, generated_destinations.get<fhicl::ParameterSet>(dest));
75 0 : }
76 0 : existing_message_config.put_or_replace("destinations", existing_destinations);
77 0 : of << " services.message: { " + existing_message_config.to_string() << "} ";
78 0 : }
79 : else
80 : {
81 16 : of << " services.message: { " << generateMessageFacilityConfiguration(mf::GetApplicationName().c_str(), true, false, "-art") << "} ";
82 : }
83 :
84 16 : if (shm_key > 0 || broadcast_key > 0) TLOG(TLVL_INFO, "ArtConfigFile") << "Inserting Shared memory keys (0x" << std::hex << shm_key << ", 0x" << std::hex << broadcast_key << ") into source config";
85 16 : if (shm_key > 0) of << " source.shared_memory_key: 0x" << std::hex << shm_key;
86 16 : if (broadcast_key > 0) of << " source.broadcast_shared_memory_key: 0x" << std::hex << broadcast_key;
87 :
88 16 : of.flush();
89 16 : of.close();
90 16 : }
91 16 : ~art_config_file()
92 : {
93 16 : remove(file_name_.c_str());
94 16 : rmdir(dir_name_.c_str()); // Will only delete directory if no config files are left over
95 16 : }
96 : /**
97 : * \brief Get the path of the temporary file
98 : * \return The path of the temporary file
99 : */
100 6 : std::string getFileName() const { return file_name_; }
101 :
102 : private:
103 : art_config_file(art_config_file const&) = delete;
104 : art_config_file(art_config_file&&) = delete;
105 : art_config_file& operator=(art_config_file const&) = delete;
106 : art_config_file& operator=(art_config_file&&) = delete;
107 :
108 : std::string dir_name_;
109 : std::string file_name_;
110 : };
111 :
112 : /**
113 : * \brief The SharedMemoryEventManager is a SharedMemoryManger which tracks events as they are built
114 : */
115 : class SharedMemoryEventManager : public SharedMemoryManager
116 : {
117 : public:
118 : static const std::string FRAGMENTS_RECEIVED_STAT_KEY; ///< Key for Fragments Received MonitoredQuantity
119 : static const std::string EVENTS_RELEASED_STAT_KEY; ///< Key for the Events Released MonitoredQuantity
120 :
121 : typedef RawEvent::run_id_t run_id_t; ///< Copy RawEvent::run_id_t into local scope
122 : typedef RawEvent::subrun_id_t subrun_id_t; ///< Copy RawEvent::subrun_id_t into local scope
123 : typedef RawEvent::event_id_t event_id_t; ///< Copy RawEvent::event_id_t into local scope
124 : typedef Fragment::sequence_id_t sequence_id_t; ///< Copy Fragment::sequence_id_t into local scope
125 : typedef std::map<sequence_id_t, RawEvent_ptr> EventMap; ///< An EventMap is a map of RawEvent_ptr objects, keyed by sequence ID
126 :
127 : /// <summary>
128 : /// Configuration of the SharedMemoryEventManager. May be used for parameter validation
129 : /// </summary>
130 : struct Config
131 : {
132 : /// "max_event_size_bytes" REQUIRED: Maximum event size(all Fragments), in bytes
133 : /// Either max_fragment_size_bytes or max_event_size_bytes must be specified
134 : fhicl::Atom<size_t> max_event_size_bytes{fhicl::Name{"max_event_size_bytes"}, fhicl::Comment{"Maximum event size (all Fragments), in bytes"}};
135 : /// "stale_buffer_timeout_usec" (Default: event_queue_wait_time * 1, 000, 000) : Maximum amount of time elapsed before a buffer is marked as abandoned.Time is reset each time an operation is performed on the buffer.
136 : fhicl::Atom<size_t> stale_buffer_timeout_usec{fhicl::Name{"stale_buffer_timeout_usec"}, fhicl::Comment{"Maximum amount of time elapsed before a buffer is marked as abandoned. Time is reset each time an operation is performed on the buffer."}, 5000000};
137 : /// "overwite_mode" (Default: false): Whether new data is allowed to overwrite buffers in the "Full" state
138 : fhicl::Atom<bool> overwrite_mode{fhicl::Name{"overwrite_mode"}, fhicl::Comment{"Whether buffers are allowed to be overwritten when safe (state == Full or Reading)"}, false};
139 : /// "shared_memory_ordering" (Default: false): Whether sequence IDs should be used to determine the order of events in shared memory, or whether buffers should be filled in the order they are received. WARNING: May disrupt round-robin distribution of events to art processes
140 : fhicl::Atom<bool> shared_memory_ordering{fhicl::Name{"shared_memory_ordering"}, fhicl::Comment{"Whether sequence IDs should be used to determine the order of events in shared memory, or whether buffers should be filled in the order they are received. WARNING: May disrupt round-robin distribution of events to art processes"}, false};
141 : /// "restart_crashed_art_processes" (Default: true) : Whether to automatically restart art processes that fail for any reason
142 : fhicl::Atom<bool> restart_crashed_art_processes{fhicl::Name{"restart_crashed_art_processes"}, fhicl::Comment{"Whether to automatically restart art processes that fail for any reason"}, true};
143 : /// "shared_memory_key" (Default 0xBEE70000 + PID) : Key to use for shared memory access
144 : fhicl::Atom<uint32_t> shared_memory_key{fhicl::Name{"shared_memory_key"}, fhicl::Comment{"Key to use for shared memory access"}, 0xBEE70000 + getpid()};
145 : /// "buffer_count" REQUIRED: Number of events in the Shared Memory(incomplete + pending art)
146 : fhicl::Atom<size_t> buffer_count{fhicl::Name{"buffer_count"}, fhicl::Comment{"Number of events in the Shared Memory (incomplete + pending art)"}};
147 : /// "max_fragment_size_bytes" REQURIED: Maximum Fragment size, in bytes
148 : /// Either max_fragment_size_bytes or max_event_size_bytes must be specified
149 : fhicl::Atom<size_t> max_fragment_size_bytes{fhicl::Name{"max_fragment_size_bytes"}, fhicl::Comment{" Maximum Fragment size, in bytes"}};
150 : /// "event_queue_wait_time" (Default: 5) : Amount of time(in seconds) an event can exist in shared memory before being released to art.Used as input to default parameter of "stale_buffer_timeout_usec".
151 : fhicl::Atom<size_t> event_queue_wait_time{fhicl::Name{"event_queue_wait_time"}, fhicl::Comment{"Amount of time (in seconds) an event can exist in shared memory before being released to art. Used as input to default parameter of \"stale_buffer_timeout_usec\"."}, 5};
152 : /// "broadcast_mode" (Default: false) : When true, buffers are not marked Empty when read, but return to Full state.Buffers are overwritten in order received.
153 : fhicl::Atom<bool> broadcast_mode{fhicl::Name{"broadcast_mode"}, fhicl::Comment{"When true, buffers are not marked Empty when read, but return to Full state. Buffers are overwritten in order received."}, false};
154 : /// "art_analyzer_count" (Default: 1) : Number of art procceses to start
155 : fhicl::Atom<size_t> art_analyzer_count{fhicl::Name{"art_analyzer_count"}, fhicl::Comment{"Number of art procceses to start"}, 1};
156 : /// "expected_fragments_per_event" (REQUIRED) : Number of Fragments to expect per event
157 : fhicl::Atom<size_t> expected_fragments_per_event{fhicl::Name{"expected_fragments_per_event"}, fhicl::Comment{"Number of Fragments to expect per event"}};
158 : /// "maximum_oversize_fragment_count" (Default: 1): Maximum number of over-size Fragments to drop before throwing an exception. Default is 1, which means to throw an exception if any over-size Fragments are dropped. Set to 0 to disable.
159 : fhicl::Atom<int> maximum_oversize_fragment_count{fhicl::Name{"maximum_oversize_fragment_count"}, fhicl::Comment{"Maximum number of over-size Fragments to drop before throwing an exception. Default is 1, which means to throw an exception if any over-size Fragments are dropped. Set to 0 to disable."}, 1};
160 : /// "use_sequence_id_for_event_number" (Default: true): Whether to use the artdaq Sequence ID (true) or the Timestamp (false) for art Event numbers
161 : fhicl::Atom<bool> use_sequence_id_for_event_number{fhicl::Name{"use_sequence_id_for_event_number"}, fhicl::Comment{"Whether to use the artdaq Sequence ID (true) or the Timestamp (false) for art Event numbers"}, true};
162 : /// "reset_event_number_for_subruns" (Default: false): Whether to start new subruns at Event ID 1 (true), or keep counting from the beginning of the run (false)
163 : fhicl::Atom<bool> reset_event_number_for_subruns{fhicl::Name{"reset_event_number_for_subruns"}, fhicl::Comment{"Whether to start new subruns at Event ID 1 (true), or keep counting from the beginning of the run (false)"}, false};
164 : /// "max_subrun_lookup_table_size" (Default: 100): The maximum number of entries to store in the sequence ID-SubRun ID lookup table
165 : fhicl::Atom<size_t> max_subrun_lookup_table_size{fhicl::Name{"max_subrun_lookup_table_size"}, fhicl::Comment{"The maximum number of entries to store in the sequence ID-SubRun ID lookup table"}, 100};
166 : /// "max_event_list_length" (Default: 100): The maximum number of entries to store in the released events list
167 : fhicl::Atom<size_t> max_event_list_length{fhicl::Name{"max_event_list_length"}, fhicl::Comment{" The maximum number of entries to store in the released events list"}, 100};
168 : /// "send_init_fragments" (Default: true): Whether Init Fragments are expected to be sent to art. If true, a Warning message is printed when an Init Fragment is requested but none are available.
169 : fhicl::Atom<bool> send_init_fragments{fhicl::Name{"send_init_fragments"}, fhicl::Comment{"Whether Init Fragments are expected to be sent to art. If true, a Warning message is printed when an Init Fragment is requested but none are available."}, true};
170 : /// "open_event_report_interval_ms" (Default: -1): Interval at which an open event report should be written
171 : fhicl::Atom<int> open_event_report_interval_ms{fhicl::Name{"open_event_report_interval_ms"}, fhicl::Comment{"Interval at which an open event report should be written"}, -1};
172 : /// "fragment_broadcast_timeout_ms" (Default: 3000): Amount of time broadcast fragments should live in the broadcast shared memory segment
173 : /// A "Broadcast shared memory segment" is used for all system-level fragments, such as Init, Start/End Run, Start/End Subrun and EndOfData
174 : fhicl::Atom<int> fragment_broadcast_timeout_ms{fhicl::Name{"fragment_broadcast_timeout_ms"}, fhicl::Comment{"Amount of time broadcast fragments should live in the broadcast shared memory segment"}, 3000};
175 : /// "art_command_line" (Default: "art -c \#CONFIG_FILE\#"): Command line used to start analysis processes. Supports two special sequences: \#CONFIG_FILE\# will be replaced with the fhicl config file. \#PROCESS_INDEX\# will be replaced by the index of the art process.
176 : fhicl::Atom<std::string> art_command_line{fhicl::Name{"art_command_line"}, fhicl::Comment{"Command line used to start analysis processes. Supports two special sequences: #CONFIG_FILE# will be replaced with the fhicl config file. #PROCESS_INDEX# will be replaced by the index of the art process."}, "art -c #CONFIG_FILE#"};
177 : /// "art_index_offset" (Default: 0): Offset to add to art process index when replacing \#PROCESS_INDEX\#
178 : fhicl::Atom<size_t> art_index_offset{fhicl::Name{"art_index_offset"}, fhicl::Comment{"Offset to add to art process index when replacing #PROCESS_INDEX#"}, 0};
179 : /// "minimum_art_lifetime_s" (Default: 2 seconds): Amount of time that an art process should run to not be considered "DOA"
180 : fhicl::Atom<double> minimum_art_lifetime_s{fhicl::Name{"minimum_art_lifetime_s"}, fhicl::Comment{"Amount of time that an art process should run to not be considered \"DOA\""}, 2.0};
181 : /// "expected_art_event_processing_time_us" (Default: 100000 us): During shutdown, SMEM will wait for this amount of time while it is checking that the art threads are done reading buffers.
182 : /// (TUNING: Should be slightly longer than the mean art processing time, but not so long that the Stop transition times out)
183 : fhicl::Atom<size_t> expected_art_event_processing_time_us{fhicl::Name{"expected_art_event_processing_time_us"}, fhicl::Comment{"During shutdown, SMEM will wait for this amount of time while it is checking that the art threads are done reading buffers."}, 100000};
184 : /// "broadcast_shared_memory_key" (Default: 0xCEE7000 + PID): Key to use for broadcast shared memory access
185 : fhicl::Atom<uint32_t> broadcast_shared_memory_key{fhicl::Name{"broadcast_shared_memory_key"}, fhicl::Comment{""}, 0xCEE70000 + getpid()};
186 : /// "broadcast_buffer_count" (Default: 10): Buffers in the broadcast shared memory segment
187 : fhicl::Atom<size_t> broadcast_buffer_count{fhicl::Name{"broadcast_buffer_count"}, fhicl::Comment{"Buffers in the broadcast shared memory segment"}, 10};
188 : /// "broadcast_buffer_size" (Default: 0x100000): Size of the buffers in the broadcast shared memory segment
189 : fhicl::Atom<size_t> broadcast_buffer_size{fhicl::Name{"broadcast_buffer_size"}, fhicl::Comment{"Size of the buffers in the broadcast shared memory segment"}, 0x100000};
190 : /// "use_art" (Default: true): Whether to start and manage art threads (Sets art_analyzer count to 0 and overwrite_mode to true when false)
191 : fhicl::Atom<bool> use_art{fhicl::Name{"use_art"}, fhicl::Comment{"Whether to start and manage art threads (Sets art_analyzer count to 0 and overwrite_mode to true when false)"}, true};
192 : /// "capture_art_stdout" (Default: true): Whether to capture the stdout and relay it to TLOG_INFO
193 : fhicl::Atom<bool> capture_art_stdout{fhicl::Name{"capture_art_stdout"}, fhicl::Comment{"Whether to capture the stdout and relay it to TLOG_INFO"}, true};
194 : /// "capture_art_stderr" (Default: true): Whether to capture the stderr and relay it to TLOG_ERROR
195 : fhicl::Atom<bool> capture_art_stderr{fhicl::Name{"capture_art_stderr"}, fhicl::Comment{"Whether to capture the stderr and relay it to TLOG_ERROR"}, true};
196 : /// "manual_art" (Default: false): Prints the startup command line for the art process so that the user may (for example) run it in GDB or valgrind
197 : fhicl::Atom<bool> manual_art{fhicl::Name{"manual_art"}, fhicl::Comment{"Prints the startup command line for the art process so that the user may (for example) run it in GDB or valgrind"}, false};
198 : /// Configuration of the RequestSender. See artdaq::RequestSender::Config
199 : fhicl::TableFragment<artdaq::RequestSender::Config> requestSenderConfig;
200 : /// Configuration of the TokenSender. See artdaq::TokenSender::Config
201 : fhicl::OptionalTable<artdaq::TokenSender::Config> tokenSenderConfig{fhicl::Name{"routing_token_config"}, fhicl::Comment{"Configuration for the Routing TokenSender"}};
202 : };
203 : /// Used for ParameterSet validation (if desired)
204 : using Parameters = fhicl::WrappedTable<Config>;
205 :
206 : /**
207 : * \brief SharedMemoryEventManager Constructor
208 : * \param pset ParameterSet used to configure SharedMemoryEventManager. See artdaq::SharedMemoryEventManager::Config for description of parameters
209 : * \param art_pset ParameterSet used to configure art. See art::Config for description of expected document format
210 : */
211 : SharedMemoryEventManager(const fhicl::ParameterSet& pset, fhicl::ParameterSet art_pset);
212 : /**
213 : * \brief SharedMemoryEventManager Destructor
214 : */
215 : virtual ~SharedMemoryEventManager() noexcept;
216 :
217 : private:
218 : /**
219 : * \brief Add a Fragment to the SharedMemoryEventManager
220 : * \param frag Header of the Fragment (seq ID and size info)
221 : * \param dataPtr Pointer to the fragment's data (i.e. Fragment::headerAddress())
222 : * \return Whether the Fragment was successfully added
223 : */
224 : bool AddFragment(detail::RawFragmentHeader frag, void* dataPtr);
225 :
226 : public:
227 : /**
228 : * \brief Copy a Fragment into the SharedMemoryEventManager
229 : * \param frag FragmentPtr object
230 : * \param timeout_usec Timeout for adding Fragment to the Shared Memory
231 : * \param [out] outfrag Rejected Fragment if timeout occurs
232 : * \return Whether the Fragment was successfully added
233 : */
234 : bool AddFragment(FragmentPtr frag, size_t timeout_usec, FragmentPtr& outfrag);
235 :
236 : /**
237 : * \brief Get a pointer to a reserved memory area for the given Fragment header
238 : * \param frag Fragment header (contains sequence ID and size information)
239 : * \param dropIfNoBuffersAvailable Whether to drop the fragment (instead of returning nullptr) when no buffers are available (Default: false)
240 : * \return Pointer to memory location for Fragment body (Header is copied into buffer here)
241 : */
242 : RawDataType* WriteFragmentHeader(detail::RawFragmentHeader frag, bool dropIfNoBuffersAvailable = false);
243 :
244 : /**
245 : * \brief Used to indicate that the given Fragment is now completely in the buffer. Will check for buffer completeness, and unset the pending flag.
246 : * \param frag Fragment that is now completely in the buffer.
247 : */
248 : void DoneWritingFragment(detail::RawFragmentHeader frag);
249 :
250 : /**
251 : * \brief Returns the number of buffers which contain data but are not yet complete
252 : * \return The number of buffers which contain data but are not yet complete
253 : */
254 41 : size_t GetOpenEventCount() { return active_buffers_.size(); }
255 :
256 : /**
257 : * \brief Returns the number of events which are complete but waiting on lower sequenced events to finish
258 : * \return The number of events which are complete but waiting on lower sequenced events to finish
259 : */
260 6 : size_t GetPendingEventCount() { return pending_buffers_.size(); }
261 :
262 : /**
263 : * \brief Returns the number of buffers currently owned by this manager
264 : * \return The number of buffers currently owned by this manager
265 : */
266 1 : size_t GetLockedBufferCount() { return GetBuffersOwnedByManager().size(); }
267 :
268 : /**
269 : * \brief Returns the number of events sent to art this run
270 : * \return The number of events sent to art this run
271 : */
272 19 : size_t GetArtEventCount() { return run_event_count_; }
273 :
274 : /**
275 : * \brief Get the count of Fragments of a given type in an event
276 : * \param seqID Sequence ID of Fragments
277 : * \param type Type of fragments to count. Use InvalidFragmentType to count all fragments (default)
278 : * \return Number of Fragments in event of given type
279 : */
280 : size_t GetFragmentCount(Fragment::sequence_id_t seqID, Fragment::type_t type = Fragment::InvalidFragmentType);
281 :
282 : /**
283 : * \brief Get the count of Fragments of a given type in a buffer
284 : * \param buffer Buffer to count
285 : * \param type Type of fragments to count. Use InvalidFragmentType to count all fragments (default)
286 : * \return Number of Fragments in buffer of given type
287 : */
288 : size_t GetFragmentCountInBuffer(int buffer, Fragment::type_t type = Fragment::InvalidFragmentType);
289 :
290 : void UpdateFragmentHeader(int buffer, detail::RawFragmentHeader hdr);
291 :
292 : /**
293 : * \brief Run an art instance, recording the return codes and restarting it until the end flag is raised
294 : */
295 : void RunArt(size_t process_index, const std::shared_ptr<std::atomic<pid_t>>& pid_out);
296 : /**
297 : * \brief Start all the art processes
298 : * \return Whether all art processes started correctly
299 : */
300 : bool StartArt();
301 :
302 : /**
303 : * \brief Start one art process
304 : * \param pset ParameterSet to send to this art process
305 : * \param process_index Index of this art process (when starting multiple)
306 : * \return pid_t of the started process
307 : */
308 : pid_t StartArtProcess(fhicl::ParameterSet pset, size_t process_index);
309 :
310 : /**
311 : * \brief Shutdown a set of art processes
312 : * \param pids PIDs of the art processes
313 : */
314 : void ShutdownArtProcesses(std::set<pid_t>& pids);
315 :
316 : /**
317 : * \brief Restart all art processes, using the given fhicl code to configure the new art processes
318 : * \param art_pset ParameterSet used to configure art
319 : * \param newRun New Run number for reconfigured art
320 : * \param n_art_processes Number of art processes to start, -1 (default) leaves the number unchanged
321 : */
322 : void ReconfigureArt(fhicl::ParameterSet art_pset, run_id_t newRun = 0, int n_art_processes = -1);
323 :
324 : /**
325 : * \brief Indicate that the end of input has been reached to the art processes.
326 : * \return True if the end proceeded correctly
327 : *
328 : * Put the end-of-data marker onto the RawEvent queue (if possible),
329 : * wait for the reader function to exit, and fill in the reader return
330 : * value. This scenario returns true. If the end-of-data marker
331 : * can not be pushed onto the RawEvent queue, false is returned.
332 : */
333 : bool endOfData();
334 :
335 : /**
336 : * \brief Start a Run
337 : * \param runID Run number of the new run
338 : */
339 : bool startRun(run_id_t runID);
340 :
341 : /**
342 : * \brief Get the current Run number
343 : * \return The current Run number
344 : */
345 6 : run_id_t runID() const { return run_id_; }
346 :
347 : /**
348 : * \brief Send an EndOfRunFragment to the art thread
349 : * \return True if enqueue successful
350 : */
351 : bool endRun();
352 :
353 : /**
354 : * \brief Rollover the subrun after the specified event
355 : * \param boundary sequence ID of the boundary (Event with seqID == boundary will be in new subrun)
356 : * \param subrun Subrun number of subrun after boundary
357 : * \param sendFragment Create and send an EndOfSubrun Fragment for this transition
358 : */
359 : void rolloverSubrun(sequence_id_t boundary, subrun_id_t subrun, bool sendFragment);
360 :
361 : /**
362 : * \brief Add a subrun transition immediately after the highest currently define sequence ID
363 : */
364 : void rolloverSubrun(bool sendFragment);
365 :
366 : /**
367 : * \brief Send metrics to the MetricManager, if one has been instantiated in the application
368 : */
369 : void sendMetrics();
370 :
371 : /**
372 : * \brief Set the RequestMessageMode for all outgoing data requests
373 : * \param mode Mode to set
374 : */
375 2 : void setRequestMode(detail::RequestMessageMode mode)
376 : {
377 2 : if (requests_) requests_->SetRequestMode(mode);
378 2 : }
379 :
380 : /**
381 : * \brief Set the overwrite flag (non-reliable data transfer) for the Shared Memory
382 : * \param overwrite Whether to allow the writer to overwrite data that has not yet been read
383 : */
384 : void setOverwrite(bool overwrite) { overwrite_mode_ = overwrite; }
385 :
386 : /**
387 : * \brief Set the stored Init fragment, if one has not yet been set already.
388 : */
389 : void AddInitFragment(FragmentPtr& frag);
390 :
391 : /**
392 : * @brief Add a Fragment for broadcast. May be collected with other Fragments before sending
393 : * @param frag Fragment to broadcast
394 : */
395 : void BroadcastFragment(FragmentPtr& frag);
396 :
397 : /**
398 : * \brief Gets the shared memory key of the broadcast SharedMemoryManager
399 : * \return The shared memory key of the broadcast SharedMemoryManager
400 : */
401 1 : uint32_t GetBroadcastKey() { return broadcasts_.GetKey(); }
402 :
403 : /**
404 : * \brief Gets the address of the "dropped data" fragment. Used for testing.
405 : * \param frag Fragment ID to get "dropped data" for
406 : * \return Pointer to the data payload of the "dropped data" fragment
407 : */
408 2 : RawDataType* GetDroppedDataAddress(detail::RawFragmentHeader frag)
409 : {
410 2 : for (auto it = dropped_data_.begin(); it != dropped_data_.end(); ++it)
411 : {
412 2 : if (frag.operator==(it->first)) // TODO, ELF 5/26/2023: Workaround until artdaq_core can be fixed for C++20
413 : {
414 2 : return it->second->dataBegin();
415 : }
416 : }
417 0 : return nullptr;
418 : }
419 :
420 : /**
421 : * \brief Updates the internally-stored copy of the art configuration.
422 : * \param art_pset ParameterSet used to configure art
423 : *
424 : * This method updates the internally-stored copy of the art configuration, but it does not
425 : * restart art processes. So, if this method is called while art processes are running, it will
426 : * have no effect until the next restart, such as the next Start of run. Typically, this
427 : * method is intended to be called between runs, when no art processes are running.
428 : */
429 : void UpdateArtConfiguration(fhicl::ParameterSet art_pset);
430 :
431 : /**
432 : * \brief Check for buffers which are ready to be marked incomplete and released to art and issue tokens for any buffers which are avaialble
433 : */
434 : void CheckPendingBuffers();
435 :
436 : /**
437 : * \brief Get the subrun number that the given Sequence ID would be assigned to
438 : * \param seqID Sequence ID to check
439 : * \return Subrun number that the given sequence ID will be associated with
440 : */
441 : subrun_id_t GetSubrunForSequenceID(Fragment::sequence_id_t seqID);
442 :
443 : /**
444 : * \brief Get the event ID that the given Sequence ID and Timestamp would be assigned to
445 : * \param seqID Sequence ID to check
446 : * \param timestamp Timestamp to check
447 : * \return Event ID that the given sequence ID and/or timestamp will be associated with. Uses Sequence ID if use_sequence_id_for_event_number is true, Timestamp otherwise
448 : */
449 : event_id_t GetEventIDForFragment(Fragment::sequence_id_t seqID, Fragment::timestamp_t timestamp);
450 :
451 : /**
452 : * \brief Get the current subrun number (Gets the last defined subrun)
453 : * \return Number of the subrun that corresponds to events with the maximum possible sequence ID.
454 : */
455 4 : subrun_id_t GetCurrentSubrun() { return GetSubrunForSequenceID(Fragment::InvalidSequenceID); }
456 :
457 0 : std::string BuildStatisticsString() const { return buildStatisticsString_(); };
458 :
459 : private:
460 : SharedMemoryEventManager(SharedMemoryEventManager const&) = delete;
461 : SharedMemoryEventManager(SharedMemoryEventManager&&) = delete;
462 : SharedMemoryEventManager& operator=(SharedMemoryEventManager const&) = delete;
463 : SharedMemoryEventManager& operator=(SharedMemoryEventManager&&) = delete;
464 :
465 46 : size_t get_art_process_count_()
466 : {
467 46 : std::unique_lock<std::mutex> lk(art_process_mutex_);
468 92 : return art_processes_.size();
469 46 : }
470 :
471 : std::string buildStatisticsString_() const;
472 :
473 : private:
474 : size_t num_art_processes_;
475 : size_t const num_fragments_per_event_;
476 : size_t const queue_size_;
477 : run_id_t run_id_;
478 :
479 : std::map<sequence_id_t, subrun_id_t> subrun_event_map_;
480 : subrun_id_t subrun_id_;
481 : size_t max_subrun_event_map_length_;
482 : static std::mutex subrun_event_map_mutex_;
483 : double subrun_transition_hold_time_s_;
484 : std::chrono::steady_clock::time_point last_event_time_;
485 :
486 : std::set<int> active_buffers_;
487 : std::set<int> pending_buffers_;
488 : std::vector<Fragment::sequence_id_t> buffer_sequence_ids_;
489 : std::unordered_map<Fragment::sequence_id_t, size_t> released_incomplete_events_;
490 : std::set<Fragment::sequence_id_t> released_events_;
491 : size_t max_event_list_length_;
492 :
493 : bool use_sequence_id_for_event_number_;
494 : bool reset_event_number_for_subruns_;
495 : bool overwrite_mode_;
496 : bool shared_memory_ordering_;
497 : size_t init_fragment_count_;
498 : std::atomic<bool> running_;
499 :
500 : std::unordered_map<int, std::atomic<int>> buffer_writes_pending_;
501 : std::unordered_map<int, std::mutex> buffer_mutexes_;
502 : static std::mutex sequence_id_mutex_;
503 :
504 : int open_event_report_interval_ms_;
505 : std::chrono::steady_clock::time_point last_open_event_report_time_;
506 : std::chrono::steady_clock::time_point last_backpressure_report_time_;
507 : std::chrono::steady_clock::time_point last_fragment_header_write_time_;
508 : std::vector<std::chrono::steady_clock::time_point> event_timing_;
509 :
510 : StatisticsHelper statsHelper_;
511 :
512 : int broadcast_timeout_ms_;
513 :
514 : std::atomic<int> run_event_count_;
515 : std::atomic<int> run_incomplete_event_count_;
516 : std::atomic<int> subrun_event_count_;
517 : std::atomic<int> subrun_incomplete_event_count_;
518 : std::atomic<int> oversize_fragment_count_;
519 : int maximum_oversize_fragment_count_;
520 :
521 : mutable std::mutex art_process_mutex_;
522 : std::set<pid_t> art_processes_;
523 : std::atomic<bool> restart_art_;
524 : bool always_restart_art_;
525 : std::atomic<bool> manual_art_;
526 : fhicl::ParameterSet current_art_pset_;
527 : std::shared_ptr<art_config_file> current_art_config_file_;
528 : std::string art_cmdline_;
529 : size_t art_process_index_offset_;
530 : double minimum_art_lifetime_s_;
531 : size_t art_event_processing_time_us_;
532 : bool capture_art_stdout_;
533 : bool capture_art_stderr_;
534 :
535 : std::unique_ptr<RequestSender> requests_;
536 : std::unique_ptr<TokenSender> tokens_;
537 : fhicl::ParameterSet data_pset_;
538 :
539 : std::mutex init_fragments_mutex_;
540 : std::unordered_map<Fragment::fragment_id_t, std::unordered_map<int, FragmentPtr>> init_fragment_map_;
541 : bool init_frags_sent_{false};
542 : std::list<std::pair<detail::RawFragmentHeader, FragmentPtr>> dropped_data_;
543 :
544 : mutable std::mutex broadcast_mutex_;
545 : struct BroadcastEntry
546 : {
547 : Fragment::type_t type;
548 : Fragment::sequence_id_t sequence_id;
549 : subrun_id_t subrun_id;
550 : FragmentPtrs fragments;
551 : std::chrono::steady_clock::time_point deadline;
552 : };
553 : std::vector<BroadcastEntry> pending_broadcasts_;
554 : void check_pending_broadcasts_();
555 :
556 : bool broadcastFragments_(FragmentPtrs& frags);
557 :
558 : detail::RawEventHeader* getEventHeader_(int buffer);
559 :
560 : int getBufferForSequenceID_(Fragment::sequence_id_t seqID, bool create_new, Fragment::timestamp_t timestamp = Fragment::InvalidTimestamp);
561 : bool hasFragments_(int buffer);
562 : void complete_buffer_(int buffer);
563 : bool bufferComparator(int bufA, int bufB);
564 : void check_pending_buffers_(std::unique_lock<std::mutex> const& lock);
565 : std::vector<char*> parse_art_command_line_(const std::shared_ptr<art_config_file>& config_file, size_t process_index);
566 :
567 : void send_init_frags_();
568 : size_t init_fragment_map_size_() const;
569 : SharedMemoryManager broadcasts_;
570 : };
571 : } // namespace artdaq
572 :
573 : #endif // ARTDAQ_DAQRATE_SHAREDMEMORYEVENTMANAGER_HH
|
