Shamrock 2025.10.0
Astrophysical Code
Loading...
Searching...
No Matches
PatchScheduler.cpp
Go to the documentation of this file.
1// -------------------------------------------------------//
2//
3// SHAMROCK code for hydrodynamics
4// Copyright (c) 2021-2026 Timothée David--Cléris <tim.shamrock@proton.me>
5// SPDX-License-Identifier: CeCILL Free Software License Agreement v2.1
6// Shamrock is licensed under the CeCILL 2.1 License, see LICENSE for more information
7//
8// -------------------------------------------------------//
9
16
19#include "shambase/string.hpp"
20#include "shambase/time.hpp"
21#include "nlohmann/json.hpp"
22#include "shambackends/math.hpp"
32#include <ctime>
33#include <memory>
34#include <optional>
35#include <sstream>
36#include <stdexcept>
37#include <vector>
38
39// TODO move types init out
40void PatchScheduler::init_mpi_required_types() {
41
42 // if(!patch::is_mpi_patch_type_active()){
43 // patch::create_MPI_patch_type();
44 // }
45}
46
47void PatchScheduler::free_mpi_required_types() {
48
49 // if(patch::is_mpi_patch_type_active()){
50 // patch::free_MPI_patch_type();
51 // }
52}
53
54template<u32 dim>
55void PatchScheduler::make_patch_base_grid(std::array<u32, dim> patch_count) {
56
57 static_assert(dim == 3, "this is not implemented for dim != 3");
58
59 u32 max_lin_patch_count = 0;
60 for (u32 i = 0; i < dim; i++) {
61 max_lin_patch_count = sycl::max(max_lin_patch_count, patch_count[i]);
62 }
63
64 u64 coord_div_fact = sham::roundup_pow2_clz(max_lin_patch_count);
65
66 u64 sz_root_patch = PatchScheduler::max_axis_patch_coord_length / coord_div_fact;
67
68 std::vector<shamrock::patch::PatchCoord<3>> coords;
69 for (u32 x = 0; x < patch_count[0]; x++) {
70 for (u32 y = 0; y < patch_count[1]; y++) {
71 for (u32 z = 0; z < patch_count[2]; z++) {
72 shamrock::patch::PatchCoord coord;
73
74 coord.coord_min[0] = sz_root_patch * (x);
75 coord.coord_min[1] = sz_root_patch * (y);
76 coord.coord_min[2] = sz_root_patch * (z);
77 coord.coord_max[0] = sz_root_patch * (x + 1) - 1;
78 coord.coord_max[1] = sz_root_patch * (y + 1) - 1;
79 coord.coord_max[2] = sz_root_patch * (z + 1) - 1;
80
81 coords.push_back(coord);
82 }
83 }
84 }
85
86 shamrock::patch::PatchCoord bounds;
87 bounds.coord_min[0] = 0;
88 bounds.coord_min[1] = 0;
89 bounds.coord_min[2] = 0;
90 bounds.coord_max[0] = sz_root_patch * patch_count[0] - 1;
91 bounds.coord_max[1] = sz_root_patch * patch_count[1] - 1;
92 bounds.coord_max[2] = sz_root_patch * patch_count[2] - 1;
93
94 get_sim_box().set_patch_coord_bounding_box(bounds);
95
96 add_root_patches(coords);
97}
98
99template void PatchScheduler::make_patch_base_grid<3>(std::array<u32, 3> patch_count);
100
102 std::vector<shamrock::patch::PatchCoord<3>> coords) {
103
104 using namespace shamrock::patch;
105
106 std::vector<u64> ret;
107
108 for (auto coord : coords) {
109
110 u32 node_owner_id = 0;
111
112 Patch root;
113 root.id_patch = patch_list._next_patch_id;
115 root.load_value = 0;
116 root.coord_min[0] = coord.coord_min[0];
117 root.coord_min[1] = coord.coord_min[1];
118 root.coord_min[2] = coord.coord_min[2];
119 root.coord_max[0] = coord.coord_max[0];
120 root.coord_max[1] = coord.coord_max[1];
121 root.coord_max[2] = coord.coord_max[2];
122 root.node_owner_id = node_owner_id;
123
124 patch_list.global.push_back(root);
125 patch_list._next_patch_id++;
126
127 if (shamcomm::world_rank() == node_owner_id) {
128 patch_data.owned_data.add_obj(root.id_patch, PatchDataLayer(get_layout_ptr_old()));
129 shamlog_debug_sycl_ln("Scheduler", "adding patch data");
130 } else {
131 shamlog_debug_sycl_ln(
132 "Scheduler",
133 "patch data wasn't added rank =",
135 " ower =",
136 node_owner_id);
137 }
138
139 patch_tree.insert_root_node(root.id_patch, coord);
140
141 ret.push_back(root.id_patch);
142
143 // auto [bmin,bmax] = get_sim_box().patch_coord_to_domain<u64_3>(root);
144 //
145 //
146 // shamlog_debug_ln("Scheduler", "adding patch : [ (",
147 // coord.x_min,
148 // coord.y_min,
149 // coord.z_min,") ] [ (",
150 // coord.x_max,
151 // coord.y_max,
152 // coord.z_max,") ]", bmin,bmax
153 //);
154 }
155
156 // build_local() is declared as nodiscard
157 (void) patch_list.build_local();
158 patch_list.reset_local_pack_index();
159 patch_list.build_local_idx_map();
160 patch_list.build_global_idx_map();
161
162 patch_list.invalidate_load_values();
163
164 return ret;
165}
166
168
169 shamlog_debug_ln("Scheduler", "pushing data obj cnt =", pdat.get_obj_cnt());
170
171 for_each_patch_data([&](u64 id_patch,
174 auto variant_main = pdl_old().get_main_field_any();
175
176 variant_main.visit([&](auto &arg) {
177 using base_t = typename std::remove_reference<decltype(arg)>::type::field_T;
178
179 if constexpr (shambase::VectorProperties<base_t>::dimension == 3) {
180 auto [bmin, bmax] = get_sim_box().patch_coord_to_domain<base_t>(cur_p);
181
182 shamlog_debug_sycl_ln(
183 "Scheduler", "pushing data in patch ", id_patch, "search range :", bmin, bmax);
184
185 pdat_sched.insert_elements_in_range(pdat, bmin, bmax);
186 } else {
187 throw std::runtime_error("this does not yet work with dimension different from 3");
188 }
189 });
190 });
191}
192
194 using namespace shamrock::patch;
195
196 PatchCoord coord;
197 coord.coord_min[0] = 0;
198 coord.coord_min[1] = 0;
199 coord.coord_min[2] = 0;
200 coord.coord_max[0] = max_axis_patch_coord;
201 coord.coord_max[1] = max_axis_patch_coord;
202 coord.coord_max[2] = max_axis_patch_coord;
203
204 add_root_patches({coord});
205
206 patch_list.invalidate_load_values();
207}
208
209PatchScheduler::PatchScheduler(
210 const std::shared_ptr<shamrock::patch::PatchDataLayerLayout> &pdl_ptr,
211 u64 crit_split,
212 u64 crit_merge)
213 : pdl_ptr(pdl_ptr),
214 patch_data(
215 pdl_ptr, {{0, 0, 0}, {max_axis_patch_coord, max_axis_patch_coord, max_axis_patch_coord}}),
216 synchronized_data() {
217
218 crit_patch_split = crit_split;
219 crit_patch_merge = crit_merge;
220}
221
222PatchScheduler::~PatchScheduler() {}
223
224bool PatchScheduler::should_resize_box(bool node_in) {
225 u16 tmp = node_in;
226 u16 out = 0;
227 shamcomm::mpi::Allreduce(&tmp, &out, 1, mpi_type_u16, MPI_MAX, MPI_COMM_WORLD);
228 return out;
229}
230
231// TODO move Loadbalancing function to template state
232void PatchScheduler::sync_build_LB(bool global_patch_sync, bool balance_load) {
233
234 patch_list.check_load_values_valid();
235
236 if (global_patch_sync)
237 patch_list.build_global();
238
239 if (balance_load) {
240 // real load balancing
241 shamrock::scheduler::LoadBalancingChangeList change_list
243
244 // exchange data
245 patch_data.apply_change_list(change_list, patch_list);
246 }
247
248 // rebuild local table
249 owned_patch_id = patch_list.build_local();
250}
251
252template<>
253std::tuple<f32_3, f32_3> PatchScheduler::get_box_tranform() {
254 if (!pdl_old().check_main_field_type<f32_3>())
256 "cannot query single precision box the main field is not of f32_3 type");
257
258 auto [bmin, bmax] = patch_data.sim_box.get_bounding_box<f32_3>();
259
260 f32_3 translate_factor = bmin;
261 f32_3 scale_factor = (bmax - bmin) / LoadBalancer::max_box_sz;
262
263 return {translate_factor, scale_factor};
264}
265
266template<>
267std::tuple<f64_3, f64_3> PatchScheduler::get_box_tranform() {
268 if (!pdl_old().check_main_field_type<f64_3>())
270 "cannot query single precision box the main field is not of f64_3 type");
271
272 auto [bmin, bmax] = patch_data.sim_box.get_bounding_box<f64_3>();
273
274 f64_3 translate_factor = bmin;
275 f64_3 scale_factor = (bmax - bmin) / LoadBalancer::max_box_sz;
276
277 return {translate_factor, scale_factor};
278}
279
280template<>
281std::tuple<f32_3, f32_3> PatchScheduler::get_box_volume() {
282 if (!pdl_old().check_main_field_type<f32_3>())
284 "cannot query single precision box the main field is not of f32_3 type");
285
286 return patch_data.sim_box.get_bounding_box<f32_3>();
287}
288
289template<>
290std::tuple<f64_3, f64_3> PatchScheduler::get_box_volume() {
291 if (!pdl_old().check_main_field_type<f64_3>())
293 "cannot query single precision box the main field is not of f64_3 type");
294
295 return patch_data.sim_box.get_bounding_box<f64_3>();
296}
297
298template<>
299std::tuple<i64_3, i64_3> PatchScheduler::get_box_volume() {
300 if (!pdl_old().check_main_field_type<i64_3>())
302 "cannot query single precision box the main field is not of i64_3 type");
303
304 return patch_data.sim_box.get_bounding_box<i64_3>();
305}
306
307// TODO clean the output of this function
308void PatchScheduler::scheduler_step(bool do_split_merge, bool do_load_balancing) {
309 StackEntry stack_loc{};
310
311 // std::cout << dump_status();
312
313 if (!is_mpi_sycl_interop_active())
315 "sycl mpi interop not initialized");
316
317 shambase::Timer timer;
318 shamlog_debug_ln("Scheduler", "running scheduler step");
319
320 struct SchedulerStepTimers {
321 shambase::Timer global_timer;
322 shambase::Timer metadata_sync;
323 std::optional<shambase::Timer> global_idx_map_build = {};
324 std::optional<shambase::Timer> patch_tree_count_reduce = {};
325 std::optional<shambase::Timer> gen_merge_split_rq = {};
326 std::optional<u32_2> split_merge_cnt = {};
327 std::optional<shambase::Timer> apply_splits = {};
328 std::optional<shambase::Timer> load_balance_compute = {};
329 std::optional<u32> load_balance_move_op_cnt = {};
330 std::optional<shambase::Timer> load_balance_apply = {};
331
332 void print_stats() {
333 if (shamcomm::world_rank() == 0) {
334 f64 total = global_timer.nanosec;
335 std::string str = "";
336 str += "Scheduler step timings : ";
337 str += shambase::format(
338 "\n metadata sync : {:<10} ({:2.1f}%)",
339 metadata_sync.get_time_str(),
340 f64(100 * (metadata_sync.nanosec / total)));
341 if (patch_tree_count_reduce) {
342 str += shambase::format(
343 "\n patch tree reduce : {:<10} ({:2.1f}%)",
344 patch_tree_count_reduce->get_time_str(),
345 100 * (patch_tree_count_reduce->nanosec / total));
346 }
347 if (gen_merge_split_rq) {
348 str += shambase::format(
349 "\n gen split merge : {:<10} ({:2.1f}%)",
350 gen_merge_split_rq->get_time_str(),
351 100 * (gen_merge_split_rq->nanosec / total));
352 }
353 if (split_merge_cnt) {
354 str += shambase::format(
355 "\n split / merge op : {}/{}",
356 split_merge_cnt->x(),
357 split_merge_cnt->y());
358 }
359 if (apply_splits) {
360 str += shambase::format(
361 "\n apply split merge : {:<10} ({:2.1f}%)",
362 apply_splits->get_time_str(),
363 100 * (apply_splits->nanosec / total));
364 }
365 if (load_balance_compute) {
366 str += shambase::format(
367 "\n LB compute : {:<10} ({:2.1f}%)",
368 load_balance_compute->get_time_str(),
369 100 * (load_balance_compute->nanosec / total));
370 }
371 if (load_balance_move_op_cnt) {
372 str += shambase::format(
373 "\n LB move op cnt : {}", *load_balance_move_op_cnt);
374 }
375 if (load_balance_apply) {
376 str += shambase::format(
377 "\n LB apply : {:<10} ({:2.1f}%)",
378 load_balance_apply->get_time_str(),
379 100 * (load_balance_apply->nanosec / total));
380 }
381 logger::info_ln("Scheduler", str);
382 }
383 }
384 } timers;
385
386 timers.global_timer.start();
387
388 patch_list.check_load_values_valid();
389
390 timers.metadata_sync.start();
391 patch_list.build_global();
392 timers.metadata_sync.stop();
393
394 // std::cout << dump_status();
395
396 std::unordered_set<u64> split_rq;
397 std::unordered_set<u64> merge_rq;
398
399 if (do_split_merge) {
400 // std::cout << dump_status() << std::endl;
401
402 // std::cout << "build_global_idx_map" <<std::endl;
403 timers.global_idx_map_build = shambase::Timer{};
404 timers.global_idx_map_build->start(); // TODO check if it it used outside of split merge ->
405 // maybe need to be put before the if
406 patch_list.build_global_idx_map();
407 timers.global_idx_map_build->stop();
408
409 // std::cout << dump_status() << std::endl;
410
411 // std::cout << "tree partial_values_reduction" <<std::endl;
412 timers.patch_tree_count_reduce = shambase::Timer{};
413 timers.patch_tree_count_reduce->start();
414 patch_tree.partial_values_reduction(patch_list.global, patch_list.id_patch_to_global_idx);
415 timers.patch_tree_count_reduce->stop();
416
417 // std::cout << dump_status() << std::endl;
418
419 // Generate merge and split request
420 timers.gen_merge_split_rq = shambase::Timer{};
421 timers.gen_merge_split_rq->start();
422 split_rq = patch_tree.get_split_request(crit_patch_split);
423 merge_rq = patch_tree.get_merge_request(crit_patch_merge);
424 timers.gen_merge_split_rq->stop();
425
426 timers.split_merge_cnt = u32_2{split_rq.size(), merge_rq.size()};
427 /*
428 std::cout << " |-> split rq : ";
429 for(u64 i : split_rq){
430 std::cout << i << " ";
431 }std::cout << std::endl;
432 //*/
433
434 /*
435 std::cout << " |-> merge rq : ";
436 for(u64 i : merge_rq){
437 std::cout << i << " ";
438 }std::cout << std::endl;
439 //*/
440
441 // std::cout << dump_status() << std::endl;
442
443 // std::cout << "split_patches" <<std::endl;
444 timers.apply_splits = shambase::Timer{};
445 timers.apply_splits->start();
446 split_patches(split_rq);
447 timers.apply_splits->stop();
448
449 // std::cout << dump_status() << std::endl;
450
451 // check not necessary if no splits
452 patch_list.build_global_idx_map();
453
454 set_patch_pack_values(merge_rq);
455 }
456
457 if (do_load_balancing) {
458 StackEntry stack_loc{};
459 timers.load_balance_compute = shambase::Timer{};
460 timers.load_balance_compute->start();
461 // generate LB change list
464 timers.load_balance_compute->stop();
465
466 timers.load_balance_move_op_cnt = change_list.change_ops.size();
467
468 timers.load_balance_apply = shambase::Timer{};
469 timers.load_balance_apply->start();
470 // apply LB change list
471 patch_data.apply_change_list(change_list, patch_list);
472 timers.load_balance_apply->stop();
473 }
474
475 // std::cout << dump_status();
476
477 if (do_split_merge) {
478 patch_list.build_local_idx_map();
479 merge_patches(merge_rq);
480 }
481
482 // TODO should be moved out of the scheduler step
483 owned_patch_id = patch_list.build_local();
484 patch_list.reset_local_pack_index();
485 patch_list.build_local_idx_map();
486 patch_list.build_global_idx_map(); // TODO check if required : added because possible bug
487 // because of for each patch & serial patch tree
488 // update_local_dtcnt_value();
489 // update_local_load_value(); disable the load value compute it should be done only in the
490 // models
491
492 if (split_rq.size() > 0 || merge_rq.size() > 0) {
493 patch_list.invalidate_load_values();
494 }
495
496 // std::cout << dump_status();
497
498 timers.global_timer.stop();
499 timers.print_stats();
500}
501
502/*
503void SchedulerMPI::scheduler_step(bool do_split_merge,bool do_load_balancing){
504
505 // update patch list
506 patch_list.sync_global();
507
508
509 if(do_split_merge){
510 // rebuild patch index map
511 patch_list.build_global_idx_map();
512
513 // apply reduction on leafs and corresponding parents
514 patch_tree.partial_values_reduction(
515 patch_list.global,
516 patch_list.id_patch_to_global_idx);
517
518 // Generate merge and split request
519 std::unordered_set<u64> split_rq = patch_tree.get_split_request(crit_patch_split);
520 std::unordered_set<u64> merge_rq = patch_tree.get_merge_request(crit_patch_merge);
521
522
523 // apply split requests
524 // update patch_list.global same on every node
525 // and split patchdata accordingly if owned
526 // & update tree
527 split_patches(split_rq);
528
529 // update packing index
530 // same operation on every cluster nodes
531 set_patch_pack_values(merge_rq);
532
533 // update patch list
534 // necessary to update load values in splitted patches
535 // alternative : disable this step and set fake load values (load parent / 8)
536 //alternative impossible if gravity because we have to compute the multipole
537 owned_patch_id = patch_list.build_local();
538 patch_list.sync_global();
539 }
540
541 if(do_load_balancing){
542 // generate LB change list
543 std::vector<std::tuple<u64, i32, i32,i32>> change_list =
544 make_change_list(patch_list.global);
545
546 // apply LB change list
547 patch_data.apply_change_list(change_list, patch_list);
548 }
549
550 if(do_split_merge){
551 // apply merge requests
552 // & update tree
553 merge_patches(merge_rq);
554
555
556
557 // if(Merge) update patch list
558 if(! merge_rq.empty()){
559 owned_patch_id = patch_list.build_local();
560 patch_list.sync_global();
561 }
562 }
563
564 //rebuild local table
565 owned_patch_id = patch_list.build_local();
566}
567//*/
568
569std::string PatchScheduler::dump_status() {
570
571 using namespace shamrock::patch;
572
573 std::stringstream ss;
574
575 ss << "----- MPI Scheduler dump -----\n\n";
576 ss << " -> SchedulerPatchList\n";
577
578 ss << " len global : " << patch_list.global.size() << "\n";
579 ss << " len local : " << patch_list.local.size() << "\n";
580
581 ss << " global content : \n";
582 for (Patch &p : patch_list.global) {
583
584 ss << " -> " << p.id_patch << " : " << p.load_value << " " << p.node_owner_id << " "
585 << p.pack_node_index << " "
586 << "( [" << p.coord_min[0] << "," << p.coord_max[0] << "] "
587 << " [" << p.coord_min[1] << "," << p.coord_max[1] << "] "
588 << " [" << p.coord_min[2] << "," << p.coord_max[2] << "] )\n";
589 }
590 ss << " local content : \n";
591 for (Patch &p : patch_list.local) {
592
593 ss << " -> id : " << p.id_patch << " : " << p.load_value << " " << p.node_owner_id
594 << " " << p.pack_node_index << " "
595 << "( [" << p.coord_min[0] << "," << p.coord_max[0] << "] "
596 << " [" << p.coord_min[1] << "," << p.coord_max[1] << "] "
597 << " [" << p.coord_min[2] << "," << p.coord_max[2] << "] )\n";
598 }
599
600 ss << shambase::format(
601 "patch_list.id_patch_to_global_idx :\n{}\n", patch_list.id_patch_to_global_idx);
602 ss << shambase::format(
603 "patch_list.id_patch_to_local_idx :\n{}\n", patch_list.id_patch_to_local_idx);
604
605 ss << " -> SchedulerPatchData\n";
606 ss << " owned data : \n";
607
608 patch_data.for_each_patchdata([&](u64 patch_id, shamrock::patch::PatchDataLayer &pdat) {
609 ss << "patch id : " << patch_id << " len = " << pdat.get_obj_cnt() << "\n";
610 });
611
612 /*
613 for(auto & [k,pdat] : patch_data.owned_data){
614 ss << " -> id : " << k << " len : (" <<
615 pdat.pos_s.size() << " " <<pdat.pos_d.size() << " " <<
616 pdat.U1_s.size() << " " <<pdat.U1_d.size() << " " <<
617 pdat.U3_s.size() << " " <<pdat.U3_d.size() << " "
618 << ")\n";
619 }
620 */
621
622 ss << " -> SchedulerPatchTree\n";
623
624 for (auto &[k, pnode] : patch_tree.tree) {
625 ss << shambase::format(
626 " -> id : {} -> ({}) <=> {} [{}, {}] (cl={} il={} l={} pid={})\n",
627 k,
628 pnode.tree_node.childs_nid,
629 pnode.linked_patchid,
630 pnode.patch_coord.coord_min,
631 pnode.patch_coord.coord_max,
632 pnode.tree_node.child_are_all_leafs,
633 pnode.tree_node.is_leaf,
634 pnode.tree_node.level,
635 pnode.tree_node.parent_nid);
636 }
637
638 return ss.str();
639}
640
641std::string PatchScheduler::format_patch_coord(shamrock::patch::Patch p) {
642 std::string ret;
643 if (pdl_old().check_main_field_type<f32_3>()) {
644 auto [bmin, bmax] = patch_data.sim_box.patch_coord_to_domain<f32_3>(p);
645 ret = shambase::format("coord = {} {}", bmin, bmax);
646 } else if (pdl_old().check_main_field_type<f64_3>()) {
647 auto [bmin, bmax] = patch_data.sim_box.patch_coord_to_domain<f64_3>(p);
648 ret = shambase::format("coord = {} {}", bmin, bmax);
649 } else if (pdl_old().check_main_field_type<u32_3>()) {
650 auto [bmin, bmax] = patch_data.sim_box.patch_coord_to_domain<u32_3>(p);
651 ret = shambase::format("coord = {} {}", bmin, bmax);
652 } else if (pdl_old().check_main_field_type<u64_3>()) {
653 auto [bmin, bmax] = patch_data.sim_box.patch_coord_to_domain<u64_3>(p);
654 ret = shambase::format("coord = {} {}", bmin, bmax);
655 } else {
657 "the main field does not match any");
658 }
659 return ret;
660}
661
662template<class vec>
663void check_locality_t(PatchScheduler &sched) {
664
665 StackEntry stack_loc{};
666
667 using namespace shamrock::patch;
669 PatchDataField<vec> &main_field = pdat.get_field<vec>(0);
670 auto [bmin_p0, bmax_p0] = sched.patch_data.sim_box.patch_coord_to_domain<vec>(p);
671
672 main_field.check_err_range(
673 [&](vec val, vec vmin, vec vmax) {
674 return Patch::is_in_patch_converted(val, vmin, vmax);
675 },
676 bmin_p0,
677 bmax_p0,
678 shambase::format("patch id = {}", pid));
679 });
680}
681
682void PatchScheduler::check_patchdata_locality_correctness() {
683
684 StackEntry stack_loc{};
685
686 if (pdl_old().check_main_field_type<f32_3>()) {
687 check_locality_t<f32_3>(*this);
688 } else if (pdl_old().check_main_field_type<f64_3>()) {
689 check_locality_t<f64_3>(*this);
690 } else if (pdl_old().check_main_field_type<u32_3>()) {
691 check_locality_t<u32_3>(*this);
692 } else if (pdl_old().check_main_field_type<u64_3>()) {
693 check_locality_t<u64_3>(*this);
694 } else if (pdl_old().check_main_field_type<i64_3>()) {
695 check_locality_t<i64_3>(*this);
696 } else {
698 "the main field does not match any");
699 }
700}
701
702void PatchScheduler::split_patches(std::unordered_set<u64> split_rq) {
703 StackEntry stack_loc{};
704 for (u64 tree_id : split_rq) {
705
706 patch_tree.split_node(tree_id);
707 PatchTree::Node &splitted_node = patch_tree.tree[tree_id];
708
709 shamrock::patch::Patch old_patch
710 = patch_list.global[patch_list.id_patch_to_global_idx[splitted_node.linked_patchid]];
711
712 auto [idx_p0, idx_p1, idx_p2, idx_p3, idx_p4, idx_p5, idx_p6, idx_p7]
713 = patch_list.split_patch(splitted_node.linked_patchid);
714
715 u64 old_patch_id = splitted_node.linked_patchid;
716
717 splitted_node.linked_patchid = u64_max;
718 patch_tree.tree[splitted_node.tree_node.childs_nid[0]].linked_patchid
719 = patch_list.global[idx_p0].id_patch;
720 patch_tree.tree[splitted_node.tree_node.childs_nid[1]].linked_patchid
721 = patch_list.global[idx_p1].id_patch;
722 patch_tree.tree[splitted_node.tree_node.childs_nid[2]].linked_patchid
723 = patch_list.global[idx_p2].id_patch;
724 patch_tree.tree[splitted_node.tree_node.childs_nid[3]].linked_patchid
725 = patch_list.global[idx_p3].id_patch;
726 patch_tree.tree[splitted_node.tree_node.childs_nid[4]].linked_patchid
727 = patch_list.global[idx_p4].id_patch;
728 patch_tree.tree[splitted_node.tree_node.childs_nid[5]].linked_patchid
729 = patch_list.global[idx_p5].id_patch;
730 patch_tree.tree[splitted_node.tree_node.childs_nid[6]].linked_patchid
731 = patch_list.global[idx_p6].id_patch;
732 patch_tree.tree[splitted_node.tree_node.childs_nid[7]].linked_patchid
733 = patch_list.global[idx_p7].id_patch;
734
735 try {
736 patch_data.split_patchdata(
737 old_patch_id,
738 {patch_list.global[idx_p0],
739 patch_list.global[idx_p1],
740 patch_list.global[idx_p2],
741 patch_list.global[idx_p3],
742 patch_list.global[idx_p4],
743 patch_list.global[idx_p5],
744 patch_list.global[idx_p6],
745 patch_list.global[idx_p7]});
746 } catch (const PatchDataRangeCheckError &e) {
747 logger::err_ln("SchedulerPatchData", "catched range issue with patchdata split");
748
749 logger::raw_ln(" old patch", old_patch.id_patch, format_patch_coord(old_patch));
750
751 logger::err_ln("Scheduler", "global patch list :");
752 for (shamrock::patch::Patch &p : patch_list.global) {
753 logger::raw_ln(" patch", p.id_patch, format_patch_coord(p));
754 }
755
757 "\n Initial error : "
758 + shambase::increase_indent(std::string("\n") + e.what(), "\n |"));
759 }
760 }
761}
762
763inline void PatchScheduler::merge_patches(std::unordered_set<u64> merge_rq) {
764 StackEntry stack_loc{};
765 for (u64 tree_id : merge_rq) {
766
767 PatchTree::Node &to_merge_node = patch_tree.tree[tree_id];
768
769 // std::cout << "merging patch tree id : " << tree_id << "\n";
770
771 u64 patch_id0 = patch_tree.tree[to_merge_node.tree_node.childs_nid[0]].linked_patchid;
772 u64 patch_id1 = patch_tree.tree[to_merge_node.tree_node.childs_nid[1]].linked_patchid;
773 u64 patch_id2 = patch_tree.tree[to_merge_node.tree_node.childs_nid[2]].linked_patchid;
774 u64 patch_id3 = patch_tree.tree[to_merge_node.tree_node.childs_nid[3]].linked_patchid;
775 u64 patch_id4 = patch_tree.tree[to_merge_node.tree_node.childs_nid[4]].linked_patchid;
776 u64 patch_id5 = patch_tree.tree[to_merge_node.tree_node.childs_nid[5]].linked_patchid;
777 u64 patch_id6 = patch_tree.tree[to_merge_node.tree_node.childs_nid[6]].linked_patchid;
778 u64 patch_id7 = patch_tree.tree[to_merge_node.tree_node.childs_nid[7]].linked_patchid;
779
780 // print list of patch that will merge
781 // std::cout << format(" -> (%d %d %d %d %d %d %d %d)\n", patch_id0, patch_id1, patch_id2,
782 // patch_id3, patch_id4, patch_id5, patch_id6, patch_id7);
783
784 if (patch_list.global[patch_list.id_patch_to_global_idx[patch_id0]].node_owner_id
786 patch_data.merge_patchdata(
787 patch_id0,
788 {patch_id0,
789 patch_id1,
790 patch_id2,
791 patch_id3,
792 patch_id4,
793 patch_id5,
794 patch_id6,
795 patch_id7});
796 }
797
798 patch_list.merge_patch(
799 patch_list.id_patch_to_global_idx[patch_id0],
800 patch_list.id_patch_to_global_idx[patch_id1],
801 patch_list.id_patch_to_global_idx[patch_id2],
802 patch_list.id_patch_to_global_idx[patch_id3],
803 patch_list.id_patch_to_global_idx[patch_id4],
804 patch_list.id_patch_to_global_idx[patch_id5],
805 patch_list.id_patch_to_global_idx[patch_id6],
806 patch_list.id_patch_to_global_idx[patch_id7]);
807
808 patch_tree.merge_node_dm1(tree_id);
809
810 to_merge_node.linked_patchid = patch_id0;
811 }
812}
813
814inline void PatchScheduler::set_patch_pack_values(std::unordered_set<u64> merge_rq) {
815
816 for (u64 tree_id : merge_rq) {
817
818 PatchTree::Node &to_merge_node = patch_tree.tree[tree_id];
819
820 u64 idx_pack
821 = patch_list.id_patch_to_global_idx[patch_tree.tree[to_merge_node.get_child_nid(0)]
822 .linked_patchid];
823
824 // std::cout << "node id : " << patch_list.global[idx_pack].id_patch << " should merge with
825 // : ";
826
827 for (u8 i = 1; i < 8; i++) {
828 // std::cout << patch_tree.tree[to_merge_node.get_child_nid(i)].linked_patchid << " ";
830 .global[patch_list.id_patch_to_global_idx
831 [patch_tree.tree[to_merge_node.get_child_nid(i)].linked_patchid]]
832 .pack_node_index = idx_pack;
833 } // std::cout << std::endl;
834 }
835}
836
837void PatchScheduler::dump_local_patches(std::string filename) {
838
839 using namespace shamrock::patch;
840
841 std::ofstream fout(filename);
842
843 if (pdl_old().check_main_field_type<f32_3>()) {
844
845 std::tuple<f32_3, f32_3> box_transform = get_box_tranform<f32_3>();
846
847 for (const Patch &p : patch_list.local) {
848
849 f32_3 box_min
850 = f32_3{p.coord_min[0], p.coord_min[1], p.coord_min[2]} * std::get<1>(box_transform)
851 + std::get<0>(box_transform);
852 f32_3 box_max = (f32_3{p.coord_max[0], p.coord_max[1], p.coord_max[2]} + 1)
853 * std::get<1>(box_transform)
854 + std::get<0>(box_transform);
855
856 fout << p.id_patch << "|" << p.load_value << "|" << p.node_owner_id << "|"
857 << p.pack_node_index << "|" << box_min.x() << "|" << box_max.x() << "|"
858 << box_min.y() << "|" << box_max.y() << "|" << box_min.z() << "|" << box_max.z()
859 << "|" << "\n";
860 }
861
862 fout.close();
863
864 } else if (pdl_old().check_main_field_type<f64_3>()) {
865
866 std::tuple<f64_3, f64_3> box_transform = get_box_tranform<f64_3>();
867
868 for (const Patch &p : patch_list.local) {
869
870 f64_3 box_min
871 = f64_3{p.coord_min[0], p.coord_min[1], p.coord_min[2]} * std::get<1>(box_transform)
872 + std::get<0>(box_transform);
873 f64_3 box_max = (f64_3{p.coord_max[0], p.coord_max[1], p.coord_max[3]} + 1)
874 * std::get<1>(box_transform)
875 + std::get<0>(box_transform);
876
877 fout << p.id_patch << "|" << p.load_value << "|" << p.node_owner_id << "|"
878 << p.pack_node_index << "|" << box_min.x() << "|" << box_max.x() << "|"
879 << box_min.y() << "|" << box_max.y() << "|" << box_min.z() << "|" << box_max.z()
880 << "|" << "\n";
881 }
882
883 fout.close();
884
885 } else {
887 "the chosen type for the main field is not handled");
888 }
889}
890
891struct Message {
892 std::unique_ptr<shamcomm::CommunicationBuffer> buf;
893 i32 rank;
894 i32 tag;
895};
896
897void send_messages(std::vector<Message> &msgs, std::vector<MPI_Request> &rqs) {
898 for (auto &msg : msgs) {
899 rqs.push_back(MPI_Request{});
900 u32 rq_index = rqs.size() - 1;
901 auto &rq = rqs[rq_index];
902
903 u64 bsize = msg.buf->get_size();
904 if (bsize % 8 != 0) {
906 "the following mpi comm assume that we can send longs to pack 8byte");
907 }
908 u64 lcount = bsize / 8;
909 if (lcount > i32_max) {
910 shambase::throw_with_loc<std::runtime_error>("The message is too large for MPI");
911 }
912
914 msg.buf->get_ptr(),
915 lcount,
916 get_mpi_type<u64>(),
917 msg.rank,
918 msg.tag,
919 MPI_COMM_WORLD,
920 &rq);
921 }
922}
923
924void recv_probe_messages(std::vector<Message> &msgs, std::vector<MPI_Request> &rqs) {
925
926 for (auto &msg : msgs) {
927 rqs.push_back(MPI_Request{});
928 u32 rq_index = rqs.size() - 1;
929 auto &rq = rqs[rq_index];
930
931 MPI_Status st;
932 i32 cnt;
933 shamcomm::mpi::Probe(msg.rank, msg.tag, MPI_COMM_WORLD, &st);
934 shamcomm::mpi::Get_count(&st, get_mpi_type<u64>(), &cnt);
935
936 msg.buf = std::make_unique<shamcomm::CommunicationBuffer>(
937 cnt * 8, shamsys::instance::get_compute_scheduler_ptr());
938
940 msg.buf->get_ptr(), cnt, get_mpi_type<u64>(), msg.rank, msg.tag, MPI_COMM_WORLD, &rq);
941 }
942}
943
944std::vector<std::unique_ptr<shamrock::patch::PatchDataLayer>> PatchScheduler::gather_data(
945 u32 rank) {
946
947 using namespace shamrock::patch;
948
949 auto plist = this->patch_list.global;
950 auto pdata = this->patch_data.owned_data;
951
952 auto serializer = [](shamrock::patch::PatchDataLayer &pdat) {
953 shamalgs::SerializeHelper ser(shamsys::instance::get_compute_scheduler_ptr());
954 ser.allocate(pdat.serialize_buf_byte_size());
955 pdat.serialize_buf(ser);
956 return ser.finalize();
957 };
958
959 auto deserializer = [&](sham::DeviceBuffer<u8> &&buf) {
960 // exchange the buffer held by the distrib data and give it to the serializer
961 shamalgs::SerializeHelper ser(
962 shamsys::instance::get_compute_scheduler_ptr(),
963 std::forward<sham::DeviceBuffer<u8>>(buf));
964 return shamrock::patch::PatchDataLayer::deserialize_buf(ser, get_layout_ptr_old());
965 };
966
967 std::vector<Message> send_payloads;
968
969 for (u32 i = 0; i < plist.size(); i++) {
970 auto &cpatch = plist[i];
971 if (cpatch.node_owner_id == shamcomm::world_rank()) {
972 auto &patchdata = pdata.get(cpatch.id_patch);
973
974 sham::DeviceBuffer<u8> tmp = serializer(patchdata);
975
976 send_payloads.push_back(
977 Message{
978 .buf = std::make_unique<shamcomm::CommunicationBuffer>(
979 std::move(tmp), shamsys::instance::get_compute_scheduler_ptr()),
980 .rank = 0,
981 .tag = i32(i)});
982 }
983 }
984
985 std::vector<MPI_Request> rqs;
986 send_messages(send_payloads, rqs);
987
988 std::vector<Message> recv_payloads;
989
990 if (shamcomm::world_rank() == 0) {
991 for (u32 i = 0; i < plist.size(); i++) {
992 recv_payloads.push_back(
993 Message{
994 .buf = std::unique_ptr<shamcomm::CommunicationBuffer>{},
995 .rank = i32(plist[i].node_owner_id),
996 .tag = i32(i)});
997 }
998 }
999
1000 // receive
1001 recv_probe_messages(recv_payloads, rqs);
1002
1003 std::vector<MPI_Status> st_lst(rqs.size());
1004 shamcomm::mpi::Waitall(rqs.size(), rqs.data(), st_lst.data());
1005
1006 std::vector<std::unique_ptr<PatchDataLayer>> ret;
1007 for (auto &recv_msg : recv_payloads) {
1008 shamcomm::CommunicationBuffer comm_buf = shambase::extract_pointer(recv_msg.buf);
1009
1010 sham::DeviceBuffer<u8> buf
1011 = shamcomm::CommunicationBuffer::convert_usm(std::move(comm_buf));
1012
1013 ret.push_back(std::make_unique<PatchDataLayer>(deserializer(std::move(buf))));
1014 }
1015
1016 return ret;
1017}
1018
1019nlohmann::json PatchScheduler::serialize_patch_metadata() {
1020
1021 nlohmann::json jsim_box;
1022 patch_data.sim_box.to_json(jsim_box);
1023
1024 return {
1025 {"patchtree", patch_tree},
1026 {"patchlist", patch_list},
1027 {"patchdata_layout", pdl_old()},
1028 {"sim_box", jsim_box},
1029 {"crit_patch_split", crit_patch_split},
1030 {"crit_patch_merge", crit_patch_merge},
1031 {"synchronized_data", synchronized_data}};
1032}
function to run load balancing with the hilbert curve
Header file describing a Node Instance.
MPI scheduler.
double f64
Alias for double.
std::uint8_t u8
8 bit unsigned integer
std::uint32_t u32
32 bit unsigned integer
std::uint64_t u64
64 bit unsigned integer
std::uint16_t u16
16 bit unsigned integer
std::int32_t i32
32 bit integer
The MPI scheduler.
void for_each_patch_data(Function &&fct)
for each macro for patchadata example usage
SchedulerPatchData patch_data
handle the data of the patches of the scheduler
u64 crit_patch_split
splitting limit (if load value > crit_patch_split => patch split)
PatchTree patch_tree
handle the tree structure of the patches
void scheduler_step(bool do_split_merge, bool do_load_balancing)
scheduler step
SynchronizedData synchronized_data
data that is synchroneous across all ranks
SchedulerPatchList patch_list
handle the list of the patches of the scheduler
std::unordered_set< u64 > owned_patch_id
(owned_patch_id = patch_list.build_local())
std::vector< u64 > add_root_patches(std::vector< shamrock::patch::PatchCoord< 3 > > coords)
add a root patch to the scheduler
u64 crit_patch_merge
merging limit (if load value < crit_patch_merge => patch merge)
void allpush_data(shamrock::patch::PatchDataLayer &pdat)
push data in the scheduler The content of pdat as to be the same for each node
void add_root_patch()
add patch to the scheduler
std::vector< shamrock::patch::Patch > local
contain the list of patch owned by the current node
std::vector< shamrock::patch::Patch > global
contain the list of all patches in the simulation
Class Timer measures the time elapsed since the timer was started.
Definition Timer.hpp:36
std::string get_time_str() const
Converts the stored nanosecond time to a string representation.
Definition Timer.hpp:79
void start()
Starts the timer.
Definition Timer.hpp:51
f64 nanosec
Time in nanoseconds.
Definition Timer.hpp:43
static sham::DeviceBuffer< u8 > convert_usm(CommunicationBuffer &&buf)
destroy the buffer and recover the held object
PatchDataLayer container class, the layout is described in patchdata_layout.
void insert_elements_in_range(PatchDataLayer &pdat, T bmin, T bmax)
insert elements of pdat only if they are within the range
std::tuple< T, T > patch_coord_to_domain(const Patch &p) const
get the patch coordinates on the domain
Definition SimBox.hpp:300
static LoadBalancingChangeList make_change_list(std::vector< shamrock::patch::Patch > &global_patch_list)
std::array< u64, 8 > childs_nid
Array of childs node ids.
shamrock::patch::SimulationBoxInfo sim_box
simulation box geometry info
This header file contains utility functions related to exception handling in the code.
constexpr T roundup_pow2_clz(T v) noexcept
round up to the next power of two 0 is rounded up to 1 as it is not a pow of 2 every input above the ...
Definition math.hpp:805
void throw_with_loc(std::string message, SourceLocation loc=SourceLocation{})
Throw an exception and append the source location to it.
ExcptTypes make_except_with_loc(std::string message, SourceLocation loc=SourceLocation{})
Create an exception with a message and a location.
std::string increase_indent(std::string in, std::string delim="\n ")
Increase indentation of a string.
Definition string.hpp:124
auto extract_pointer(std::unique_ptr< T > &o, SourceLocation loc=SourceLocation()) -> T
extract content out of unique_ptr
Definition memory.hpp:227
i32 world_rank()
Gives the rank of the current process in the MPI communicator.
Definition worldInfo.cpp:40
constexpr u64 u64_max
u64 max value
constexpr i32 i32_max
i32 max value
header for PatchData related function and declaration
void info_ln(std::string module_name, Types... var2)
Prints a log message with multiple arguments followed by a newline.
Definition logs.hpp:133
This file contains the definition for the stacktrace related functionality.
shambase::details::BasicStackEntry StackEntry
Alias for shambase::details::BasicStackEntry.
Patch object that contain generic patch information.
Definition Patch.hpp:33
static bool is_in_patch_converted(sycl::vec< T, 3 > val, sycl::vec< T, 3 > min_val, sycl::vec< T, 3 > max_val)
check if particle is in the asked range, given the output of @convert_coord
Definition Patch.hpp:210
u64 pack_node_index
this value mean "to pack with index xxx in the global patch table" and not "to pack with id_pach == x...
Definition Patch.hpp:87
u32 node_owner_id
node rank owner of this patch
Definition Patch.hpp:93
u64 load_value
if synchronized contain the load value of the patch
Definition Patch.hpp:88
u64 id_patch
unique key that identify the patch
Definition Patch.hpp:86
header file to manage sycl
void Get_count(const MPI_Status *status, MPI_Datatype datatype, int *count)
MPI wrapper for MPI_Get_count.
Definition wrapper.cpp:222
void Irecv(void *buf, int count, MPI_Datatype datatype, int source, int tag, MPI_Comm comm, MPI_Request *request)
MPI wrapper for MPI_Irecv.
Definition wrapper.cpp:102
void Probe(int source, int tag, MPI_Comm comm, MPI_Status *status)
MPI wrapper for MPI_Probe.
Definition wrapper.cpp:201
void Allreduce(const void *sendbuf, void *recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
MPI wrapper for MPI_Allreduce.
Definition wrapper.cpp:119
void Waitall(int count, MPI_Request array_of_requests[], MPI_Status *array_of_statuses)
MPI wrapper for MPI_Waitall.
Definition wrapper.cpp:187
void Isend(const void *buf, int count, MPI_Datatype datatype, int dest, int tag, MPI_Comm comm, MPI_Request *request)
MPI wrapper for MPI_Isend.
Definition wrapper.cpp:85