Main.c 13 KB
Newer Older
1
2
3
#include <stdio.h>
#include <stdlib.h>
#include <mpi.h>
iker_martin's avatar
iker_martin committed
4
#include <fcntl.h>
5
#include <unistd.h>
iker_martin's avatar
iker_martin committed
6
#include <sys/stat.h>
7
#include "computing_func.h"
8
9
10
11
12
13
//#include "../IOcodes/read_ini.h"
//#include "../IOcodes/results.h"
//#include "../malleability/ProcessDist.h"
#include "../malleability/CommDist.h" //TODO Refactor para que no haga falta
#include "../malleability/malleabilityManager.h"
#include "../malleability/malleabilityStates.h"
14
15
16

#define ROOT 0

iker_martin's avatar
iker_martin committed
17
int work();
18
void iterate(double *matrix, int n, int async_comm);
19

20
void init_group_struct(char *argv[], int argc, int myId, int numP);
21
void init_application();
22
void obtain_op_times();
23
24
void free_application_data();

25
void print_general_info(int myId, int grp, int numP);
26
int print_local_results();
27
int print_final_results();
iker_martin's avatar
iker_martin committed
28
int create_out_file(char *nombre, int *ptr, int newstdout);
29

iker_martin's avatar
iker_martin committed
30
31
32
33
typedef struct {
  int myId;
  int numP;
  int grp;
34
  int iter_start;
35
  int argc;
iker_martin's avatar
iker_martin committed
36

37
  int numS; // Cantidad de procesos hijos
iker_martin's avatar
iker_martin committed
38
  MPI_Comm children, parents;
39
40

  char *compute_comm_array;
iker_martin's avatar
iker_martin committed
41
  char **argv;
42
  char *sync_array, *async_array;
iker_martin's avatar
iker_martin committed
43
44
45
46
} group_data;

configuration *config_file;
group_data *group;
47
results_data *results;
48
MPI_Comm comm;
49
int run_id = 0; // Utilizado para diferenciar más fácilmente ejecuciones en el análisis
50

