← Documentation
ADS-IDAC Engine 5.0.0
ParallelExecution.cpp
Go to the documentation of this file.
1#include "ParallelExecution.h"
2#include "../Scheduler/Scheduler.h"
3#include "../Notification/Notification.h"
4#include "../Global/SimulationContext.h"
5#include "../Global/OutputMerge.h"
6
7
8using namespace std;
9namespace fs = std::filesystem;
10
11parallelexec::parallelexec() {
12 // TODO: Add parallelization options here, we can also use input and read them as program options
13}
14
15// Definition of the global pointer
16Client* g_client = nullptr;
17
18// The setter function
19void set_global_client(Client* c) {
20 g_client = c;
21}
22
23void parallelexec::parallel_init (int *argc, char **argv[])
24{
25 double wtime = MPI_Wtime();
26 MPI_Comm_rank (MPI_COMM_WORLD, &PG::rank);
27 MPI_Comm_size (MPI_COMM_WORLD, &PG::size);
28 // i think the following needs to be done only once
29 g_sim.SScheduler.parseOptions(*argc, *argv);
30
31 if (!PG::rank) {
32 std::cout << "MPI execution (" << PG::size << " ranks: 1 server + "
33 << PG::size - 1 << " workers)" << std::endl;
34 std::cout << "Project: " << g_sim.SScheduler.getProjectDirectory() << std::endl;
35 }
36
37 Setup_Directories();
38
39 if (!PG::rank){
40 Server server(PG::rank, g_sim.SScheduler.getVizPort());
41 server.Simulation();
42 server.Shutdown();
43 if (g_sim.config.merge_mode == 2 && !server.stoppedAtCheckpoint()) {
44 // --merge=defer: the run ends at tree completion; the rank dirs
45 // stay the only complete copy, recorded by the manifest, and a
46 // later --merge-only invocation consolidates them.
47 adsidac::merge::write_manifest(g_sim.SScheduler.getProjectDirectory(), PG::size,
48 g_sim.config.aggregate_pe_output,
49 server.getSequencesCompleted(),
50 g_sim.config_fingerprint);
51 PU::print_message(PG::rank, "Merge deferred: rank outputs kept. Consolidate with: "
52 "adsidac --merge-only --project-dir=" + g_sim.SScheduler.getProjectDirectory()
53 + " --input-deck=" + g_sim.SScheduler.getInputDeckFileDirectoryAndName(), PG::log);
54 } else {
55 if (g_sim.config.merge_mode == 2)
56 PU::print_message(PG::rank, "[MERGE] --merge=defer ignored: a checkpoint-stop "
57 "needs its intermediate merge for resume", PG::log);
58 server.merge_outputs();
59 }
60 // Send completion notification from rank 0
61 {
62 RunSummary summary;
63 summary.project = g_sim.SScheduler.getProjectDirectory();
64 summary.sequences = server.getSequencesCompleted();
65 summary.wall_seconds = MPI_Wtime() - wtime;
66 summary.ranks = PG::size;
67 summary.solver = g_sim.config.solver_url.empty() ? "relap" : "mock";
68 summary.truncation_time = g_sim.SScheduler.getSequenceTruncationTime();
69 send_completion_notification(summary);
70 }
71 }
72 else {
73 ParallelUtilities::print_message(PG::rank, "Client Parallel Execution", PG::log);
74 // rank>0 => client
75 // Create a static or local client object
76 static Client client(PG::rank);
77
78 // Store pointer globally
79 set_global_client(&client);
80
81 // Run the client
82 client.Work();
83 }
84
85 MPI_Barrier(MPI_COMM_WORLD);
86
87 if (PG::rank == 0) {
88 PU::print_message(PG::rank, "Total simulation time: " + std::to_string(MPI_Wtime()-wtime), PG::log);
89 }
90}
91
92void parallelexec::Setup_Directories() {
93 // The rank directory is anchored to the PROJECT directory, not to the launch
94 // CWD (this runs before any chdir, so a relative "rank_N" would land
95 // wherever mpirun was started). Anchoring makes --project-dir isolate a
96 // run: two runs launched from the same directory with different projects
97 // never share rank_* directories or race inside remove_all().
98 // merge_outputs() reads "../rank_N/Output/" from inside rank_0, which
99 // resolves against the same base.
100 std::error_code ec;
101 const fs::path launch_cwd = fs::current_path(ec);
102 const fs::path base(g_sim.SScheduler.getProjectDirectory());
103 const fs::path rankDir = base / ("rank_" + std::to_string(PG::rank));
104
105 // error_code overloads throughout. These are the throwing overloads by
106 // default, and an uncaught filesystem_error here kills the rank with signal
107 // 6 and no message -- a filesystem hiccup should not be fatal.
108 fs::remove_all(rankDir, ec);
109 fs::create_directories(rankDir, ec);
110 if (ec) {
111 std::cerr << "rank " << PG::rank << ": cannot create " << rankDir
112 << " (" << ec.message() << ")" << std::endl;
113 }
114
115 // RELAP5 reads the steam tables from its working directory. Look beside the
116 // executable (the launch dir) first, then in the project dir, so both
117 // invocation styles work: `cd build/bin && mpirun ... --project-dir=<p>`
118 // and `cd <p> && mpirun <abs>/adsidac --project-dir=.`
119 for (const auto& name : {"tpfh2o", "tpfh2onew", "tpfd2o"}) {
120 for (const fs::path& dir : {launch_cwd, base}) {
121 const fs::path src = dir / name;
122 if (fs::exists(src, ec)) {
123 fs::copy(src, rankDir / name, fs::copy_options::overwrite_existing, ec);
124 if (ec)
125 std::cerr << "rank " << PG::rank << ": cannot copy " << name
126 << " (" << ec.message() << ")" << std::endl;
127 break;
128 }
129 }
130 }
131
132 // Change current working directory
133 const std::string dirName = rankDir.string();
134#ifdef _WIN32
135 if (!SetCurrentDirectory(dirName.c_str())) {
136 std::cerr << "Failed to change directory on Windows" << std::endl;
137 // Handle error
138 }
139#else
140 if (chdir(dirName.c_str()) != 0) {
141 perror("Failed to change directory on Unix/Linux");
142 // Handle error
143 }
144#endif
145
146 // Redirect stdout/stderr to per-rank log files
147 freopen("rank_output.log", "w", stdout);
148 freopen("rank_errors.log", "w", stderr);
149 setvbuf(stdout, nullptr, _IOLBF, 0); // line-buffered so last output before hang is visible
150 setvbuf(stderr, nullptr, _IOLBF, 0);
151}
152
153// // MPI_Datatype R5ParType;
154// // R5Par branches[maximum_number_of_branches];
155//
156// // void parallelexec::Setup_Types (void)
157// // {
158// // int blockcounts[] = { 4, 5 };
159// // MPI_Aint offsets[] = { 0, 4 * sizeof(double) };
160// // MPI_Datatype types[2];
161// // types[0] = MPI_DOUBLE;
162// // types[1] = MPI_INT;
163// // MPI_Type_create_struct (2, blockcounts, offsets, types, &R5ParType);
164// // MPI_Type_commit (&R5ParType);
165// // }
166//
167//
168
169//
170// // SERVER IMPLEMENTATION
171//
172//
173//
174// // R5Par parallelexec::ReceiveNumberOfBranches (void)
175// // {
176// // // R5Par r;
177// // // MPI_Recv (&r, 1, R5ParType, MPI_ANY_SOURCE, tag_branchthis, MPI_COMM_WORLD, &statusServer);
178// // // printf ("\e[1;31mP[%i] - I'm getting the number of branches from %i\n\e[0m", rank, statusServer.MPI_SOURCE);
179// // // printf ("\e[1;31mP[%i] - This is what I received from interface %i\n\e[0m", rank, r.branches);
180// // // printf ("\e[1;31mP[%i] - This is what I received from interface %i\n\e[0m", rank, r.branches);
181// //
182// // // return r;
183// // }
184//
185// // CLIENT IMPLEMENTATION
186//
187// // R5Par r;
188//
189// void parallelexec::Branch_Sequence (int i)
190// {
191// printf ("\e[1;3%imP[%i] branching...\n\e[0m", rank + 1, rank);
192// // r.branches = 1;
193// // r.time = i + 1;
194// // MPI_Send (&r, 1, R5ParType, statusClient.MPI_SOURCE, tag_branchthis, MPI_COMM_WORLD);
195// }
196//
197// int parallelexec::Client ()
198// {
199// bool fresh_restart;
200// bool expected_abort;
201// int noOfBranches = 0;
202// string str;
203// // RestartNodeInfo *restart_info_pt;
204// // auto restart_info_pt = make_shared<RestartNodeInfo>();
205// while (true) {
206// MPI_Probe (MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &statusClient);
207// if (statusClient.MPI_TAG == tag_simulatethis) {
208// noOfBranches++;
209// // MPI_Recv (&r, 1, R5ParType, statusClient.MPI_SOURCE, tag_simulatethis, MPI_COMM_WORLD, &statusClient);
210// printf ("\e[1;3%imP[%i] Started simulating!\n\e[0m", rank + 1, rank);
211// ParallelUtilities::print_message(rank, "Started simulating!", true);
212// // Simulate_Sequence (&r);
213// // ++++++++++++++++++++++++++++++
214// // set system mode
215// r5pars_.tsimmode = CONSOLE_Mode;
216// int i;
217// time(&t_current);
218// time_begin = "";
219// time_begin += ctime(&t_current);
220// // seed random number generator using system time
221// srand((unsigned) time(0));
222// //
223// SScheduler.parseOptions(argc, argv);
224// SScheduler.initiationForSimulation();
225// Plant.initiationForSimulation();
226// Translator.initiationForSimulation();
227// Branch_center.initiationForSimulation();
228// //
229// fresh_restart = false;
230// do { /*!< do-loop to run all sequences */
231// i = Branch_center.getIDOfRecentGeneratedES() + 1;
232// cout << "*** Sequence " << i << " ****" << endl;
233// expected_abort = false;
234// relap5_(); /*!<<run one sequence */
235// i = remove("out.o"); /*!< remove the side product file generated by RELAP5 */
236// //! After the first run of RELAP5, change the input deck name to indeck.i
237// if (!expected_abort) { //sequesce termination was not detectedS
238// es_pt = new BranchEndState();
239// str = "Sequence is terminated for unknown reason.";
240// cout << "Sequence is terminated for unknown reason." << endl;
241// es_pt->setTypeOfConsequence(Consequence::NoKB);
242// es_pt->setConsequenceDescription(str);
243// Branch_center.addAnEndState(es_pt);
244// }
245// if (r5pars_.cnewrst) {
246// strcpy(r5pars_.carg2, "indeck.i");
247// r5pars_.carg2len = (int) strlen(r5pars_.carg2);
248// r5pars_.cnewrst = false; //finished first scenario, starts second scenario
249// }
250//
251//
252// if (rank == 1) {
253// // restart_info_pt = new RestartNodeInfo();
254// // restart_info_pt->saveRestartNodeInfo();
255// // restart_info_pt->setCorrespondingBranchClusterID(
256// // Branch_center.getIDOfRecentGeneratedBranchCluster());
257// //// Branch_center.branch_cluster.
258// // printf ("\e[1;3%imP[%i] - Writing branch center to file\n\e[0m", rank+1, rank);
259// // printf ("\e[1;3%imP[%i] - Sent === Corresponding brunch cluster ID : %i\n\e[0m", rank+1, rank, (*restart_info_pt).getCorrespondingBranchClusterID());
260// // MPI_Send(restart_info_pt, sizeof(*restart_info_pt), MPI_BYTE, 0, tag_branchthis, MPI_COMM_WORLD);
261// // MPI_Send(&Branch_center, sizeof(Branch_center), MPI_BYTE, 0, tag_branchthis, MPI_COMM_WORLD);
262// // MPI_Send(&SScheduler, sizeof(SScheduler), MPI_BYTE, 0, tag_branchthis, MPI_COMM_WORLD);
263// // MPI_Send(&Plant, sizeof(Plant), MPI_BYTE, 0, tag_branchthis, MPI_COMM_WORLD);
264// // MPI_Send(&Translator, sizeof(Translator), MPI_BYTE, 0, tag_branchthis, MPI_COMM_WORLD);
265// // saveToFile(*restart_info_pt, "restart_info.txt");
266// // restartforanewsequence_();
267// // cout << "rank_1 restart_info " << sizeof(restart_info) << endl;
268//
269// MPI_Send (nullptr, 0, MPI_INT, 2, tag_restart, MPI_COMM_WORLD); // BEWARE THIS IS NOT SUPPOSED TO BE tag_end bu ttag_restart
270// }
271//
272//
273// // d
274// cout << "terminateall: " << r5pars_.terminateall << endl;
275// } while (false);/*! if r5pars_.terminateall is true, then the simulation is terminated */
276// printf ("\e[1;3%imP[%i] Finished simulating!\n\e[0m", rank + 1, rank);
277// MPI_Send (nullptr, 0, MPI_INT, statusClient.MPI_SOURCE, tag_donesimulating, MPI_COMM_WORLD);
278// printf ("\e[1;3%imP[%i] Informed server I'm done!\n\e[0m", rank + 1, rank);
279// MPI_Send (nullptr, 0, MPI_INT, 0, tag_end, MPI_COMM_WORLD);
280// // printf ("\e[1;3%imP[%i] Number of branches simulated: %i!\n\e[0m", rank + 1, rank, noOfBranches);
281//
282// } else if (statusClient.MPI_TAG == tag_restart) {
283// if (rank == 3) {
284// // MPI_Recv(restart_info_pt, sizeof(*restart_info_pt), MPI_BYTE, 1, tag_branchthis, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
285// MPI_Recv(nullptr, 0, MPI_INT, 1, tag_restart, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
286// printf ("\e[1;3%imP[%i] Started simulating!\n\e[0m", rank + 1, rank);
287// // Simulate_Sequence (&r);
288// // ++++++++++++++++++++++++++++++
289// // set system mode
290// r5pars_.tsimmode = CONSOLE_Mode;
291// int i;
292// time(&t_current);
293// time_begin = "";
294// time_begin += ctime(&t_current);
295// // seed random number generator using system time
296// srand((unsigned) time(0));
297// //
298// // SScheduler.readInfoFromFile();
299// SScheduler.parseOptions(argc, argv);
300// SScheduler.initiationForSimulation();
301// Plant.initiationForSimulation();
302// Translator.initiationForSimulation();
303// Branch_center.initiationForSimulation();
304// //
305// fresh_restart = false;
306// // BranchRecord new_branch_center_from_file ;
307// printf ("\e[1;3%imP[%i] - Reading branch center from file\n\e[0m", rank+1, rank);
308// loadFromFile(Branch_center, "../branch_center.txt");
309//
310// i = Branch_center.getIDOfRecentGeneratedES() + 1;
311// cout << "*** Sequence " << i << " ****" << endl;
312// expected_abort = false;
313// i = remove("out.o"); /*!< remove the side product file generated by RELAP5 */
314// // relap5_();
315// // r5pars_.terminateall = 1;
316// i = remove("out.o"); /*!< remove the side product file generated by RELAP5 */
317// //! After the first run of RELAP5, change the input deck name to indeck.i
318// // if (!expected_abort) { //sequesce termination was not detected
319// // es_pt = new BranchEndState();
320// // str = "Sequence is terminated for unknown reason.";
321// // cout << "Sequence is terminated for unknown reason." << endl;
322// // es_pt->setTypeOfConsequence(Consequence::NoKB);
323// // es_pt->setConsequenceDescription(str);
324// // Branch_center.addAnEndState(es_pt);
325// // }
326// if (r5pars_.cnewrst) {
327// strcpy(r5pars_.carg2, "indeck.i");
328// r5pars_.carg2len = (int) strlen(r5pars_.carg2);
329// r5pars_.cnewrst = false; //finished first scenario, starts second scenario
330// }
331//
332// // file based
333// // std::shared_ptr<RestartNodeInfo> read_restart_info = std::make_shared<RestartNodeInfo>();
334// // loadFromFile(*read_restart_info, "../rank_1/restart_info.txt");
335// // restart_info.push_back(read_restart_info);
336//
337// int alp = 1;
338// while (!alp)
339// sleep(5);
340//
341// // Debugging: Check the value after pushing into vector
342// auto debug_ptr = restart_info.back();
343// cout << "Stored BranchClusterID: " << debug_ptr->getCorrespondingBranchClusterID() << endl;
344//
345// restartforanewsequence_();
346// fs::path src = "../rank_1/restart.r";
347// fs::path dest = "restart.r";
348//
349// try {
350// // Copy the file
351// fs::copy_file(src, dest, fs::copy_options::overwrite_existing);
352// std::cout << "File copied successfully.\n";
353// } catch (fs::filesystem_error &e) {
354// std::cerr << e.what() << '\n';
355// }
356// // std::filesystem::path filePath("../rank_2/restart.r");
357// //
358// // // Create an empty file
359// // std::ofstream file(filePath);
360// // file.close();
361// //
362// // if (std::filesystem::exists(filePath)) {
363// // std::cout << "File 'restart.r' created successfully." << std::endl;
364// // } else {
365// // std::cout << "Failed to create the file 'restart.r'." << std::endl;
366// // }
367// relap5_();
368// // r5pars_.terminateall = 1;
369// printf ("\e[1;3%imP[%i] Finished simulating!\n\e[0m", rank + 1, rank);
370// printf ("\e[1;3%imP[%i] Informed server I'm done!\n\e[0m", rank + 1, rank);
371// MPI_Send (nullptr, 0, MPI_INT, 0, tag_end, MPI_COMM_WORLD);
372// MPI_Send (nullptr, 0, MPI_INT, 0, tag_flush, MPI_COMM_WORLD);
373//
374// break;
375// }
376// }
377//
378// else if (statusClient.MPI_TAG == tag_flush) {
379// MPI_Recv (nullptr, 0, MPI_INT, statusClient.MPI_SOURCE, tag_flush, MPI_COMM_WORLD, &statusClient);
380// printf ("\e[1;3%imP[%i] Flushing!\n\e[0m", rank + 1, rank);
381// //
382// // Branch_center.endSimulation();
383// // Communication_place.clearCommunicationPlaceAllPointers();
384// // Command_handler.clearCommandControlCenteFromSimulation();
385// // Action_Taker.clearOperatorATFinal();
386// // Decision_Maker.clearOperatorDMFinal();
387// // vector<RestartNodeInfo *>::iterator restart_info_itr;
388// // for (restart_info_itr = restart_info.begin();
389// // restart_info_itr != restart_info.end(); restart_info_itr++) {
390// // delete (*restart_info_itr);
391// // }
392// // restart_info.clear();
393// // time(&t_current);
394// // time_end = "";
395// // time_end += ctime(&t_current);
396// // Aux_info.addRecordWithTime(-999.0, "Start_time:");
397// // Aux_info.addRecordWithTime(-999.0, time_begin);
398// // Aux_info.addRecordWithTime(-999.0, "End_time:");
399// // Aux_info.addRecordWithTime(-999.0, time_end);
400// // Aux_info.saveInfoToFile();
401// std::cout << "size of restart_info = " << restart_info.size() << std::endl;
402// cout << "Simulation Finished!\nStart at: " << time_begin << "\nEnd at: " << time_end << endl;
403// } else if (statusClient.MPI_TAG == tag_shutdown) {
404// MPI_Recv (nullptr, 0, MPI_INT, statusClient.MPI_SOURCE, tag_shutdown, MPI_COMM_WORLD, &statusClient);
405// printf ("\e[1;3%imP[%i] Shutting down!\n\e[0m", rank + 1, rank);
406// return 0;
407// } else {
408// printf ("Unknown tag %d on node %d.\n", statusClient.MPI_TAG, rank);
409// }
410// }
411// return 0;
412// }
413//
414// void parallelexec::parallel_final()
415// {
416// MPI_Finalize();
417// }
418//
419// template<typename T>
420// void parallelexec::saveToFile(const T& object, const std::string& filePath) {
421// std::ofstream file(filePath);
422// if (!file.is_open()) {
423// throw std::runtime_error("Failed to open the file for writing: " + filePath);
424// }
425// boost::archive::text_oarchive oa(file, ios::binary);
426// std::cout << "Serializing object of type: " << typeid(T).name() << std::endl;
427// oa << object;
428// std::cout << "Serialization complete for type: " << typeid(T).name() << std::endl;
429// }
430//
431// template <typename T>
432// void parallelexec::loadFromFile(T& object, const std::string& filePath) {
433// std::cout << "Filebased DESERIALIZATION STARTING:" << std::endl;
434// std::ifstream file(filePath);
435// if (!file.is_open()) {
436// throw std::runtime_error("Failed to open the file for reading: " + filePath);
437// }
438// boost::archive::text_iarchive ia(file, ios::binary);
439// ia >> object;
440// }
441//
442// // Explicit instantiation
443// template void parallelexec::saveToFile<RestartNodeInfo>(const RestartNodeInfo& object, const std::string& filePath);
444// template void parallelexec::saveToFile<BranchRecord>(const BranchRecord& object, const std::string& filePath);
445// //template void parallelexec::loadFromFile<RestartNodeInfo>(const RestartNodeInfo& object, const std::string& filePath);
446//
447// template<typename T>
448// void parallelexec::send_object(const T& obj, int dest_rank, int tag, MPI_Comm comm) {
449// // Serialize the object
450// std::cout << "StringStream SERIALIZING STARTING:" << std::endl;
451// std::ostringstream oss;
452// boost::archive::text_oarchive oa(oss, ios::binary);
453// oa << obj;
454// std::string serialized_str = oss.str();
455// std::cout << "Serialized string length: " << serialized_str.size() << std::endl;
456// // std::cout << "Serialized string: " << serialized_str << std::endl;
457//
458// // Send the size of the serialized string
459// unsigned long long num_bytes = serialized_str.size();
460// std::cerr << "\nSending size: " << num_bytes << " with tag: " << tag << std::endl;
461// MPI_Send(&num_bytes, 1, MPI_UNSIGNED_LONG_LONG, dest_rank, tag, comm);
462// // Send the serialized string
463// MPI_Send(serialized_str.c_str(), static_cast<int>(num_bytes), MPI_CHAR, dest_rank, tag, comm);
464// std::cout << "StringStream SERIALIZING ENDING:" << std::endl;
465// }
466//
467// template<typename T>
468// void parallelexec::receive_object(T& obj, int src_rank, int tag, MPI_Comm comm) {
469// try {
470// std::cout << "StringStream DESERIALIZING STARTING:" << std::endl;
471//
472// // Receive the size of the serialized string
473// unsigned long long size;
474// std::cout << "Receiving size with tag: " << tag << std::endl;
475// MPI_Recv(&size, 1, MPI_UNSIGNED_LONG_LONG, src_rank, tag, comm, MPI_STATUS_IGNORE);
476// std::cerr << "\nExpected receive buffer size: " << size << std::endl;
477//
478// // Receive the serialized string
479// std::vector<char> buffer(size);
480// MPI_Recv(buffer.data(), static_cast<int>(size), MPI_CHAR, src_rank, tag, comm, MPI_STATUS_IGNORE);
481// std::string serialized_str(buffer.begin(), buffer.end());
482//
483// std::cerr << "\nrecv_buff_str:\t" <<"\ntype_id:\t"<<typeid(obj).name()<<std::endl;
484//
485// // Deserialize the object
486// std::istringstream iss(serialized_str.c_str());
487// boost::archive::text_iarchive ia(iss, ios::binary);
488// ia >> obj;
489//
490//
491//
492// std::cout << "StringStream DESERIALIZING ENDING:" << std::endl;
493// }
494// catch (const boost::archive::archive_exception& ex) {
495// std::cerr << "Deserialization error: " << ex.what() <<":"<<ex.code<<std::endl;
496// }
497// }
498//
499// #include <boost/archive/binary_oarchive.hpp>
500// #include <boost/archive/binary_iarchive.hpp>
501//
502// // template<typename T>
503// // void parallelexec::send_object(const T& obj, int dest_rank, int tag, MPI_Comm comm) {
504// // // Serialize the object
505// // std::cout << "Binary SERIALIZING STARTING:" << std::endl;
506// // std::ostringstream oss(std::ios::binary);
507// // boost::archive::binary_oarchive oa(oss);
508// // oa << obj;
509// // std::string serialized_str = oss.str();
510// // std::cout << "Serialized string length: " << serialized_str.size() << std::endl;
511// //
512// // // Send the size of the serialized string
513// // int size = static_cast<int>(serialized_str.size());
514// // std::cout << "Sending size: " << size << " with tag: " << tag << std::endl;
515// // MPI_Send(&size, 1, MPI_INT, dest_rank, tag, comm);
516// //
517// // // Send the serialized string
518// // MPI_Send(serialized_str.c_str(), size, MPI_CHAR, dest_rank, tag, comm);
519// // std::cout << "Binary SERIALIZING ENDING:" << std::endl;
520// // }
521// //
522// // template<typename T>
523// // void parallelexec::receive_object(T& obj, int src_rank, int tag, MPI_Comm comm) {
524// // try {
525// // std::cout << "Binary DESERIALIZING STARTING:" << std::endl;
526// // int alp = 1;
527// // while (!alp) {
528// // sleep(5);
529// // }
530// // // Receive the size of the serialized string
531// // int size = 0;
532// // std::cout << "Receiving size with tag: " << tag << std::endl;
533// // MPI_Recv(&size, 1, MPI_INT, src_rank, tag, comm, MPI_STATUS_IGNORE);
534// // std::cout << "Received size: " << size << std::endl;
535// //
536// // // Receive the serialized string
537// // std::vector<char> buffer(size);
538// // MPI_Recv(buffer.data(), size, MPI_CHAR, src_rank, tag, comm, MPI_STATUS_IGNORE);
539// // std::cout << "Received serialized string length: " << buffer.size() << std::endl;
540// //
541// // // Deserialize the object
542// // std::string serialized_str(buffer.begin(), buffer.end());
543// // std::istringstream iss(serialized_str, std::ios::binary);
544// // boost::archive::binary_iarchive ia(iss);
545// // ia >> obj;
546// //
547// // std::cout << "Binary DESERIALIZING ENDING:" << std::endl;
548// // }
549// // catch (const boost::archive::archive_exception& ex) {
550// // std::cerr << "Deserialization error: " << ex.what() << std::endl;
551// // }
552// // }
553//
554// // Explicit instantiation for specific types
555// template void parallelexec::send_object<RestartNodeInfo>(const RestartNodeInfo& obj, int dest_rank, int tag, MPI_Comm comm);
556// template void parallelexec::send_object<BranchRecord>(const BranchRecord& obj, int dest_rank, int tag, MPI_Comm comm);
557//
558// template void parallelexec::receive_object<RestartNodeInfo>(RestartNodeInfo& obj, int src_rank, int tag, MPI_Comm comm);
559// template void parallelexec::receive_object<BranchRecord>(BranchRecord& obj, int src_rank, int tag, MPI_Comm comm);
560//
send_completion_notification
void send_completion_notification(const RunSummary &summary)
Send completion notifications to all configured targets. Non-fatal: prints a warning if curl fails,...
Definition: Notification.cpp:111
set_global_client
void set_global_client(Client *c)
Definition: ParallelExecution.cpp:19
g_client
Client * g_client
Definition: ParallelExecution.cpp:16
ParallelExecution.h
MPI parallel execution manager — initialization, Server/Client dispatch, serialization.
g_sim
SimulationContext g_sim
Global simulation context: the single instance, defined in SimulationContext.cpp.
Definition: SimulationContext.cpp:7
Client
MPI worker client that executes simulation sequences.
Definition: Client.h:42
Scheduler::getVizPort
int getVizPort() const
Definition: Scheduler.h:227
Scheduler::getProjectDirectory
std::string getProjectDirectory()
Definition: Scheduler.cpp:595
Scheduler::getSequenceTruncationTime
double getSequenceTruncationTime()
Definition: Scheduler.cpp:646
Scheduler::parseOptions
void parseOptions(int argc, char *argv[])
Parses the command line and ZiniADS.txt, fills this scheduler and g_sim.config, and reads ZCtrlPar....
Definition: Scheduler.cpp:59
Scheduler::getInputDeckFileDirectoryAndName
std::string getInputDeckFileDirectoryAndName()
Definition: Scheduler.cpp:610
Server
MPI server that manages the DET task queue and client dispatch.
Definition: Server.h:62
PG::log
bool log
Enable/disable rank-specific logging.
Definition: ParallelGlobal.h:174
PG::size
int size
Total number of MPI processes.
Definition: ParallelGlobal.h:173
PG::rank
int rank
MPI rank of this process.
Definition: ParallelGlobal.h:172
adsidac::merge::write_manifest
void write_manifest(const std::string &project_dir, int ranks, bool aggregate, long sequences, unsigned long long fingerprint)
Write the –merge=defer manifest <project_dir>/Output/MERGE_PENDING.txt (Output/ is created if needed;...
Definition: OutputMerge.h:327
RunSummary
Summary of a completed simulation run.
Definition: Notification.h:8
RunSummary::solver
std::string solver
Definition: Notification.h:13
RunSummary::project
std::string project
Definition: Notification.h:9
RunSummary::truncation_time
double truncation_time
Definition: Notification.h:14
RunSummary::wall_seconds
double wall_seconds
Definition: Notification.h:11
SimConfig::merge_mode
int merge_mode
Definition: SimConfig.h:38
SimConfig::solver_url
std::string solver_url
REST solver URL (empty = use RELAP5). –solver=mock gives "http://localhost:<--solver-port,...
Definition: SimConfig.h:27
SimConfig::aggregate_pe_output
bool aggregate_pe_output
Definition: SimConfig.h:23
SimulationContext::SScheduler
Scheduler SScheduler
Simulation scheduling, parses CLI options.
Definition: SimulationContext.h:41
SimulationContext::config_fingerprint
uint64_t config_fingerprint
FNV-1a of immutable config inputs (see ConfigFingerprint.h); 0 = not computed.
Definition: SimulationContext.h:69