malleabilityManager.c 31.5 KB
Newer Older
1
#include <pthread.h>
2
#include <string.h>
3
4
#include "malleabilityManager.h"
#include "malleabilityStates.h"
5
#include "malleabilityDataStructures.h"
6
#include "malleabilityTypes.h"
iker_martin's avatar
iker_martin committed
7
#include "malleabilityZombies.h"
8
#include "spawn_methods/GenericSpawn.h"
9
10
11
12
13
14
15
16
17
18
19
20
21
22
#include "CommDist.h"

#define MALLEABILITY_USE_SYNCHRONOUS 0
#define MALLEABILITY_USE_ASYNCHRONOUS 1


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);

void Children_init();
int spawn_step();
int start_redistribution();
int check_redistribution();
int end_redistribution();
iker_martin's avatar
iker_martin committed
23
int shrink_redistribution();
24

25
26
27
void comm_node_data(int rootBcast, int is_child_group);
void def_nodeinfo_type(MPI_Datatype *node_type);

28
29
int thread_creation();
int thread_check();
30
void* thread_async_work();
31

32
33
void print_comms_state();

34
typedef struct {
35
  int spawn_method;
36
  int spawn_dist;
37
  int spawn_strategies;
38
39
  int red_method;
  int red_strategies;
40

41
  int grp;
42
43
44
45
  configuration *config_file;
  results_data *results;
} malleability_config_t;

iker_martin's avatar
iker_martin committed
46
typedef struct { //FIXME numC_spawned no se esta usando
47
  int myId, numP, numC, numC_spawned, root, root_parents;
48
49
50
  pthread_t async_thread;
  MPI_Comm comm, thread_comm;
  MPI_Comm intercomm;
51
  MPI_Comm user_comm;
52
  
53
  char *name_exec, *nodelist;
54
  int num_cpus, num_nodes, nodelist_len;
55
56
} malleability_t;

57
int state = MALL_UNRESERVED; //FIXME Mover a otro lado
58
59
60
61
62
63
64
65
66

malleability_config_t *mall_conf;
malleability_t *mall;

malleability_data_t *rep_s_data;
malleability_data_t *dist_s_data;
malleability_data_t *rep_a_data;
malleability_data_t *dist_a_data;

67
/*
68
69
70
71
72
73
74
75
 * 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.
76
 */
77
int init_malleability(int myId, int numP, int root, MPI_Comm comm, char *name_exec, char *nodelist, int num_cpus, int num_nodes) {
78
79
80
81
82
83
84
85
86
  MPI_Comm dup_comm, thread_comm;

  mall_conf = (malleability_config_t *) malloc(sizeof(malleability_config_t));
  mall = (malleability_t *) malloc(sizeof(malleability_t));
  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
90
  MPI_Comm_set_name(dup_comm, "MPI_COMM_MALL");
  MPI_Comm_set_name(thread_comm, "MPI_COMM_MALL_THREAD");
91
92
93
94

  mall->myId = myId;
  mall->numP = numP;
  mall->root = root;
95
  mall->comm = dup_comm;
96
  mall->thread_comm = thread_comm;
97
  mall->user_comm = comm;
98

99
  mall->name_exec = name_exec;
100
101
102
  mall->nodelist = nodelist;
  mall->num_cpus = num_cpus;
  mall->num_nodes = num_nodes;
103
104
105
106
107
108

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

109
  state = MALL_NOT_STARTED;
110

111
112
  zombies_service_init();

113
114
115
116
  // Si son el primer grupo de procesos, obtienen los datos de los padres
  MPI_Comm_get_parent(&(mall->intercomm));
  if(mall->intercomm != MPI_COMM_NULL ) { 
    Children_init();
117
    return MALLEABILITY_CHILDREN;
118
  }
iker_martin's avatar
iker_martin committed
119

120
121
122
123
124
125
126
  if(nodelist != NULL) { //TODO To be deprecated by using Slurm or else statement
    mall->nodelist_len = strlen(nodelist);
  } else { // If no nodelist is detected, get it from the actual run
    mall->nodelist = malloc(MPI_MAX_PROCESSOR_NAME * sizeof(char));
    MPI_Get_processor_name(mall->nodelist, &mall->nodelist_len);
    //TODO Get name of each process and create real nodelist
  }
127

128
  return MALLEABILITY_NOT_CHILDREN;
129
130
}

131
132
133
134
135
/*
 * Elimina toda la memoria reservado por el modulo
 * de maleabilidad y asegura que los zombies
 * despierten si los hubiese.
 */
136
void free_malleability() {	  
137
138
139
140
141
142
143
144
145
146
  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);

