MAM_Manager.c 40.9 KB
Newer Older
1
#include <pthread.h>
2
#include <string.h>
3
#include "MAM.h"
4
5
6
7
8
9
#include "MAM_Constants.h"
#include "MAM_DataStructures.h"
#include "MAM_Types.h"
#include "MAM_Zombies.h"
#include "MAM_Times.h"
#include "MAM_RMS.h"
10
#include "MAM_Init_Configuration.h"
11
#include "spawn_methods/GenericSpawn.h"
12
#include "distribution_methods/Distributed_CommDist.h"
13

14
15
#define MAM_USE_SYNCHRONOUS 0
#define MAM_USE_ASYNCHRONOUS 1
16

17
void MAM_Commit(int *mam_state);
18
19
20
21

void send_data(int numP_children, malleability_data_t *data_struct, int is_asynchronous);
void recv_data(int numP_parents, malleability_data_t *data_struct, int is_asynchronous);

22

23
24
int MAM_St_rms(int *mam_state);
int MAM_St_spawn_start();
25
26
int MAM_St_spawn_pending(int wait_completed);
int MAM_St_red_start();
27
int MAM_St_red_pending(int wait_completed);
28
int MAM_St_user_start(int *mam_state);
29
30
31
32
33
34
35
36
int MAM_St_user_pending(int *mam_state, int wait_completed, void (*user_function)(void *), void *user_args);
int MAM_St_user_completed();
int MAM_St_spawn_adapt_pending(int wait_completed);
int MAM_St_spawn_adapted(int *mam_state);
int MAM_St_red_completed(int *mam_state);
int MAM_St_completed(int *mam_state);


37
void Children_init(void (*user_function)(void *), void *user_args);
38
39
int spawn_step();
int start_redistribution();
40
int check_redistribution(int wait_completed);
41
int end_redistribution();
iker_martin's avatar
iker_martin committed
42
int shrink_redistribution();
43
44

int thread_creation();
45
int thread_check(int wait_completed);
46
void* thread_async_work();
47

48
int MAM_I_convert_key(char *key);
49
void MAM_I_create_user_struct(int is_children_group);
50

51
52
53
54
55
malleability_data_t *rep_s_data;
malleability_data_t *dist_s_data;
malleability_data_t *rep_a_data;
malleability_data_t *dist_a_data;

56
57
mam_user_reconf_t *user_reconf;

58
/*
59
60
61
62
63
64
65
66
 * Inicializa la reserva de memoria para el modulo de maleabilidad
 * creando todas las estructuras necesarias y copias de comunicadores
 * para no interferir en la aplicación.
 *
 * Si es llamada por un grupo de procesos creados de forma dinámica,
 * inicializan la comunicacion con sus padres. En este caso, al terminar 
 * la comunicacion los procesos hijo estan preparados para ejecutar la
 * aplicacion.
67
 */
68
int MAM_Init(int root, MPI_Comm *comm, char *name_exec, void (*user_function)(void *), void *user_args) {
69
  MPI_Comm dup_comm, thread_comm, original_comm;
70
71
72

  mall_conf = (malleability_config_t *) malloc(sizeof(malleability_config_t));
  mall = (malleability_t *) malloc(sizeof(malleability_t));
73
74
75
76
77
  user_reconf = (mam_user_reconf_t *) malloc(sizeof(mam_user_reconf_t));

  MPI_Comm_rank(*comm, &(mall->myId));
  MPI_Comm_size(*comm, &(mall->numP));

78
  #if MAM_DEBUG
79
80
    DEBUG_FUNC("Initializing MaM", mall->myId, mall->numP); fflush(stdout); MPI_Barrier(*comm);
  #endif
81

82
83
84
85
86
  rep_s_data = (malleability_data_t *) malloc(sizeof(malleability_data_t));
  dist_s_data = (malleability_data_t *) malloc(sizeof(malleability_data_t));
  rep_a_data = (malleability_data_t *) malloc(sizeof(malleability_data_t));
  dist_a_data = (malleability_data_t *) malloc(sizeof(malleability_data_t));

87
88
  MPI_Comm_dup(*comm, &dup_comm);
  MPI_Comm_dup(*comm, &thread_comm);
89
  MPI_Comm_dup(*comm, &original_comm);
90
91
  MPI_Comm_set_name(dup_comm, "MAM_MAIN");
  MPI_Comm_set_name(thread_comm, "MAM_THREAD");
92
  MPI_Comm_set_name(original_comm, "MAM_ORIGINAL");
93
94

  mall->root = root;
iker_martin's avatar
iker_martin committed
95
  mall->root_parents = root;
96
  mall->zombie = 0;
97
  mall->comm = dup_comm;
98
  mall->thread_comm = thread_comm;
99
  mall->original_comm = original_comm;
100
101
  mall->user_comm = comm; 
  mall->tmp_comm = MPI_COMM_NULL;
102

103
  mall->name_exec = name_exec;
104
  mall->nodelist = NULL;
105
  mall->nodelist_len = 0;
106
107
108
109
110
111

  rep_s_data->entries = 0;
  rep_a_data->entries = 0;
  dist_s_data->entries = 0;
  dist_a_data->entries = 0;

112
  state = MAM_I_NOT_STARTED;
113

114
  MAM_Init_configuration();
115
  MAM_Zombies_service_init();
116
  init_malleability_times();
117
  MAM_Def_main_datatype();
118

119
120
  // Si son el primer grupo de procesos, obtienen los datos de los padres
  MPI_Comm_get_parent(&(mall->intercomm));
121
  if(mall->intercomm != MPI_COMM_NULL) { 
122
    Children_init(user_function, user_args);
123
    return MAM_TARGETS;
124
  }
125

126
  //TODO Check potential improvement - If check_hosts does not use slurm, internode_group could be obtained there
127
  MAM_check_hosts();
128
  mall->internode_group = MAM_Is_internode_group();
129
  MAM_Set_initial_configuration();
iker_martin's avatar
iker_martin committed
130

131
  #if MAM_USE_BARRIERS && MAM_DEBUG
132
133
134
135
    if(mall->myId == mall->root)
      printf("MaM: Using barriers to record times.\n");
  #endif

136
  #if MAM_DEBUG
137
    DEBUG_FUNC("MaM has been initialized correctly as parents", mall->myId, mall->numP); fflush(stdout); MPI_Barrier(*comm);
138
139
  #endif

140
  return MAM_SOURCES;
141
142
}

143
144
145
146
147
/*
 * Elimina toda la memoria reservado por el modulo
 * de maleabilidad y asegura que los zombies
 * despierten si los hubiese.
 */