51
int main(int argc, char *argv[]) {
52
    int numP, myId, res;
iker_martin's avatar
iker_martin committed
53
    int req;
54
    int im_child;
55

56
    MPI_Init_thread(&argc, &argv, MPI_THREAD_MULTIPLE, &req);
57
    MPI_Comm_size(MPI_COMM_WORLD, &numP);
iker_martin's avatar
iker_martin committed
58
    MPI_Comm_rank(MPI_COMM_WORLD, &myId);
59
    comm = MPI_COMM_WORLD;
iker_martin's avatar
iker_martin committed
60

61
62
63
64
    if(req != MPI_THREAD_MULTIPLE) {
      printf("No se ha obtenido la configuración de hilos necesaria\nSolicitada %d -- Devuelta %d\n", req, MPI_THREAD_MULTIPLE);
    }

65
    init_group_struct(argv, argc, myId, numP);
66
    im_child = init_malleability(myId, numP, ROOT, comm, argv[0]);
67

68
    if(!im_child) {
69
70
      init_application();

71
      set_benchmark_grp(group->grp);
72
73
74
      set_benchmark_configuration(config_file);
      set_benchmark_results(results);

75
      MPI_Barrier(comm);
76
      results->exec_start = MPI_Wtime();
77
78
    } else { //Init hijos
      get_malleability_user_comm(&comm);
79
80
      get_benchmark_configuration(&config_file); //No se obtiene bien el archivo
      get_benchmark_results(&results); //No se obtiene bien el archivo
81
      set_results_post_reconfig(results, group->grp, config_file->sdr, config_file->adr); //TODO Cambio al añadir nueva redistribucion
82

83
84
85
      if(config_file->comm_tam) {
        group->compute_comm_array = malloc(config_file->comm_tam * sizeof(char));
      }
86
87
88
89
90
91
92
93

      void *value = NULL;
      malleability_get_data(&value, 0, 1, 1);
      group->grp = *((int *)value);
      free(value);
      malleability_get_data(&value, 1, 1, 1);
      run_id = *((int *)value);
      free(value);
94
95
96
97
      
      malleability_get_data(&value, 2, 1, 1);
      group->iter_start = *((int *)value);
      free(value);
98
99

      group->grp = group->grp + 1;
100
101
    }

102
103
    group->grp = group->grp - 1; // TODO REFACTOR???
    do {
104

105
106
107
108
109
110
111
      group->grp = group->grp + 1;
      set_benchmark_grp(group->grp);
      get_malleability_user_comm(&comm);
      MPI_Comm_size(comm, &(group->numP));
      MPI_Comm_rank(comm, &(group->myId));

      if(config_file->resizes != group->grp + 1) { 
112
        set_malleability_configuration(config_file->cst, config_file->css, config_file->phy_dist[group->grp+1], -1, config_file->aib, -1);
113
114
115
116
117
        set_children_number(config_file->procs[group->grp+1]); // TODO TO BE DEPRECATED

        if(group->grp == 0) {
          malleability_add_data(&(group->grp), 1, MAL_INT, 1, 1);
          malleability_add_data(&run_id, 1, MAL_INT, 1, 1);
118
          malleability_add_data(&(group->iter_start), 1, MAL_INT, 1, 1);
119
        }
120
      }
121
122

      res = work();
iker_martin's avatar
iker_martin committed
123
      if(res == MAL_ZOMBIE) break;
124
125

      print_local_results();
126
127
      reset_results_index(results);
    } while((config_file->resizes > group->grp + 1) && (config_file->cst == COMM_SPAWN_MERGE || config_file->cst == COMM_SPAWN_MERGE_PTHREAD));
128

129

iker_martin's avatar
iker_martin committed
130
    if(res==1) { // Se he llegado al final de la aplicacion
131
      MPI_Barrier(comm); // TODO Posible error al utilizar SHRINK
132
      results->exec_time = MPI_Wtime() - results->exec_start;
133
    }
134

135
    print_final_results(); // Pasado este punto ya no pueden escribir los procesos
136
137
138
139

    if(comm != MPI_COMM_WORLD && comm != MPI_COMM_NULL) {
      MPI_Comm_free(&comm);
    }
iker_martin's avatar
iker_martin committed
140
    free_application_data();
141

142
    if(group->myId == ROOT) MPI_Abort(MPI_COMM_WORLD, -100);
143
    MPI_Finalize();
144

145
146
147
148
    return 0;
}

/*
149
150
151
152
153
154
155
156
157
 * Función de trabajo principal.
 *
 * Incializa los datos para realizar el computo y a continuacion
 * pasa a realizar "maxiter" iteraciones de computo.
 *
 * Terminadas las iteraciones realiza el redimensionado de procesos.
 * Si el redimensionado se realiza de forma asincrona se 
 * siguen realizando iteraciones de computo hasta que termine la 
 * comunicacion asincrona y realizar entonces la sincrona.
158
159
160
161
 *
 * Si el grupo de procesos es el ultimo que va a ejecutar, se devuelve
 * el valor 1 para indicar que no se va a seguir trabajando con nuevos grupos
 * de procesos. En caso contrario se devuelve 0.
162
 */
iker_martin's avatar
iker_martin committed
163
int work() {
164
  int iter, maxiter, state, res;
165
  double *matrix = NULL;
166

iker_martin's avatar
iker_martin committed
167
  maxiter = config_file->iters[group->grp];
168
  //initMatrix(&matrix, config_file->matrix_tam);
169
  state = MAL_NOT_STARTED;
170
  
171
  res = 0;
172
  for(iter=group->iter_start; iter < maxiter; iter++) {
173
    iterate(matrix, config_file->matrix_tam, state);
174
  }
175

176
177
178
  if(config_file->iters[group->grp] == iter && config_file->resizes != group->grp + 1)
    state = malleability_checkpoint();

179
  iter = 0;
180
  while(state == MAL_DIST_PENDING || state == MAL_SPAWN_PENDING) {
181
    iterate(matrix, config_file->matrix_tam, state);
182
    iter++;
183
    state = malleability_checkpoint();
184
  }
185
  group->iter_start = iter;
186
  
187
  if(config_file->resizes - 1 == group->grp) res=1;
iker_martin's avatar
iker_martin committed
188
  if(state == MAL_ZOMBIE) res=state;
189
  return res;
190
191
}

