WIP. Refactoring. Ahora se ha separado completamente la logica de la...

WIP. Refactoring. Ahora se ha separado completamente la logica de la maleabilidad de la aplicación sintética

Codes/IOcodes/read_ini.c

@@ -129,7 +129,7 @@ void free_config(configuration *user_config) {
     }
 }
 
-/* TODO Añadir COMMTAM
+/*
  * Imprime por salida estandar toda la informacion que contiene
  * la configuracion pasada como argumento
  */

Codes/IOcodes/results.c

@@ -11,7 +11,7 @@ void def_results_type(results_data *results, int resizes, MPI_Datatype *results_
 //======================================================||
 //======================================================||
 
-
+//TODO Generalizar ambas funciones en una sola
 /*
  * Envia una estructura de resultados al grupo de procesos al que se 
  * enlaza este grupo a traves del intercomunicador pasado como argumento.
@@ -57,7 +57,7 @@ void recv_results(results_data *results, int root, int resizes, MPI_Comm interco
  * Define un tipo derivado de MPI para mandar los tiempos
  * con una sola comunicacion.
  *
- * En concreto son tres escales y un vector de tamaño "resizes"
+ * En concreto son tres escalares y dos vectores de tamaño "resizes"
  */
 void def_results_type(results_data *results, int resizes, MPI_Datatype *results_type) {
   int i, counts = 5;
@@ -83,6 +83,30 @@ void def_results_type(results_data *results, int resizes, MPI_Datatype *results_
   MPI_Type_create_struct(counts, blocklengths, displs, types, results_type);
   MPI_Type_commit(results_type);
 }
+//======================================================||
+//======================================================||
+//================SET RESULTS FUNCTIONS=================||
+//======================================================||
+//======================================================||
+
+/*
+ * Guarda los resultados respecto a la redistribución de datos
+ * tras una reconfiguración. A llamar por los hijos tras
+ * terminar la redistribución y obtener la configuración.
+ */
+void set_results_post_reconfig(results_data *results, int grp, int sdr, int adr) {
+  if(sdr) { // Si no hay datos sincronos, el tiempo es 0
+    results->sync_time[grp]  = results->sync_end - results->sync_start;
+  } else {
+    results->sync_time[grp]  = 0;
+  }
+  if(adr) { // Si no hay datos asincronos, el tiempo es 0
+    results->async_time[grp]  = results->async_end - results->async_start;
+  } else {
+    results->async_time[grp]  = 0;
+  }
+}
+
 
 //======================================================||
 //======================================================||
@@ -96,22 +120,22 @@ void def_results_type(results_data *results, int resizes, MPI_Datatype *results_
  * por iteracion, el tipo (Normal o durante communicacion asincrona)
  * y cuantas operaciones internas se han realizado en cada iteracion.
  */
-void print_iter_results(results_data *results, int last_normal_iter_index) {
+void print_iter_results(results_data results, int last_normal_iter_index) {
   int i, aux;
 
   printf("Titer: ");
-  for(i=0; i< results->iter_index; i++) {
-    printf("%lf ", results->iters_time[i]);
+  for(i=0; i< results.iter_index; i++) {
+    printf("%lf ", results.iters_time[i]);
   }
 
   printf("\nTtype: "); //FIXME modificar a imprimir solo la cantidad de asincronas
-  for(i=0; i< results->iter_index; i++) {
-    printf("%d ", results->iters_type[i] == 0);
+  for(i=0; i< results.iter_index; i++) {
+    printf("%d ", results.iters_type[i] == 0);
   }
 
   printf("\nTop: "); //TODO modificar a imprimir solo cuantas operaciones cuestan una iteracion?
-  for(i=0; i< results->iter_index; i++) {
-    aux = results->iters_type[i] == 0 ? results->iters_type[last_normal_iter_index] : results->iters_type[i];
+  for(i=0; i< results.iter_index; i++) {
+    aux = results.iters_type[i] == 0 ? results.iters_type[last_normal_iter_index] : results.iters_type[i];
     printf("%d ", aux);
   }
   printf("\n");
@@ -122,30 +146,30 @@ void print_iter_results(results_data *results, int last_normal_iter_index) {
  * Estos son el tiempo de creacion de procesos, los de comunicacion
  * asincrona y sincrona y el tiempo total de ejecucion.
  */
-void print_global_results(results_data *results, int resizes) {
+void print_global_results(results_data results, int resizes) {
   int i;
 
   printf("Tspawn: ");
   for(i=0; i< resizes - 1; i++) {
-    printf("%lf ", results->spawn_time[i]);
+    printf("%lf ", results.spawn_time[i]);
   }
 
   printf("\nTthread: ");
   for(i=0; i< resizes - 1; i++) {
-    printf("%lf ", results->spawn_thread_time[i]);
+    printf("%lf ", results.spawn_thread_time[i]);
   }
 
   printf("\nTsync: ");
   for(i=1; i < resizes; i++) {
-    printf("%lf ", results->sync_time[i]);
+    printf("%lf ", results.sync_time[i]);
   }
 
   printf("\nTasync: ");
   for(i=1; i < resizes; i++) {
-    printf("%lf ", results->async_time[i]);
+    printf("%lf ", results.async_time[i]);
   }
 
-  printf("\nTex: %lf\n", results->exec_time);
+  printf("\nTex: %lf\n", results.exec_time);
 }
 
 //======================================================||
@@ -160,18 +184,18 @@ void print_global_results(results_data *results, int resizes) {
  * Los argumentos "resizes" y "iters_size" se necesitan para obtener el tamaño
  * de los vectores de resultados.
  */
-void init_results_data(results_data **results, int resizes, int iters_size) {
-  *results = malloc(1 * sizeof(results_data));
-
-  (*results)->spawn_time = calloc(resizes, sizeof(double));
-  (*results)->spawn_thread_time = calloc(resizes, sizeof(double));
-  (*results)->sync_time = calloc(resizes, sizeof(double));
-  (*results)->async_time = calloc(resizes, sizeof(double));
-
-  (*results)->iters_size = iters_size + 100;
-  (*results)->iters_time = calloc(iters_size + 100, sizeof(double)); //FIXME Numero magico
-  (*results)->iters_type = calloc(iters_size + 100, sizeof(int));
-  (*results)->iter_index = 0;
+void init_results_data(results_data *results, int resizes, int iters_size) {
+  //*results = malloc(1 * sizeof(results_data)); FIXME Borrar
+
+  results->spawn_time = calloc(resizes, sizeof(double));
+  results->spawn_thread_time = calloc(resizes, sizeof(double));
+  results->sync_time = calloc(resizes, sizeof(double));
+  results->async_time = calloc(resizes, sizeof(double));
+
+  results->iters_size = iters_size + 100;
+  results->iters_time = calloc(iters_size + 100, sizeof(double)); //FIXME Numero magico
+  results->iters_type = calloc(iters_size + 100, sizeof(int));
+  results->iter_index = 0;
 }
 
 void realloc_results_iters(results_data *results, int needed) {
@@ -193,13 +217,13 @@ void realloc_results_iters(results_data *results, int needed) {
 /*
  * Libera toda la memoria asociada con una estructura de resultados.
  */
-void free_results_data(results_data **results) {
-    free((*results)->spawn_time);
-    free((*results)->spawn_thread_time);
-    free((*results)->sync_time);
-    free((*results)->async_time);
-
-    free((*results)->iters_time);
-    free((*results)->iters_type);
-    free(*results);
+void free_results_data(results_data *results) {
+    free(results->spawn_time);
+    free(results->spawn_thread_time);
+    free(results->sync_time);
+    free(results->async_time);
+
+    free(results->iters_time);
+    free(results->iters_type);
+    //free(*results); FIXME Borrar
 }

Codes/IOcodes/results.h

@@ -2,6 +2,8 @@
 #include <stdlib.h>
 #include <mpi.h>
 
+#define RESULTS_INIT_DATA_QTY 100
+
 typedef struct {
   // Iters data
   double *iters_time;
@@ -9,8 +11,8 @@ typedef struct {
 
   // Spawn, Thread, Sync, Async and Exec time
   double spawn_start, *spawn_time, *spawn_thread_time;
-  double sync_start,  *sync_time;
-  double async_start, *async_time;
+  double sync_start, sync_end,  *sync_time;
+  double async_start, async_end, *async_time;
   double exec_start, exec_time;
 } results_data;
 
@@ -18,8 +20,10 @@ typedef struct {
 void send_results(results_data *results, int root, int resizes, MPI_Comm intercomm);
 void recv_results(results_data *results, int root, int resizes, MPI_Comm intercomm);
 
-void print_iter_results(results_data *results, int last_normal_iter_index);
-void print_global_results(results_data *results, int resizes);
-void init_results_data(results_data **results, int resizes, int iters_size);
+void set_results_post_reconfig(results_data *results, int grp, int sdr, int adr);
+
+void print_iter_results(results_data results, int last_normal_iter_index);
+void print_global_results(results_data results, int resizes);
+void init_results_data(results_data *results, int resizes, int iters_size);
 void realloc_results_iters(results_data *results, int needed);
-void free_results_data(results_data **results);
+void free_results_data(results_data *results);

Codes/Main/Main.c

@@ -2,18 +2,20 @@
 #include <stdlib.h>
 #include <mpi.h>
 #include <fcntl.h>
+#include <unistd.h>
 #include <sys/stat.h>
-#include <pthread.h>
 #include "computing_func.h"
-#include "../IOcodes/read_ini.h"
-#include "../IOcodes/results.h"
-#include "../malleability/ProcessDist.h"
-#include "../malleability/CommDist.h"
+//#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"
 
 #define ROOT 0
 
 int work();
-void Sons_init();
+/*void Sons_init();
 
 int checkpoint(int iter, int state, MPI_Request **comm_req);
 int TC(int numS, int comm_type);
@@ -24,7 +26,7 @@ int end_redistribution(int iter);
 int thread_creation();
 int thread_check(int iter);
 void* thread_async_work(void* void_arg);
-
+*/
 void iterate(double *matrix, int n, int async_comm);
 
 void init_group_struct(char *argv[], int argc, int myId, int numP);
@@ -44,7 +46,7 @@ typedef struct {
   int argc;
 
   int numS; // Cantidad de procesos hijos
-  int commAsync;
+  //int commAsync; FIXME REMOVE
   MPI_Comm children, parents;
 
   char *compute_comm_array;
@@ -52,18 +54,20 @@ typedef struct {
   char *sync_array, *async_array;
 } group_data;
 
+/*
 typedef struct {
   int myId, numP, numS, adr;
   MPI_Comm children;
   char *sync_array;
 } thread_data;
+*/
 
 configuration *config_file;
 group_data *group;
 results_data *results;
 int run_id = 0; // Utilizado para diferenciar más fácilmente ejecuciones en el análisis
 
-pthread_t async_thread; // TODO Cambiar de sitio?
+//pthread_t async_thread; // TODO Cambiar de sitio?
 
 int main(int argc, char *argv[]) {
     int numP, myId, res;
@@ -78,17 +82,37 @@ int main(int argc, char *argv[]) {
     }
 
     init_group_struct(argv, argc, myId, numP);
+    init_malleability(myId, numP, ROOT, MPI_COMM_WORLD, argv[0]); //FIXME Cambiar el uso de MPI_COMM_WORLD?
 
-    MPI_Comm_get_parent(&(group->parents));
-    if(group->parents == MPI_COMM_NULL ) { // Si son el primer grupo de procesos, recogen la configuracion inicial
-      init_application();
-    } else { // Si son procesos hijos deben comunicarse con las padres
-      Sons_init();
+    MPI_Comm_get_parent(&(group->parents)); //FIXME No usar esto
+    if(group->parents != MPI_COMM_NULL ) {
+      group->grp++;
+      MPI_Comm_disconnect(&(group->parents)); //FIXME Volver a poner cuando se arregle MAIN.c
     }
 
     if(group->grp == 0) {
+      init_application();
+
+      set_benchmark_configuration(config_file);
+      set_benchmark_results(results);
+
       MPI_Barrier(MPI_COMM_WORLD);
       results->exec_start = MPI_Wtime();
+    } else {
+      // TODO Que habría que hacer aqui?
+      get_benchmark_configuration(&config_file); //No se obtiene bien el archivo
+      get_benchmark_results(&results); //No se obtiene bien el archivo
+
+      set_results_post_reconfig(results, group->grp, config_file->sdr, config_file->adr);
+      if(config_file->comm_tam) {
+        group->compute_comm_array = malloc(config_file->comm_tam * sizeof(char));
+      }
+    }
+
+    if(config_file->resizes != group->grp + 1) {
+      int spawn_type = COMM_SPAWN_SERIAL; // TODO Pasar a CONFIG
+      set_malleability_configuration(spawn_type, config_file->phy_dist[group->grp+1], -1, config_file->aib, -1);
+      set_children_number(config_file->procs[group->grp+1]); // TODO TO BE DEPRECATED
     }
 
     res = work();
@@ -97,8 +121,8 @@ int main(int argc, char *argv[]) {
       MPI_Barrier(MPI_COMM_WORLD);
       results->exec_time = MPI_Wtime() - results->exec_start;
     }
-    print_final_results();
 
+    print_final_results(); // Pasado este punto ya no pueden escribir los procesos
     free_application_data();
     MPI_Finalize();
     return 0;
@@ -121,30 +145,252 @@ int main(int argc, char *argv[]) {
  */
 int work() {
   int iter, maxiter, state, res;
-  double *matrix;
-  MPI_Request *async_comm;
+  double *matrix = NULL;
 
   maxiter = config_file->iters[group->grp];
   //initMatrix(&matrix, config_file->matrix_tam);
-  state = MAL_COMM_UNINITIALIZED;
+  state = MAL_NOT_STARTED;
 
   res = 0;
   for(iter=group->iter_start; iter < maxiter; iter++) {
     iterate(matrix, config_file->matrix_tam, state);
   }
-  state = checkpoint(iter, state, &async_comm);
-  
+  if(config_file->iters[group->grp] == iter && config_file->resizes != group->grp + 1)
+    state = malleability_checkpoint();
+
   iter = 0;
-  while(state == MAL_ASYNC_PENDING || state == COMM_IN_PROGRESS) {
+  while(state == MAL_DIST_PENDING || state == MAL_SPAWN_PENDING) {
     iterate(matrix, config_file->matrix_tam, state);
     iter++;
-    state = checkpoint(iter, state, &async_comm);
+    state = malleability_checkpoint();
   }
   
   if(config_file->resizes - 1 == group->grp) res=1;
   return res;
 }
 
+
+/////////////////////////////////////////
+/////////////////////////////////////////
+//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) {
+    MPI_Bcast(group->compute_comm_array, config_file->comm_tam, MPI_CHAR, ROOT, MPI_COMM_WORLD);
+  }
+
+  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);
+}
+
+/*
+ * Pide al proceso raiz imprimir los datos sobre las iteraciones realizadas por el grupo de procesos.
+ *
+ * Si es el ultimo grupo de procesos, muestra los datos obtenidos de tiempo de ejecucion, creacion de procesos
+ * y las comunicaciones.
+ */
+int print_final_results() {
+  int ptr_local, ptr_global, err;
+  char *file_name;
+
+  if(group->myId == ROOT) {
+    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);
+
+    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->commAsync   = MAL_COMM_UN; FIXME REMOVE
+  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
+  MPI_Bcast(&(config_file->Top), 1, MPI_DOUBLE, ROOT, MPI_COMM_WORLD); 
+}
+
+/*
+ * 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);
+  }
+  
+  if(group->grp == 0) { //FIXME Revisar porque cuando es diferente a 0 no funciona
+    //free_config(config_file);
+    //free_results_data(results);
+  }
+  free(group);
+
+  free_malleability();
+  free(results);
+}
+
+
+/* 
+ * 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;
+}
+
+
+
+
 /*
  * Se realiza el redimensionado de procesos por parte de los padres.
  *
@@ -160,11 +406,12 @@ int work() {
  * Si solo hay datos sincronos se envian tras la creacion de los procesos
  * y finalmente se desconectan los dos grupos de procesos.
  */
+/*
 int checkpoint(int iter, int state, MPI_Request **comm_req) {
   
   if(state == MAL_COMM_UNINITIALIZED) {
     // Comprobar si se tiene que realizar un redimensionado
-    if(config_file->iters[group->grp] > iter || config_file->resizes == group->grp + 1) {return MAL_COMM_UNINITIALIZED;}
+    //if(config_file->iters[group->grp] > iter || config_file->resizes == group->grp + 1) {return MAL_COMM_UNINITIALIZED;}
 
     group->numS = config_file->procs[group->grp +1];
     int comm_type = COMM_SPAWN_SERIAL; // TODO Pasar a CONFIG
@@ -193,18 +440,21 @@ int checkpoint(int iter, int state, MPI_Request **comm_req) {
 
   return state;
 }
-
+*/
 /*
  * Se encarga de realizar la creacion de los procesos hijos.
  * Si se pide en segundo plano devuelve el estado actual.
  */
+/*
 int TC(int numS, int comm_type){
   // Inicialización de la comunicación con SLURM
   int dist = config_file->phy_dist[group->grp +1];
   int comm_state;
+  MPI_Comm *new_comm = malloc(sizeof(MPI_Comm));
 
       results->spawn_start = MPI_Wtime();
-  comm_state = init_slurm_comm(group->argv, group->myId, numS, ROOT, dist, comm_type, MPI_COMM_WORLD, &(group->children));
+  MPI_Comm_dup(MPI_COMM_WORLD, new_comm);
+  comm_state = init_slurm_comm(group->argv, group->myId, numS, ROOT, dist, comm_type, *new_comm, &(group->children));
   if(comm_type == COMM_SPAWN_SERIAL)
       results->spawn_time[group->grp] = MPI_Wtime() - results->spawn_start;
   else if(comm_type == COMM_SPAWN_PTHREAD) {
@@ -213,7 +463,7 @@ int TC(int numS, int comm_type){
   }
   return comm_state;
 }
-
+*/
 /*
  * Comienza la redistribucion de los datos con el nuevo grupo de procesos.
  *
@@ -228,6 +478,7 @@ int TC(int numS, int comm_type){
  * Finalmente se envian datos sobre los resultados a los hijos y se desconectan ambos
  * grupos de procesos.
  */
+/*
 int start_redistribution(int iter, int numS, MPI_Request **comm_req) {
   int rootBcast = MPI_PROC_NULL;
   if(group->myId == ROOT) rootBcast = MPI_ROOT;
@@ -248,10 +499,11 @@ int start_redistribution(int iter, int numS, MPI_Request **comm_req) {
   } 
   return end_redistribution(iter);
 }
-
+*/
 /*
  * Crea una hebra para ejecutar una comunicación en segundo plano.
  */
+/*
 int thread_creation() {
   if(pthread_create(&async_thread, NULL, thread_async_work, NULL)) {
     printf("Error al crear el hilo\n");
@@ -260,13 +512,14 @@ int thread_creation() {
   }
   return MAL_ASYNC_PENDING;
 }
-
+*/
 /*
  * 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(int iter) {
   int all_completed = 0;
 
@@ -281,7 +534,7 @@ int thread_check(int iter) {
   } 
   return end_redistribution(iter);
 }
-
+*/
 /*
  * Función ejecutada por una hebra.
  * Ejecuta una comunicación síncrona con los hijos que
@@ -290,12 +543,13 @@ int thread_check(int iter) {
  * Cuando termina la comunicación la hebra maestra puede comprobarlo
  * por el valor "commAsync".
  */
+/*
 void* thread_async_work(void* void_arg) {
   send_sync(group->async_array, config_file->adr, group->myId, group->numP, ROOT, group->children, group->numS);
   group->commAsync = MAL_COMM_COMPLETED;
   pthread_exit(NULL);
 }
-
+*/
 /*
  * @deprecated
  * Comprueba si la redistribucion asincrona ha terminado. 
@@ -310,6 +564,7 @@ void* thread_async_work(void* void_arg) {
  * Si se utiliza el modo "MAL_USE_IBARRIER", se considera terminada cuando
  * los hijos han terminado de recibir.
  */
+/*
 int check_redistribution(int iter, MPI_Request **comm_req) {
   int completed, all_completed, test_err;
   MPI_Request *req_completed;
@@ -342,7 +597,7 @@ int check_redistribution(int iter, MPI_Request **comm_req) {
   free(*comm_req);
   return end_redistribution(iter);
 }
-
+*/
 
 /*
  * Termina la redistribución de los datos con los hijos, comprobando
@@ -352,30 +607,30 @@ int check_redistribution(int iter, MPI_Request **comm_req) {
  * Además se realizan las comunicaciones síncronas se las hay.
  * Finalmente termina enviando los datos temporales a los hijos.
  */ 
+/*
 int end_redistribution(int iter) {
   int rootBcast = MPI_PROC_NULL;
   if(group->myId == ROOT) rootBcast = MPI_ROOT;
 
-  if(iter > 0) { // Mandar a los hijos iteracion en la que comenzar
-    MPI_Bcast(&iter, 1, MPI_INT, rootBcast, group->children);
-  }
   if(config_file->sdr > 0) { // Realizar envio sincrono
       results->sync_start = MPI_Wtime();
     send_sync(group->sync_array, config_file->sdr, group->myId, group->numP, ROOT, group->children, group->numS);
   }
 
+  MPI_Bcast(&iter, 1, MPI_INT, rootBcast, group->children);
   send_results(results, rootBcast, config_file->resizes, group->children);
   // Desconectar intercomunicador con los hijos
   MPI_Comm_disconnect(&(group->children));
   return MAL_COMM_COMPLETED;
 }
-
+*/
 /*
  * 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 Sons_init() {
 
   // Enviar a los hijos que grupo de procesos son
@@ -385,11 +640,9 @@ void Sons_init() {
 
   config_file = recv_config_file(ROOT, group->parents);
   int numP_parents = config_file->procs[group->grp -1];
-  init_results_data(&results, config_file->resizes - 1, config_file->iters[group->grp]);
+  results = malloc(sizeof(results_data));
+  init_results_data(results, config_file->resizes - 1, config_file->iters[group->grp]);
 
-  if(config_file->comm_tam) {
-    group->compute_comm_array = malloc(config_file->comm_tam * sizeof(char));
-  }
   if(config_file->adr) { // Recibir datos asincronos
     if(config_file->aib == MAL_USE_NORMAL || config_file->aib == MAL_USE_IBARRIER || config_file->aib == MAL_USE_POINT) {
       recv_async(&(group->async_array), config_file->adr, group->myId, group->numP, ROOT, group->parents, numP_parents, config_file->aib);
@@ -397,243 +650,18 @@ void Sons_init() {
       recv_sync(&(group->async_array), config_file->adr, group->myId, group->numP, ROOT, group->parents, numP_parents);
     }
 
-      results->async_time[group->grp] = MPI_Wtime();
-    MPI_Bcast(&(group->iter_start), 1, MPI_INT, ROOT, group->parents);
+      results->async_end = MPI_Wtime();
   }
-    //MPI_Bcast(&(group->iter_start), 1, MPI_INT, ROOT, group->parents); //FIXME Quitar -- Que tenga en cuenta Pthread y async
   if(config_file->sdr) { // Recibir datos sincronos
     recv_sync(&(group->sync_array), config_file->sdr, group->myId, group->numP, ROOT, group->parents, numP_parents);
-    results->sync_time[group->grp] = MPI_Wtime();
+    results->sync_end = MPI_Wtime();
   }
+  MPI_Bcast(&(group->iter_start), 1, MPI_INT, ROOT, group->parents);
 
   // Guardar los resultados de esta transmision
   recv_results(results, ROOT, config_file->resizes, group->parents);
-  if(config_file->sdr) { // Si no hay datos sincronos, el tiempo es 0
-    results->sync_time[group->grp]  = MPI_Wtime() - results->sync_start;
-  } else {
-    results->sync_time[group->grp]  = 0;
-  }
-  if(config_file->adr) { // Si no hay datos asincronos, el tiempo es 0
-    results->async_time[group->grp]  = MPI_Wtime() - results->async_start;
-  } else {
-    results->async_time[group->grp]  = 0;
-  }
 
   // Desconectar intercomunicador con los hijos
   MPI_Comm_disconnect(&(group->parents));
 }
-
-
-/////////////////////////////////////////
-/////////////////////////////////////////
-//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;
-  for(i=0; i < operations; i++) {
-    aux += computePiSerial(n);
-  }
-
-  if(config_file->comm_tam) {
-    MPI_Bcast(group->compute_comm_array, config_file->comm_tam, MPI_CHAR, ROOT, MPI_COMM_WORLD);
-  }
-
-  actual_time = MPI_Wtime(); // Guardar tiempos
-  // TODO Que diferencie entre ambas en el IO
-  if(async_comm == MAL_ASYNC_PENDING || async_comm == COMM_IN_PROGRESS) { // 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);
-}
-
-/*
- * Pide al proceso raiz imprimir los datos sobre las iteraciones realizadas por el grupo de procesos.
- *
- * Si es el ultimo grupo de procesos, muestra los datos obtenidos de tiempo de ejecucion, creacion de procesos
- * y las comunicaciones.
- */
-int print_final_results() {
-  int ptr_local, ptr_global, err;
-  char *file_name;
-
-  if(group->myId == ROOT) {
-    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);
-
-    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->commAsync   = MAL_COMM_UNINITIALIZED;
-  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");
-    exit(0);
-  }
-  if(group->argc > 2) {
-    run_id = atoi(group->argv[2]);
-  }
-
-  config_file = read_ini_file(group->argv[1]);
-  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
-  MPI_Bcast(&(config_file->Top), 1, MPI_DOUBLE, ROOT, MPI_COMM_WORLD); 
-}
-
-/*
- * 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);
-  }
-  
-  free(group);
-  free_config(config_file);
-  free_results_data(&results);
-}
-
-
-/* 
- * 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;
-}
+*/

Codes/compila.sh

 module load mpich-3.4.1-noucx
 
-mpicc -Wall Main/Main.c Main/computing_func.c IOcodes/results.c IOcodes/read_ini.c IOcodes/ini.c malleability/ProcessDist.c malleability/CommDist.c -pthread -lslurm -lm
+#mpicc -Wall Main/Main.c Main/computing_func.c IOcodes/results.c IOcodes/read_ini.c IOcodes/ini.c malleability/ProcessDist.c malleability/CommDist.c -pthread -lslurm -lm
+mpicc -Wall Main/Main.c Main/computing_func.c IOcodes/results.c IOcodes/read_ini.c IOcodes/ini.c malleability/malleabilityManager.c malleability/malleabilityTypes.c malleability/ProcessDist.c malleability/CommDist.c -pthread -lslurm -lm
 
 if [ $# -gt 0 ]
 then

Codes/malleability/CommDist.c

@@ -181,7 +181,7 @@ void recv_sync_arrays(struct Dist_data dist_data, char *array, int root, int num
 
 //================================================================================
 //================================================================================
-//========================ASINCHRONOUS FUNCTIONS==================================
+//========================ASYNCHRONOUS FUNCTIONS==================================
 //================================================================================
 //================================================================================
 
@@ -212,18 +212,18 @@ int send_async(char *array, int qty, int myId, int numP, int root, MPI_Comm inte
 
     // MAL_USE_THREAD sigue el camino sincrono
     if(parents_wait == MAL_USE_NORMAL) {
-      *comm_req = (MPI_Request *) malloc(sizeof(MPI_Request));
+      //*comm_req = (MPI_Request *) malloc(sizeof(MPI_Request));
       *comm_req[0] = MPI_REQUEST_NULL;
       send_async_arrays(dist_data, array, rootBcast, numP_child, idS[0], idS[1], counts, &(*comm_req[0])); 
 
     } else if (parents_wait == MAL_USE_IBARRIER){
-      *comm_req = (MPI_Request *) malloc(2 * sizeof(MPI_Request));
-      (*comm_req)[0] = MPI_REQUEST_NULL;
-      (*comm_req)[1] = MPI_REQUEST_NULL;
+      //*comm_req = (MPI_Request *) malloc(2 * sizeof(MPI_Request));
+      *comm_req[0] = MPI_REQUEST_NULL;
+      *comm_req[1] = MPI_REQUEST_NULL;
       send_async_arrays(dist_data, array, rootBcast, numP_child, idS[0], idS[1], counts, &((*comm_req)[1])); 
       MPI_Ibarrier(intercomm, &((*comm_req)[0]) );
     } else if (parents_wait == MAL_USE_POINT){
-      *comm_req = (MPI_Request *) malloc(numP_child * sizeof(MPI_Request));
+      //*comm_req = (MPI_Request *) malloc(numP_child * sizeof(MPI_Request));
       for(i=0; i<numP_child; i++){
         (*comm_req)[i] = MPI_REQUEST_NULL;
       }

Codes/malleability/CommDist.h

@@ -2,15 +2,16 @@
 #include <stdlib.h>
 #include <mpi.h>
 #include <string.h>
+#include "malleabilityStates.h"
 
-#define MAL_COMM_COMPLETED 0
-#define MAL_COMM_UNINITIALIZED 2
-#define MAL_ASYNC_PENDING 1
+//#define MAL_COMM_COMPLETED 0
+//#define MAL_COMM_UNINITIALIZED 2
+//#define MAL_ASYNC_PENDING 1
 
-#define MAL_USE_NORMAL 0
-#define MAL_USE_IBARRIER 1
-#define MAL_USE_POINT 2
-#define MAL_USE_THREAD 3
+//#define MAL_USE_NORMAL 0
+//#define MAL_USE_IBARRIER 1
+//#define MAL_USE_POINT 2
+//#define MAL_USE_THREAD 3
 
 int send_sync(char *array, int qty, int myId, int numP, int root, MPI_Comm intercomm, int numP_child);
 void recv_sync(char **array, int qty, int myId, int numP, int root, MPI_Comm intercomm, int numP_parents);

Codes/malleability/ProcessDist.c

@@ -8,9 +8,7 @@
 #include <slurm/slurm.h>
 #include "ProcessDist.h"
 
-#define ROOT 0
-
-int commSlurm = COMM_UNRESERVED;
+int commSlurm = MAL_NOT_STARTED;
 struct Slurm_data *slurm_data;  
 pthread_t slurm_thread;
 MPI_Comm *returned_comm;
@@ -23,7 +21,7 @@ struct Slurm_data {
 };
 
 typedef struct {
-  char **argv;
+  char *argv;
   int numP_childs, myId, root, type_dist;
   MPI_Comm comm;
 }Creation_data;
@@ -33,21 +31,17 @@ typedef struct {
 void* thread_work(void* creation_data_arg);
 
 //--------------PRIVATE DECLARATIONS---------------//
-
-void processes_dist(char *argv[], int numP_childs, int type_dist);
+void processes_dist(char *argv, int numP_childs, int type_dist);
 int create_processes(int myId, int root, MPI_Comm *child, MPI_Comm comm);
 void node_dist(slurm_job_info_t job_record, int type, int total_procs, int **qty, int *used_nodes);
 
-int create_hostfile(char *jobId, char **file_name);
-int write_hostfile_node(int ptr, int qty, char *node_name);
-void fill_hostfile(slurm_job_info_t job_record, int ptr, int *qty, int used_nodes);
-
-//TESTS
 void fill_str_hostfile(slurm_job_info_t job_record, int *qty, int used_nodes, char **hostfile_str);
 int write_str_node(char **hostfile_str, int len_og, int qty, char *node_name);
-//
 
-void print_Info(MPI_Info info);
+//@deprecated functions
+int create_hostfile(char *jobId, char **file_name);
+int write_hostfile_node(int ptr, int qty, char *node_name);
+void fill_hostfile(slurm_job_info_t job_record, int ptr, int *qty, int used_nodes);
 
 //--------------PUBLIC FUNCTIONS---------------//
 
@@ -67,7 +61,7 @@ void print_Info(MPI_Info info);
  * Devuelve el estado de el procedimiento. Si no devuelve "COMM_FINISHED", es necesario llamar a
  * "check_slurm_comm()".
  */
-int init_slurm_comm(char **argv, int myId, int numP, int root, int type_dist, int type_creation, MPI_Comm comm, MPI_Comm *child) {
+int init_slurm_comm(char *argv, int myId, int numP, int root, int type_dist, int type_creation, MPI_Comm comm, MPI_Comm *child) {
 
   slurm_data = malloc(sizeof(struct Slurm_data));
 
@@ -88,10 +82,10 @@ int init_slurm_comm(char **argv, int myId, int numP, int root, int type_dist, in
     free(slurm_data->cmd);
     free(slurm_data);
 
-    commSlurm = COMM_FINISHED;
+    commSlurm = MAL_SPAWN_COMPLETED;
 
   } else if(type_creation == COMM_SPAWN_PTHREAD) {
-    commSlurm = COMM_IN_PROGRESS;
+    commSlurm = MAL_SPAWN_PENDING;
 
     Creation_data *creation_data = (Creation_data *) malloc(sizeof(Creation_data));
     creation_data->argv = argv;
@@ -120,30 +114,9 @@ int check_slurm_comm(int myId, int root, int numP, MPI_Comm *child) { // TODO Bo
 
   if(slurm_data->type_creation == COMM_SPAWN_PTHREAD) {
 
-    //MPI_Allreduce(&commSlurm, &state, 1, MPI_INT, MPI_MIN, MPI_COMM_WORLD);
-    
-    if(myId == root) {
-      int i, recv_state;
-      state = commSlurm;
-      for(i=0; i<numP; i++) { //Recv states
-	if(i != myId) {
-          MPI_Recv(&recv_state, 1, MPI_INT, i, 120, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
-          if(recv_state == COMM_IN_PROGRESS) {
-	    state = recv_state;
-	  }
-	}
-      }
-      for(i=0; i<numP; i++) { //Send state
-	if(i != myId) {
-          MPI_Send(&state, 1, MPI_INT, i, 120, MPI_COMM_WORLD);
-	}
-      }
-    } else {
-      MPI_Send(&commSlurm, 1, MPI_INT, root, 120, MPI_COMM_WORLD);
-      MPI_Recv(&state, 1, MPI_INT, root, 120, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
-    }
+    MPI_Allreduce(&commSlurm, &state, 1, MPI_INT, MPI_MIN, MPI_COMM_WORLD);
 
-    if(state != COMM_FINISHED) return state; // Continue only if asynchronous process creation has ended 
+    if(state != MAL_SPAWN_COMPLETED) return state; // Continue only if asynchronous process creation has ended 
 
   } else { 
     return commSlurm;
@@ -187,7 +160,7 @@ void* thread_work(void* creation_data_arg) {
   }
 
   create_processes(creation_data->myId, creation_data->root, returned_comm, creation_data->comm);
-  commSlurm = COMM_FINISHED;
+  commSlurm = MAL_SPAWN_COMPLETED;
 
   free(creation_data);
   pthread_exit(NULL);
@@ -200,8 +173,8 @@ void* thread_work(void* creation_data_arg) {
  * para una llamada a MPI_Comm_spawn, obteniendo una distribucion fisica
  * para los procesos y creando un fichero hostfile.
  */
-void processes_dist(char *argv[], int numP_childs, int type) {
-    int jobId, ptr;
+void processes_dist(char *argv, int numP_childs, int type) {
+    int jobId;
     char *tmp;
     job_info_msg_t *j_info;
     slurm_job_info_t last_record;
@@ -217,8 +190,8 @@ void processes_dist(char *argv[], int numP_childs, int type) {
     last_record = j_info->job_array[j_info->record_count - 1];
 
     //COPY PROGRAM NAME
-    slurm_data->cmd = malloc(strlen(argv[0]) * sizeof(char));
-    strcpy(slurm_data->cmd, argv[0]);
+    slurm_data->cmd = malloc(strlen(argv) * sizeof(char));
+    strcpy(slurm_data->cmd, argv);
 
     // GET NEW DISTRIBUTION 
     node_dist(last_record, type, numP_childs, &procs_array, &used_nodes);
@@ -226,6 +199,7 @@ void processes_dist(char *argv[], int numP_childs, int type) {
 
     /*
     // CREATE/UPDATE HOSTFILE 
+    int ptr;
     ptr = create_hostfile(tmp, &hostfile);
     MPI_Info_create(&(slurm_data->info));
     MPI_Info_set(slurm_data->info, "hostfile", hostfile);
@@ -236,7 +210,6 @@ void processes_dist(char *argv[], int numP_childs, int type) {
     close(ptr);
     */
     
-
     // TEST 
     fill_str_hostfile(last_record, procs_array, used_nodes, &hostfile);
     MPI_Info_create(&(slurm_data->info));
@@ -319,6 +292,70 @@ void node_dist(slurm_job_info_t job_record, int type, int total_procs, int **qty
 }
 
 /*
+ * Crea y devuelve una cadena para ser utilizada por la llave "hosts"
+ * al crear procesos e indicar donde tienen que ser creados.
+ */
+void fill_str_hostfile(slurm_job_info_t job_record, int *qty, int used_nodes, char **hostfile_str) {
+  int i=0, len=0;
+  char *host;
+  hostlist_t hostlist;
+  
+  hostlist = slurm_hostlist_create(job_record.nodes);
+  while ( (host = slurm_hostlist_shift(hostlist)) && i < used_nodes) {
+    len = write_str_node(hostfile_str, len, qty[i], host);
+    i++;
+    free(host);
+  }
+  slurm_hostlist_destroy(hostlist);
+
+}
+
+/*
+ * Añade en una cadena "qty" entradas de "node_name".
+ * Realiza la reserva de memoria y la realoja si es necesario.
+ */
+int write_str_node(char **hostfile_str, int len_og, int qty, char *node_name) {
+  int err, len_node, len, i;
+  char *ocurrence;
+
+  len_node = strlen(node_name);
+  len = qty * (len_node + 1);
+
+  if(len_og == 0) { // Memoria no reservada
+    *hostfile_str = (char *) malloc(len * sizeof(char) - (1 * sizeof(char)));
+  } else { // Cadena ya tiene datos
+    *hostfile_str = (char *) realloc(*hostfile_str, (len_og + len) * sizeof(char) - (1 * sizeof(char)));
+  }
+  if(hostfile_str == NULL) return -1; // No ha sido posible alojar la memoria
+
+  ocurrence = (char *) malloc((len_node+1) * sizeof(char));
+  if(ocurrence == NULL) return -1; // No ha sido posible alojar la memoria
+  err = sprintf(ocurrence, ",%s", node_name);
+  if(err < 0) return -2; // No ha sido posible escribir sobre la variable auxiliar
+
+  i=0;
+  if(len_og == 0) { // Si se inicializa, la primera es una copia
+    i++;
+    strcpy(*hostfile_str, node_name);
+  }
+  for(; i<qty; i++){ // Las siguientes se conctanenan
+    strcat(*hostfile_str, ocurrence);
+  }
+
+  
+  free(ocurrence);
+  return len+len_og;
+}
+
+//====================================================
+//====================================================
+//============DEPRECATED FUNCTIONS====================
+//====================================================
+//====================================================
+
+
+/*
+ * @deprecated
  * Crea un fichero que se utilizara como hostfile
  * para un nuevo grupo de procesos. 
  *
@@ -346,6 +383,7 @@ int create_hostfile(char *jobId, char **file_name) {
 }
 
 /*
+ * @deprecated
  * Rellena un fichero hostfile indicado por ptr con los nombres
  * de los nodos a utilizar indicados por "job_record" y la cantidad 
  * de procesos que alojara cada nodo indicado por "qty".
@@ -365,6 +403,7 @@ void fill_hostfile(slurm_job_info_t job_record, int ptr, int *qty, int used_node
 }
 
 /*
+ * @deprecated
  * Escribe en el fichero hostfile indicado por ptr una nueva linea.
  *
  * Esta linea indica el nombre de un nodo y la cantidad de procesos a
@@ -389,53 +428,3 @@ int write_hostfile_node(int ptr, int qty, char *node_name) {
 
   return 0;
 }
-
-
-
-void fill_str_hostfile(slurm_job_info_t job_record, int *qty, int used_nodes, char **hostfile_str) {
-  int i=0, len=0;
-  char *host;
-  hostlist_t hostlist;
-  
-  hostlist = slurm_hostlist_create(job_record.nodes);
-  while ( (host = slurm_hostlist_shift(hostlist)) && i < used_nodes) {
-    len = write_str_node(hostfile_str, len, qty[i], host);
-    i++;
-    free(host);
-  }
-  slurm_hostlist_destroy(hostlist);
-
-}
-
-int write_str_node(char **hostfile_str, int len_og, int qty, char *node_name) {
-  int err, len_node, len, i;
-  char *ocurrence;
-
-  len_node = strlen(node_name);
-  len = qty * (len_node + 1);
-
-  if(len_og == 0) { // Memoria no reservada
-    *hostfile_str = (char *) malloc(len * sizeof(char) - (1 * sizeof(char)));
-  } else { // Cadena ya tiene datos
-    *hostfile_str = (char *) realloc(*hostfile_str, (len_og + len) * sizeof(char) - (1 * sizeof(char)));
-  }
-  if(hostfile_str == NULL) return -1; // No ha sido posible alojar la memoria
-
-  ocurrence = (char *) malloc((len_node+1) * sizeof(char));
-  if(ocurrence == NULL) return -1; // No ha sido posible alojar la memoria
-  err = sprintf(ocurrence, ",%s", node_name);
-  if(err < 0) return -2; // No ha sido posible escribir sobre la variable auxiliar
-
-  i=0;
-  if(len_og == 0) { // Si se inicializa, la primera es una copia
-    i++;
-    strcpy(*hostfile_str, node_name);
-  }
-  for(; i<qty; i++){ // Las siguientes se realizan con concatenan
-    strcat(*hostfile_str, ocurrence);
-  }
-
-  
-  free(ocurrence);
-  return len+len_og;
-}

Codes/malleability/ProcessDist.h

@@ -3,16 +3,17 @@
 #include <mpi.h>
 #include <string.h>
 #include <slurm/slurm.h>
+#include "malleabilityStates.h"
 
-#define COMM_UNRESERVED -2
-#define COMM_IN_PROGRESS -1
-#define COMM_FINISHED 0
+//#define COMM_UNRESERVED -2
+//#define COMM_IN_PROGRESS -1
+//#define COMM_FINISHED 0
 
-#define COMM_PHY_NODES 1
-#define COMM_PHY_CPU 2
+//#define COMM_PHY_NODES 1
+//#define COMM_PHY_CPU 2
 
-#define COMM_SPAWN_SERIAL 0
-#define COMM_SPAWN_PTHREAD 1
+//#define COMM_SPAWN_SERIAL 0
+//#define COMM_SPAWN_PTHREAD 1
 
-int init_slurm_comm(char **argv, int myId, int numP, int root, int type_dist, int type_creation, MPI_Comm comm, MPI_Comm *child);
+int init_slurm_comm(char *argv, int myId, int numP, int root, int type_dist, int type_creation, MPI_Comm comm, MPI_Comm *child);
 int check_slurm_comm(int myId, int root, int numP, MPI_Comm *child);

Codes/malleability/malleabilityManager.c

+#include <pthread.h>
+#include "malleabilityManager.h"
+#include "malleabilityStates.h"
+#include "malleabilityTypes.h"
+#include "ProcessDist.h"
+#include "CommDist.h"
+
+#define MALLEABILITY_ROOT 0
+#define MALLEABILITY_CHILDREN 1
+#define MALLEABILITY_NOT_CHILDREN 0
+#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();
+
+int thread_creation();
+int thread_check();
+void* thread_async_work(void* void_arg);
+
+typedef struct {
+  int spawn_type;
+  int spawn_dist;
+  int spawn_threaded;
+  int comm_type;
+  int comm_threaded;
+
+  configuration *config_file;
+  results_data *results;
+  int grp;
+} malleability_config_t;
+
+typedef struct {
+  int myId, numP, numC, root, root_parents;
+  pthread_t async_thread;
+  MPI_Comm comm, thread_comm;
+  MPI_Comm intercomm;
+  
+  char *name_exec;
+} malleability_t;
+
+int state = MAL_UNRESERVED; //FIXME Mover a otro lado
+
+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;
+
+
+void init_malleability(int myId, int numP, int root, MPI_Comm comm, char *name_exec) {
+  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));
+
+  MPI_Comm_dup(comm, &dup_comm);
+  MPI_Comm_dup(comm, &thread_comm);
+
+  mall->myId = myId;
+  mall->numP = numP;
+  mall->root = root;
+  mall->comm = dup_comm;
+  mall->comm = thread_comm; // TODO Refactor -- Crear solo si es necesario?
+  mall->name_exec = name_exec;
+
+  rep_s_data->entries = 0;
+  rep_a_data->entries = 0;
+  dist_s_data->entries = 0;
+  dist_a_data->entries = 0;
+
+  state = MAL_NOT_STARTED;
+
+  // 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();
+  }
+}
+
+void free_malleability() {	
+  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);
+
+  //MPI_Comm_free(&(mall->comm)); // TODO Revisar si hace falta?
+  //MPI_Comm_free(&(mall->thread_comm));
+  free(mall);
+  free(mall_conf);
+  state = MAL_UNRESERVED;
+}
+
+/*
+ * 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() {
+  
+  if(state == MAL_UNRESERVED) return MAL_UNRESERVED;
+
+  if(state == MAL_NOT_STARTED) {
+    // Comprobar si se tiene que realizar un redimensionado
+    //if(CHECK_RMS()) {return MAL_DENIED;}
+
+    state = spawn_step();
+
+    if (state == MAL_SPAWN_COMPLETED){
+      state = start_redistribution();
+    }
+
+  } else if(state == MAL_SPAWN_PENDING) { // Comprueba si el spawn ha terminado y comienza la redistribucion
+    state = check_slurm_comm(mall->myId, mall->root, mall->numP, &(mall->intercomm));
+
+    if (state == MAL_SPAWN_COMPLETED) {  
+        mall_conf->results->spawn_time[mall_conf->grp] = MPI_Wtime() - mall_conf->results->spawn_start;
+      state = start_redistribution();
+    }
+
+  } else if(state == MAL_DIST_PENDING) {
+    if(mall_conf->comm_type == MAL_USE_THREAD) {
+      state = thread_check();
+    } else {
+      state = check_redistribution();
+    }
+  }
+
+  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;
+}
+
+void get_benchmark_configuration(configuration **config_file) { //FIXME Revisar posible error de memoria
+  *config_file = mall_conf->config_file;
+}
+
+void set_benchmark_results(results_data *results) {
+  mall_conf->results = results;
+}
+
+void get_benchmark_results(results_data **results) { //FIXME Revisar posible error de memoria
+  *results = mall_conf->results;
+}
+//-------------------------------------------------------------------------------------------------------------
+
+void set_malleability_configuration(int spawn_type, int spawn_dist, int spawn_threaded, int comm_type, int comm_threaded) {
+  mall_conf->spawn_type = spawn_type;
+  mall_conf->spawn_dist = spawn_dist;
+  mall_conf->spawn_threaded = spawn_threaded;
+  mall_conf->comm_type = comm_type;
+  mall_conf->comm_threaded = comm_threaded;
+}
+
+/*
+ * To be deprecated
+ */
+void set_children_number(int numC){
+  mall->numC = numC;
+}
+
+/*
+ * 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".
+ */
+void malleability_add_data(void *data, int total_qty, int type, int is_replicated, int is_constant) {
+
+  if(is_constant) {
+    if(is_replicated) {
+      add_data(data, total_qty, type, 0, rep_s_data); //FIXME Numero magico
+    } else {
+      add_data(data, total_qty, type, 0, dist_s_data); //FIXME Numero magico
+    }
+  } else {
+    if(is_replicated) {
+      add_data(data, total_qty, type, 0, rep_a_data); //FIXME Numero magico || Un request?
+    } else {
+      int total_reqs = 0;
+      
+      if(mall_conf->comm_type  == MAL_USE_NORMAL) {
+        total_reqs = 1;
+      } else if(mall_conf->comm_type  == MAL_USE_IBARRIER) {
+        total_reqs = 2;
+      } else if(mall_conf->comm_type  == MAL_USE_POINT) {
+        total_reqs = mall->numC;
+      }
+      
+      add_data(data, total_qty, type, total_reqs, dist_a_data);
+    }
+  }
+}
+
+/*
+ * Devuelve el numero de entradas para la estructura de descripcion de 
+ * datos elegida.
+ */
+void malleability_get_entries(int *entries, int is_replicated, int is_constant){
+  
+  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
+ */
+void malleability_get_data(void *data, int index, int is_replicated, int is_constant) {
+  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;
+    }
+  }
+
+  data = (void *) data_struct->arrays[index];
+}
+
+
+//======================================================||
+//================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) {
+  int i;
+  char *aux;
+
+  if(is_asynchronous) {
+    for(i=0; i < data_struct->entries; i++) {
+      aux = (char *) data_struct->arrays[i]; //TODO Comprobar que realmente es un char
+      send_async(aux, data_struct->qty[i], mall->myId, mall->numP, mall->root, mall->intercomm, numP_children, data_struct->requests, mall_conf->comm_type);
+    }
+  } else {
+    for(i=0; i < data_struct->entries; i++) {
+      aux = (char *) data_struct->arrays[i]; //TODO Comprobar que realmente es un char
+      send_sync(aux, data_struct->qty[i], mall->myId, mall->numP, mall->root, mall->intercomm, numP_children);
+    }
+  }
+}
+
+/*
+ * 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) {
+  int i;
+  char *aux;
+
+  if(is_asynchronous) {
+    for(i=0; i < data_struct->entries; i++) {
+      aux = (char *) data_struct->arrays[i]; //TODO Comprobar que realmente es un char
+      recv_async(&aux, data_struct->qty[i], mall->myId, mall->numP, mall->root, mall->intercomm, numP_parents, mall_conf->comm_type);
+      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
+      recv_sync(&aux, data_struct->qty[i], mall->myId, mall->numP, mall->root, mall->intercomm, numP_parents);
+      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() {
+
+  /* FIXME
+   * grp -- a constante replicado || TODO Acordarse de sumar 1
+   * run_id -- a constante replicado
+   * iter_start -- a constante replicado || TODO Setear valor segun adr==0
+   */
+  int numP_parents, root_parents, i;
+  //int *aux;
+
+  MPI_Bcast(&root_parents, 1, MPI_INT, MALLEABILITY_ROOT, mall->intercomm); 
+  MPI_Bcast(&numP_parents, 1, MPI_INT, root_parents, mall->intercomm);
+
+  mall_conf->config_file = recv_config_file(mall->root, mall->intercomm);
+  mall_conf->results = (results_data *) malloc(sizeof(results_data));
+  init_results_data(mall_conf->results, mall_conf->config_file->resizes - 1, RESULTS_INIT_DATA_QTY);
+
+  if(dist_a_data->entries || rep_a_data->entries) { // Recibir datos asincronos
+    comm_data_info(rep_a_data, dist_a_data, MALLEABILITY_CHILDREN, mall->myId, root_parents, mall->intercomm);
+
+    if(mall_conf->comm_type == MAL_USE_NORMAL || mall_conf->comm_type == MAL_USE_IBARRIER || mall_conf->comm_type == MAL_USE_POINT) {
+      recv_data(numP_parents, dist_a_data, 1);
+
+    } else if (mall_conf->comm_type == MAL_USE_THREAD) { //TODO Modificar uso para que tenga sentido comm_threaded
+      recv_data(numP_parents, dist_a_data, 0);
+    }
+    mall_conf->results->async_end= MPI_Wtime(); // Obtener timestamp de cuando termina comm asincrona
+  }
+  
+  if(dist_s_data->entries || rep_s_data->entries) { // Recibir datos sincronos
+    comm_data_info(rep_s_data, dist_s_data, MALLEABILITY_CHILDREN, mall->myId, root_parents, mall->intercomm);
+    recv_data(numP_parents, dist_s_data, 0);
+
+    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++) {
+      //aux = (int *) rep_s_data->arrays[i]; //TODO Comprobar que realmente es un int
+      MPI_Bcast(rep_s_data->arrays[i], 1, MPI_INT, root_parents, mall->intercomm);
+      //rep_s_data->arrays[i] = (void *) aux;
+    } 
+  }
+
+  // Guardar los resultados de esta transmision
+  recv_results(mall_conf->results, mall->root, mall_conf->config_file->resizes, mall->intercomm);
+
+  //MPI_Comm_disconnect(&(mall->intercomm)); //FIXME Volver a poner cuando se arregle MAIN.c
+}
+
+//======================================================||
+//================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();
+  state = init_slurm_comm(mall->name_exec, mall->myId, mall->numC, mall->root, mall_conf->spawn_dist, mall_conf->spawn_type, mall->comm, &(mall->intercomm));
+
+  if(mall_conf->spawn_type == COMM_SPAWN_SERIAL)
+      mall_conf->results->spawn_time[mall_conf->grp] = MPI_Wtime() - mall_conf->results->spawn_start;
+  else if(mall_conf->spawn_type == COMM_SPAWN_PTHREAD) {
+      mall_conf->results->spawn_thread_time[mall_conf->grp] = MPI_Wtime() - mall_conf->results->spawn_start;
+      mall_conf->results->spawn_start = MPI_Wtime();
+  }
+  return state;
+}
+
+/*
+ * 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() {
+  int rootBcast = MPI_PROC_NULL;
+  if(mall->myId == mall->root) rootBcast = MPI_ROOT;
+
+  MPI_Bcast(&(mall->root), 1, MPI_INT, rootBcast, mall->intercomm);
+  MPI_Bcast(&(mall->numP), 1, MPI_INT, rootBcast, mall->intercomm);
+  send_config_file(mall_conf->config_file, rootBcast, mall->intercomm);
+
+  if(dist_a_data->entries || rep_a_data->entries) { // Recibir datos asincronos
+    mall_conf->results->async_start = MPI_Wtime();
+    comm_data_info(rep_a_data, dist_a_data, MALLEABILITY_NOT_CHILDREN, mall->myId, mall->root, mall->intercomm);
+    if(mall_conf->comm_type == MAL_USE_THREAD) {
+      return thread_creation();
+    } else {
+//      send_async(group->async_array, config_file->adr, group->myId, group->numP, ROOT, group->children, group->numS, comm_req, config_file->aib);
+      send_data(mall->numC, dist_a_data, MALLEABILITY_USE_ASYNCHRONOUS);
+      return MAL_DIST_PENDING;
+    }
+  } 
+  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() {
+  int completed, all_completed, test_err;
+  MPI_Request *req_completed;
+//dist_a_data->requests[0][X] //FIXME Numero magico 0 -- Modificar para que sea un for?
+
+  if (mall_conf->comm_type == MAL_USE_POINT) {
+    test_err = MPI_Testall(mall->numC, dist_a_data->requests[0], &completed, MPI_STATUSES_IGNORE);
+  } else {
+    if(mall_conf->comm_type == MAL_USE_NORMAL) {
+      req_completed = &(dist_a_data->requests[0][0]);
+    } else if (mall_conf->comm_type == MAL_USE_IBARRIER) {
+      req_completed = &(dist_a_data->requests[0][1]);
+    }
+
+    test_err = MPI_Test(req_completed, &completed, MPI_STATUS_IGNORE);
+  }
+ 
+  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);
+  if(!all_completed) return MAL_DIST_PENDING; // Continue only if asynchronous send has ended 
+  
+
+  if(mall_conf->comm_type == MAL_USE_IBARRIER) {
+    MPI_Wait(&(dist_a_data->requests[0][0]), MPI_STATUS_IGNORE); // Indicar como completado el envio asincrono
+    //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
+  }
+  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() {
+  int i, rootBcast = MPI_PROC_NULL;
+  if(mall->myId == mall->root) rootBcast = MPI_ROOT;
+
+  if(dist_s_data->entries || rep_s_data->entries) { // Recibir datos sincronos
+    mall_conf->results->sync_start = MPI_Wtime();
+    comm_data_info(rep_s_data, dist_s_data, MALLEABILITY_NOT_CHILDREN, mall->myId, mall->root, mall->intercomm);
+    send_data(mall->numC, dist_s_data, MALLEABILITY_USE_SYNCHRONOUS);
+
+    // 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++) {
+      //aux = (int *) rep_s_data->arrays[i]; //TODO Comprobar que realmente es un int
+      MPI_Bcast(rep_s_data->arrays[i], 1, MPI_INT, rootBcast, mall->intercomm);
+    } 
+  }
+
+  send_results(mall_conf->results, rootBcast, mall_conf->config_file->resizes, mall->intercomm);
+
+  MPI_Comm_disconnect(&(mall->intercomm));
+  state = MAL_NOT_STARTED;
+  return MAL_DIST_COMPLETED;
+}
+
+// TODO MOVER A OTRO LADO??
+//======================================================||
+//================PRIVATE FUNCTIONS=====================||
+//===============COMM PARENTS THREADS===================||
+//======================================================||
+//======================================================||
+
+/*
+ * Crea una hebra para ejecutar una comunicación en segundo plano.
+ */
+int thread_creation() {
+  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;
+  }
+  return MAL_DIST_PENDING;
+}
+
+/*
+ * 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() {
+  int all_completed = 0;
+
+  // Comprueba que todos los hilos han terminado la distribucion (Mismo valor en commAsync)
+  MPI_Allreduce(&state, &all_completed, 1, MPI_INT, MPI_MAX, mall->comm);
+  if(all_completed != MAL_DIST_COMPLETED) return MAL_DIST_PENDING; // Continue only if asynchronous send has ended 
+
+  if(pthread_join(mall->async_thread, NULL)) {
+    printf("Error al esperar al hilo\n");
+    MPI_Abort(MPI_COMM_WORLD, -1);
+    return -2;
+  } 
+  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".
+ */
+void* thread_async_work(void* void_arg) {
+  send_data(mall->numC, dist_a_data, MALLEABILITY_USE_SYNCHRONOUS);
+  state = MAL_DIST_COMPLETED;
+  pthread_exit(NULL);
+}

Codes/malleability/malleabilityManager.h

+#include <stdio.h>
+#include <stdlib.h>
+#include <mpi.h>
+#include <fcntl.h>
+#include <sys/stat.h>
+#include <mpi.h>
+#include "../IOcodes/read_ini.h"
+#include "../IOcodes/results.h"
+#include "malleabilityStates.h"
+
+void init_malleability(int myId, int numP, int root, MPI_Comm comm, char *name_exec);
+void free_malleability();
+int malleability_checkpoint();
+void set_benchmark_grp(int grp);
+
+void set_malleability_configuration(int spawn_type, int spawn_dist, int spawn_threaded, int comm_type, int comm_threaded);
+void set_children_number(int numC); // TODO TO BE DEPRECATED
+
+void malleability_add_data(void *data, int total_qty, int type, int is_replicated, int is_constant);
+void malleability_get_entries(int *entries, int is_replicated, int is_constant);
+void malleability_get_data(void *data, int index, int is_replicated, int is_constant);
+
+void set_benchmark_configuration(configuration *config_file);
+void get_benchmark_configuration(configuration **config_file);
+void set_benchmark_results(results_data *results);
+void get_benchmark_results(results_data **results);

Codes/malleability/malleabilityStates.h

+#define MAL_UNRESERVED -1
+#define MAL_DENIED -2
+#define MAL_NOT_STARTED 0
+#define MAL_SPAWN_PENDING 1
+#define MAL_SPAWN_COMPLETED 2
+#define MAL_DIST_PENDING 3
+#define MAL_DIST_COMPLETED 4
+
+
+// TODO Refactor
+#define COMM_PHY_NODES 1
+#define COMM_PHY_CPU 2
+
+#define COMM_SPAWN_SERIAL 0
+#define COMM_SPAWN_PTHREAD 1
+
+#define MAL_USE_NORMAL 0
+#define MAL_USE_IBARRIER 1
+#define MAL_USE_POINT 2
+#define MAL_USE_THREAD 3

Codes/malleability/malleabilityTypes.c

+#include "malleabilityTypes.h"
+
+
+void init_malleability_data_struct(malleability_data_t *data_struct, int size);
+void realloc_malleability_data_struct(malleability_data_t *data_struct, int qty_to_add);
+
+void def_malleability_entries(malleability_data_t *data_struct_rep, malleability_data_t *data_struct_dist, MPI_Datatype *new_type);
+void def_malleability_qty_type(malleability_data_t *data_struct_rep, malleability_data_t *data_struct_dist, MPI_Datatype *new_type);
+
+
+//======================================================||
+//======================================================||
+//===================PUBLIC FUNCTIONS===================||
+//======================================================||
+//======================================================||
+
+/*
+ * Anyade en la estructura de datos a comunicar con los hijos
+ * un nuevo set de datos de un total "total_qty" distribuido entre
+ * todos los padres. La nueva serie "data" solo representa los datos
+ * que tiene este padre.
+ */
+void add_data(void *data, int total_qty, int type, int request_qty, malleability_data_t *data_struct) {
+  int i;
+  
+  if(data_struct->entries == 0) {
+    init_malleability_data_struct(data_struct, MALLEABILITY_INIT_DATA_QTY);
+  } else if(data_struct->entries == data_struct->max_entries) {
+    realloc_malleability_data_struct(data_struct, MALLEABILITY_INIT_DATA_QTY);
+  }
+  
+  data_struct->qty[data_struct->entries] = total_qty;
+  data_struct->types[data_struct->entries] = type;
+  data_struct->arrays[data_struct->entries] = data;
+
+  data_struct->requests[data_struct->entries] = (MPI_Request *) malloc(request_qty * sizeof(MPI_Request));
+  for(i=0; i < request_qty; i++) {
+    data_struct->requests[data_struct->entries][i] = MPI_REQUEST_NULL;
+  }
+  data_struct->entries+=1;
+}
+
+
+/*
+ * Comunicar desde los padres a los hijos las estructuras de datos sincronas o asincronas
+ * No es necesario que las estructuras esten inicializadas para un buen funcionamiento.
+ *
+ * En el argumento "root" todos tienen que indicar quien es el proceso raiz de los padres
+ * unicamente.
+ */
+void comm_data_info(malleability_data_t *data_struct_rep, malleability_data_t *data_struct_dist, int is_children_group, int myId, int root, MPI_Comm intercomm) {
+  int rootBcast = MPI_PROC_NULL;
+  MPI_Datatype entries_type, struct_type;
+
+  if(is_children_group) {
+    rootBcast = root;
+  } else {
+    if(myId == root) rootBcast = MPI_ROOT;
+  }
+
+  // Mandar primero numero de entradas
+  def_malleability_entries(data_struct_dist, data_struct_rep, &entries_type);
+  MPI_Bcast(&(data_struct_rep->entries), 1, entries_type, rootBcast, intercomm);
+
+  if(is_children_group) {
+    if(data_struct_rep->entries == 0) init_malleability_data_struct(data_struct_rep, data_struct_rep->entries);
+    if(data_struct_dist->entries == 0) init_malleability_data_struct(data_struct_dist, data_struct_dist->entries);
+  }
+
+  def_malleability_qty_type(data_struct_dist, data_struct_rep, &struct_type);
+  MPI_Bcast(&data_struct_rep, 1, struct_type, rootBcast, intercomm);
+
+  MPI_Type_free(&entries_type);
+  MPI_Type_free(&struct_type);
+}
+
+//======================================================||
+//======================================================||
+//=========INIT/REALLOC/FREE RESULTS FUNCTIONS==========||
+//======================================================||
+//======================================================||
+
+/*
+ * Inicializa la estructura que describe una serie de datos con las mismas
+ * caracteristicas de localización y uso. Se inicializa para utilizar hasta
+ * "size" elementos.
+ */
+void init_malleability_data_struct(malleability_data_t *data_struct, int size) {
+  data_struct->max_entries = size;
+  data_struct->qty = (int *) malloc(size * sizeof(int));
+  data_struct->types = (int *) malloc(size * sizeof(int));
+  data_struct->requests = (MPI_Request **) malloc(size * sizeof(MPI_Request *));
+  data_struct->arrays = (void **) malloc(size * sizeof(void *));
+
+  data_struct->request_ibarrier = MPI_REQUEST_NULL;
+}
+
+/*
+ * Realoja la estructura que describe una serie de datos con las mismas
+ * caracteristicas de localización y uso. Se anyaden "size" entradas nuevas
+ * a las ya existentes.
+ */
+void realloc_malleability_data_struct(malleability_data_t *data_struct, int qty_to_add) {
+  int *qty_aux, *types_aux, needed;
+  MPI_Request **requests_aux;
+  void **arrays_aux;
+
+  needed = data_struct->max_entries + qty_to_add;
+  qty_aux = (int *) realloc(data_struct->qty, needed * sizeof(int));
+  types_aux = (int *) realloc(data_struct->types, needed * sizeof(int));
+  requests_aux = (MPI_Request **) realloc(data_struct->requests, needed * sizeof(MPI_Request *));
+  arrays_aux = (void **) realloc(data_struct->arrays, needed * sizeof(void *));
+
+  if(qty_aux == NULL || arrays_aux == NULL || requests_aux == NULL || types_aux == NULL) {
+    fprintf(stderr, "Fatal error - No se ha podido realojar la memoria constante de datos a redistribuir/comunicar\n");
+    MPI_Abort(MPI_COMM_WORLD, 1);
+  }
+
+  data_struct->qty = qty_aux;
+  data_struct->types = types_aux;
+  data_struct->requests = requests_aux;
+  data_struct->arrays = arrays_aux;
+  data_struct->max_entries = needed;
+}
+
+void free_malleability_data_struct(malleability_data_t *data_struct) {
+  int i, max;
+
+  max = data_struct->entries;
+  if(max != 0) {
+    for(i=0; i<max; i++) {
+      free(data_struct->requests[i]);
+    }
+
+    free(data_struct->qty);
+    free(data_struct->types);
+    free(data_struct->requests);
+    free(data_struct->arrays);
+  }
+}
+
+//======================================================||
+//======================================================||
+//================MPI DERIVED DATATYPES=================||
+//======================================================||
+//======================================================||
+
+/*
+ * Crea un tipo derivado para mandar el numero de entradas
+ * en dos estructuras de descripcion de datos.
+ */
+void def_malleability_entries(malleability_data_t *data_struct_rep, malleability_data_t *data_struct_dist, MPI_Datatype *new_type) {
+  int counts = 2;
+  int blocklength = 1;
+  MPI_Aint displs, dir;
+
+  // Obtener direccion base
+  MPI_Get_address(&(data_struct_rep->entries), &dir);
+  MPI_Get_address(&(data_struct_dist->entries), &displs);
+  displs -= dir;
+
+  MPI_Type_create_hvector(counts, blocklength, displs, MPI_INT, new_type);
+  MPI_Type_commit(new_type);
+}
+
+/*
+ * Crea un tipo derivado para mandar las cantidades y tipo
+ * de datos de dos estructuras de descripcion de datos.
+ * El vector de "requests" no es enviado ya que solo es necesario
+ * en los padres.
+ */
+void def_malleability_qty_type(malleability_data_t *data_struct_rep, malleability_data_t *data_struct_dist, MPI_Datatype *new_type) {
+  int i, counts = 4;
+  int blocklengths[counts];
+  MPI_Aint displs[counts], dir;
+  MPI_Datatype types[counts];
+
+  types[0] = types[1] = types[2] = types[3] = MPI_INT;
+  blocklengths[0] = blocklengths[1] = data_struct_rep->entries;
+  blocklengths[2] = blocklengths[3] = data_struct_dist->entries;
+
+  // Obtener direccion base
+  MPI_Get_address(data_struct_rep, &dir);
+
+  MPI_Get_address(&(data_struct_rep->qty), &displs[0]);
+  MPI_Get_address(&(data_struct_rep->types), &displs[1]);
+  MPI_Get_address(&(data_struct_dist->qty), &displs[2]);
+  MPI_Get_address(&(data_struct_dist->types), &displs[3]);
+
+  for(i=0;i<counts;i++) displs[i] -= dir;
+
+  MPI_Type_create_struct(counts, blocklengths, displs, types, new_type);
+  MPI_Type_commit(new_type);
+}

Codes/malleability/malleabilityTypes.h

+#include <stdio.h>
+#include <stdlib.h>
+#include <mpi.h>
+#include <fcntl.h>
+#include <sys/stat.h>
+#include "malleabilityStates.h"
+
+#define MALLEABILITY_INIT_DATA_QTY 100
+#define MAL_INT 0
+#define MAL_CHAR 1
+
+typedef struct {
+  int entries; // Indica numero de vectores a comunicar (replicated data)
+  int max_entries;
+  MPI_Request request_ibarrier; // Request para indicar que los padres esperan a que los hijos terminen de recibir
+  int *qty; // Indica numero de elementos en cada subvector de sync_array
+  int *types;
+
+  // Vector de vectores de request. En cada elemento superior se indican los requests a comprobar para dar por finalizada 
+  // la comunicacion de ese dato
+  MPI_Request **requests; 
+  void **arrays; // Cada subvector es una serie de datos a comunicar
+
+} malleability_data_t;
+
+void add_data(void *data, int total_qty, int type, int request_qty, malleability_data_t *data_struct);
+void comm_data_info(malleability_data_t *data_struct_rep, malleability_data_t *data_struct_dist, int is_children_group, int myId, int root, MPI_Comm intercomm);
+void free_malleability_data_struct(malleability_data_t *data_struct);