148
149
int MAM_Finalize() {	  
  int request_abort;
150
151
152
153
154
155
156
157
158
  free_malleability_data_struct(rep_s_data);
  free_malleability_data_struct(rep_a_data);
  free_malleability_data_struct(dist_s_data);
  free_malleability_data_struct(dist_a_data);

  free(rep_s_data);
  free(rep_a_data);
  free(dist_s_data);
  free(dist_a_data);
159
  if(mall->nodelist != NULL) free(mall->nodelist);
160

161
  MAM_Free_main_datatype();
162
  request_abort = MAM_Zombies_service_free();
163
  free_malleability_times();
164
165
  if(mall->comm != MPI_COMM_WORLD && mall->comm != MPI_COMM_NULL) MPI_Comm_disconnect(&(mall->comm));
  if(mall->thread_comm != MPI_COMM_WORLD && mall->thread_comm != MPI_COMM_NULL) MPI_Comm_disconnect(&(mall->thread_comm));
166
  if(mall->intercomm != MPI_COMM_WORLD && mall->intercomm != MPI_COMM_NULL) { MPI_Comm_disconnect(&(mall->intercomm)); } //FIXME Error en OpenMPI + Merge
167
  if(mall->original_comm != MPI_COMM_WORLD && mall->original_comm != MPI_COMM_NULL) MPI_Comm_free(&(mall->original_comm));
168
169
  free(mall);
  free(mall_conf);
170
  free(user_reconf);
iker_martin's avatar
iker_martin committed
171

172
  state = MAM_I_UNRESERVED;
173
  return request_abort;
174
175
}

176
177
/* 
 * TODO Reescribir
178
179
180
181
 * Comprueba el estado de la maleabilidad. Intenta avanzar en la misma
 * si es posible. Funciona como una máquina de estados.
 * Retorna el estado de la maleabilidad concreto y modifica el argumento
 * "mam_state" a uno generico.
182
 *
183
184
 * El argumento "wait_completed" se utiliza para esperar a la finalización de
 * las tareas llevadas a cabo por parte de MAM.
185
186
 *
 */
187
int MAM_Checkpoint(int *mam_state, int wait_completed, void (*user_function)(void *), void *user_args) {
188
  int call_checkpoint = 0;
189

190
  //TODO This could be changed to an array with the functions to call in each case
191
  switch(state) {
192
    case MAM_I_UNRESERVED:
193
      *mam_state = MAM_UNRESERVED;
194
      break;
195
    case MAM_I_NOT_STARTED:
196
197
      call_checkpoint = MAM_St_rms(mam_state);
      break;
198
    case MAM_I_RMS_COMPLETED:
199
      call_checkpoint = MAM_St_spawn_start();
200
      break;
201

202
203
    case MAM_I_SPAWN_PENDING: // Comprueba si el spawn ha terminado
    case MAM_I_SPAWN_SINGLE_PENDING:
204
      call_checkpoint = MAM_St_spawn_pending(wait_completed);
205
      break;
206

207
208
    case MAM_I_SPAWN_ADAPT_POSTPONE:
    case MAM_I_SPAWN_COMPLETED:
209
      call_checkpoint = MAM_St_red_start();
210
      break;
211

212
    case MAM_I_DIST_PENDING:
213
      call_checkpoint = MAM_St_red_pending(wait_completed);
214
215
      break;

216
    case MAM_I_USER_START:
217
218
219
      call_checkpoint = MAM_St_user_start(mam_state);
      break;

220
    case MAM_I_USER_PENDING:
221
      call_checkpoint = MAM_St_user_pending(mam_state, wait_completed, user_function, user_args);
222
      break;
223

224
    case MAM_I_USER_COMPLETED:
225
      call_checkpoint = MAM_St_user_completed();
226
      break;
227

228
    case MAM_I_SPAWN_ADAPT_PENDING:
229
      call_checkpoint = MAM_St_spawn_adapt_pending(wait_completed);
230
231
      break;

232
233
    case MAM_I_SPAWN_ADAPTED:
    case MAM_I_DIST_COMPLETED:
234
      call_checkpoint = MAM_St_completed(mam_state);
235
236
      break;
  }
237

238
  if(call_checkpoint) { MAM_Checkpoint(mam_state, wait_completed, user_function, user_args); }
239
  if(state > MAM_I_NOT_STARTED && state < MAM_I_COMPLETED) *mam_state = MAM_PENDING;
240
241
242
  return state;
}

243
244
245
/*
 * TODO
 */
246
void MAM_Resume_redistribution(int *mam_state) {
247
  state = MAM_I_USER_COMPLETED;
248
  if(mam_state != NULL) *mam_state = MAM_PENDING;
249
250
251
252
253
}

/*
 * TODO
 */
254
void MAM_Commit(int *mam_state) {
255
  int request_abort;
256
  #if MAM_DEBUG
257
    if(mall->myId == mall->root){ DEBUG_FUNC("Trying to commit", mall->myId, mall->numP); } fflush(stdout);
258
259
  #endif

260
  // Get times before commiting
261
  if(mall_conf->spawn_method == MAM_SPAWN_BASELINE) {
262
    // This communication is only needed when the root process will become a zombie
263
    malleability_times_broadcast(mall->root_collectives);
264
  }
265

266
  // Free unneded communicators
267
268
  if(mall->tmp_comm != MPI_COMM_WORLD && mall->tmp_comm != MPI_COMM_NULL) MPI_Comm_disconnect(&(mall->tmp_comm));
  if(*(mall->user_comm) != MPI_COMM_WORLD && *(mall->user_comm) != MPI_COMM_NULL) MPI_Comm_disconnect(mall->user_comm);
269

270
271
  // Zombies Treatment
  MAM_Zombies_update();
272
  if(mall->zombie) {
273
    #if MAM_DEBUG >= 1
274
275
      DEBUG_FUNC("Is terminating as zombie", mall->myId, mall->numP); fflush(stdout);
    #endif
276
277
    request_abort = MAM_Finalize();
    if(request_abort) { MPI_Abort(MPI_COMM_WORLD, -101); }
278
    MPI_Finalize();
279
280
281
    exit(0);
  }

282
  // Reset/Free communicators
283
  if(mall_conf->spawn_method == MAM_SPAWN_MERGE) { MAM_comms_update(mall->intercomm); }
284
285
  if(mall->intercomm != MPI_COMM_NULL && mall->intercomm != MPI_COMM_WORLD) { MPI_Comm_disconnect(&(mall->intercomm)); } //FIXME Error en OpenMPI + Merge

iker_martin's avatar
iker_martin committed
286
287
  MPI_Comm_rank(mall->comm, &mall->myId);
  MPI_Comm_size(mall->comm, &mall->numP);
288
  mall->root = mall_conf->spawn_method == MAM_SPAWN_BASELINE ? mall->root : mall->root_parents;
289
  mall->root_parents = mall->root;
290
  state = MAM_I_NOT_STARTED;
291
  if(mam_state != NULL) *mam_state = MAM_COMPLETED;
292

293
  // Set new communicator
294
  MPI_Comm_dup(mall->comm, mall->user_comm);
295
296
  //if(mall_conf->spawn_method == MAM_SPAWN_BASELINE) { *(mall->user_comm) = MPI_COMM_WORLD; }
  //else if(mall_conf->spawn_method == MAM_SPAWN_MERGE) { MPI_Comm_dup(mall->comm, mall->user_comm); }
297
  #if MAM_DEBUG
298
    if(mall->myId == mall->root) DEBUG_FUNC("Reconfiguration has been commited", mall->myId, mall->numP); fflush(stdout);
299
  #endif
300

301
  #if MAM_USE_BARRIERS
302
303
304
    MPI_Barrier(mall->comm);
  #endif
  mall_conf->times->malleability_end = MPI_Wtime();
305
306
}