192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218

/////////////////////////////////////////
/////////////////////////////////////////
//COMPUTE FUNCTIONS
/////////////////////////////////////////
/////////////////////////////////////////


/*
 * Simula la ejecucción de una iteración de computo en la aplicación
 * que dura al menos un tiempo de "time" segundos.
 */
void iterate(double *matrix, int n, int async_comm) {
  double start_time, actual_time;
  double time = config_file->general_time * config_file->factors[group->grp];
  double Top = config_file->Top;
  int i, operations = 0;
  double aux = 0;

  start_time = actual_time = MPI_Wtime();

  operations = time / Top; //FIXME Calcular una sola vez
  for(i=0; i < operations; i++) {
    aux += computePiSerial(n);
  }

  if(config_file->comm_tam) {
219
    MPI_Bcast(group->compute_comm_array, config_file->comm_tam, MPI_CHAR, ROOT, comm);
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
  }

  actual_time = MPI_Wtime(); // Guardar tiempos
  // TODO Que diferencie entre ambas en el IO
  if(async_comm == MAL_DIST_PENDING || async_comm == MAL_SPAWN_PENDING) { // Se esta realizando una redistribucion de datos asincrona
    operations=0;
  }

  if(results->iter_index == results->iters_size) { // Aumentar tamaño de ambos vectores de resultados
    realloc_results_iters(results, results->iters_size + 100);
  }
  results->iters_time[results->iter_index] = actual_time - start_time;
  results->iters_type[results->iter_index] = operations;
  results->iter_index = results->iter_index + 1;
}

//======================================================||
//======================================================||
//=============INIT/FREE/PRINT FUNCTIONS================||
//======================================================||
//======================================================||

/*
 * Muestra datos generales sobre los procesos, su grupo,
 * en que nodo residen y la version de MPI utilizada.
 */
void print_general_info(int myId, int grp, int numP) {
  int len;
  char *name = malloc(MPI_MAX_PROCESSOR_NAME * sizeof(char));
  char *version = malloc(MPI_MAX_LIBRARY_VERSION_STRING * sizeof(char));
  MPI_Get_processor_name(name, &len);
  MPI_Get_library_version(version, &len);
  printf("P%d Nuevo GRUPO %d de %d procs en nodo %s con %s\n", myId, grp, numP, name, version);

  free(name);
  free(version);
}

258

259
260
261
/*
 * Pide al proceso raiz imprimir los datos sobre las iteraciones realizadas por el grupo de procesos.
 */
262
263
int print_local_results() {
  int ptr_local, ptr_out, err;
264
265
266
  char *file_name;

  if(group->myId == ROOT) {
267
268
    ptr_out = dup(1);

269
270
271
272
273
274
275
276
277
278
279
    file_name = NULL;
    file_name = malloc(40 * sizeof(char));
    if(file_name == NULL) return -1; // No ha sido posible alojar la memoria
    err = snprintf(file_name, 40, "R%d_G%dNP%dID%d.out", run_id, group->grp, group->numP, group->myId);
    if(err < 0) return -2; // No ha sido posible obtener el nombre de fichero
    create_out_file(file_name, &ptr_local, 1);
  
    print_config_group(config_file, group->grp);
    print_iter_results(*results, config_file->iters[group->grp] -1);
    free(file_name);

280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
    close(1);
    dup(ptr_out);
  }
  return 0;
}

/*
 * Si es el ultimo grupo de procesos, pide al proceso raiz mostrar los datos obtenidos de tiempo de ejecucion, creacion de procesos
 * y las comunicaciones.
 */
