GCC Code Coverage Report

Directory: src/
File: src/mngo/mngo.c
Date: 2026-09-15 07:37:49
Exec Total Coverage
Lines: 147 187 78.6%
Functions: 8 10 80.0%
Branches: 63 108 58.3%
Line Branch Exec Source
1 /*********************************************************************************/
2 /* Copyright 2009-2026 Barcelona Supercomputing Center */
3 /* */
4 /* This file is part of the DLB library. */
5 /* */
6 /* DLB is free software: you can redistribute it and/or modify */
7 /* it under the terms of the GNU Lesser General Public License as published by */
8 /* the Free Software Foundation, either version 3 of the License, or */
9 /* (at your option) any later version. */
10 /* */
11 /* DLB is distributed in the hope that it will be useful, */
12 /* but WITHOUT ANY WARRANTY; without even the implied warranty of */
13 /* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the */
14 /* GNU Lesser General Public License for more details. */
15 /* */
16 /* You should have received a copy of the GNU Lesser General Public License */
17 /* along with DLB. If not, see <https://www.gnu.org/licenses/>. */
18 /*********************************************************************************/
19
20 /*
21 * Here we describe the logic of the DLB module MNGO.
22 *
23 * The concept of MNGO is that it is in charge of periodically sampling the
24 * applications performance and take action acordingly by
25 * activating/deactivating other other DLB modules to improve the overall
26 * applications efficiency.
27 *
28 * The general idea is to use TALP to collect different metrics (both within
29 * the same node, and across). Then use these metrics to decide wheather or not
30 * activating LeWI or DROM would improve the applications performance or
31 * efficiency, and activate the corresponding module based on different
32 * polices.
33 */
34
35 /*
36 * For activating the DLB modules in a coordinated fashion we have implemented a
37 * shared memory interface.
38 */
39 #include "LB_comm/shmem_barrier.h"
40 #include "LB_comm/shmem_mngo.h"
41
42 /*
43 * To use DROM we use the procinfo shared memory interface to change the process
44 * masks according to polices.
45 */
46 #include "LB_comm/shmem_procinfo.h"
47
48 /*
49 * To activate and deactivate LeWI we use the `set_lewi_enabled` function
50 * included in the DLB_kernel.h
51 */
52 #include "LB_core/DLB_kernel.h"
53
54 // Used to set the thread to observer when using the helper-thread mode
55 #include "LB_core/thread_ctx.h"
56
57 /*
58 * The MODULE communicates with TALP to gather performance metrics of the
59 * application.
60 */
61 #include "LB_core/node_barrier.h"
62 #include "LB_core/thread_ctx.h"
63 #include "apis/dlb_mngo.h"
64 #include "dlb_talp.h"
65 #include "mngo/mngo_drom.h"
66 #include "mngo/mngo_resources.h"
67 #include "talp/regions.h"
68 #include "talp/talp.h"
69 /*
70 *
71 */
72 #include "LB_core/spd.h"
73 /*
74 *
75 */
76 #include "mngo/mngo.h"
77 #include "mngo/mngo_balancer.h"
78 /*
79 * All the functions where something could go wrong will report an error defined
80 * as an enum value in DLBErrorCodes.
81 */
82 #include "apis/dlb_errors.h"
83 /*
84 * We have used the verbose function to report extra information about what the
85 * MODULE is doing when the debug flag is enabled.
86 */
87 #include "support/debug.h"
88 /*
89 * Similarly we use the Extrae API to generate MNGO events to visualize the
90 * MODULE in action in Paraver traces.
91 */
92 #include "support/queues.h"
93 #include "support/tracing.h"
94 /*
95 * When we use DROM we need to transmit information to the threads that are part
96 * of the OpenMP team so them can change the mask. To communicate between
97 * different threads of the same process we have used atomic variables.
98 */
99 #include "apis/dlb_types.h"
100 #include "support/types.h"
101
102 #include "support/mask_utils.h"
103
104 #include <sched.h>
105 #include <stdlib.h>
106 #include <string.h>
107 #include <unistd.h>
108
109 /*
110 * To ceil the drom balancing.
111 */
112 #include <math.h>
113
114 #include "mngo/mngo_talp.h"
115
116 /*
117 * True when mngo has been initialized correctly
118 */
119 static bool have_mngo = false;
120 static bool have_drom = false;
121 static bool have_lewi = false;
122
123 static bool have_helper_thread = false;
124 static __thread bool im_helper_thread = false;
125
126 static const char *helper_thread_region_name = "helper-thread";
127
128 /*
129 * To take decisions we need a history of performance to make an educated guess
130 * on what is the best action to improve performance.
131 */
132
133 #include <LB_comm/shmem_cpuinfo.h>
134
135 12 static mngo_actions_t mngo_actions(const subprocess_descriptor_t *spd,
136 const mngo_reduced_metric_t *metrics) {
137 12 mngo_info_t *mngo_info = spd->mngo_info;
138
139 // Load balance in deviation: is the percentage of parallel efficiency
140 // difference of this process with relation to the node average.
141 12 float lb_in_deviation = (metrics->self_pe / metrics->shared_pe) - 1;
142
143 // We consider that there is LB_IN if the deviation absolute value of
144 // deviation is greater than deviation.
145
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 6 times.
 18 bool lb_in = (lb_in_deviation > +mngo_info->lb_in_threshold ||
146
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6 lb_in_deviation < -mngo_info->lb_in_threshold);
147
148 /*
149 * For now only one decision can be taken per process each time. So when
150 * any policy is activated we return.
151 */
152 12 mngo_actions_t action = MNGO_ACTION_NONE;
153
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 12 times.
 12 if (lb_in) {
154
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 6 times.
 12 if (have_drom) {
155 // TODO: in the event where both LeWI and DROM are enabled we try
156 // first with DROM and then if we observe performance degradation/no
157 // performance increase we skip this and try with LeWI.
158 6 action = MNGO_ACTION_DROM;
159 6 goto policy_applied;
160 }
161
162
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (have_lewi) {
163 6 action = MNGO_ACTION_LEWI_START;
164 6 goto policy_applied;
165 }
166 }
167 ✗ policy_applied:
168
169 // Share the actions accross nodes to aggree on the new decission
170 12 shmem_mngo__alltoall_actions(mngo_info->mid, &action);
171
172 12 return action;
173 }
174
175 /*
176 * Finally we take the actions that have been collectively decided and we apply
177 * them.
178 */
179 12 static void mngo_apply_actions(subprocess_descriptor_t *spd,
180 mngo_actions_t *actions, cpu_set_t *next_mask) {
181 12 mngo_info_t *mngo_info = spd->mngo_info;
182 mngo_state_t *state =
183 12 mngo_regions__get_current_resources(&mngo_info->region_handler);
184
185
2/4
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 6 times.
✗ Branch 3 not taken.
 12 switch (*actions) {
186 6 case MNGO_ACTION_LEWI_START: {
187
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6 verbose(VB_MNGO, "Starting LeWI");
188 instrument_event(MNGO_MANAGER, EVENT_MNGO_LEWI_ON, EVENT_BEGINEND);
189 6 state->lewi_on = true;
190 6 break;
191 }
192 ✗ case MNGO_ACTION_LEWI_STOP: {
193 ✗ verbose(VB_MNGO, "Stopping LeWI");
194 instrument_event(MNGO_MANAGER, EVENT_MNGO_LEWI_OFF, EVENT_BEGINEND);
195 ✗ state->lewi_on = false;
196 ✗ break;
197 }
198 6 case MNGO_ACTION_DROM: {
199 // instrument_event done outside this function, in mngo_manger
200 6 memcpy(&state->drom_mask, next_mask, sizeof(cpu_set_t));
201 6 break;
202 }
203 ✗ default:
204 ✗ break;
205 }
206 12 }
207
208 /*
209 * The MPI collectives mode will run the mngo_manager in the same thread that
210 * is executing while an MPI collective call with the MPI_COMM_WORLD
211 * communicator is used. Since the most applications usually have some
212 * consecutive collective calls. To prevent calling the mngo_manager to
213 * continuously we use the mngo interval option as a cool-down time while all
214 * collectives will be ignored and no manager will be called.
215 *
216 * This mode will directly call the mngo_manager function to run the MNGO core
217 * functionality.
218 */
219 32 int mngo_manager(subprocess_descriptor_t *spd, dlb_mngo_region_t *region,
220 mngo_manager_entrypoint_t entrypoint) {
221
222 32 int error = DLB_SUCCESS;
223
224 32 mngo_info_t *mngo_info = spd->mngo_info;
225 32 mngo_regions_manager_t *region_manager = &mngo_info->region_handler;
226
227
2/2
✓ Branch 0 taken 16 times.
✓ Branch 1 taken 16 times.
 32 if (entrypoint & MANAGER_END) {
228
2/2
✓ Branch 1 taken 4 times.
✓ Branch 2 taken 12 times.
 16 if (!mngo_regions__check_top(region_manager, region)) {
229 4 return DLB_NOUPDT;
230 }
231
232 12 mngo_region_info_t *region_info = &region->region_info;
233
234 // Collect metrics and update metrics history
235 {
236 instrument_event(MNGO_MANAGER, EVENT_MNGO_TALP, EVENT_BEGIN);
237 12 region_stop(spd, region_info->talp.monitor); // TALP region stop
238 mngo_talp_metrics_t *metrics =
239 12 mngo_metrics__history_emplace(&region_info->metrics_history);
240 12 mngo_talp__take_metrics(spd, &region_info->talp, metrics);
241 instrument_event(MNGO_MANAGER, EVENT_MNGO_TALP, EVENT_END);
242 }
243
244 {
245 instrument_event(MNGO_MANAGER, EVENT_MNGO_HISTORY, EVENT_BEGIN);
246 mngo_reduced_metric_t metrics;
247 12 mngo_metrics__history_tendency(&region_info->metrics_history,
248 &metrics);
249 instrument_event(MNGO_MANAGER, EVENT_MNGO_HISTORY, EVENT_END);
250
251 instrument_event(MNGO_MANAGER, EVENT_MNGO_DECIDE, EVENT_BEGIN);
252 12 mngo_actions_t actions = mngo_actions(spd, &metrics);
253 instrument_event(MNGO_MANAGER, EVENT_MNGO_DECIDE, EVENT_END);
254
255 cpu_set_t new_cpu_mask;
256
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 6 times.
 12 if (MNGO_ACTION_DROM == actions) {
257 instrument_event(MNGO_MANAGER, EVENT_MNGO_DROM_IN, EVENT_BEGIN);
258 6 mngo_drom__balance(spd, metrics.self_pe, metrics.shared_pe,
259 &new_cpu_mask);
260 }
261
262 12 mngo_apply_actions(spd, &actions, &new_cpu_mask);
263
264 12 if (MNGO_ACTION_DROM == actions) {
265 instrument_event(MNGO_MANAGER, EVENT_MNGO_DROM_IN, EVENT_END);
266 }
267 }
268
269 12 mngo_regions__pop_active(region_manager, region);
270 12 mngo_state__load(spd,
271 mngo_regions__get_current_resources(region_manager));
272
273 12 shmem_mngo_barrier_wait();
274 }
275
276
2/2
✓ Branch 0 taken 16 times.
✓ Branch 1 taken 12 times.
 28 if (entrypoint & MANAGER_BEGIN) {
277
2/2
✓ Branch 1 taken 4 times.
✓ Branch 2 taken 12 times.
 16 if (mngo_regions__check_top(region_manager, region)) {
278 4 return DLB_NOUPDT;
279 }
280
281 12 mngo_regions__push_active(region_manager, region);
282 12 mngo_state__load(spd,
283 mngo_regions__get_current_resources(region_manager));
284
285 12 mngo_region_info_t *region_info = &region->region_info;
286
287 12 region_start(spd, region_info->talp.monitor); // TALP region start
288
289 12 shmem_mngo_barrier_wait();
290 }
291
292 24 return error; // TODO: track errors
293 }
294
295 /*
296 * The regions mode will implement two calls, one to start the regions and the
297 * other to end it. Performance information will then be stored for each
298 * region, and each region will have its own profile. The profile will contain
299 * the information about which DLB modules are being used in the specific
300 * region.
301 */
302
303 12 dlb_mngo_region_t *mngo_region_register(const subprocess_descriptor_t *spd,
304 const char *name) {
305
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 12 times.
 12 if (!have_mngo) return NULL;
306
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 12 if (have_helper_thread && !im_helper_thread) return NULL;
307
308
1/2
✓ Branch 0 taken 12 times.
✗ Branch 1 not taken.
 12 verbose(VB_MNGO, "Registering region %s", name);
309
310 12 mngo_info_t *mngo_info = spd->mngo_info;
311 12 mngo_regions_manager_t *region_manager = &mngo_info->region_handler;
312
313 12 dlb_mngo_region_t *region_found = mngo_regions__find(region_manager, name);
314
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 8 times.
 12 if (NULL != region_found) {
315 4 return region_found;
316 }
317
318 /**
319 * If the region does not exists we create a new one.
320 */
321 mngo_state_t *current_state =
322 8 mngo_regions__get_current_resources(region_manager);
323
324 8 dlb_mngo_region_t *region = mngo_regions__alloc(region_manager);
325
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8 times.
 8 if (NULL == region) {
326 ✗ return NULL;
327 }
328
329 // The new region inherits the current state
330 8 region->region_info.state = *current_state;
331
332 // We create a monitoring region for the new region
333 int region_register_error =
334 8 mngo_talp__region_register(spd, name, &region->region_info.talp);
335
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8 times.
 8 if (DLB_SUCCESS != region_register_error) {
336 ✗ return NULL;
337 }
338
339 // The region name lives in the TALP monitor.
340 8 region->name = mngo_talp__get_region_name(&region->region_info.talp);
341
342 // We initialize the performance history of the region
343 8 mngo_metrics__history_init(&region->region_info.metrics_history);
344
345 8 return region;
346 }
347
348 16 int mngo_region_start(subprocess_descriptor_t *spd, dlb_mngo_region_t *region) {
349 16 int error = DLB_SUCCESS;
350
351
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 16 times.
 16 if (!have_mngo) return DLB_ERR_NOMNGO;
352
353
1/2
✓ Branch 0 taken 16 times.
✗ Branch 1 not taken.
 16 if (!have_helper_thread) {
354
1/2
✓ Branch 0 taken 16 times.
✗ Branch 1 not taken.
 16 verbose(VB_MNGO, "Starting region %s", region->name);
355
356 16 error = mngo_manager(spd, region, MANAGER_BEGIN);
357
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 12 times.
 16 if (error != DLB_SUCCESS) goto bail_on_error;
358
359 // the if stops here because if we are running with helper-thread we
360 // will take the oportunity to call poll_drom_update.
361 }
362
363
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 6 times.
 12 if (have_drom) {
364 6 error = poll_drom_update(spd);
365 // No change is spossible so we translate to success
366
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6 if (error == DLB_NOUPDT) error = DLB_SUCCESS;
367
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6 if (error != DLB_SUCCESS) goto bail_on_error;
368 }
369
370 12 bail_on_error:
371 16 return error;
372 }
373
374 16 int mngo_region_stop(subprocess_descriptor_t *spd, dlb_mngo_region_t *region) {
375 16 int error = DLB_SUCCESS;
376
377
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 16 times.
 16 if (!have_mngo) return DLB_ERR_NOMNGO;
378
379
1/2
✓ Branch 0 taken 16 times.
✗ Branch 1 not taken.
 16 if (!have_helper_thread) {
380
1/2
✓ Branch 0 taken 16 times.
✗ Branch 1 not taken.
 16 verbose(VB_MNGO, "Stopping region %s", region->name);
381 16 mngo_info_t *mngo_info = spd->mngo_info;
382 16 shmem_barrier__barrier(mngo_info->dlb_node_barrier);
383
384 16 error = mngo_manager(spd, region, MANAGER_END);
385
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 12 times.
 16 if (error != DLB_SUCCESS) goto bail_on_error;
386
387 // the if stops here because if we are running with helper-thread we
388 // will take the oportunity to call poll_drom_update.
389 }
390
391
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 6 times.
 12 if (have_drom) {
392 6 error = poll_drom_update(spd);
393 // No change is spossible so we translate to success
394
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6 if (error == DLB_NOUPDT) error = DLB_SUCCESS;
395
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6 if (error != DLB_SUCCESS) goto bail_on_error;
396 }
397
398 12 bail_on_error:
399 16 return error;
400 }
401
402 /*
403 * The helper thread mode will spawn a pthread that will sleep for an interval
404 * of time defined in the options. After the interval of time the thread will
405 * wakeup, execute the mngo_manger, which will apply selected MNGO policy.
406 */
407 ✗ static void mngo_helper_thread(subprocess_descriptor_t *spd) {
408
409 ✗ im_helper_thread = true;
410 ✗ thread_spd = spd;
411
412 ✗ thread_ctx_set_observer(true);
413
414 ✗ mngo_info_t *mngo_info = spd->mngo_info;
415
416 ✗ verbose(VB_MNGO, "Rank %d started helper thread", mngo_info->rank);
417
418 dlb_mngo_region_t *region =
419 ✗ mngo_region_register(spd, helper_thread_region_name);
420
421 while (1) {
422 /**
423 * Only the node representative manager gathers the metrics, takes
424 * decisions and sends them to the other managers.
425 */
426 ✗ if (mngo_info->mid == 0) {
427 /** Time between MNGO managment events.
428 */
429 struct timeval now;
430 ✗ gettimeofday(&now, NULL);
431
432 ✗ struct timespec abstime = {
433 ✗ .tv_sec = now.tv_sec + thread_spd->options.mngo_interval_time,
434 .tv_nsec = 0,
435 };
436
437 ✗ shmem_mngo_manager_wait(&abstime);
438 }
439
440 ✗ shmem_mngo_barrier_wait();
441
442 ✗ if (shmem_mngo_check_stop_flag()) {
443 ✗ break;
444 }
445
446 ✗ verbose(VB_MNGO, "Rank %d helper thread has woken up", mngo_info->rank);
447
448 ✗ mngo_manager(spd, region, (mngo_manager_entrypoint_t)(MANAGER_BEGIN | MANAGER_END));
449 }
450
451 ✗ verbose(VB_MNGO, "Rank %d helper thread stopping", mngo_info->rank);
452
453 /**
454 * Wait all the managers from the node to finalize.
455 */
456 ✗ shmem_mngo_fini_sync();
457
458 /**
459 * Close the shared memory used by the manager thread.
460 */
461 ✗ shmem_mngo_fini(mngo_info->mid);
462
463 // Finalize thread
464 ✗ pthread_exit(0);
465 }
466
467 /*
468 * For the MODULE to work properly some initialization steps have to be taken.
469 * We provide the mngo_init function to prepare the MODULE for use. This
470 * function will first allocate and initialize the mngo_info, initialize the
471 * shared memory, and provide a unique (locally in the node) identifier to each
472 * caller, create a TALP region for the MODULE, and also a private MPI
473 * communicator for the MODULE.
474 *
475 * This module will finally spawn a helper thread if the `mngo-mode` option is
476 * set to `helper-thread`.
477 */
478 static void mngo_helper_thread(subprocess_descriptor_t *spd);
479 /*
480 * For now the `helper-thread` option is the only one available, and the helper
481 * thread is created always. In the future other modes will implemented such a
482 * `mpi-collectives` mode which will call the mngo manager on MPI collective
483 * calls, or the `region` mode, where using an API the developer will define
484 * different regions in the execution, and then MNGO will create automatic
485 * profiles for the different regions.
486 */
487 14 void mngo_init(subprocess_descriptor_t *spd) {
488
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 14 times.
 14 if (have_mngo) return;
489
490
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 10 times.
 14 if (spd->options.lewi) {
491 4 have_lewi = true;
492 }
493
494
2/2
✓ Branch 0 taken 10 times.
✓ Branch 1 taken 4 times.
 14 if (spd->options.drom) {
495 10 have_drom = true;
496 }
497
498
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 14 times.
 14 if (spd->options.mngo_mode == MNGO_HELPER_THREAD) {
499 ✗ have_helper_thread = true;
500 #ifdef MPI_LIB
501 int any_abort_init, abort_init = 0;
502 int mpi_thread_provided;
503 int error = PMPI_Query_thread(&mpi_thread_provided);
504
505 if (error != MPI_SUCCESS) {
506 warning0("Could not check the MPI thread level, required for "
507 "helper-thread mode.");
508 abort_init = 1;
509 }
510
511 if (mpi_thread_provided < MPI_THREAD_MULTIPLE) {
512 abort_init = 1;
513 }
514
515 PMPI_Allreduce(&abort_init, &any_abort_init, 1, MPI_INT, MPI_SUM,
516 MPI_COMM_WORLD);
517 if (any_abort_init > 0) {
518 warning0("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"
519 "!!!!!!!!!");
520 warning0("!!! Trying to initialize MNGO with helper-thread mode, "
521 "but the !!!");
522 warning0("!!! MPI thread level is lower than MPI_THREAD_MULTIPLE. "
523 " !!!");
524 warning0("!!! ABORTING MNGO INITIALIZATION "
525 " !!!");
526 warning0("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"
527 "!!!!!!!!!");
528 spd->mngo_info = NULL;
529 return;
530 }
531 #endif // MPI_LIB
532 }
533
534 14 mngo_info_t *mngo_info = (mngo_info_t *)calloc(1, sizeof(mngo_info_t));
535
536 #ifdef MPI_LIB
537 int rank;
538 PMPI_Comm_rank(MPI_COMM_WORLD, &rank);
539 PMPI_Comm_dup(MPI_COMM_WORLD, &mngo_info->comm);
540 #else
541 14 int rank = 0;
542 #endif
543
544 /**
545 * Initialize this process manegers shared memory.
546 */
547 14 pid_t pid = spd->id;
548
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 14 times.
 14 if (shmem_mngo_init(spd->options.shm_key, pid, rank, &mngo_info->mid) !=
549 DLB_SUCCESS) {
550 // DO SOMETHING
551 ✗ fatal("MNGO Shared memory initialization failed.");
552 }
553
554 14 mngo_info->rank = rank;
555
556 cpu_set_t mask;
557 14 shmem_procinfo__getprocessmask(shmem_mngo_get_pid(mngo_info->mid), &mask,
558 DLB_DROM_FLAGS_NONE);
559
560 14 mngo_info->default_state = (const mngo_state_t){
561 .lewi_on = false,
562 };
563 14 memcpy(&mngo_info->default_state.drom_mask, &mask, sizeof(cpu_set_t));
564
565 14 mngo_regions__manager_init(&mngo_info->region_handler,
566 &mngo_info->default_state);
567
568 /**
569 * Configure MNGO based on options.
570 */
571 14 mngo_info->lb_in_threshold = ((float) spd->options.mngo_lb_in_threshold) / 100.0;
572 14 mngo_info->lb_out_threshold = ((float) spd->options.mngo_lb_out_threshold) / 100.0;
573
574 /**
575 * Create a barrier_t to limit regions
576 */
577 14 mngo_info->dlb_node_barrier =
578 14 node_barrier_register(spd, "__mngo", DLB_BARRIER_LEWI_ON);
579 14 shmem_mngo_barrier_wait(); // Ensure that all processes of the node attached
580
581 /**
582 * Store the MNGO info to the SubProcessDescriptor.
583 */
584 14 spd->mngo_info = mngo_info;
585
586 /**
587 * Initialize the MNGO manager thread for this process.
588 */
589
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 14 times.
 14 if (spd->options.mngo_mode == MNGO_HELPER_THREAD) {
590 ✗ pthread_create(&mngo_info->thread, NULL,
591 (void *(*)(void *))mngo_helper_thread, (void *)spd);
592 }
593
594 // MNGO initialized successfully
595 14 have_mngo = 1;
596
2/2
✓ Branch 0 taken 8 times.
✓ Branch 1 taken 6 times.
 14 verbose(VB_MNGO, "MNGO Initialized successfully.");
597 }
598
599 14 void mngo_fini(const subprocess_descriptor_t *spd) {
600
601
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 14 times.
 14 if (!have_mngo) return;
602 14 have_mngo = false;
603
604 14 mngo_info_t *mngo_info = spd->mngo_info;
605
606 // If MNGO was never initialized just return.
607
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 14 times.
 14 if (mngo_info == NULL) {
608 ✗ return;
609 }
610
611 if (true /* TODO mngo_info->shmem_active */) {
612
613 #ifdef MPI_LIB
614 PMPI_Barrier(MPI_COMM_WORLD);
615 #endif
616
617 /**
618 * Send the stop message throug the shm and signal the manager thread to
619 * wake up.
620 */
621
2/2
✓ Branch 0 taken 7 times.
✓ Branch 1 taken 7 times.
 14 if (mngo_info->mid == 0) {
622 7 shmem_mngo_manager_wake_finalize();
623 }
624 }
625
626 /**
627 * Join with the manager thread.
628 */
629
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 14 times.
 14 if (spd->options.mngo_mode == MNGO_HELPER_THREAD) {
630 int *retval;
631 ✗ pthread_join(mngo_info->thread, (void **)&retval);
632 }
633
634 /**
635 * Free the TALP monitoring regions.
636 */
637 queue_iter_head2tail_t iter =
638 14 queue__into_head2tail_iter(mngo_info->region_handler.regions);
639 14 dlb_mngo_region_t **current = NULL;
640
2/2
✓ Branch 1 taken 8 times.
✓ Branch 2 taken 14 times.
 22 while ((current = queue_iter__get_nth(&iter, 0)) != NULL) {
641 8 region_stop(spd,
642 8 (*current)->region_info.talp.monitor); // TALP region stop
643 8 mngo_metrics__history_fini(&(*current)->region_info.metrics_history);
644 }
645
646 14 mngo_regions__manager_finalize(&mngo_info->region_handler);
647
648
1/2
✓ Branch 0 taken 14 times.
✗ Branch 1 not taken.
 14 if (!have_helper_thread) {
649 // The shared memory is managed by the helper-thread if available
650
651 /**
652 * Wait all the managers from the node to finalize.
653 */
654 14 shmem_mngo_fini_sync();
655
656 /**
657 * Close the shared memory used by the manager thread.
658 */
659 14 shmem_mngo_fini(mngo_info->mid);
660 }
661
662 /**
663 * Free the mngo_info_t struct from the sub-process descriptor.
664 */
665 14 free(mngo_info);
666 14 mngo_info = NULL;
667 }
668
669 ✗ int mngo_manager_wake(const subprocess_descriptor_t *spd) {
670
671 ✗ mngo_info_t *mngo_info = spd->mngo_info;
672
673 ✗ if (mngo_info->mid == 0) {
674 ✗ shmem_mngo_manager_wake();
675 }
676
677 ✗ return DLB_SUCCESS;
678 }
679