307
/*
308
309
310
311
312
313
314
315
316
 * This function adds data to a data structure based on whether the operation is synchronous or asynchronous,
 * and whether the data is replicated or distributed. It takes the following parameters:
 * - data: a pointer to the data to be added
 * - index: a pointer to a size_t variable where the index of the added data will be stored
 * - total_qty: the amount of elements in data
 * - type: the MPI datatype of the data
 * - is_replicated: a flag indicating whether the data is replicated (MAM_DATA_REPLICATED) or not (MAM_DATA_DISTRIBUTED)
 * - is_constant: a flag indicating whether the operation is asynchronous (MAM_DATA_CONSTANT) or synchronous (MAM_DATA_VARIABLE)
 * Finally, it updates the index with the index of the last added data if index is not NULL.
317
 */
318
319
void MAM_Data_add(void *data, size_t *index, size_t total_qty, MPI_Datatype type, int is_replicated, int is_constant) {
  size_t total_reqs = 0, returned_index;
320

321
  if(is_constant) { //Async
322
    if(is_replicated) {
323
      total_reqs = 1;
324
      add_data(data, total_qty, type, total_reqs, rep_a_data);
325
      returned_index = rep_a_data->entries-1;
326
    } else {
327
      if(mall_conf->red_method  == MAM_RED_BASELINE) {
328
        total_reqs = 1;
329
      } else if(mall_conf->red_method  == MAM_RED_POINT || mall_conf->red_method  == MAM_RED_RMA_LOCK || mall_conf->red_method  == MAM_RED_RMA_LOCKALL) {
330
        total_reqs = mall->numC;
331
      } 
332
333
      
      add_data(data, total_qty, type, total_reqs, dist_a_data);
334
      returned_index = dist_a_data->entries-1;
335
    }
336
  } else { //Sync
337
338
    if(is_replicated) {
      add_data(data, total_qty, type, total_reqs, rep_s_data);
339
      returned_index = rep_s_data->entries-1;
340
341
    } else {
      add_data(data, total_qty, type, total_reqs, dist_s_data);
342
      returned_index = dist_s_data->entries-1;
343
    }
344
  }
345
346

  if(index != NULL) *index = returned_index;
347
348
}

349
/*
350
351
352
353
354
355
356
357
 * This function modifies a data entry to a data structure based on whether the operation is synchronous or asynchronous,
 * and whether the data is replicated or distributed. It takes the following parameters:
 * - data: a pointer to the data to be added
 * - index: a value indicating which entry will be modified
 * - total_qty: the amount of elements in data
 * - type: the MPI datatype of the data
 * - is_replicated: a flag indicating whether the data is replicated (MAM_DATA_REPLICATED) or not (MAM_DATA_DISTRIBUTED)
 * - is_constant: a flag indicating whether the operation is asynchronous (MAM_DATA_CONSTANT) or synchronous (MAM_DATA_VARIABLE)
358
 */
359
void MAM_Data_modify(void *data, size_t index, size_t total_qty, MPI_Datatype type, int is_replicated, int is_constant) {
360
361
  size_t total_reqs = 0;

362
363
  if(is_constant) {
    if(is_replicated) {
364
      total_reqs = 1;
365
366
      modify_data(data, index, total_qty, type, total_reqs, rep_a_data); //FIXME total_reqs==0 ??? 
    } else {    
367
      if(mall_conf->red_method  == MAM_RED_BASELINE) {
368
        total_reqs = 1;
369
      } else if(mall_conf->red_method  == MAM_RED_POINT || mall_conf->red_method  == MAM_RED_RMA_LOCK || mall_conf->red_method  == MAM_RED_RMA_LOCKALL) {
370
371
372
        total_reqs = mall->numC;
      }
      
373
      modify_data(data, index, total_qty, type, total_reqs, dist_a_data);
374
    }
375
376
377
378
379
380
  } else {
    if(is_replicated) {
      modify_data(data, index, total_qty, type, total_reqs, rep_s_data);
    } else {
      modify_data(data, index, total_qty, type, total_reqs, dist_s_data);
    }
381
382
383
  }
}

384
/*
385
386
387
388
389
 * This functions returns how many data entries are available for one of the specific data structures.
 * It takes the following parameters:
 * - is_replicated: a flag indicating whether the structure is replicated (MAM_DATA_REPLICATED) or not (MAM_DATA_DISTRIBUTED)
 * - is_constant: a flag indicating whether the operation is asynchronous (MAM_DATA_CONSTANT) or synchronous (MAM_DATA_VARIABLE)
 * - entries: a pointer where the amount of entries will be stored
390
 */
391
void MAM_Data_get_entries(int is_replicated, int is_constant, size_t *entries){
392
393
394
  
  if(is_constant) {
    if(is_replicated) {
395
      *entries = rep_a_data->entries;
396
    } else {
397
      *entries = dist_a_data->entries;
398
399
400
    }
  } else {
    if(is_replicated) {
401
      *entries = rep_s_data->entries;
402
    } else {
403
      *entries = dist_s_data->entries;
404
405
406
407
408
    }
  }
}

/*
409
410
411
412
413
414
415
416
 * This function returns a data entry to a data structure based on whether the operation is synchronous or asynchronous,
 * and whether the data is replicated or distributed. It takes the following parameters:
 * - index: a value indicating which entry will be modified
 * - is_replicated: a flag indicating whether the data is replicated (MAM_DATA_REPLICATED) or not (MAM_DATA_DISTRIBUTED)
 * - is_constant: a flag indicating whether the operation is asynchronous (MAM_DATA_CONSTANT) or synchronous (MAM_DATA_VARIABLE)
 * - data: a pointer where the data will be stored. The user must free it
 * - total_qty: the amount of elements in data for all ranks
 * - local_qty: the amount of elements in data for this rank
417
 */
418
void MAM_Data_get_pointer(void **data, size_t index, size_t *total_qty, MPI_Datatype *type, int is_replicated, int is_constant) {
419
420
421
422
  malleability_data_t *data_struct;

  if(is_constant) {
    if(is_replicated) {
423
      data_struct = rep_a_data;
424
    } else {
425
      data_struct = dist_a_data;
426
427
428
    }
  } else {
    if(is_replicated) {
429
      data_struct = rep_s_data;
430
    } else {
431
      data_struct = dist_s_data;
432
433
434
    }
  }

435
  *data = data_struct->arrays[index];
436
437
438
  *total_qty = data_struct->qty[index];
  *type = data_struct->types[index];
  //get_block_dist(qty, mall->myId, mall->numP, &dist_data); //FIXME Asegurar que numP es correcto
439
440
}