147
148
  if(mall->comm != MPI_COMM_WORLD) MPI_Comm_free(&(mall->comm));
  if(mall->thread_comm != MPI_COMM_WORLD) MPI_Comm_free(&(mall->thread_comm));
149
150
  free(mall);
  free(mall_conf);
iker_martin's avatar
iker_martin committed
151
152
153
154

  zombies_awake();
  zombies_service_free();

155
  state = MALL_UNRESERVED;
156
157
}

158
159
/* 
 * TODO Reescribir
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
 * Se realiza el redimensionado de procesos por parte de los padres.
 *
 * Se crean los nuevos procesos con la distribucion fisica elegida y
 * a continuacion se transmite la informacion a los mismos.
 *
 * Si hay datos asincronos a transmitir, primero se comienza a
 * transmitir estos y se termina la funcion. Se tiene que comprobar con
 * llamando a la función de nuevo que se han terminado de enviar
 *
 * Si hay ademas datos sincronos a enviar, no se envian aun.
 *
 * Si solo hay datos sincronos se envian tras la creacion de los procesos
 * y finalmente se desconectan los dos grupos de procesos.
 */
int malleability_checkpoint() {
175
176
177
178
179
180
181
182
  double end_real_time;

  switch(state) {
    case MALL_UNRESERVED:
      break;
    case MALL_NOT_STARTED:
      // Comprobar si se tiene que realizar un redimensionado
      //if(CHECK_RMS()) {return MALL_DENIED;}
183

184
185
186
187
188
189
      state = spawn_step();

      if (state == MALL_SPAWN_COMPLETED || state == MALL_SPAWN_ADAPT_POSTPONE){
        malleability_checkpoint();
      }
      break;
190

191
192
193
194
195
196
    case MALL_SPAWN_PENDING: // Comprueba si el spawn ha terminado y comienza la redistribucion
    case MALL_SPAWN_SINGLE_PENDING:
      state = check_spawn_state(&(mall->intercomm), mall->comm, &end_real_time);
      if (state == MALL_SPAWN_COMPLETED || state == MALL_SPAWN_ADAPTED) {
        mall_conf->results->spawn_time[mall_conf->grp] = MPI_Wtime() - mall_conf->results->spawn_start;
        mall_conf->results->spawn_real_time[mall_conf->grp] = end_real_time - mall_conf->results->spawn_start;
197

198
199
200
        malleability_checkpoint();
      }
      break;
201

202
203
204
205
206
    case MALL_SPAWN_ADAPT_POSTPONE:
    case MALL_SPAWN_COMPLETED:
      state = start_redistribution();
      malleability_checkpoint();
      break;
207

208
    case MALL_DIST_PENDING:
209
      if(malleability_red_contains_strat(mall_conf->red_strategies, MALL_RED_THREAD, NULL)) {
210
211
212
213
        state = thread_check();
      } else {
        state = check_redistribution();
      }
214
      if(state != MALL_DIST_PENDING) { 
215
216
217
218
219
220
221
222
        malleability_checkpoint();
      }
      break;

    case MALL_SPAWN_ADAPT_PENDING:
      mall_conf->results->spawn_start = MPI_Wtime();
      unset_spawn_postpone_flag(state);
      state = check_spawn_state(&(mall->intercomm), mall->comm, &end_real_time);
223

224
      if(!malleability_spawn_contains_strat(mall_conf->spawn_strategies, MALL_SPAWN_PTHREAD, NULL)) {
225
226
        mall_conf->results->spawn_time[mall_conf->grp] = MPI_Wtime() - mall_conf->results->spawn_start;
      }
227
      break;
228

229
    case MALL_SPAWN_ADAPTED:
230
      state = shrink_redistribution();
231
      malleability_checkpoint();
232
      break;
233

234
235
236
237
    case MALL_DIST_COMPLETED: //TODO No es esto muy feo?
      state = MALL_COMPLETED;
      break;
  }
238
239
240
241
242
243
244
245
246
247
248
249
250
  return state;
}

// Funciones solo necesarias por el benchmark
//-------------------------------------------------------------------------------------------------------------
void set_benchmark_grp(int grp) {
  mall_conf->grp = grp;
}

void set_benchmark_configuration(configuration *config_file) {
  mall_conf->config_file = config_file;
}

251
void get_benchmark_configuration(configuration **config_file) {
252
253
254
255
256
257
258
  *config_file = mall_conf->config_file;
}

void set_benchmark_results(results_data *results) {
  mall_conf->results = results;
}

259
void get_benchmark_results(results_data **results) {
260
261
262
263
  *results = mall_conf->results;
}
//-------------------------------------------------------------------------------------------------------------