int print_final_results() {
  int ptr_global, err;
  char *file_name;

  if(group->myId == ROOT) {

296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
    if(group->grp == config_file->resizes -1) {
      file_name = NULL;
      file_name = malloc(20 * sizeof(char));
      if(file_name == NULL) return -1; // No ha sido posible alojar la memoria
      err = snprintf(file_name, 20, "R%d_Global.out", run_id);
      if(err < 0) return -2; // No ha sido posible obtener el nombre de fichero

      create_out_file(file_name, &ptr_global, 1);
      print_config(config_file, group->grp);
      print_global_results(*results, config_file->resizes);
      free(file_name);
      
    }
  }
  return 0;
}

/*
 * Inicializa la estructura group
 */
void init_group_struct(char *argv[], int argc, int myId, int numP) {
  group = malloc(1 * sizeof(group_data));
  group->myId        = myId;
  group->numP        = numP;
  group->grp         = 0;
  group->iter_start  = 0;
  group->argc        = argc;
  group->argv        = argv;
}

/*
 * Inicializa los datos para este grupo de procesos.
 *
 * En caso de ser el primer grupo de procesos, lee el fichero de configuracion
 * e inicializa los vectores de comunicacion.
 *
 * En caso de ser otro grupo de procesos entra a la funcion "Sons_init()" donde
 * se comunican con los padres para inicializar sus datos.
 */
void init_application() {
  if(group->argc < 2) {
    printf("Falta el fichero de configuracion. Uso:\n./programa config.ini id\nEl argumento numerico id es opcional\n");
    MPI_Abort(MPI_COMM_WORLD, -1);
  }
  if(group->argc > 2) {
    run_id = atoi(group->argv[2]);
  }

  config_file = read_ini_file(group->argv[1]);
  results = malloc(sizeof(results_data));
  init_results_data(results, config_file->resizes, config_file->iters[group->grp]);
  if(config_file->comm_tam) {
    group->compute_comm_array = malloc(config_file->comm_tam * sizeof(char));
  }
  if(config_file->sdr) {
    malloc_comm_array(&(group->sync_array), config_file->sdr , group->myId, group->numP);
  }
  if(config_file->adr) {
    malloc_comm_array(&(group->async_array), config_file->adr , group->myId, group->numP);
  }
   
  obtain_op_times();
}

/*
 * Obtiene cuanto tiempo es necesario para realizar una operacion de PI
 */
void obtain_op_times() {
  double result, start_time = MPI_Wtime();
  int i, qty = 20000;
  result = 0;
  for(i=0; i<qty; i++) {
    result += computePiSerial(config_file->matrix_tam);
  }
  //printf("Creado Top con valor %lf\n", result);
  //fflush(stdout);

  config_file->Top = (MPI_Wtime() - start_time) / qty; //Tiempo de una operacion
374
  MPI_Bcast(&(config_file->Top), 1, MPI_DOUBLE, ROOT, comm);
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
}

/*
 * Libera toda la memoria asociada con la aplicacion
 */
void free_application_data() {
  if(config_file->comm_tam) {
    free(group->compute_comm_array);
  }
  if(config_file->sdr) {
    free(group->sync_array);
  }
  if(config_file->adr) {
    free(group->async_array);
  }
  
391
392
393
  free_malleability();
  free_config(config_file);

394
  if(group->grp == 0) { //FIXME Revisar porque cuando es diferente a 0 no funciona
395
    free_results_data(results);
396
    free(results);
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
  }
  free(group);

}


/* 
 * Función para crear un fichero con el nombre pasado como argumento.
 * Si el nombre ya existe, se escribe la informacion a continuacion.
 *
 * El proceso que llama a la función pasa a tener como salida estandar
 * dicho fichero si el valor "newstdout" es verdadero.
 *
 */
int create_out_file(char *nombre, int *ptr, int newstdout) {
  int err;

  *ptr = open(nombre, O_WRONLY | O_CREAT | O_APPEND, 0644);
  if(*ptr < 0) return -1; // No ha sido posible crear el fichero

  if(newstdout) {
    err = close(1);
    if(err < 0) return -2; // No es posible modificar la salida estandar
    err = dup(*ptr);
    if(err < 0) return -3; // No es posible modificar la salida estandar
  }

  return 0;
}