441
442
443
/*
 * @brief Returns a structure to perform data redistribution during a reconfiguration.
 *
444
 * This function is intended to be called when the state of MaM is MAM_I_USER_PENDING only. 
445
446
447
448
449
450
451
 * It is designed to provide the necessary information for the user to perform data redistribution.
 *
 * Parameters:
 *   - mam_user_reconf_t *reconf_info: A pointer to a mam_user_reconf_t structure where the function will store the required information for data redistribution.
 *
 * Return Value:
 *   - MAM_OK: If the function successfully retrieves the reconfiguration information.
452
 *   - MAM_DENIED: If the function is called when the state of the MaM is not MAM_I_USER_PENDING.
453
454
 */
int MAM_Get_Reconf_Info(mam_user_reconf_t *reconf_info) {
455
  if(state != MAM_I_USER_PENDING) return MAM_DENIED;
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470

  *reconf_info = *user_reconf;
  return MAM_OK;
}

/*
 * @brief Returns the times used for the different steps of last reconfiguration.
 *
 * This function is intended to be called when a reconfiguration has ended. 
 * It is designed to provide the necessary information for the user to perform data redistribution.
 *
 * Parameters:
 *  - double *sp_time:   A pointer where the spawn time will be saved.
 *  - double *sy_time:   A pointer where the sychronous data redistribution time will be saved.
 *  - double *asy_time:  A pointer where the asychronous data redistribution time will be saved.
471
 *  - double *user_time: A pointer where the user data redistribution time will be saved.
472
473
 *  - double *mall_time: A pointer where the malleability time will be saved.
 */
474
475
void MAM_Retrieve_times(double *sp_time, double *sy_time, double *asy_time, double *user_time, double *mall_time) {
  MAM_I_retrieve_times(sp_time, sy_time, asy_time, user_time, mall_time);
476
}
477
478
479
480
481
482
483
484
485
486
487
488
489

//======================================================||
//================PRIVATE FUNCTIONS=====================||
//================DATA COMMUNICATION====================||
//======================================================||
//======================================================||

/*
 * Funcion generalizada para enviar datos desde los hijos.
 * La asincronizidad se refiere a si el hilo padre e hijo lo hacen
 * de forma bloqueante o no. El padre puede tener varios hilos.
 */
void send_data(int numP_children, malleability_data_t *data_struct, int is_asynchronous) {
490
  size_t i;
491
  void *aux_send, *aux_recv;
492
493
494

  if(is_asynchronous) {
    for(i=0; i < data_struct->entries; i++) {
495
      aux_send = data_struct->arrays[i];
496
      aux_recv = NULL;
497
      async_communication_start(aux_send, &aux_recv, data_struct->qty[i], data_struct->types[i], mall->numP, numP_children, MAM_SOURCES,  
498
		      mall->intercomm, &(data_struct->requests[i]), &(data_struct->request_qty[i]), &(data_struct->windows[i]));
499
      if(aux_recv != NULL) data_struct->arrays[i] = aux_recv;
500
501
502
    }
  } else {
    for(i=0; i < data_struct->entries; i++) {
503
      aux_send = data_struct->arrays[i];
504
      aux_recv = NULL;
505
      sync_communication(aux_send, &aux_recv, data_struct->qty[i], data_struct->types[i], mall->numP, numP_children, MAM_SOURCES, mall->intercomm);
506
      if(aux_recv != NULL) data_struct->arrays[i] = aux_recv;
507
508
509
510
511
512
513
514
515
516
    }
  }
}

/*
 * Funcion generalizada para recibir datos desde los hijos.
 * La asincronizidad se refiere a si el hilo padre e hijo lo hacen
 * de forma bloqueante o no. El padre puede tener varios hilos.
 */
void recv_data(int numP_parents, malleability_data_t *data_struct, int is_asynchronous) {
517
  size_t i;
518
  void *aux, *aux_s = NULL;
519
520
521

  if(is_asynchronous) {
    for(i=0; i < data_struct->entries; i++) {
522
      aux = data_struct->arrays[i];
523
      async_communication_start(aux_s, &aux, data_struct->qty[i], data_struct->types[i], mall->numP, numP_parents, MAM_TARGETS,
524
		      mall->intercomm, &(data_struct->requests[i]), &(data_struct->request_qty[i]), &(data_struct->windows[i]));
525
      data_struct->arrays[i] = aux;
526
527
528
    }
  } else {
    for(i=0; i < data_struct->entries; i++) {
529
      aux = data_struct->arrays[i];
530
      sync_communication(aux_s, &aux, data_struct->qty[i], data_struct->types[i], mall->numP, numP_parents, MAM_TARGETS, mall->intercomm);
531
      data_struct->arrays[i] = aux;
532
533
534
535
    }
  }
}

536
537
538
539
540
541

//======================================================||
//================PRIVATE FUNCTIONS=====================||
//====================MAM STAGES========================||
//======================================================||
//======================================================||
542
543
544
545
//======================================================||
//======================================================||
//======================================================||
//======================================================||
546

547
int MAM_St_rms(int *mam_state) {
548
  reset_malleability_times();
549
  #if MAM_USE_BARRIERS
550
551
552
    MPI_Barrier(mall->comm);
  #endif
  mall_conf->times->malleability_start = MPI_Wtime();
553

554
  MAM_Check_configuration();
555
  *mam_state = MAM_NOT_STARTED;
556
  state = MAM_I_RMS_COMPLETED;
557
558
  mall->wait_targets_posted = 0;

559
  //if(CHECK_RMS()) {return MAM_DENIED;}    
560
561
  return 1;
}
562

563
int MAM_St_spawn_start() {
564
  mall->num_parents = mall->numP;
565
566
  state = spawn_step();
  //FIXME Esto es necesario pero feo
567
568
  if(mall_conf->spawn_method == MAM_SPAWN_MERGE && mall->myId >= mall->numC){ mall->zombie = 1; }
  else if(mall_conf->spawn_method == MAM_SPAWN_BASELINE){ mall->zombie = 1; }
569

570
  if (state == MAM_I_SPAWN_COMPLETED || state == MAM_I_SPAWN_ADAPT_POSTPONE){
571
572
573
574
575
576
577
    return 1;
  }
  return 0;
}

int MAM_St_spawn_pending(int wait_completed) {
  state = check_spawn_state(&(mall->intercomm), mall->comm, wait_completed);
578
  if (state == MAM_I_SPAWN_COMPLETED || state == MAM_I_SPAWN_ADAPTED) {
579
    #if MAM_USE_BARRIERS
580
581
582
583
584
585
586
587
588
      MPI_Barrier(mall->comm);
    #endif
    mall_conf->times->spawn_time = MPI_Wtime() - mall_conf->times->malleability_start;
    return 1;
  }
  return 0;
}