264
void set_malleability_configuration(int spawn_method, int spawn_strategies, int spawn_dist, int red_method, int red_strategies) {
265
266
  mall_conf->spawn_method = spawn_method;
  mall_conf->spawn_strategies = spawn_strategies;
267
  mall_conf->spawn_dist = spawn_dist;
268
269
  mall_conf->red_method = red_method;
  mall_conf->red_strategies = red_strategies;
270
271
272
273
}

/*
 * To be deprecated
274
 * Tiene que ser llamado despues de setear la config
275
276
 */
void set_children_number(int numC){
277
  if((mall_conf->spawn_method == MALL_SPAWN_MERGE) && (numC >= mall->numP)) {
278
279
280
281
282
    mall->numC = numC;
    mall->numC_spawned = numC - mall->numP;

    if(numC == mall->numP) { // Migrar
      mall->numC_spawned = numC;
283
      mall_conf->spawn_method = MALL_SPAWN_BASELINE;
284
285
286
287
288
289
290
291
292
293
294
295
    }
  } else {
    mall->numC = numC;
    mall->numC_spawned = numC;
  }
}

/*
 * TODO
 */
void get_malleability_user_comm(MPI_Comm *comm) {
  *comm = mall->user_comm;
296
297
298
299
300
301
302
303
304
}

/*
 * Anyade a la estructura concreta de datos elegida
 * el nuevo set de datos "data" de un total de "total_qty" elementos.
 *
 * Los datos variables se tienen que anyadir cuando quieran ser mandados, no antes
 *
 * Mas informacion en la funcion "add_data".
305
306
 *
 * //FIXME Si es constante se debería ir a asincrono, no sincrono
307
 */
308
void malleability_add_data(void *data, size_t total_qty, int type, int is_replicated, int is_constant) {
309
  size_t total_reqs = 0;
310
311
312

  if(is_constant) {
    if(is_replicated) {
313
      add_data(data, total_qty, type, total_reqs, rep_s_data);
314
    } else {
315
      add_data(data, total_qty, type, total_reqs, dist_s_data);
316
317
318
    }
  } else {
    if(is_replicated) {
319
      add_data(data, total_qty, type, total_reqs, rep_a_data); //FIXME total_reqs==0 ??? 
320
    } else {
321
      if(mall_conf->red_method  == MALL_RED_BASELINE) {
322
        total_reqs = 1;
323
      } else if(mall_conf->red_method  == MALL_RED_IBARRIER) { //TODO This is a strategy, not a method
324
        total_reqs = 2;
325
      } else if(mall_conf->red_method  == MALL_RED_POINT) {
326
        total_reqs = mall->numC;
327
328
329
330
331
332
333
      }
      
      add_data(data, total_qty, type, total_reqs, dist_a_data);
    }
  }
}

334
335
336
337
338
339
340
/*
 * Modifica en la estructura concreta de datos elegida en el indice "index"
 * con el set de datos "data" de un total de "total_qty" elementos.
 *
 * Los datos variables se tienen que modificar cuando quieran ser mandados, no antes
 *
 * Mas informacion en la funcion "modify_data".
341
 * //FIXME Si es constante se debería ir a asincrono, no sincrono
342
343
 */
void malleability_modify_data(void *data, size_t index, size_t total_qty, int type, int is_replicated, int is_constant) {
344
345
  size_t total_reqs = 0;

346
347
  if(is_constant) {
    if(is_replicated) {
348
      modify_data(data, index, total_qty, type, total_reqs, rep_s_data);
349
    } else {
350
      modify_data(data, index, total_qty, type, total_reqs, dist_s_data);
351
352
353
    }
  } else {
    if(is_replicated) {
354
355
356
      modify_data(data, index, total_qty, type, total_reqs, rep_a_data); //FIXME total_reqs==0 ??? 
    } else {    
      if(mall_conf->red_method  == MALL_RED_BASELINE) {
357
        total_reqs = 1;
358
      } else if(mall_conf->red_method  == MALL_RED_IBARRIER) { //TODO This is a strategy, not a method
359
        total_reqs = 2;
360
      } else if(mall_conf->red_method  == MALL_RED_POINT) {
361
362
363
        total_reqs = mall->numC;
      }
      
364
      modify_data(data, index, total_qty, type, total_reqs, dist_a_data);
365
366
367
368
    }
  }
}

369
370
371
/*
 * Devuelve el numero de entradas para la estructura de descripcion de 
 * datos elegida.
372
 * //FIXME Si es constante se debería ir a asincrono, no sincrono
373
 */
