Line data Source code
1 0 : // Distributed under the MIT License.
2 : // See LICENSE.txt for details.
3 :
4 : #pragma once
5 :
6 : #include <cstddef>
7 : #include <mutex>
8 : #include <optional>
9 : #include <string>
10 : #include <tuple>
11 : #include <unordered_map>
12 : #include <utility>
13 : #include <vector>
14 :
15 : #include "DataStructures/DataBox/DataBox.hpp"
16 : #include "DataStructures/TaggedTuple.hpp"
17 : #include "IO/H5/AccessType.hpp"
18 : #include "IO/H5/Dat.hpp"
19 : #include "IO/H5/File.hpp"
20 : #include "IO/Observer/Helpers.hpp"
21 : #include "IO/Observer/ObservationId.hpp"
22 : #include "IO/Observer/Protocols/ReductionDataFormatter.hpp"
23 : #include "IO/Observer/Tags.hpp"
24 : #include "Parallel/ArrayComponentId.hpp"
25 : #include "Parallel/ArrayIndex.hpp"
26 : #include "Parallel/GlobalCache.hpp"
27 : #include "Parallel/Info.hpp"
28 : #include "Parallel/Invoke.hpp"
29 : #include "Parallel/Local.hpp"
30 : #include "Parallel/NodeLock.hpp"
31 : #include "Parallel/ParallelComponentHelpers.hpp"
32 : #include "Parallel/Printf/Printf.hpp"
33 : #include "Parallel/Reduction.hpp"
34 : #include "Utilities/ErrorHandling/Assert.hpp"
35 : #include "Utilities/ErrorHandling/Error.hpp"
36 : #include "Utilities/GetOutput.hpp"
37 : #include "Utilities/Gsl.hpp"
38 : #include "Utilities/PrettyType.hpp"
39 : #include "Utilities/ProtocolHelpers.hpp"
40 : #include "Utilities/Requires.hpp"
41 : #include "Utilities/Serialization/PupStlCpp17.hpp"
42 : #include "Utilities/StdHelpers.hpp"
43 : #include "Utilities/TMPL.hpp"
44 :
45 : namespace observers {
46 :
47 : /// \cond
48 : template <class Metavariables>
49 : struct ObserverWriter;
50 : /// \endcond
51 :
52 0 : namespace ThreadedActions {
53 : /// \cond
54 : struct CollectReductionDataOnNode;
55 : struct WriteReductionData;
56 : /// \endcond
57 : } // namespace ThreadedActions
58 :
59 : /// Indicates no formatter is selected
60 1 : struct NoFormatter {
61 0 : void pup(PUP::er& /*p*/) {}
62 : };
63 :
64 : /*!
65 : * \brief Rows of data to append to one `h5::Dat` subfile of the reductions
66 : * file with `ThreadedActions::WriteMultipleReductionDataRows`.
67 : */
68 1 : struct DatRows {
69 : /// Name of the `h5::Dat` subfile, including a leading slash
70 1 : std::string subfile_name{};
71 : /// Column labels of the subfile
72 1 : std::vector<std::string> legend{};
73 : /// Rows to append. Each row must have as many entries as the `legend`.
74 1 : std::vector<std::vector<double>> rows{};
75 :
76 : // NOLINTNEXTLINE(google-runtime-references)
77 0 : void pup(PUP::er& p);
78 : };
79 :
80 0 : bool operator==(const DatRows& lhs, const DatRows& rhs);
81 0 : bool operator!=(const DatRows& lhs, const DatRows& rhs);
82 :
83 : namespace Actions {
84 : /// \cond
85 : struct ContributeReductionDataToWriter;
86 : /// \endcond
87 :
88 : /*!
89 : * \ingroup ObserversGroup
90 : * \brief Send reduction data to the observer group.
91 : *
92 : * Once everything at a specific `ObservationId` has been contributed to the
93 : * reduction, the groups reduce to their local nodegroup.
94 : *
95 : * The caller of this Action (which is to be invoked on the Observer parallel
96 : * component) must pass in an `observation_id` used to uniquely identify the
97 : * observation in time, the name of the `h5::Dat` subfile in the HDF5 file (e.g.
98 : * `/element_data`, where the slash is important), a `std::vector<std::string>`
99 : * of names of the quantities being reduced (e.g. `{"Time", "L1ErrorDensity",
100 : * "L2ErrorDensity"}`), and the `Parallel::ReductionData` that holds the
101 : * `ReductionDatums` containing info on how to do the reduction.
102 : *
103 : * The observer components need to know all expected reduction data types by
104 : * compile-time, so they rely on the
105 : * `Metavariables::observed_reduction_data_tags` alias to collect them in one
106 : * place. To this end, each Action that contributes reduction data must expose
107 : * the type alias as:
108 : *
109 : * \snippet ObserverHelpers.hpp make_reduction_data_tags
110 : *
111 : * Then, in the `Metavariables` collect them from all observing Actions using
112 : * the `observers::collect_reduction_data_tags` metafunction.
113 : *
114 : * This action also accepts a "formatter" that will be forwarded along with the
115 : * reduction data and used to print an informative message when the reduction is
116 : * complete. The formatter must conform to
117 : * `observers::protocols::ReductionDataFormatter`.
118 : *
119 : * This action also supports observing the intermediate stage of the reduction
120 : * over just the processing element that the element is currently on. This can
121 : * be useful e.g. to measure performance metric to assess load-balancing such as
122 : * the number of grid points on each core. Enable per-core observations by
123 : * passing `true` for the `observe_per_core` argument (default: `false`). The
124 : * data will be written in one H5 file per _node_ prefixed with the
125 : * `observers::Tags::ReductionFileName`, in a `Core{core_id}` subfile, where
126 : * `core_id` is an integer identifying the core across all nodes (see
127 : * `Parallel::my_proc`). For example, when running on 2 nodes with 2 cores each
128 : * you will end up with `Reductions0.h5` containing `/Core0/{subfile_name}.dat`
129 : * and `/Core1/{subfile_name}.dat`, and `Reductions1.h5` containing
130 : * `/Core2/{subfile_name}.dat` and `/Core3/{subfile_name}.dat`. This is in
131 : * addition to the usual reduction output over _all_ registered elements,
132 : * written to `Reductions.h5` (no node ID suffix in the file name).
133 : */
134 1 : struct ContributeReductionData {
135 : template <typename ParallelComponent, typename DbTagsList,
136 : typename Metavariables, typename ArrayIndex, typename... Ts,
137 : typename Formatter = observers::NoFormatter>
138 0 : static void apply(db::DataBox<DbTagsList>& box,
139 : Parallel::GlobalCache<Metavariables>& cache,
140 : const ArrayIndex& array_index,
141 : const observers::ObservationId& observation_id,
142 : const Parallel::ArrayComponentId& sender_array_id,
143 : const std::string& subfile_name,
144 : const std::vector<std::string>& reduction_names,
145 : Parallel::ReductionData<Ts...>&& reduction_data,
146 : std::optional<Formatter>&& formatter = std::nullopt,
147 : const bool observe_per_core = false) {
148 : db::mutate<Tags::ReductionData<Ts...>, Tags::ReductionDataNames<Ts...>,
149 : Tags::ContributorsOfReductionData>(
150 : [&array_index, &cache, &observation_id,
151 : reduction_data = std::move(reduction_data), &reduction_names,
152 : &sender_array_id, &subfile_name, &formatter, &observe_per_core](
153 : const gsl::not_null<std::unordered_map<
154 : ObservationId, Parallel::ReductionData<Ts...>>*>
155 : reduction_data_map,
156 : const gsl::not_null<
157 : std::unordered_map<ObservationId, std::vector<std::string>>*>
158 : reduction_names_map,
159 : const gsl::not_null<std::unordered_map<
160 : ObservationId, std::unordered_set<Parallel::ArrayComponentId>>*>
161 : reduction_observers_contributed,
162 : const std::unordered_map<
163 : ObservationKey,
164 : std::unordered_set<Parallel::ArrayComponentId>>&
165 : observations_registered) mutable { // NOLINT(spectre-mutable)
166 : ASSERT(
167 : observations_registered.find(observation_id.observation_key()) !=
168 : observations_registered.end(),
169 : "Couldn't find registration key "
170 : << observation_id.observation_key()
171 : << " in the registered observers. Known IDs are "
172 : << keys_of(observations_registered)
173 : << "\nIt could be that you are using a nodegroup DG "
174 : "collection but the observation code for the event hasn't "
175 : "been updated to work with the nodegroup yet.");
176 :
177 : auto& contributed_array_ids =
178 : (*reduction_observers_contributed)[observation_id];
179 : if (UNLIKELY(contributed_array_ids.find(sender_array_id) !=
180 : contributed_array_ids.end())) {
181 : ERROR("Already received reduction data to observation id "
182 : << observation_id << " from array component id "
183 : << sender_array_id);
184 : }
185 : contributed_array_ids.insert(sender_array_id);
186 :
187 : if (reduction_data_map->count(observation_id) == 0) {
188 : reduction_data_map->emplace(observation_id,
189 : std::move(reduction_data));
190 : reduction_names_map->emplace(observation_id, reduction_names);
191 : } else {
192 : if (UNLIKELY(reduction_names_map->at(observation_id) !=
193 : reduction_names)) {
194 : using ::operator<<;
195 : ERROR("Reduction names differ at ObservationId "
196 : << observation_id << " with the expected names being "
197 : << reduction_names_map->at(observation_id)
198 : << " and the received names being " << reduction_names);
199 : }
200 : reduction_data_map->operator[](observation_id)
201 : .combine(std::move(reduction_data));
202 : }
203 :
204 : // Check if we have received all reduction data from the registered
205 : // elements. If so, we reduce to the local ObserverWriter nodegroup.
206 : if (UNLIKELY(
207 : contributed_array_ids.size() ==
208 : observations_registered.at(observation_id.observation_key())
209 : .size())) {
210 : auto& local_writer = *Parallel::local_branch(
211 : Parallel::get_parallel_component<ObserverWriter<Metavariables>>(
212 : cache));
213 : auto& my_proxy =
214 : Parallel::get_parallel_component<ParallelComponent>(cache);
215 : const std::optional<int> observe_with_core_id =
216 : observe_per_core ? std::make_optional(Parallel::my_proc<int>(
217 : *Parallel::local_branch(my_proxy)))
218 : : std::nullopt;
219 : Parallel::threaded_action<
220 : ThreadedActions::CollectReductionDataOnNode>(
221 : local_writer, observation_id,
222 : Parallel::make_array_component_id<ParallelComponent>(
223 : array_index),
224 : subfile_name, (*reduction_names_map)[observation_id],
225 : std::move((*reduction_data_map)[observation_id]),
226 : std::move(formatter), observe_with_core_id);
227 : reduction_data_map->erase(observation_id);
228 : reduction_names_map->erase(observation_id);
229 : reduction_observers_contributed->erase(observation_id);
230 : }
231 : },
232 : make_not_null(&box),
233 : db::get<Tags::ExpectedContributorsForObservations>(box));
234 : // Silence a gcc <= 9 unused-variable warning
235 : (void)observe_per_core;
236 : }
237 : };
238 : } // namespace Actions
239 :
240 : namespace ThreadedActions {
241 :
242 : namespace ReductionActions_detail {
243 : void append_to_reduction_data(
244 : const gsl::not_null<std::vector<double>*> all_reduction_data,
245 : const double t);
246 :
247 : template <typename T>
248 : void append_to_reduction_data(
249 : const gsl::not_null<std::vector<double>*> all_reduction_data,
250 : const std::vector<T>& t);
251 :
252 : void append_to_reduction_data(
253 : gsl::not_null<std::vector<double>*> all_reduction_data,
254 : const std::array<double, 3>& t);
255 :
256 : template <typename... Ts, size_t... Is>
257 : void write_data(const std::string& subfile_name,
258 : const std::string& input_source,
259 : std::vector<std::string> legend, const std::tuple<Ts...>& data,
260 : const std::string& file_prefix,
261 : std::index_sequence<Is...> /*meta*/) {
262 : static_assert(sizeof...(Ts) > 0,
263 : "Must be reducing at least one piece of data");
264 : std::vector<double> data_to_append{};
265 : EXPAND_PACK_LEFT_TO_RIGHT(
266 : append_to_reduction_data(&data_to_append, std::get<Is>(data)));
267 :
268 : if (legend.size() != data_to_append.size()) {
269 : ERROR(
270 : "There must be one name provided for each piece of data. You provided "
271 : << legend.size() << " names: '" << get_output(legend)
272 : << "' but there are " << data_to_append.size()
273 : << " pieces of data being reduced");
274 : }
275 :
276 : h5::H5File<h5::AccessType::ReadWrite> h5file(file_prefix + ".h5", true,
277 : input_source);
278 : constexpr size_t version_number = 0;
279 : auto& time_series_file = h5file.try_insert<h5::Dat>(
280 : subfile_name, std::move(legend), version_number);
281 : time_series_file.append(data_to_append);
282 : }
283 :
284 : void write_rows(const std::vector<DatRows>& data,
285 : const std::string& input_source,
286 : const std::string& file_prefix);
287 : } // namespace ReductionActions_detail
288 :
289 : /*!
290 : * \brief Gathers all the reduction data from all processing elements/cores on a
291 : * node.
292 : */
293 1 : struct CollectReductionDataOnNode {
294 : public:
295 : template <typename ParallelComponent, typename DbTagsList,
296 : typename Metavariables, typename ArrayIndex,
297 : typename... ReductionDatums,
298 : typename Formatter = observers::NoFormatter>
299 0 : static void apply(
300 : db::DataBox<DbTagsList>& box, Parallel::GlobalCache<Metavariables>& cache,
301 : const ArrayIndex& /*array_index*/,
302 : const gsl::not_null<Parallel::NodeLock*> node_lock,
303 : const observers::ObservationId& observation_id,
304 : Parallel::ArrayComponentId observer_group_id,
305 : const std::string& subfile_name,
306 : std::vector<std::string>&& reduction_names,
307 : Parallel::ReductionData<ReductionDatums...>&& received_reduction_data,
308 : std::optional<Formatter>&& formatter = std::nullopt,
309 : const std::optional<int> observe_with_core_id = std::nullopt) {
310 : // The below gymnastics with pointers is done in order to minimize the
311 : // time spent locking the entire node, which is necessary because the
312 : // DataBox does not allow any functions calls, both get and mutate, during
313 : // a mutate. This design choice in DataBox is necessary to guarantee a
314 : // consistent state throughout mutation. Here, however, we need to be
315 : // reasonable efficient in parallel and so we manually guarantee that
316 : // consistent state. To this end, we create pointers and assign to them
317 : // the data in the DataBox which is guaranteed to be pointer stable. The
318 : // data itself is guaranteed to be stable inside the ReductionDataLock.
319 : std::unordered_map<observers::ObservationId,
320 : Parallel::ReductionData<ReductionDatums...>>*
321 : reduction_data = nullptr;
322 : std::unordered_map<ObservationId, std::vector<std::string>>*
323 : reduction_names_map = nullptr;
324 : std::unordered_map<observers::ObservationId,
325 : std::unordered_set<Parallel::ArrayComponentId>>*
326 : reduction_observers_contributed = nullptr;
327 : Parallel::NodeLock* reduction_data_lock = nullptr;
328 : Parallel::NodeLock* reduction_file_lock = nullptr;
329 : size_t observations_registered_with_id = std::numeric_limits<size_t>::max();
330 :
331 : {
332 : const std::lock_guard hold_lock(*node_lock);
333 : db::mutate<Tags::ReductionData<ReductionDatums...>,
334 : Tags::ReductionDataNames<ReductionDatums...>,
335 : Tags::ContributorsOfReductionData, Tags::ReductionDataLock,
336 : Tags::H5FileLock>(
337 : [&reduction_data, &reduction_names_map,
338 : &reduction_observers_contributed, &reduction_data_lock,
339 : &reduction_file_lock, &observation_id, &observer_group_id,
340 : &observations_registered_with_id](
341 : const gsl::not_null<std::unordered_map<
342 : observers::ObservationId,
343 : Parallel::ReductionData<ReductionDatums...>>*>
344 : reduction_data_ptr,
345 : const gsl::not_null<
346 : std::unordered_map<ObservationId, std::vector<std::string>>*>
347 : reduction_names_map_ptr,
348 : const gsl::not_null<std::unordered_map<
349 : observers::ObservationId,
350 : std::unordered_set<Parallel::ArrayComponentId>>*>
351 : reduction_observers_contributed_ptr,
352 : const gsl::not_null<Parallel::NodeLock*> reduction_data_lock_ptr,
353 : const gsl::not_null<Parallel::NodeLock*> reduction_file_lock_ptr,
354 : const std::unordered_map<
355 : ObservationKey,
356 : std::unordered_set<Parallel::ArrayComponentId>>&
357 : observations_registered) {
358 : const ObservationKey& key{observation_id.observation_key()};
359 : const auto& registered_group_ids = observations_registered.at(key);
360 : if (UNLIKELY(registered_group_ids.find(observer_group_id) ==
361 : registered_group_ids.end())) {
362 : ERROR("The observer group id "
363 : << observer_group_id
364 : << " was not registered for the observation id "
365 : << observation_id);
366 : }
367 : reduction_data = &*reduction_data_ptr;
368 : reduction_names_map = &*reduction_names_map_ptr;
369 : reduction_observers_contributed =
370 : &*reduction_observers_contributed_ptr;
371 : reduction_data_lock = &*reduction_data_lock_ptr;
372 : reduction_file_lock = &*reduction_file_lock_ptr;
373 : observations_registered_with_id =
374 : observations_registered.at(key).size();
375 : },
376 : make_not_null(&box),
377 : db::get<Tags::ExpectedContributorsForObservations>(box));
378 : }
379 :
380 : ASSERT(
381 : observations_registered_with_id != std::numeric_limits<size_t>::max(),
382 : "Failed to set observations_registered_with_id when mutating the "
383 : "DataBox. This is a bug in the code.");
384 :
385 : bool send_data = false;
386 : // Now that we've retrieved pointers to the data in the DataBox we wish to
387 : // manipulate, lock the data and manipulate it.
388 : {
389 : const std::lock_guard hold_data_lock(*reduction_data_lock);
390 : auto& contributed_group_ids =
391 : (*reduction_observers_contributed)[observation_id];
392 :
393 : if (UNLIKELY(contributed_group_ids.find(observer_group_id) !=
394 : contributed_group_ids.end())) {
395 : ERROR("Already received reduction data to observation id "
396 : << observation_id << " from array component id "
397 : << observer_group_id);
398 : }
399 : contributed_group_ids.insert(observer_group_id);
400 :
401 : // If requested, write the intermediate reduction data from the particular
402 : // core to one file per node. This allows measuring reduction data
403 : // per-core, e.g. performance metrics to assess load balancing.
404 : if (observe_with_core_id.has_value()) {
405 : auto reduction_data_this_core = received_reduction_data;
406 : reduction_data_this_core.finalize();
407 : auto reduction_names_this_core = reduction_names;
408 : auto& my_proxy =
409 : Parallel::get_parallel_component<ParallelComponent>(cache);
410 : const std::lock_guard hold_file_lock(*reduction_file_lock);
411 : ReductionActions_detail::write_data(
412 : "/Core" + std::to_string(observe_with_core_id.value()) +
413 : subfile_name,
414 : observers::input_source_from_cache(cache),
415 : std::move(reduction_names_this_core),
416 : std::move(reduction_data_this_core.data()),
417 : Parallel::get<Tags::ReductionFileName>(cache) +
418 : std::to_string(
419 : Parallel::my_node<int>(*Parallel::local_branch(my_proxy))),
420 : std::make_index_sequence<sizeof...(ReductionDatums)>{});
421 : }
422 :
423 : if (reduction_data->find(observation_id) == reduction_data->end()) {
424 : // This Action has been called for the first time,
425 : // so all we need to do is move the input data to the
426 : // reduction_data in the DataBox.
427 : reduction_data->operator[](observation_id) =
428 : std::move(received_reduction_data);
429 : } else {
430 : // This Action is being called at least the second time
431 : // (but not the final time if on node 0).
432 : reduction_data->at(observation_id)
433 : .combine(std::move(received_reduction_data));
434 : }
435 :
436 : if (UNLIKELY(reduction_names.empty())) {
437 : ERROR(
438 : "The reduction names, which is a std::vector of the names of "
439 : "the columns in the file, must be non-empty.");
440 : }
441 : if (auto current_names = reduction_names_map->find(observation_id);
442 : current_names == reduction_names_map->end()) {
443 : reduction_names_map->emplace(observation_id,
444 : std::move(reduction_names));
445 : } else if (UNLIKELY(current_names->second != reduction_names)) {
446 : ERROR(
447 : "The reduction names passed in must match the currently "
448 : "known reduction names.");
449 : }
450 :
451 : // Check if we have received all reduction data from the Observer
452 : // group. If so we reduce to node 0 for writing to disk. We use a bool
453 : // `send_data` to allow us to defer the send call until after we've
454 : // unlocked the lock.
455 : if (reduction_observers_contributed->at(observation_id).size() ==
456 : observations_registered_with_id) {
457 : send_data = true;
458 : // We intentionally move the data out of the map and erase it
459 : // before call `WriteReductionData` since if the call to
460 : // `WriteReductionData` is inlined and we erase data from the maps
461 : // afterward we would lose data.
462 : reduction_names =
463 : std::move(reduction_names_map->operator[](observation_id));
464 : received_reduction_data =
465 : std::move(reduction_data->operator[](observation_id));
466 : reduction_observers_contributed->erase(observation_id);
467 : reduction_data->erase(observation_id);
468 : reduction_names_map->erase(observation_id);
469 : }
470 : }
471 :
472 : if (send_data) {
473 : auto& my_proxy =
474 : Parallel::get_parallel_component<ParallelComponent>(cache);
475 : Parallel::threaded_action<WriteReductionData>(
476 : Parallel::get_parallel_component<ObserverWriter<Metavariables>>(
477 : cache)[0],
478 : observation_id,
479 : Parallel::my_node<size_t>(*Parallel::local_branch(my_proxy)),
480 : subfile_name,
481 : // NOLINTNEXTLINE(bugprone-use-after-move)
482 : std::move(reduction_names), std::move(received_reduction_data),
483 : std::move(formatter));
484 : }
485 : }
486 : };
487 :
488 : /*!
489 : * \ingroup ObserversGroup
490 : * \brief Write reduction data to disk from node 0.
491 : */
492 1 : struct WriteReductionData {
493 : template <typename ParallelComponent, typename DbTagsList,
494 : typename Metavariables, typename ArrayIndex,
495 : typename... ReductionDatums,
496 : typename Formatter = observers::NoFormatter>
497 0 : static void apply(
498 : db::DataBox<DbTagsList>& box, Parallel::GlobalCache<Metavariables>& cache,
499 : const ArrayIndex& /*array_index*/,
500 : const gsl::not_null<Parallel::NodeLock*> node_lock,
501 : const observers::ObservationId& observation_id,
502 : const size_t sender_node_number, const std::string& subfile_name,
503 : std::vector<std::string>&& reduction_names,
504 : Parallel::ReductionData<ReductionDatums...>&& received_reduction_data,
505 : std::optional<Formatter>&& formatter = std::nullopt) {
506 : if constexpr (not std::is_same_v<Formatter, observers::NoFormatter>) {
507 : static_assert(
508 : tt::assert_conforms_to_v<Formatter,
509 : protocols::ReductionDataFormatter>);
510 : static_assert(
511 : std::is_same_v<typename Formatter::reduction_data,
512 : Parallel::ReductionData<ReductionDatums...>>,
513 : "Mismatch between the formatter's `reduction_data` type alias and "
514 : "the reduction data that is being reduced.");
515 : }
516 : // The below gymnastics with pointers is done in order to minimize the
517 : // time spent locking the entire node, which is necessary because the
518 : // DataBox does not allow any functions calls, both get and mutate, during
519 : // a mutate. This design choice in DataBox is necessary to guarantee a
520 : // consistent state throughout mutation. Here, however, we need to be
521 : // reasonable efficient in parallel and so we manually guarantee that
522 : // consistent state. To this end, we create pointers and assign to them
523 : // the data in the DataBox which is guaranteed to be pointer stable. The
524 : // data itself is guaranteed to be stable inside the ReductionDataLock.
525 : std::unordered_map<observers::ObservationId,
526 : Parallel::ReductionData<ReductionDatums...>>*
527 : reduction_data = nullptr;
528 : std::unordered_map<ObservationId, std::vector<std::string>>*
529 : reduction_names_map = nullptr;
530 : std::unordered_map<observers::ObservationId, std::unordered_set<size_t>>*
531 : nodes_contributed = nullptr;
532 : Parallel::NodeLock* reduction_data_lock = nullptr;
533 : Parallel::NodeLock* reduction_file_lock = nullptr;
534 : size_t observations_registered_with_id = std::numeric_limits<size_t>::max();
535 :
536 : {
537 : const std::lock_guard hold_lock(*node_lock);
538 : db::mutate<Tags::ReductionData<ReductionDatums...>,
539 : Tags::ReductionDataNames<ReductionDatums...>,
540 : Tags::NodesThatContributedReductions, Tags::ReductionDataLock,
541 : Tags::H5FileLock>(
542 : [&nodes_contributed, &reduction_data, &reduction_names_map,
543 : &reduction_data_lock, &reduction_file_lock, &observation_id,
544 : &observations_registered_with_id, &sender_node_number](
545 : const gsl::not_null<
546 : typename Tags::ReductionData<ReductionDatums...>::type*>
547 : reduction_data_ptr,
548 : const gsl::not_null<
549 : std::unordered_map<ObservationId, std::vector<std::string>>*>
550 : reduction_names_map_ptr,
551 : const gsl::not_null<std::unordered_map<
552 : ObservationId, std::unordered_set<size_t>>*>
553 : nodes_contributed_ptr,
554 : const gsl::not_null<Parallel::NodeLock*> reduction_data_lock_ptr,
555 : const gsl::not_null<Parallel::NodeLock*> reduction_file_lock_ptr,
556 : const std::unordered_map<ObservationKey, std::set<size_t>>&
557 : nodes_registered_for_reductions) {
558 : const ObservationKey& key{observation_id.observation_key()};
559 : ASSERT(nodes_registered_for_reductions.find(key) !=
560 : nodes_registered_for_reductions.end(),
561 : "Performing reduction with unregistered ID key "
562 : << observation_id.observation_key());
563 : const auto& registered_nodes =
564 : nodes_registered_for_reductions.at(key);
565 :
566 : if (UNLIKELY(registered_nodes.find(sender_node_number) ==
567 : registered_nodes.end())) {
568 : ERROR("Node " << sender_node_number
569 : << " was not registered for the observation id "
570 : << observation_id);
571 : }
572 :
573 : reduction_data = &*reduction_data_ptr;
574 : reduction_names_map = &*reduction_names_map_ptr;
575 : nodes_contributed = &*nodes_contributed_ptr;
576 : reduction_data_lock = &*reduction_data_lock_ptr;
577 : reduction_file_lock = &*reduction_file_lock_ptr;
578 : observations_registered_with_id =
579 : nodes_registered_for_reductions.at(key).size();
580 : },
581 : make_not_null(&box),
582 : db::get<Tags::NodesExpectedToContributeReductions>(box));
583 : }
584 :
585 : ASSERT(
586 : observations_registered_with_id != std::numeric_limits<size_t>::max(),
587 : "Failed to set observations_registered_with_id when mutating the "
588 : "DataBox. This is a bug in the code.");
589 :
590 : bool write_to_disk = false;
591 : // Now that we've retrieved pointers to the data in the DataBox we wish to
592 : // manipulate, lock the data and manipulate it.
593 : {
594 : const std::lock_guard hold_lock(*reduction_data_lock);
595 : auto& nodes_contributed_to_observation =
596 : (*nodes_contributed)[observation_id];
597 : if (nodes_contributed_to_observation.find(sender_node_number) !=
598 : nodes_contributed_to_observation.end()) {
599 : ERROR("Already received reduction data at observation id "
600 : << observation_id << " from node " << sender_node_number);
601 : }
602 : nodes_contributed_to_observation.insert(sender_node_number);
603 :
604 : if (UNLIKELY(reduction_names.empty())) {
605 : ERROR(
606 : "The reduction names, which is a std::vector of the names of "
607 : "the columns in the file, must be non-empty.");
608 : }
609 : if (auto current_names = reduction_names_map->find(observation_id);
610 : current_names == reduction_names_map->end()) {
611 : reduction_names_map->emplace(observation_id,
612 : std::move(reduction_names));
613 : } else if (UNLIKELY(current_names->second != reduction_names)) {
614 : using ::operator<<;
615 : ERROR(
616 : "The reduction names passed in must match the currently "
617 : "known reduction names. Current ones are "
618 : << current_names->second << " while the received are "
619 : << reduction_names);
620 : }
621 :
622 : if (reduction_data->find(observation_id) == reduction_data->end()) {
623 : // This Action has been called for the first time,
624 : // so all we need to do is move the input data to the
625 : // reduction_data in the DataBox.
626 : reduction_data->operator[](observation_id) =
627 : std::move(received_reduction_data);
628 : } else {
629 : // This Action is being called at least the second time
630 : // (but not the final time if on node 0).
631 : reduction_data->at(observation_id)
632 : .combine(std::move(received_reduction_data));
633 : }
634 :
635 : // We use a bool `write_to_disk` to allow us to defer the data writing
636 : // until after we've unlocked the lock. For the same reason, we move the
637 : // final, reduced result into `received_reduction_data` and
638 : // `reduction_names`.
639 : if (nodes_contributed_to_observation.size() ==
640 : observations_registered_with_id) {
641 : write_to_disk = true;
642 : received_reduction_data =
643 : std::move(reduction_data->operator[](observation_id));
644 : reduction_names =
645 : std::move(reduction_names_map->operator[](observation_id));
646 : reduction_data->erase(observation_id);
647 : reduction_names_map->erase(observation_id);
648 : nodes_contributed->erase(observation_id);
649 : }
650 : }
651 :
652 : if (write_to_disk) {
653 : const std::lock_guard hold_lock(*reduction_file_lock);
654 : // NOLINTNEXTLINE(bugprone-use-after-move)
655 : received_reduction_data.finalize();
656 : if constexpr (not std::is_same_v<Formatter, NoFormatter>) {
657 : if (formatter.has_value()) {
658 : Parallel::printf(
659 : std::apply(*formatter, received_reduction_data.data()));
660 : }
661 : }
662 : ReductionActions_detail::write_data(
663 : subfile_name, observers::input_source_from_cache(cache),
664 : // NOLINTNEXTLINE(bugprone-use-after-move)
665 : std::move(reduction_names), std::move(received_reduction_data.data()),
666 : Parallel::get<Tags::ReductionFileName>(cache),
667 : std::make_index_sequence<sizeof...(ReductionDatums)>{});
668 : }
669 : }
670 : };
671 :
672 : /*!
673 : * \brief Write a single row of data to the reductions file without the need to
674 : * register or reduce anything, e.g. from a singleton component or from a
675 : * specific chare.
676 : *
677 : * Use observers::Actions::ContributeReductionData instead if you need to
678 : * perform a reduction before writing to the file. Use
679 : * `WriteMultipleReductionDataRows` to write to several subfiles, or several
680 : * rows, in a single call.
681 : *
682 : * Invoke this action on the observers::ObserverWriter component on node 0. Pass
683 : * the following arguments when invoking this action:
684 : *
685 : * - `subfile_name`: the name of the `h5::Dat` subfile in the HDF5 file. Include
686 : * a leading slash, e.g., `/element_data`.
687 : * - `legend`: a `std::vector<std::string>` of column labels for the quantities
688 : * being observed (e.g. `{"Time", "L1ErrorDensity", "L2ErrorDensity"}`).
689 : * - `reduction_data`: a `std::tuple<...>` with the data to write. The tuple can
690 : * hold either `double`s or `std::vector<double>`s and is flattened before it
691 : * is written to the file to form a single row of values. The total number of
692 : * values must match the length of the `legend`.
693 : */
694 1 : struct WriteReductionDataRow {
695 : /// \brief The apply call for the threaded action
696 : template <typename ParallelComponent, typename DbTagsList,
697 : typename Metavariables, typename ArrayIndex, typename... Ts>
698 1 : static void apply(db::DataBox<DbTagsList>& box,
699 : Parallel::GlobalCache<Metavariables>& cache,
700 : const ArrayIndex& /*array_index*/,
701 : const gsl::not_null<Parallel::NodeLock*> node_lock,
702 : const std::string& subfile_name,
703 : std::vector<std::string> legend,
704 : std::tuple<Ts...>&& reduction_data) {
705 : apply<ParallelComponent>(box, node_lock, cache, subfile_name,
706 : std::move(legend), std::move(reduction_data));
707 : }
708 :
709 : // The local synchronous action
710 0 : using return_type = void;
711 :
712 : /// \brief The apply call for the local synchronous action
713 : template <typename ParallelComponent, typename DbTagList,
714 : typename Metavariables, typename... Ts>
715 1 : static return_type apply(
716 : db::DataBox<DbTagList>& box,
717 : const gsl::not_null<Parallel::NodeLock*> /*node_lock*/,
718 : Parallel::GlobalCache<Metavariables>& cache,
719 : const std::string& subfile_name, std::vector<std::string> legend,
720 : std::tuple<Ts...>&& reduction_data) {
721 : auto& reduction_file_lock =
722 : db::get_mutable_reference<Tags::H5FileLock>(make_not_null(&box));
723 : const std::lock_guard hold_lock(reduction_file_lock);
724 : ThreadedActions::ReductionActions_detail::write_data(
725 : subfile_name, observers::input_source_from_cache(cache),
726 : std::move(legend), std::move(reduction_data),
727 : Parallel::get<Tags::ReductionFileName>(cache),
728 : std::make_index_sequence<sizeof...(Ts)>{});
729 : }
730 : };
731 :
732 : /*!
733 : * \brief Write rows of data to one or more `h5::Dat` subfiles of the
734 : * reductions file, opening the file only once.
735 : *
736 : * Like `WriteReductionDataRow`, this action requires no registration or
737 : * reduction. Each `WriteReductionDataRow` call opens and closes the reductions
738 : * file, so prefer this action when writing to many subfiles at once (e.g. one
739 : * subfile per component of a function of time) or several rows to one subfile
740 : * at once.
741 : *
742 : * Invoke this action on the observers::ObserverWriter component on node 0 with
743 : * a `std::vector<observers::DatRows>`, one entry per subfile to write to. The
744 : * `legend` of each entry must have one name per column, and each of its `rows`
745 : * must have one value per column.
746 : */
747 1 : struct WriteMultipleReductionDataRows {
748 : /// \brief The apply call for the threaded action
749 : template <typename ParallelComponent, typename DbTagsList,
750 : typename Metavariables, typename ArrayIndex>
751 1 : static void apply(db::DataBox<DbTagsList>& box,
752 : Parallel::GlobalCache<Metavariables>& cache,
753 : const ArrayIndex& /*array_index*/,
754 : const gsl::not_null<Parallel::NodeLock*> node_lock,
755 : const std::vector<DatRows>& data) {
756 : apply<ParallelComponent>(box, node_lock, cache, data);
757 : }
758 :
759 : // The local synchronous action
760 0 : using return_type = void;
761 :
762 : /// \brief The apply call for the local synchronous action
763 : template <typename ParallelComponent, typename DbTagList,
764 : typename Metavariables>
765 1 : static return_type apply(
766 : db::DataBox<DbTagList>& box,
767 : const gsl::not_null<Parallel::NodeLock*> /*node_lock*/,
768 : Parallel::GlobalCache<Metavariables>& cache,
769 : const std::vector<DatRows>& data) {
770 : auto& reduction_file_lock =
771 : db::get_mutable_reference<Tags::H5FileLock>(make_not_null(&box));
772 : const std::lock_guard hold_lock(reduction_file_lock);
773 : ThreadedActions::ReductionActions_detail::write_rows(
774 : data, observers::input_source_from_cache(cache),
775 : Parallel::get<Tags::ReductionFileName>(cache));
776 : }
777 : };
778 :
779 : } // namespace ThreadedActions
780 : } // namespace observers
|