int MAM_St_red_start() {
589
590
591
592
593
594
  if(MAM_Contains_strat(MAM_SPAWN_STRATEGIES, MAM_STRAT_SPAWN_INTERCOMM, NULL)) {
    mall->root_collectives = mall->myId == mall->root ? MPI_ROOT : MPI_PROC_NULL;
  } else {
    mall->root_collectives = mall->root;
  }

595
596
597
598
  state = start_redistribution();
  return 1;
}

599
int MAM_St_red_pending(int wait_completed) {
600
  if(MAM_Contains_strat(MAM_RED_STRATEGIES, MAM_STRAT_RED_PTHREAD, NULL)) {
601
602
603
604
605
    state = thread_check(wait_completed);
  } else {
    state = check_redistribution(wait_completed);
  }

606
607
  if(state != MAM_I_DIST_PENDING) { 
    state = MAM_I_USER_START;
608
609
610
611
612
    return 1;
  }
  return 0;
}

613
int MAM_St_user_start(int *mam_state) {
614
  #if MAM_USE_BARRIERS
615
616
617
    MPI_Barrier(mall->intercomm);
  #endif
  mall_conf->times->user_start = MPI_Wtime(); // Obtener timestamp de cuando termina user redist
618
  if(MAM_Contains_strat(MAM_SPAWN_STRATEGIES, MAM_STRAT_SPAWN_INTERCOMM, NULL)) {
619
    MPI_Intercomm_merge(mall->intercomm, MAM_SOURCES, &mall->tmp_comm); //El que pone 0 va primero
620
621
622
623
  } else {
    MPI_Comm_dup(mall->intercomm, &mall->tmp_comm);
  }
  MPI_Comm_set_name(mall->tmp_comm, "MAM_USER_TMP");
624
  state = MAM_I_USER_PENDING;
625
626
627
628
  *mam_state = MAM_USER_PENDING;
  return 1;
}

629
int MAM_St_user_pending(int *mam_state, int wait_completed, void (*user_function)(void *), void *user_args) {
630
  #if MAM_DEBUG
631
632
633
    if(mall->myId == mall->root) DEBUG_FUNC("Starting USER redistribution", mall->myId, mall->numP); fflush(stdout);
  #endif
  if(user_function != NULL) {
634
    MAM_I_create_user_struct(MAM_SOURCES);
635
636
    do {
      user_function(user_args);
637
    } while(wait_completed && state == MAM_I_USER_PENDING);
638
639
640
641
  } else {
    MAM_Resume_redistribution(mam_state);
  }

642
  if(state != MAM_I_USER_PENDING) {
643
    #if MAM_USE_BARRIERS
644
645
      MPI_Barrier(mall->intercomm);
    #endif
646
    if(mall_conf->spawn_method == MAM_SPAWN_MERGE) mall_conf->times->user_end = MPI_Wtime(); // Obtener timestamp de cuando termina user redist
647
    #if MAM_DEBUG
648
649
650
651
652
653
654
655
656
657
658
659
660
      if(mall->myId == mall->root) DEBUG_FUNC("Ended USER redistribution", mall->myId, mall->numP); fflush(stdout);
    #endif
    return 1;
  }
  return 0;
}

int MAM_St_user_completed() {
  state = end_redistribution();
  return 1;
}

int MAM_St_spawn_adapt_pending(int wait_completed) {
661
  wait_completed = MAM_WAIT_COMPLETION;
662
  #if MAM_USE_BARRIERS
663
664
665
666
667
    MPI_Barrier(mall->comm);
  #endif
  mall_conf->times->spawn_start = MPI_Wtime();
  unset_spawn_postpone_flag(state);
  state = check_spawn_state(&(mall->intercomm), mall->comm, wait_completed);
668
669
/* TODO Comentar problema, basicamente indicar que no es posible de la forma actual
 * Ademas es solo para una operación que hemos visto como "extremadamente" rápida
670
671
 * NO es posible debido a que solo se puede hacer tras enviar los datos variables 
 * y por tanto pierden validez dichos datos
672
  if(!MAM_Contains_strat(MAM_SPAWN_STRATEGIES, MAM_STRAT_SPAWN_PTHREAD, NULL)) {
673
    #if MAM_USE_BARRIERS
674
675
      MPI_Barrier(mall->comm);
    #endif
676
    mall_conf->times->spawn_time = MPI_Wtime() - mall_conf->times->spawn_start;
677
678
679
    return 1;
  }
  return 0;
680
  */
681
  #if MAM_USE_BARRIERS
682
683
684
685
    MPI_Barrier(mall->comm);
  #endif
  mall_conf->times->spawn_time = MPI_Wtime() - mall_conf->times->spawn_start;
  return 1;
686
687
688
}

int MAM_St_completed(int *mam_state) {
689
  MAM_Commit(mam_state);
690
691
692
693
  return 0;
}


694
695
696
697
698
//======================================================||
//================PRIVATE FUNCTIONS=====================||
//=====================CHILDREN=========================||
//======================================================||
//======================================================||
699
700
701
702
//======================================================||
//======================================================||
//======================================================||
//======================================================||
703
704
705
706
707
708
/*
 * Inicializacion de los datos de los hijos.
 * En la misma se reciben datos de los padres: La configuracion
 * de la ejecucion a realizar; y los datos a recibir de los padres
 * ya sea de forma sincrona, asincrona o ambas.
 */