374
void malleability_get_entries(size_t *entries, int is_replicated, int is_constant){
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
  
  if(is_constant) {
    if(is_replicated) {
      *entries = rep_s_data->entries;
    } else {
      *entries = dist_s_data->entries;
    }
  } else {
    if(is_replicated) {
      *entries = rep_a_data->entries;
    } else {
      *entries = dist_a_data->entries;
    }
  }
}

/*
 * Devuelve el elemento de la lista "index" al usuario.
 * La devolución es en el mismo orden que lo han metido los padres
 * con la funcion "malleability_add_data()".
 * Es tarea del usuario saber el tipo de esos datos.
 * TODO Refactor a que sea automatico
397
 * //FIXME Si es constante se debería ir a asincrono, no sincrono
398
 */
399
void malleability_get_data(void **data, int index, int is_replicated, int is_constant) {
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
  malleability_data_t *data_struct;

  if(is_constant) {
    if(is_replicated) {
      data_struct = rep_s_data;
    } else {
      data_struct = dist_s_data;
    }
  } else {
    if(is_replicated) {
      data_struct = rep_a_data;
    } else {
      data_struct = dist_a_data;
    }
  }

416
  *data = data_struct->arrays[index];
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
}


//======================================================||
//================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) {
433
  size_t i;
434
  char *aux_send, *aux_recv;
435
436
437

  if(is_asynchronous) {
    for(i=0; i < data_struct->entries; i++) {
438
      aux_send = (char *) data_struct->arrays[i]; //TODO Comprobar que realmente es un char
439
      aux_recv = NULL;
440
441
      async_communication(aux_send, &aux_recv, data_struct->qty[i], mall->myId, mall->numP, numP_children, MALLEABILITY_NOT_CHILDREN, mall_conf->red_method, mall_conf->red_strategies, mall->intercomm, 
		      &(data_struct->requests[i]), &(data_struct->request_qty[i]));
442
      if(aux_recv != NULL) data_struct->arrays[i] = (void *) aux_recv;
443
444
445
    }
  } else {
    for(i=0; i < data_struct->entries; i++) {
446
447
      aux_send = (char *) data_struct->arrays[i]; //TODO Comprobar que realmente es un char
      aux_recv = NULL;
448
      sync_communication(aux_send, &aux_recv, data_struct->qty[i], mall->myId, mall->numP, numP_children, MALLEABILITY_NOT_CHILDREN, mall_conf->red_method, mall->intercomm);
449
      if(aux_recv != NULL) data_struct->arrays[i] = (void *) aux_recv;
450
451
452
453
454
455
456
457
458
459
    }
  }
}

/*
 * 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) {
460
  size_t i;
461
  char *aux, aux_s;
462
463
464
465

  if(is_asynchronous) {
    for(i=0; i < data_struct->entries; i++) {
      aux = (char *) data_struct->arrays[i]; //TODO Comprobar que realmente es un char
466
467
      async_communication(&aux_s, &aux, data_struct->qty[i], mall->myId, mall->numP, numP_parents, MALLEABILITY_CHILDREN, mall_conf->red_method, mall_conf->red_strategies, mall->intercomm, 
		      &(data_struct->requests[i]), &(data_struct->request_qty[i]));
468
469
470
471
472
      data_struct->arrays[i] = (void *) aux;
    }
  } else {
    for(i=0; i < data_struct->entries; i++) {
      aux = (char *) data_struct->arrays[i]; //TODO Comprobar que realmente es un char
473
      sync_communication(&aux_s, &aux, data_struct->qty[i], mall->myId, mall->numP, numP_parents, MALLEABILITY_CHILDREN, mall_conf->red_method, mall->intercomm);
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
      data_struct->arrays[i] = (void *) aux;
    }
  }
}

//======================================================||
//================PRIVATE FUNCTIONS=====================||
//=====================CHILDREN=========================||
//======================================================||
//======================================================||

/*
 * 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.
 */
