Skip to content

Commit d1b975a

Browse files
mgoversmgovers
authored
Merge pull request #1055 from PowerGridModel/bugfix/calculation-info-memory-creep
Performance: fix memory creep + false sharing in calculation info
2 parents 07d30c8 + 199bcb6 commit d1b975a

File tree

2 files changed

+64
-49
lines changed

2 files changed

+64
-49
lines changed

‎power_grid_model_c/power_grid_model/include/power_grid_model/job_dispatch.hpp

Lines changed: 56 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
#include "main_core/calculation_info.hpp"
1010
#include "main_core/update.hpp"
1111

12+
#include <mutex>
1213
#include <thread>
1314

1415
namespace power_grid_model {
@@ -60,27 +61,35 @@ template <class MainModel, class... ComponentType> class JobDispatch {
6061

6162
// error messages
6263
std::vector<std::string> exceptions(n_scenarios, "");
63-
std::vector<CalculationInfo> infos(n_scenarios);
64+
65+
// thread-safe handling of calculation info
66+
std::mutex calculation_info_mutex;
67+
auto const thread_safe_add_calculation_info = [&calculation_info,
68+
&calculation_info_mutex](CalculationInfo const& info) {
69+
std::lock_guard const lock{calculation_info_mutex};
70+
main_core::merge_into(calculation_info, info);
71+
};
6472

6573
// lambda for sub batch calculation
6674
main_core::utils::SequenceIdx<ComponentType...> all_scenarios_sequence;
67-
auto sub_batch = sub_batch_calculation_(model, std::forward<Calculate>(calculation_fn), result_data,
68-
update_data, all_scenarios_sequence, exceptions, infos);
75+
auto sub_batch =
76+
sub_batch_calculation_(model, std::forward<Calculate>(calculation_fn), result_data, update_data,
77+
all_scenarios_sequence, exceptions, thread_safe_add_calculation_info);
6978

7079
job_dispatch(sub_batch, n_scenarios, threading);
7180

7281
handle_batch_exceptions(exceptions);
73-
calculation_info = main_core::merge_calculation_info(infos);
7482

7583
return BatchParameter{};
7684
}
7785

78-
template <typename Calculate>
86+
template <typename Calculate, typename AddCalculationInfo>
7987
requires std::invocable<std::remove_cvref_t<Calculate>, MainModel&, MutableDataset const&, Idx>
8088
static auto sub_batch_calculation_(MainModel const& base_model, Calculate&& calculation_fn,
8189
MutableDataset const& result_data, ConstDataset const& update_data,
8290
main_core::utils::SequenceIdx<ComponentType...>& all_scenarios_sequence,
83-
std::vector<std::string>& exceptions, std::vector<CalculationInfo>& infos) {
91+
std::vector<std::string>& exceptions,
92+
AddCalculationInfo&& thread_safe_add_calculation_info) {
8493
// cache component update order where possible.
8594
// the order for a cacheable (independent) component by definition is the same across all scenarios
8695
auto const components_to_update = base_model.get_components_to_update(update_data);
@@ -89,56 +98,55 @@ template <class MainModel, class... ComponentType> class JobDispatch {
8998
all_scenarios_sequence = main_core::update::get_all_sequence_idx_map<ComponentType...>(
9099
base_model.state(), update_data, 0, components_to_update, update_independence, false);
91100

92-
return [&base_model, &exceptions, &infos, calculation_fn_ = std::forward<Calculate>(calculation_fn),
93-
&result_data, &update_data, &all_scenarios_sequence_ = std::as_const(all_scenarios_sequence),
94-
components_to_update, update_independence](Idx start, Idx stride, Idx n_scenarios) {
101+
return [&base_model, &exceptions, &thread_safe_add_calculation_info,
102+
calculation_fn_ = std::forward<Calculate>(calculation_fn), &result_data, &update_data,
103+
&all_scenarios_sequence_ = std::as_const(all_scenarios_sequence), components_to_update,
104+
update_independence](Idx start, Idx stride, Idx n_scenarios) {
95105
assert(n_scenarios <= narrow_cast<Idx>(exceptions.size()));
96-
assert(n_scenarios <= narrow_cast<Idx>(infos.size()));
97106

98-
Timer const t_total(infos[start], 0000, "Total in thread");
107+
CalculationInfo thread_info;
99108

100-
auto const copy_model_functor = [&base_model, &infos](Idx scenario_idx) {
101-
Timer const t_copy_model_functor(infos[scenario_idx], 1100, "Copy model");
109+
Timer t_total(thread_info, 0000, "Total in thread");
110+
111+
auto const copy_model_functor = [&base_model, &thread_info] {
112+
Timer const t_copy_model_functor(thread_info, 1100, "Copy model");
102113
return MainModel{base_model};
103114
};
104-
auto model = copy_model_functor(start);
115+
auto model = copy_model_functor();
105116

106117
auto current_scenario_sequence_cache = main_core::utils::SequenceIdx<ComponentType...>{};
107118
auto [setup, winddown] =
108119
scenario_update_restore(model, update_data, components_to_update, update_independence,
109-
all_scenarios_sequence_, current_scenario_sequence_cache, infos);
120+
all_scenarios_sequence_, current_scenario_sequence_cache, thread_info);
110121

111122
auto calculate_scenario = JobDispatch::call_with<Idx>(
112-
[&model, &calculation_fn_, &result_data, &infos](Idx scenario_idx) {
123+
[&model, &calculation_fn_, &result_data, &thread_info](Idx scenario_idx) {
113124
calculation_fn_(model, result_data, scenario_idx);
114-
infos[scenario_idx].merge(model.calculation_info());
125+
main_core::merge_into(thread_info, model.calculation_info());
115126
},
116-
std::move(setup), std::move(winddown), scenario_exception_handler(model, exceptions, infos),
117-
[&model, &copy_model_functor](Idx scenario_idx) { model = copy_model_functor(scenario_idx); });
127+
std::move(setup), std::move(winddown), scenario_exception_handler(model, exceptions, thread_info),
128+
[&model, &copy_model_functor](Idx /*scenario_idx*/) { model = copy_model_functor(); });
118129

119130
for (Idx scenario_idx = start; scenario_idx < n_scenarios; scenario_idx += stride) {
120-
Timer const t_total_single(infos[scenario_idx], 0100, "Total single calculation in thread");
121-
131+
Timer const t_total_single(thread_info, 0100, "Total single calculation in thread");
122132
calculate_scenario(scenario_idx);
123133
}
134+
135+
t_total.stop();
136+
thread_safe_add_calculation_info(thread_info);
124137
};
125138
}
126139

127-
// run sequential if
128-
// specified threading < 0
129-
// use hardware threads, but it is either unknown (0) or only has one thread (1)
130-
// specified threading = 1
131140
template <typename RunSubBatchFn>
132141
requires std::invocable<std::remove_cvref_t<RunSubBatchFn>, Idx /*start*/, Idx /*stride*/, Idx /*n_scenarios*/>
133142
static void job_dispatch(RunSubBatchFn sub_batch, Idx n_scenarios, Idx threading) {
134143
// run batches sequential or parallel
135-
auto const hardware_thread = static_cast<Idx>(std::thread::hardware_concurrency());
136-
if (threading < 0 || threading == 1 || (threading == 0 && hardware_thread < 2)) {
144+
auto const n_thread = n_threads(n_scenarios, threading);
145+
if (n_thread == 1) {
137146
// run all in sequential
138147
sub_batch(0, 1, n_scenarios);
139148
} else {
140149
// create parallel threads
141-
Idx const n_thread = std::min(threading == 0 ? hardware_thread : threading, n_scenarios);
142150
std::vector<std::thread> threads;
143151
threads.reserve(n_thread);
144152
for (Idx thread_number = 0; thread_number < n_thread; ++thread_number) {
@@ -151,6 +159,18 @@ template <class MainModel, class... ComponentType> class JobDispatch {
151159
}
152160
}
153161

162+
// run sequential if
163+
// specified threading < 0
164+
// use hardware threads, but it is either unknown (0) or only has one thread (1)
165+
// specified threading = 1
166+
static Idx n_threads(Idx n_scenarios, Idx threading) {
167+
auto const hardware_thread = static_cast<Idx>(std::thread::hardware_concurrency());
168+
if (threading < 0 || threading == 1 || (threading == 0 && hardware_thread < 2)) {
169+
return 1; // sequential
170+
}
171+
return std::min(threading == 0 ? hardware_thread : threading, n_scenarios);
172+
}
173+
154174
template <typename... Args, typename RunFn, typename SetupFn, typename WinddownFn, typename HandleExceptionFn,
155175
typename RecoverFromBadFn>
156176
requires std::invocable<std::remove_cvref_t<RunFn>, Args const&...> &&
@@ -184,7 +204,7 @@ template <class MainModel, class... ComponentType> class JobDispatch {
184204
main_core::update::independence::UpdateIndependence<ComponentType...> const& do_update_cache,
185205
main_core::utils::SequenceIdx<ComponentType...> const& all_scenario_sequence,
186206
main_core::utils::SequenceIdx<ComponentType...>& current_scenario_sequence_cache,
187-
std::vector<CalculationInfo>& infos) noexcept {
207+
CalculationInfo& info) noexcept {
188208
main_core::utils::ComponentFlags<ComponentType...> independence_flags{};
189209
std::ranges::transform(do_update_cache, independence_flags.begin(),
190210
[](auto const& comp) { return comp.is_independent(); });
@@ -202,15 +222,15 @@ template <class MainModel, class... ComponentType> class JobDispatch {
202222

203223
return std::make_pair(
204224
[&model, &update_data, scenario_sequence, &current_scenario_sequence_cache, &components_to_store,
205-
do_update_cache_ = std::move(do_update_cache), &infos](Idx scenario_idx) {
206-
Timer const t_update_model(infos[scenario_idx], 1200, "Update model");
225+
do_update_cache_ = std::move(do_update_cache), &info](Idx scenario_idx) {
226+
Timer const t_update_model(info, 1200, "Update model");
207227
current_scenario_sequence_cache = main_core::update::get_all_sequence_idx_map<ComponentType...>(
208228
model.state(), update_data, scenario_idx, components_to_store, do_update_cache_, true);
209229

210230
model.template update_components<cached_update_t>(update_data, scenario_idx, scenario_sequence());
211231
},
212-
[&model, scenario_sequence, &current_scenario_sequence_cache, &infos](Idx scenario_idx) {
213-
Timer const t_update_model(infos[scenario_idx], 1201, "Restore model");
232+
[&model, scenario_sequence, &current_scenario_sequence_cache, &info](Idx /*scenario_idx*/) {
233+
Timer const t_update_model(info, 1201, "Restore model");
214234

215235
model.restore_components(scenario_sequence());
216236
std::ranges::for_each(current_scenario_sequence_cache,
@@ -220,8 +240,8 @@ template <class MainModel, class... ComponentType> class JobDispatch {
220240

221241
// Lippincott pattern
222242
static auto scenario_exception_handler(MainModel& model, std::vector<std::string>& messages,
223-
std::vector<CalculationInfo>& infos) {
224-
return [&model, &messages, &infos](Idx scenario_idx) {
243+
CalculationInfo& info) {
244+
return [&model, &messages, &info](Idx scenario_idx) {
225245
std::exception_ptr const ex_ptr = std::current_exception();
226246
try {
227247
std::rethrow_exception(ex_ptr);
@@ -230,7 +250,7 @@ template <class MainModel, class... ComponentType> class JobDispatch {
230250
} catch (...) {
231251
messages[scenario_idx] = "unknown exception";
232252
}
233-
infos[scenario_idx].merge(model.calculation_info());
253+
info.merge(model.calculation_info());
234254
};
235255
}
236256

‎power_grid_model_c/power_grid_model/include/power_grid_model/main_core/calculation_info.hpp

Lines changed: 8 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -10,21 +10,16 @@
1010

1111
namespace power_grid_model::main_core {
1212

13-
inline CalculationInfo merge_calculation_info(std::vector<CalculationInfo> const& infos) {
14-
CalculationInfo result;
15-
16-
auto const key = Timer::make_key(2226, "Max number of iterations");
17-
for (auto const& info : infos) {
18-
for (auto const& [k, v] : info) {
19-
if (k == key) {
20-
result[k] = std::max(result[k], v);
21-
} else {
22-
result[k] += v;
23-
}
13+
inline CalculationInfo& merge_into(CalculationInfo& destination, CalculationInfo const& source) {
14+
static auto const key = Timer::make_key(2226, "Max number of iterations");
15+
for (auto const& [k, v] : source) {
16+
if (k == key) {
17+
destination[k] = std::max(destination[k], v);
18+
} else {
19+
destination[k] += v;
2420
}
2521
}
26-
27-
return result;
22+
return destination;
2823
}
2924

3025
} // namespace power_grid_model::main_core

0 commit comments

Comments
 (0)