709
void Children_init(void (*user_function)(void *), void *user_args) {
710
  size_t i;
711

712
  #if MAM_DEBUG
713
    DEBUG_FUNC("MaM will now initialize spawned processes", mall->myId, mall->numP); fflush(stdout); MPI_Barrier(MPI_COMM_WORLD);
714
715
  #endif

716
  malleability_connect_children(&(mall->intercomm));
717
  if(mall_conf->spawn_method == MAM_SPAWN_MERGE) { // For Merge Method, these processes will be added
iker_martin's avatar
iker_martin committed
718
719
    MPI_Comm_rank(mall->intercomm, &mall->myId);
    MPI_Comm_size(mall->intercomm, &mall->numP);
720
  }
721
  mall->root_collectives = mall->root_parents;
722

723
724
725
726
727
728
  if(MAM_Contains_strat(MAM_SPAWN_STRATEGIES, MAM_STRAT_SPAWN_MULTIPLE, NULL)) {
    mall->internode_group = 0;
  } else {
    mall->internode_group = MAM_Is_internode_group();
  }

729
  #if MAM_DEBUG
730
    DEBUG_FUNC("Spawned have completed spawn step", mall->myId, mall->numP); fflush(stdout); MPI_Barrier(MPI_COMM_WORLD);
731
732
  #endif

733
  comm_data_info(rep_a_data, dist_a_data, MAM_TARGETS);
734
  if(dist_a_data->entries || rep_a_data->entries) { // Recibir datos asincronos
735
    #if MAM_DEBUG >= 2
736
      DEBUG_FUNC("Spawned start asynchronous redistribution", mall->myId, mall->numP); fflush(stdout); MPI_Barrier(MPI_COMM_WORLD);
737
    #endif
738
    #if MAM_USE_BARRIERS
739
740
      MPI_Barrier(mall->intercomm);
    #endif
741

742
    if(MAM_Contains_strat(MAM_RED_STRATEGIES, MAM_STRAT_RED_PTHREAD, NULL)) {
743
      recv_data(mall->num_parents, dist_a_data, MAM_USE_SYNCHRONOUS);
744
      for(i=0; i<rep_a_data->entries; i++) {
745
        MPI_Bcast(rep_a_data->arrays[i], rep_a_data->qty[i], rep_a_data->types[i], mall->root_collectives, mall->intercomm);
746
      } 
747
    } else {
748
      recv_data(mall->num_parents, dist_a_data, MAM_USE_ASYNCHRONOUS); 
749

750
      for(i=0; i<rep_a_data->entries; i++) {
751
        MPI_Ibcast(rep_a_data->arrays[i], rep_a_data->qty[i], rep_a_data->types[i], mall->root_collectives, mall->intercomm, &(rep_a_data->requests[i][0]));
752
      } 
753
      #if MAM_DEBUG >= 2
754
        DEBUG_FUNC("Spawned started asynchronous redistribution", mall->myId, mall->numP); fflush(stdout); MPI_Barrier(MPI_COMM_WORLD);
755
      #endif
756

757
      for(i=0; i<rep_a_data->entries; i++) {
758
        async_communication_wait(rep_a_data->requests[i], rep_a_data->request_qty[i]);
759
      }
760
      for(i=0; i<dist_a_data->entries; i++) {
761
        async_communication_wait(dist_a_data->requests[i], dist_a_data->request_qty[i]);
762
      }
763
764
765
766
767
768
      if(MAM_Contains_strat(MAM_RED_STRATEGIES, MAM_STRAT_RED_WAIT_TARGETS, NULL)) {
        MPI_Ibarrier(mall->intercomm, &mall->wait_targets);
        mall->wait_targets_posted = 1;
        MPI_Wait(&mall->wait_targets, MPI_STATUS_IGNORE);
      }

769
      #if MAM_DEBUG >= 2
770
        DEBUG_FUNC("Spawned waited for all asynchronous redistributions", mall->myId, mall->numP); fflush(stdout); MPI_Barrier(MPI_COMM_WORLD);
771
      #endif
772
      for(i=0; i<dist_a_data->entries; i++) {
773
        async_communication_end(dist_a_data->requests[i], dist_a_data->request_qty[i], &(dist_a_data->windows[i]));
774
      }
775
      for(i=0; i<rep_a_data->entries; i++) {
776
        async_communication_end(rep_a_data->requests[i], rep_a_data->request_qty[i], &(rep_a_data->windows[i]));
777
      }
778
    }
779

780
    #if MAM_USE_BARRIERS
781
782
      MPI_Barrier(mall->intercomm);
    #endif
783
    mall_conf->times->async_end= MPI_Wtime(); // Obtener timestamp de cuando termina comm asincrona
784
  }
785
  #if MAM_DEBUG
786
    DEBUG_FUNC("Spawned have completed asynchronous data redistribution step", mall->myId, mall->numP); fflush(stdout); MPI_Barrier(MPI_COMM_WORLD);
787
  #endif
788

789
  #if MAM_USE_BARRIERS
790
791
    MPI_Barrier(mall->intercomm);
  #endif
792
  if(MAM_Contains_strat(MAM_SPAWN_STRATEGIES, MAM_STRAT_SPAWN_INTERCOMM, NULL)) {
793
    MPI_Intercomm_merge(mall->intercomm, MAM_TARGETS, &mall->tmp_comm); //El que pone 0 va primero
794
795
796
797
798
  } else {
    MPI_Comm_dup(mall->intercomm, &mall->tmp_comm);
  }
  MPI_Comm_set_name(mall->tmp_comm, "MAM_USER_TMP");
  if(user_function != NULL) {
799
    state = MAM_I_USER_PENDING;
800
    MAM_I_create_user_struct(MAM_TARGETS);
801
802
    user_function(user_args);
  }
803
  #if MAM_USE_BARRIERS
804
805
806
    MPI_Barrier(mall->intercomm);
  #endif
  mall_conf->times->user_end = MPI_Wtime(); // Obtener timestamp de cuando termina user redist
807

808
  comm_data_info(rep_s_data, dist_s_data, MAM_TARGETS);
809
  if(dist_s_data->entries || rep_s_data->entries) { // Recibir datos sincronos
810
    #if MAM_USE_BARRIERS
811
812
      MPI_Barrier(mall->intercomm);
    #endif
813
    recv_data(mall->num_parents, dist_s_data, MAM_USE_SYNCHRONOUS);
814
815

    for(i=0; i<rep_s_data->entries; i++) {
816
      MPI_Bcast(rep_s_data->arrays[i], rep_s_data->qty[i], rep_s_data->types[i], mall->root_collectives, mall->intercomm);
817
    } 
818
    #if MAM_USE_BARRIERS
819
820
      MPI_Barrier(mall->intercomm);
    #endif
821
    mall_conf->times->sync_end = MPI_Wtime(); // Obtener timestamp de cuando termina comm sincrona
822
  }
823
  #if MAM_DEBUG
824
    DEBUG_FUNC("Targets have completed synchronous data redistribution step", mall->myId, mall->numP); fflush(stdout); MPI_Barrier(MPI_COMM_WORLD);
825
  #endif
826

827
  MAM_Commit(NULL);
828

829
  #if MAM_DEBUG
830
    DEBUG_FUNC("MaM has been initialized correctly for new ranks", mall->myId, mall->numP); fflush(stdout); MPI_Barrier(MPI_COMM_WORLD);
831
  #endif
832
833
834
835
836
837
838
}

//======================================================||
//================PRIVATE FUNCTIONS=====================||
//=====================PARENTS==========================||
//======================================================||
//======================================================||
839
840
//======================================================||
//======================================================||
841
842
843
844
845
846

/*
 * Se encarga de realizar la creacion de los procesos hijos.
 * Si se pide en segundo plano devuelve el estado actual.
 */
int spawn_step(){
847
  #if MAM_USE_BARRIERS
848
849
    MPI_Barrier(mall->comm);
  #endif
850
  mall_conf->times->spawn_start = MPI_Wtime();
851
 
iker_martin's avatar
iker_martin committed
852
  state = init_spawn(mall->thread_comm, &(mall->intercomm));
853

854
  if(!MAM_Contains_strat(MAM_SPAWN_STRATEGIES, MAM_STRAT_SPAWN_PTHREAD, NULL)) {
855
      #if MAM_USE_BARRIERS
856
857
        MPI_Barrier(mall->comm);
      #endif
858
      mall_conf->times->spawn_time = MPI_Wtime() - mall_conf->times->malleability_start;
859
860
861
862
  }
  return state;
}