void Children_init() {
492
493
  size_t i;
  int numP_parents, root_parents;
494
  int is_intercomm;
495

496
  malleability_connect_children(mall->myId, mall->numP, mall->root, mall->comm, &numP_parents, &root_parents, &(mall->intercomm));
497
498
499
500
501
  MPI_Comm_test_inter(mall->intercomm, &is_intercomm);
  if(!is_intercomm) { // For intracommunicators, these processes will be added
    MPI_Comm_rank(mall->intercomm, &(mall->myId));
    MPI_Comm_size(mall->intercomm, &(mall->numP));
  }
502

503
  recv_config_file(mall->root, mall->intercomm, &(mall_conf->config_file));
504
  comm_node_data(root_parents, MALLEABILITY_CHILDREN);
505
506
  MPI_Bcast(&(mall_conf->red_method), 1, MPI_INT, root_parents, mall->intercomm);
  MPI_Bcast(&(mall_conf->red_strategies), 1, MPI_INT, root_parents, mall->intercomm);
507

508
  mall_conf->results = (results_data *) malloc(sizeof(results_data));
509
  init_results_data(mall_conf->results, mall_conf->config_file->n_resizes, mall_conf->config_file->n_stages, RESULTS_INIT_DATA_QTY);
510

511
  comm_data_info(rep_a_data, dist_a_data, MALLEABILITY_CHILDREN, mall->myId, root_parents, mall->intercomm);
512
513
  if(dist_a_data->entries || rep_a_data->entries) { // Recibir datos asincronos

514
    if(malleability_red_contains_strat(mall_conf->red_strategies, MALL_RED_THREAD, NULL)) {
515
      recv_data(numP_parents, dist_a_data, MALLEABILITY_USE_SYNCHRONOUS);
516
517
    } else {
      recv_data(numP_parents, dist_a_data, MALLEABILITY_USE_ASYNCHRONOUS);
518
    }
519

520
521
    mall_conf->results->async_end= MPI_Wtime(); // Obtener timestamp de cuando termina comm asincrona
  }
522

523
  comm_data_info(rep_s_data, dist_s_data, MALLEABILITY_CHILDREN, mall->myId, root_parents, mall->intercomm);
524
  if(dist_s_data->entries || rep_s_data->entries) { // Recibir datos sincronos
525
    recv_data(numP_parents, dist_s_data, MALLEABILITY_USE_SYNCHRONOUS);
526
527
528
529
530
531

    mall_conf->results->sync_end = MPI_Wtime(); // Obtener timestamp de cuando termina comm sincrona

    // TODO Crear funcion especifica y anyadir para Asinc
    // TODO Tener en cuenta el tipo y qty
    for(i=0; i<rep_s_data->entries; i++) {
532
533
534
535
536
537
      MPI_Datatype datatype;
      if(rep_s_data->types[i] == MAL_INT) {
        datatype = MPI_INT;
      } else {
        datatype = MPI_CHAR;
      }
538
      MPI_Bcast(rep_s_data->arrays[i], rep_s_data->qty[i], datatype, root_parents, mall->intercomm);
539
540
541
    } 
  }

542
  // Guardar los resultados de esta transmision
543
  comm_results(mall_conf->results, mall->root, mall_conf->config_file->n_resizes, mall->intercomm);
544
545
546
547
548
549
550
  if(!is_intercomm) {
    if(mall->thread_comm != MPI_COMM_WORLD) MPI_Comm_free(&(mall->thread_comm));
    if(mall->comm != MPI_COMM_WORLD) MPI_Comm_free(&(mall->comm));
    if(mall->user_comm != MPI_COMM_WORLD) MPI_Comm_free(&(mall->user_comm)); //TODO No es peligroso?

    MPI_Comm_dup(mall->intercomm, &(mall->thread_comm));
    MPI_Comm_dup(mall->intercomm, &(mall->comm));
551
    MPI_Comm_dup(mall->intercomm, &(mall->user_comm)); 
552
  }
553

554
  MPI_Comm_disconnect(&(mall->intercomm));
555
556
557
558
559
560
561
562
563
564
565
566
567
568
}

//======================================================||
//================PRIVATE FUNCTIONS=====================||
//=====================PARENTS==========================||
//======================================================||
//======================================================||

/*
 * Se encarga de realizar la creacion de los procesos hijos.
 * Si se pide en segundo plano devuelve el estado actual.
 */
int spawn_step(){
  mall_conf->results->spawn_start = MPI_Wtime();
569
 
570
  state = init_spawn(mall->name_exec, mall->num_cpus, mall->num_nodes, mall->nodelist, mall->myId, mall->numP, mall->numC, mall->root, mall_conf->spawn_dist, mall_conf->spawn_method, mall_conf->spawn_strategies, mall->thread_comm, &(mall->intercomm));
571

572
  if(!malleability_spawn_contains_strat(mall_conf->spawn_strategies, MALL_SPAWN_PTHREAD, NULL)) {
573
574
575
576
577
      mall_conf->results->spawn_time[mall_conf->grp] = MPI_Wtime() - mall_conf->results->spawn_start;
  }
  return state;
}

578

