Line data Source code
1 0 : // Distributed under the MIT License.
2 : // See LICENSE.txt for details.
3 :
4 : #pragma once
5 :
6 : #include <numeric>
7 : #include <string>
8 : #include <tuple>
9 : #include <utility>
10 :
11 : #include "DataStructures/DataBox/DataBox.hpp"
12 : #include "IO/Observer/ObserverComponent.hpp"
13 : #include "IO/Observer/ReductionActions.hpp"
14 : #include "Parallel/GlobalCache.hpp"
15 : #include "Parallel/Info.hpp"
16 : #include "Parallel/Invoke.hpp"
17 : #include "Parallel/Local.hpp"
18 : #include "Parallel/MemoryMonitor/MemoryMonitor.hpp"
19 : #include "Parallel/MemoryMonitor/Tags.hpp"
20 : #include "Utilities/ErrorHandling/Assert.hpp"
21 : #include "Utilities/GetOutput.hpp"
22 : #include "Utilities/Gsl.hpp"
23 : #include "Utilities/Serialization/Serialize.hpp"
24 :
25 : namespace mem_monitor {
26 : /*!
27 : * \brief Simple action meant to be run on the MemoryMonitor component that
28 : * collects sizes from Groups and Nodegroups.
29 : *
30 : * \details This action collects the sizes of all the local branches of a group
31 : * or nodegroup component, computes the total memory usage on a node for each,
32 : * then writes it to disk. For groups, the proc with the maximum memory usage is
33 : * also reported along with the size on the proc.
34 : *
35 : * The columns in the dat file for a nodegroup when running on 3 nodes will be
36 : *
37 : * - %Time
38 : * - Size on node 0 (MB)
39 : * - Size on node 1 (MB)
40 : * - Size on node 2 (MB)
41 : * - Average size per node (MB)
42 : *
43 : * The columns in the dat file for a group when running on 3 nodes will be
44 : *
45 : * - %Time
46 : * - Size on node 0 (MB)
47 : * - Size on node 1 (MB)
48 : * - Size on node 2 (MB)
49 : * - Proc of max size
50 : * - Size on proc of max size (MB)
51 : * - Average size per node (MB)
52 : *
53 : * The dat file will be placed in the `/MemoryMonitors/` group in the reduction
54 : * file. The name of the dat file is the `pretty_type::name` of the component.
55 : */
56 : template <typename ContributingComponent>
57 1 : struct ContributeMemoryData {
58 : template <typename ParallelComponent, typename DbTags, typename Metavariables,
59 : typename ArrayIndex>
60 0 : static void apply(db::DataBox<DbTags>& box,
61 : Parallel::GlobalCache<Metavariables>& cache,
62 : const ArrayIndex& /*array_index*/, const double time,
63 : const int node_or_proc, const double size_in_megabytes) {
64 : static_assert(Parallel::is_group_v<ContributingComponent> or
65 : Parallel::is_nodegroup_v<ContributingComponent>);
66 :
67 : using tag = Tags::MemoryHolder;
68 : db::mutate<tag>(
69 : [&cache, &time, &node_or_proc, &size_in_megabytes](
70 : const gsl::not_null<std::unordered_map<
71 : std::string,
72 : std::unordered_map<double, std::unordered_map<int, double>>>*>
73 : memory_holder_all) {
74 : auto memory_holder_pair = memory_holder_all->try_emplace(
75 : pretty_type::name<ContributingComponent>());
76 : auto& memory_holder = (*memory_holder_pair.first).second;
77 :
78 : memory_holder.try_emplace(time);
79 : memory_holder.at(time)[node_or_proc] = size_in_megabytes;
80 :
81 : // If we have received data for every node/proc at a given
82 : // time, get all the data, write it to disk, then remove the current
83 : // time from the stored times as it's no longer needed
84 :
85 : auto& mem_monitor_proxy =
86 : Parallel::get_parallel_component<MemoryMonitor<Metavariables>>(
87 : cache);
88 :
89 : constexpr bool is_group = Parallel::is_group_v<ContributingComponent>;
90 :
91 : const size_t num_nodes = Parallel::number_of_nodes<size_t>(
92 : *Parallel::local(mem_monitor_proxy));
93 : const size_t num_procs = Parallel::number_of_procs<size_t>(
94 : *Parallel::local(mem_monitor_proxy));
95 : const size_t expected_number = is_group ? num_procs : num_nodes;
96 : ASSERT(memory_holder.at(time).size() <= expected_number,
97 : "ContributeMemoryData received more data than it was "
98 : "expecting. Was expecting "
99 : << expected_number << " calls but instead got "
100 : << memory_holder.at(time).size());
101 : if (memory_holder.at(time).size() == expected_number) {
102 : // First column is always time
103 : std::vector<double> data_to_append{time};
104 : std::vector<std::string> legend{{"Time"}};
105 :
106 : // Append a column for each node, and keep track of cumulative
107 : // total. If we have proc data (from groups) do an additional loop
108 : // over the procs to get the total on that node and get the proc
109 : // of the maximum memory usage
110 : double avg_size_per_node = 0.0;
111 : double max_usage_on_proc = -std::numeric_limits<double>::max();
112 : int proc_of_max = 0;
113 : for (size_t node = 0; node < num_nodes; node++) {
114 : double size_on_node = 0.0;
115 : if (not is_group) {
116 : size_on_node = memory_holder.at(time).at(node);
117 : } else {
118 : const int first_proc = Parallel::first_proc_on_node<int>(
119 : node, *Parallel::local(mem_monitor_proxy));
120 : const int procs_on_node = Parallel::procs_on_node<int>(
121 : node, *Parallel::local(mem_monitor_proxy));
122 : const int last_proc = first_proc + procs_on_node;
123 : for (int proc = first_proc; proc < last_proc; proc++) {
124 : size_on_node += memory_holder.at(time).at(proc);
125 : if (memory_holder.at(time).at(proc) > max_usage_on_proc) {
126 : max_usage_on_proc = memory_holder.at(time).at(proc);
127 : proc_of_max = proc;
128 : }
129 : }
130 : }
131 :
132 : data_to_append.push_back(size_on_node);
133 : avg_size_per_node += size_on_node;
134 : legend.emplace_back("Size on node " + get_output(node) + " (MB)");
135 : }
136 :
137 : // If we have proc data, write the proc with the maximum usage to
138 : // disk along with how much memory it's using
139 : if (is_group) {
140 : data_to_append.push_back(static_cast<double>(proc_of_max));
141 : data_to_append.push_back(max_usage_on_proc);
142 : legend.emplace_back("Proc of max size");
143 : legend.emplace_back("Size on proc of max size (MB)");
144 : }
145 :
146 : avg_size_per_node /= static_cast<double>(num_nodes);
147 :
148 : // Last column is average over all nodes
149 : data_to_append.push_back(avg_size_per_node);
150 : legend.emplace_back("Average size per node (MB)");
151 :
152 : auto& observer_writer_proxy = Parallel::get_parallel_component<
153 : observers::ObserverWriter<Metavariables>>(cache);
154 :
155 : Parallel::threaded_action<
156 : observers::ThreadedActions::WriteReductionDataRow>(
157 : // Node 0 is always the writer
158 : observer_writer_proxy[0], subfile_name<ContributingComponent>(),
159 : legend, std::make_tuple(data_to_append));
160 :
161 : // Clean up finished time
162 : auto finished_time_iter = memory_holder.find(time);
163 : memory_holder.erase(finished_time_iter);
164 : }
165 : },
166 : make_not_null(&box));
167 : }
168 : };
169 : } // namespace mem_monitor
|