863

864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
/*
 * Comienza la redistribucion de los datos con el nuevo grupo de procesos.
 *
 * Primero se envia la configuracion a utilizar al nuevo grupo de procesos y a continuacion
 * se realiza el envio asincrono y/o sincrono si lo hay.
 *
 * En caso de que haya comunicacion asincrona, se comienza y se termina la funcion 
 * indicando que se ha comenzado un envio asincrono.
 *
 * Si no hay comunicacion asincrono se pasa a realizar la sincrona si la hubiese.
 *
 * Finalmente se envian datos sobre los resultados a los hijos y se desconectan ambos
 * grupos de procesos.
 */
int start_redistribution() {
879
  size_t i;
880

881
  if(mall->intercomm == MPI_COMM_NULL) {
882
883
    // Si no tiene comunicador creado, se debe a que se ha pospuesto el Spawn
    //   y se trata del spawn Merge Shrink
884
    MPI_Comm_dup(mall->comm, &(mall->intercomm));
885
  }
886

887
  comm_data_info(rep_a_data, dist_a_data, MAM_SOURCES);
888
  if(dist_a_data->entries || rep_a_data->entries) { // Enviar datos asincronos
889
    #if MAM_USE_BARRIERS
890
891
      MPI_Barrier(mall->intercomm);
    #endif
892
    mall_conf->times->async_start = MPI_Wtime();
893
    if(MAM_Contains_strat(MAM_RED_STRATEGIES, MAM_STRAT_RED_PTHREAD, NULL)) {
894
895
      return thread_creation();
    } else {
896
      send_data(mall->numC, dist_a_data, MAM_USE_ASYNCHRONOUS);
897
      for(i=0; i<rep_a_data->entries; i++) {
898
        MPI_Ibcast(rep_a_data->arrays[i], rep_a_data->qty[i], rep_a_data->types[i], mall->root_collectives, mall->intercomm, &(rep_a_data->requests[i][0]));
899
      } 
900
901
902
903
904

      if(mall->zombie && MAM_Contains_strat(MAM_RED_STRATEGIES, MAM_STRAT_RED_WAIT_TARGETS, NULL)) {
        MPI_Ibarrier(mall->intercomm, &mall->wait_targets);
        mall->wait_targets_posted = 1;
      }
905
      return MAM_I_DIST_PENDING; 
906
907
    }
  } 
908
  return MAM_I_USER_START;
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
}


/*
 * Comprueba si la redistribucion asincrona ha terminado. 
 * Si no ha terminado la funcion termina indicandolo, en caso contrario,
 * se continua con la comunicacion sincrona, el envio de resultados y
 * se desconectan los grupos de procesos.
 *
 * Esta funcion permite dos modos de funcionamiento al comprobar si la
 * comunicacion asincrona ha terminado.
 * Si se utiliza el modo "MAL_USE_NORMAL" o "MAL_USE_POINT", se considera 
 * terminada cuando los padres terminan de enviar.
 * Si se utiliza el modo "MAL_USE_IBARRIER", se considera terminada cuando
 * los hijos han terminado de recibir.
924
 * //FIXME Modificar para que se tenga en cuenta rep_a_data
925
 */
926
int check_redistribution(int wait_completed) {
927
  int completed, local_completed, all_completed;
928
  size_t i, req_qty;
929
  MPI_Request *req_completed;
930
931
  MPI_Win window;
  local_completed = 1;
932
  #if MAM_DEBUG >= 2
933
    DEBUG_FUNC("Sources are testing for all asynchronous redistributions", mall->myId, mall->numP); fflush(stdout); MPI_Barrier(MPI_COMM_WORLD);
934
  #endif
935

936
  if(wait_completed) {
937
938
939
    if(MAM_Contains_strat(MAM_RED_STRATEGIES, MAM_STRAT_RED_WAIT_TARGETS, NULL) && !mall->wait_targets_posted) {
      MPI_Ibarrier(mall->intercomm, &mall->wait_targets);
      mall->wait_targets_posted = 1;
940
941
942
943
    }
    for(i=0; i<dist_a_data->entries; i++) {
      req_completed = dist_a_data->requests[i];
      req_qty = dist_a_data->request_qty[i];
944
      async_communication_wait(req_completed, req_qty);
945
    }
946
947
948
    for(i=0; i<rep_a_data->entries; i++) {
      req_completed = rep_a_data->requests[i];
      req_qty = rep_a_data->request_qty[i];
949
      async_communication_wait(req_completed, req_qty);
950
    }
951
952

    if(MAM_Contains_strat(MAM_RED_STRATEGIES, MAM_STRAT_RED_WAIT_TARGETS, NULL)) { MPI_Wait(&mall->wait_targets, MPI_STATUS_IGNORE); }
953
  } else {
954
955
956
957
958
959
    if(mall->wait_targets_posted) { 
      MPI_Test(&mall->wait_targets, &local_completed, MPI_STATUS_IGNORE); 
    } else {
      for(i=0; i<dist_a_data->entries; i++) {
        req_completed = dist_a_data->requests[i];
        req_qty = dist_a_data->request_qty[i];
960
        completed = async_communication_check(MAM_SOURCES, req_completed, req_qty);
961
962
963
964
965
        local_completed = local_completed && completed;
      }
      for(i=0; i<rep_a_data->entries; i++) {
        req_completed = rep_a_data->requests[i];
        req_qty = rep_a_data->request_qty[i];
966
        completed = async_communication_check(MAM_SOURCES, req_completed, req_qty);
967
968
969
970
971
972
973
974
        local_completed = local_completed && completed;
      }

      if(local_completed && MAM_Contains_strat(MAM_RED_STRATEGIES, MAM_STRAT_RED_WAIT_TARGETS, NULL)) {
        MPI_Ibarrier(mall->intercomm, &mall->wait_targets);
        mall->wait_targets_posted = 1;
        MPI_Test(&mall->wait_targets, &local_completed, MPI_STATUS_IGNORE); //TODO - Figure out if last process takes profit from calling here
      }
975
    }
976
    #if MAM_DEBUG >= 2
977
978
979
980
      DEBUG_FUNC("Sources will now check a global decision", mall->myId, mall->numP); fflush(stdout); MPI_Barrier(MPI_COMM_WORLD);
    #endif

    MPI_Allreduce(&local_completed, &all_completed, 1, MPI_INT, MPI_MIN, mall->comm);
981
    if(!all_completed) return MAM_I_DIST_PENDING; // Continue only if asynchronous send has ended 
982
983
  }

984
  #if MAM_DEBUG >= 2
985
    DEBUG_FUNC("Sources sent asynchronous redistributions", mall->myId, mall->numP); fflush(stdout); MPI_Barrier(MPI_COMM_WORLD);
986
  #endif
987

988
989
990
991
  for(i=0; i<dist_a_data->entries; i++) {
    req_completed = dist_a_data->requests[i];
    req_qty = dist_a_data->request_qty[i];
    window = dist_a_data->windows[i];
992
    async_communication_end(req_completed, req_qty, &window);
993
  }
994
995
996
997
  for(i=0; i<rep_a_data->entries; i++) {
    req_completed = rep_a_data->requests[i];
    req_qty = rep_a_data->request_qty[i];
    window = rep_a_data->windows[i];
998
    async_communication_end(req_completed, req_qty, &window);
999
  }
1000

1001
  #if MAM_USE_BARRIERS
1002
1003
    MPI_Barrier(mall->intercomm);
  #endif
1004
1005
  if(mall_conf->spawn_method == MAM_SPAWN_MERGE) mall_conf->times->async_end = MPI_Wtime(); // Merge method only
  return MAM_I_USER_START;
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
}