579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
/*
 * 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() {
594
595
596
597
598
599
600
601
  int rootBcast, is_intercomm;

  is_intercomm = 0;
  if(mall->intercomm != MPI_COMM_NULL) {
    MPI_Comm_test_inter(mall->intercomm, &is_intercomm);
  } else { 
    // Si no tiene comunicador creado, se debe a que se ha pospuesto el Spawn
    //   y se trata del spawn Merge Shrink
602
    MPI_Comm_dup(mall->comm, &(mall->intercomm));
603
  }
604

605
606
607
608
609
  if(is_intercomm) {
    rootBcast = mall->myId == mall->root ? MPI_ROOT : MPI_PROC_NULL;
  } else {
    rootBcast = mall->root;
  }
610

611
  send_config_file(mall_conf->config_file, rootBcast, mall->intercomm);
612
  comm_node_data(rootBcast, MALLEABILITY_NOT_CHILDREN);
613
614
  MPI_Bcast(&(mall_conf->red_method), 1, MPI_INT, rootBcast, mall->intercomm);
  MPI_Bcast(&(mall_conf->red_strategies), 1, MPI_INT, rootBcast, mall->intercomm);
615

616
  comm_data_info(rep_a_data, dist_a_data, MALLEABILITY_NOT_CHILDREN, mall->myId, mall->root, mall->intercomm);
617
  if(dist_a_data->entries || rep_a_data->entries) { // Enviar datos asincronos
618
	  //FIXME No se envian los datos replicados (rep_a_data)
619
    mall_conf->results->async_time[mall_conf->grp] = MPI_Wtime();
620
    if(malleability_red_contains_strat(mall_conf->red_strategies, MALL_RED_THREAD, NULL)) {
621
622
623
      return thread_creation();
    } else {
      send_data(mall->numC, dist_a_data, MALLEABILITY_USE_ASYNCHRONOUS);
624
      return MALL_DIST_PENDING; 
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
    }
  } 
  return end_redistribution();
}


/*
 * @deprecated
 * 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.
 */
int check_redistribution() {
646
  int is_intercomm, req_qty, completed, all_completed, test_err;
647
648
  MPI_Request *req_completed;

649
650
651
652
653
  //FIXME Modificar para que sea un for
  req_completed = dist_a_data->requests[0]; //FIXME Numero magico
  req_qty = dist_a_data->request_qty[0]; //FIXME Numero magico
  if(malleability_red_contains_strat(mall_conf->red_strategies, MALL_RED_IBARRIER, NULL)) { //FIXME Strategy not fully implemented
    test_err = MPI_Test(&(req_completed[req_qty-1]), &completed, MPI_STATUS_IGNORE);
654
  } else {
655
    test_err = MPI_Testall(req_qty, req_completed, &completed, MPI_STATUSES_IGNORE); //FIXME Numero magico
656
657
658
659
660
661
662
663
  }
 
  if (test_err != MPI_SUCCESS && test_err != MPI_ERR_PENDING) {
    printf("P%d aborting -- Test Async\n", mall->myId);
    MPI_Abort(MPI_COMM_WORLD, test_err);
  }

  MPI_Allreduce(&completed, &all_completed, 1, MPI_INT, MPI_MIN, mall->comm);
664
  if(!all_completed) return MALL_DIST_PENDING; // Continue only if asynchronous send has ended 
665
  
666
667
  if(malleability_red_contains_strat(mall_conf->red_strategies, MALL_RED_IBARRIER, NULL)) { //FIXME Strategy not fully implemented
    MPI_Waitall(req_qty, req_completed, MPI_STATUSES_IGNORE);
668
669
670
    //Para la desconexión de ambos grupos de procesos es necesario indicar a MPI que esta comm
    //ha terminado, aunque solo se pueda llegar a este punto cuando ha terminado
  }
671
672
673
674


  MPI_Comm_test_inter(mall->intercomm, &is_intercomm);
  if(!is_intercomm) mall_conf->results->async_end = MPI_Wtime(); // Merge method only
675
676
677
678
679
680
681
682
683
684
685
686
687
  return end_redistribution();
}


/*
 * 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() {
688
689
  size_t i;
  int is_intercomm, rootBcast, local_state;
690

691
  MPI_Comm_test_inter(mall->intercomm, &is_intercomm);
692
693
694
695
696
697
  if(is_intercomm) {
    rootBcast = mall->myId == mall->root ? MPI_ROOT : MPI_PROC_NULL;
  } else {
    rootBcast = mall->root;
  }
  
698
  comm_data_info(rep_s_data, dist_s_data, MALLEABILITY_NOT_CHILDREN, mall->myId, mall->root, mall->intercomm);
699
  if(dist_s_data->entries || rep_s_data->entries) { // Enviar datos sincronos
700
    mall_conf->results->sync_time[mall_conf->grp] = MPI_Wtime();
701
    send_data(mall->numC, dist_s_data, MALLEABILITY_USE_SYNCHRONOUS);
702
    if(!is_intercomm) mall_conf->results->sync_end = MPI_Wtime(); // Merge method only
703
704

    // TODO Crear funcion especifica y anyadir para Asinc
705
    // TODO Tener en cuenta el tipo
706
    for(i=0; i<rep_s_data->entries; i++) {
707
708
709
710
711
712
      MPI_Datatype datatype;
      if(rep_s_data->types[i] == MAL_INT) {
        datatype = MPI_INT;
      } else {
        datatype = MPI_CHAR;
      }
713
      MPI_Bcast(rep_s_data->arrays[i], rep_s_data->qty[i], datatype, rootBcast, mall->intercomm);
714
715
    } 
  }
iker_martin's avatar
iker_martin committed
716

717
  comm_results(mall_conf->results, rootBcast, mall_conf->config_file->n_resizes, mall->intercomm);
iker_martin's avatar
iker_martin committed
718

719
720
721
722
723
724
  local_state = MALL_DIST_COMPLETED;
  if(!is_intercomm) { // Merge Spawn
    if(mall->numP < mall->numC) { // Expand
      if(mall->thread_comm != MPI_COMM_WORLD) MPI_Comm_free(&(mall->thread_comm));
      if(mall->comm != MPI_COMM_WORLD) MPI_Comm_free(&(mall->comm));
      if(mall->user_comm != MPI_COMM_WORLD) MPI_Comm_free(&(mall->user_comm)); //TODO No es peligroso?
725

726
727
728
      MPI_Comm_dup(mall->intercomm, &(mall->thread_comm));
      MPI_Comm_dup(mall->intercomm, &(mall->comm));
      MPI_Comm_dup(mall->intercomm, &(mall->user_comm));
729
730
731
732

      MPI_Comm_set_name(mall->thread_comm, "MPI_COMM_MALL_THREAD");
      MPI_Comm_set_name(mall->comm, "MPI_COMM_MALL");
      MPI_Comm_set_name(mall->user_comm, "MPI_COMM_MALL_USER");
733
734
735
736
    } else { // Shrink || Merge Shrink requiere de mas tareas
      local_state = MALL_SPAWN_ADAPT_PENDING;
    }
  }
737

738
739
  if(mall->intercomm != MPI_COMM_NULL && mall->intercomm != MPI_COMM_WORLD) {
    MPI_Comm_disconnect(&(mall->intercomm));
740
  }
741

742
  return local_state;
743
744
}

745
746
747
748

///=============================================
///=============================================
///=============================================
749
//TODO Add comment
iker_martin's avatar
iker_martin committed
750
int shrink_redistribution() {
751
    double time_extra = MPI_Wtime();
752
753

    //TODO REFACTOR -- Que solo la llamada de collect iters este fuera de los hilos
754
    zombies_collect_suspended(mall->comm, mall->myId, mall->numP, mall->numC, mall->root, (void *) mall_conf->results, mall_conf->config_file->n_stages);
iker_martin's avatar
iker_martin committed
755
756
    
    if(mall->myId < mall->numC) {
757
758
759
760
761
762
763
764
      if(mall->thread_comm != MPI_COMM_WORLD) MPI_Comm_free(&(mall->thread_comm));
      if(mall->comm != MPI_COMM_WORLD) MPI_Comm_free(&(mall->comm));
      if(mall->user_comm != MPI_COMM_WORLD) MPI_Comm_free(&(mall->user_comm)); //TODO No es peligroso?

      MPI_Comm_dup(mall->intercomm, &(mall->thread_comm));
      MPI_Comm_dup(mall->intercomm, &(mall->comm));
      MPI_Comm_dup(mall->intercomm, &(mall->user_comm));

765
766
767
768
      MPI_Comm_set_name(mall->thread_comm, "MPI_COMM_MALL_THREAD");
      MPI_Comm_set_name(mall->comm, "MPI_COMM_MALL");
      MPI_Comm_set_name(mall->user_comm, "MPI_COMM_MALL_USER");

769
770
      MPI_Comm_free(&(mall->intercomm));

771
772
773
      mall_conf->results->spawn_time[mall_conf->grp] += MPI_Wtime() - time_extra;
      if(malleability_spawn_contains_strat(mall_conf->spawn_strategies,MALL_SPAWN_PTHREAD, NULL)) {
          mall_conf->results->spawn_real_time[mall_conf->grp] += MPI_Wtime() - time_extra;
774
      }
775
      return MALL_DIST_COMPLETED;
iker_martin's avatar
iker_martin committed
776
    } else {
777
      return MALL_ZOMBIE;
iker_martin's avatar
iker_martin committed
778
779
780
    }
}

781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
//======================================================||
//================PRIVATE FUNCTIONS=====================||
//=================COMM NODE INFO ======================||
//======================================================||
//======================================================||
//TODO Add comment
void comm_node_data(int rootBcast, int is_child_group) {
  MPI_Datatype node_type;

  def_nodeinfo_type(&node_type);
  MPI_Bcast(mall, 1, node_type, rootBcast, mall->intercomm);

  if(is_child_group) {
    mall->nodelist = malloc((mall->nodelist_len+1) * sizeof(char));
    mall->nodelist[mall->nodelist_len] = '\0';
  }
  MPI_Bcast(mall->nodelist, mall->nodelist_len, MPI_CHAR, rootBcast, mall->intercomm);

  MPI_Type_free(&node_type);
}

//TODO Add comment
void def_nodeinfo_type(MPI_Datatype *node_type) {
  int i, counts = 3;
  int blocklengths[3] = {1, 1, 1};
  MPI_Aint displs[counts], dir;
  MPI_Datatype types[counts];

  // Rellenar vector types
  types[0] = types[1] = types[2] = MPI_INT;

  // Rellenar vector displs
  MPI_Get_address(mall, &dir);

  MPI_Get_address(&(mall->num_cpus), &displs[0]);
  MPI_Get_address(&(mall->num_nodes), &displs[1]);
  MPI_Get_address(&(mall->nodelist_len), &displs[2]);

  for(i=0;i<counts;i++) displs[i] -= dir;

  MPI_Type_create_struct(counts, blocklengths, displs, types, node_type);
  MPI_Type_commit(node_type);
}

825
826
827
828
829
830
831
// TODO MOVER A OTRO LADO??
//======================================================||
//================PRIVATE FUNCTIONS=====================||
//===============COMM PARENTS THREADS===================||
//======================================================||
//======================================================||

832

833
int comm_state; //FIXME Usar un handler
834
835
836
837
/*
 * Crea una hebra para ejecutar una comunicación en segundo plano.
 */