/*
 * Termina la redistribución de los datos con los hijos, comprobando
 * si se han realizado iteraciones con comunicaciones en segundo plano
 * y enviando cuantas iteraciones se han realizado a los hijos.
 *
 * Además se realizan las comunicaciones síncronas se las hay.
 * Finalmente termina enviando los datos temporales a los hijos.
 */ 
int end_redistribution() {
1017
  size_t i;
1018
  int local_state;
1019

1020
  comm_data_info(rep_s_data, dist_s_data, MAM_SOURCES);
1021
  if(dist_s_data->entries || rep_s_data->entries) { // Enviar datos sincronos
1022
    #if MAM_USE_BARRIERS
1023
1024
      MPI_Barrier(mall->intercomm);
    #endif
1025
    mall_conf->times->sync_start = MPI_Wtime();
1026
    send_data(mall->numC, dist_s_data, MAM_USE_SYNCHRONOUS);
1027
1028

    for(i=0; i<rep_s_data->entries; i++) {
1029
      MPI_Bcast(rep_s_data->arrays[i], rep_s_data->qty[i], rep_s_data->types[i], mall->root_collectives, mall->intercomm);
1030
1031
    }

1032
    #if MAM_USE_BARRIERS
1033
1034
      MPI_Barrier(mall->intercomm);
    #endif
1035
    if(mall_conf->spawn_method == MAM_SPAWN_MERGE) mall_conf->times->sync_end = MPI_Wtime(); // Merge method only
1036
  }
iker_martin's avatar
iker_martin committed
1037

1038
1039
1040
  local_state = MAM_I_DIST_COMPLETED;
  if(mall_conf->spawn_method == MAM_SPAWN_MERGE && mall->numP > mall->numC) { // Merge Shrink
    local_state = MAM_I_SPAWN_ADAPT_PENDING;
1041
  }
1042

1043
  return local_state;
1044
1045
1046
1047
1048
1049
1050
1051
1052
}

// TODO MOVER A OTRO LADO??
//======================================================||
//================PRIVATE FUNCTIONS=====================||
//===============COMM PARENTS THREADS===================||
//======================================================||
//======================================================||

1053
1054

int comm_state; //FIXME Usar un handler
1055
1056
1057
1058
/*
 * Crea una hebra para ejecutar una comunicación en segundo plano.
 */
int thread_creation() {
1059
  comm_state = MAM_I_DIST_PENDING;
1060
1061
1062
1063
1064
  if(pthread_create(&(mall->async_thread), NULL, thread_async_work, NULL)) {
    printf("Error al crear el hilo\n");
    MPI_Abort(MPI_COMM_WORLD, -1);
    return -1;
  }
1065
  return comm_state;
1066
1067
1068
1069
1070
1071
1072
1073
}

/*
 * Comprobación por parte de una hebra maestra que indica
 * si una hebra esclava ha terminado su comunicación en segundo plano.
 *
 * El estado de la comunicación es devuelto al finalizar la función. 
 */
1074
int thread_check(int wait_completed) {
1075
  int all_completed = 0;
1076

1077
  if(wait_completed && comm_state == MAM_I_DIST_PENDING) {
1078
1079
1080
1081
1082
1083
1084
    if(pthread_join(mall->async_thread, NULL)) {
      printf("Error al esperar al hilo\n");
      MPI_Abort(MPI_COMM_WORLD, -1);
      return -2;
    } 
  }

1085
  // Comprueba que todos los hilos han terminado la distribucion (Mismo valor en commAsync)
1086
  MPI_Allreduce(&comm_state, &all_completed, 1, MPI_INT, MPI_MAX, mall->comm);
1087
  if(all_completed != MAM_I_DIST_COMPLETED) return MAM_I_DIST_PENDING; // Continue only if asynchronous send has ended 
1088
1089
1090
1091
1092
1093

  if(pthread_join(mall->async_thread, NULL)) {
    printf("Error al esperar al hilo\n");
    MPI_Abort(MPI_COMM_WORLD, -1);
    return -2;
  } 
1094

1095
  #if MAM_USE_BARRIERS
1096
1097
    MPI_Barrier(mall->intercomm);
  #endif
1098
1099
  if(mall_conf->spawn_method == MAM_SPAWN_MERGE) mall_conf->times->async_end = MPI_Wtime(); // Merge method only
  return MAM_I_USER_START;
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
}


/*
 * Función ejecutada por una hebra.
 * Ejecuta una comunicación síncrona con los hijos que
 * para el usuario se puede considerar como en segundo plano.
 *
 * Cuando termina la comunicación la hebra maestra puede comprobarlo
 * por el valor "commAsync".
 */
1111
void* thread_async_work() {
1112
1113
  size_t i;

1114
  send_data(mall->numC, dist_a_data, MAM_USE_SYNCHRONOUS);
1115
  for(i=0; i<rep_a_data->entries; i++) {
1116
    MPI_Bcast(rep_a_data->arrays[i], rep_a_data->qty[i], rep_a_data->types[i], mall->root_collectives, mall->intercomm);
1117
  } 
1118
  comm_state = MAM_I_DIST_COMPLETED;
1119
1120
  pthread_exit(NULL);
}
1121
1122
1123


//==============================================================================
1124
1125
1126
1127

/*
 * TODO Por hacer
 */
1128
void MAM_I_create_user_struct(int is_children_group) {
1129
1130
1131
  user_reconf->comm = mall->tmp_comm;

  if(is_children_group) {
1132
    user_reconf->rank_state = MAM_PROC_NEW_RANK;
iker_martin's avatar
iker_martin committed
1133
    user_reconf->numS = mall->num_parents;
1134
    user_reconf->numT = mall->numP;
1135
1136
1137
  } else {
    user_reconf->numS = mall->numP;
    user_reconf->numT = mall->numC;
1138
1139
    if(mall->zombie) user_reconf->rank_state = MAM_PROC_ZOMBIE;
    else user_reconf->rank_state = MAM_PROC_CONTINUE;
1140
1141
  }
}