int thread_creation() {
838
  comm_state = MALL_DIST_PENDING;
839
840
841
842
843
  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;
  }
844
  return comm_state;
845
846
847
848
849
850
851
852
853
}

/*
 * 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. 
 */
int thread_check() {
854
  int all_completed = 0, is_intercomm;
855
856

  // Comprueba que todos los hilos han terminado la distribucion (Mismo valor en commAsync)
857
  MPI_Allreduce(&comm_state, &all_completed, 1, MPI_INT, MPI_MAX, mall->comm);
858
859
  if(all_completed != MALL_DIST_COMPLETED) return MALL_DIST_PENDING; // Continue only if asynchronous send has ended 
  //FIXME No se tiene en cuenta el estado MALL_APP_ENDED
860
861
862
863
864
865

  if(pthread_join(mall->async_thread, NULL)) {
    printf("Error al esperar al hilo\n");
    MPI_Abort(MPI_COMM_WORLD, -1);
    return -2;
  } 
866
867
  MPI_Comm_test_inter(mall->intercomm, &is_intercomm);
  if(!is_intercomm) mall_conf->results->async_end = MPI_Wtime(); // Merge method only
868
869
870
871
872
873
874
875
876
877
878
879
  return end_redistribution();
}


/*
 * 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".
 */
880
void* thread_async_work() {
881
  send_data(mall->numC, dist_a_data, MALLEABILITY_USE_SYNCHRONOUS);
882
  comm_state = MALL_DIST_COMPLETED;
883
884
  pthread_exit(NULL);
}
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904


//==============================================================================
/*
 * Muestra por pantalla el estado actual de todos los comunicadores
 */
void print_comms_state() {
  int tester;
  char *test = malloc(MPI_MAX_OBJECT_NAME * sizeof(char));

  MPI_Comm_get_name(mall->comm, test, &tester);
  printf("P%d Comm=%d Name=%s\n", mall->myId, mall->comm, test);
  MPI_Comm_get_name(mall->user_comm, test, &tester);
  printf("P%d Comm=%d Name=%s\n", mall->myId, mall->user_comm, test);
  if(mall->intercomm != MPI_COMM_NULL) {
    MPI_Comm_get_name(mall->intercomm, test, &tester);
    printf("P%d Comm=%d Name=%s\n", mall->myId, mall->intercomm, test);
  }
  free(test);
}