Commit 32a4ca93 authored by Iker Martín Álvarez's avatar Iker Martín Álvarez
Browse files

Merge branch 'malleability-refactor' into 'dev'

Malleability focus refactor of Proteo

See merge request martini/malleability_benchmark!5
parents f1511cb4 06573694
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <fcntl.h>
#include <unistd.h>
#include <mpi.h>
#include <signal.h>
#include "../IOcodes/results.h"
#include "malleabilityZombies.h"
#define PIDS_QTY 320
void zombies_suspend();
int offset_pids, *pids = NULL;
void gestor_usr2() {}
void zombies_collect_suspended(MPI_Comm comm, int myId, int numP, int numC, int root, void *results_void, int n_stages) {
int pid = getpid();
int *pids_counts = malloc(numP * sizeof(int));
int *pids_displs = malloc(numP * sizeof(int));
int i, count=1;
if(myId < numC) {
count = 0;
if(myId == root) {
for(i=0; i < numC; i++) {
pids_counts[i] = 0;
}
for(i=numC; i<numP; i++) {
pids_counts[i] = 1;
pids_displs[i] = (i + offset_pids) - numC;
}
offset_pids += numP - numC;
}
}
MPI_Gatherv(&pid, count, MPI_INT, pids, pids_counts, pids_displs, MPI_INT, root, comm);
free(pids_counts);
free(pids_displs);
if(myId >= numC) {
// FIXME No deberia estar aqui
// Needed to ensure iteration times are collected before suspending these processes
results_data *results = (results_data *) results_void;
compute_results_iter(results, myId, numP,root, comm);
compute_results_stages(results, myId, numP, n_stages, root, comm);
zombies_suspend();
}
}
void zombies_service_init() {
offset_pids = 0;
pids = malloc(PIDS_QTY * sizeof(int));
for(int i=0; i<PIDS_QTY; i++) {
pids[i] = 0;
}
}
void zombies_service_free() {
free(pids);
}
void zombies_suspend() {
struct sigaction act;
sigemptyset(&act.sa_mask);
act.sa_flags=0;
act.sa_handler=gestor_usr2;
sigaction(SIGUSR2, &act, NULL);
sigset_t set;
sigprocmask(SIG_SETMASK,NULL,&set);
sigsuspend(&set);
}
void zombies_awake() {
for(int i=0; i < offset_pids; i++) { // Despertar a los zombies
kill(pids[i], SIGUSR2);
}
}
#ifndef MALLEABILITY_ZOMBIES_H
#define MALLEABILITY_ZOMBIES_H
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <fcntl.h>
#include <unistd.h>
#include <mpi.h>
#include <signal.h>
void zombies_collect_suspended(MPI_Comm comm, int myId, int numP, int numC, int root, void *results_void, int n_stages);
void zombies_service_init();
void zombies_service_free();
void zombies_awake();
#endif
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <mpi.h> #include <mpi.h>
#include "../malleabilityStates.h" #include "../MAM_Constants.h"
#include "../MAM_DataStructures.h"
#include "Baseline.h" #include "Baseline.h"
#include "Spawn_state.h" #include "Spawn_state.h"
#define MAM_TAG_STRAT_SINGLE 130
#define MAM_TAG_STRAT_MULTIPLE_FIRST 131
#define MAM_TAG_STRAT_MULTIPLE_OTHER 132
//--------------PRIVATE DECLARATIONS---------------// //--------------PRIVATE DECLARATIONS---------------//
int baseline_spawn(Spawn_data spawn_data, MPI_Comm comm, MPI_Comm *child); int baseline_spawn(Spawn_set spawn_set, MPI_Comm comm, MPI_Comm *child);
int single_strat_parents(Spawn_data spawn_data, MPI_Comm *child); void baseline_parents(Spawn_data spawn_data, MPI_Comm *child);
void single_strat_children(int myId, int root, MPI_Comm *parents);
void multiple_strat_parents(Spawn_data spawn_data, MPI_Comm comm, MPI_Comm *intercomms, MPI_Comm *child);
void multiple_strat_children(MPI_Comm *parents);
void single_strat_parents(Spawn_data spawn_data, MPI_Comm *child);
void single_strat_children(MPI_Comm *parents);
//--------------PUBLIC FUNCTIONS---------------// //--------------PUBLIC FUNCTIONS---------------//
...@@ -18,72 +27,193 @@ void single_strat_children(int myId, int root, MPI_Comm *parents); ...@@ -18,72 +27,193 @@ void single_strat_children(int myId, int root, MPI_Comm *parents);
*/ */
int baseline(Spawn_data spawn_data, MPI_Comm *child) { //TODO Tratamiento de errores int baseline(Spawn_data spawn_data, MPI_Comm *child) { //TODO Tratamiento de errores
MPI_Comm intercomm; MPI_Comm intercomm;
MPI_Comm_get_parent(&intercomm); MPI_Comm_get_parent(&intercomm); //FIXME May be a problem for third reconf or more with only expansions
if (intercomm == MPI_COMM_NULL) { // Parents path if (intercomm == MPI_COMM_NULL) { // Parents path
if (spawn_data.spawn_is_single) { baseline_parents(spawn_data, child);
single_strat_parents(spawn_data, child); } else { // Children path
} else { if(spawn_data.spawn_is_multiple) { multiple_strat_children(child); }
baseline_spawn(spawn_data, spawn_data.comm, child); if(spawn_data.spawn_is_single) { single_strat_children(child); }
}
} else if(spawn_data.spawn_is_single) { // Children path
single_strat_children(spawn_data.myId, spawn_data.root, child);
} }
return MALL_SPAWN_COMPLETED;
return MAM_I_SPAWN_COMPLETED;
} }
//--------------PRIVATE FUNCTIONS---------------// //--------------PRIVATE FUNCTIONS---------------//
/* /*
* Crea un grupo de procesos segun la configuracion indicada por la funcion * Funcion utilizada por los padres para realizar la
* "processes_dist()". * creación de procesos.
*
*/ */
int baseline_spawn(Spawn_data spawn_data, MPI_Comm comm, MPI_Comm *child) { void baseline_parents(Spawn_data spawn_data, MPI_Comm *child) {
int i;
MPI_Comm comm, *intercomms;
if (spawn_data.spawn_is_single && mall->myId != mall->root) {
single_strat_parents(spawn_data, child);
return;
}
comm = spawn_data.spawn_is_single ? MPI_COMM_SELF : spawn_data.comm;
MPI_Bcast(&spawn_data.total_spawns, 1, MPI_INT, mall->root, comm);
intercomms = (MPI_Comm*) malloc(spawn_data.total_spawns * sizeof(MPI_Comm));
if(mall->myId != mall->root) {
spawn_data.sets = (Spawn_set *) malloc(spawn_data.total_spawns * sizeof(Spawn_set));
}
#if MAM_DEBUG >= 3
DEBUG_FUNC("Starting spawning of processes", mall->myId, mall->numP); fflush(stdout);
#endif
for(i=0; i<spawn_data.total_spawns; i++) {
baseline_spawn(spawn_data.sets[i], comm, &intercomms[i]);
}
#if MAM_DEBUG >= 3
DEBUG_FUNC("Sources have created the new processes. Performing additional actions if required.", mall->myId, mall->numP); fflush(stdout);
#endif
// TODO Improvement - Deactivate Multiple spawn before spawning if total_spawns == 1
if(spawn_data.spawn_is_multiple) { multiple_strat_parents(spawn_data, comm, intercomms, child); }
else { *child = intercomms[0]; }
if(spawn_data.spawn_is_single) { single_strat_parents(spawn_data, child); }
free(intercomms);
if(mall->myId != mall->root) { free(spawn_data.sets); }
}
/*
* Funcion basica encargada de la creacion de procesos.
* Crea un set de procesos segun la configuracion obtenida
* en ProcessDist.c
* Devuelve en "child" el intercomunicador que se conecta a los hijos.
*/
int baseline_spawn(Spawn_set spawn_set, MPI_Comm comm, MPI_Comm *child) {
int rootBcast = MPI_PROC_NULL; int rootBcast = MPI_PROC_NULL;
if(spawn_data.myId == spawn_data.root) rootBcast = MPI_ROOT; if(mall->myId == mall->root) rootBcast = MPI_ROOT;
// WORK int spawn_err = MPI_Comm_spawn(spawn_set.cmd, MPI_ARGV_NULL, spawn_set.spawn_qty, spawn_set.mapping, mall->root, comm, child, MPI_ERRCODES_IGNORE);
int spawn_err = MPI_Comm_spawn(spawn_data.cmd, MPI_ARGV_NULL, spawn_data.spawn_qty, spawn_data.mapping, spawn_data.root, comm, child, MPI_ERRCODES_IGNORE);
MPI_Comm_set_name(*child, "MPI_COMM_MALL_RESIZE");
// END WORK
if(spawn_err != MPI_SUCCESS) { if(spawn_err != MPI_SUCCESS) {
printf("Error creating new set of %d procs.\n", spawn_data.spawn_qty); printf("Error creating new set of %d procs.\n", spawn_set.spawn_qty);
} }
MPI_Bcast(&spawn_data, 1, spawn_data.dtype, rootBcast, *child); MAM_Comm_main_structures(*child, rootBcast);
return spawn_err; return spawn_err;
} }
void multiple_strat_parents(Spawn_data spawn_data, MPI_Comm comm, MPI_Comm *intercomms, MPI_Comm *child) {
int i, tag;
char *port_name, aux;
if(mall->myId == mall->root) {
port_name = (char *) malloc(MPI_MAX_PORT_NAME * sizeof(char));
tag = MAM_TAG_STRAT_MULTIPLE_FIRST;
MPI_Send(&spawn_data.total_spawns, 1, MPI_INT, MAM_ROOT, tag, intercomms[0]);
MPI_Recv(port_name, MPI_MAX_PORT_NAME, MPI_CHAR, MPI_ANY_SOURCE, tag, intercomms[0], MPI_STATUS_IGNORE);
for(i=1; i<spawn_data.total_spawns; i++) {
MPI_Send(port_name, MPI_MAX_PORT_NAME, MPI_CHAR, MAM_ROOT, tag+i, intercomms[i]);
MPI_Recv(&aux, 1, MPI_CHAR, MPI_ANY_SOURCE, MAM_TAG_STRAT_MULTIPLE_FIRST, intercomms[0], MPI_STATUS_IGNORE);
}
} else { port_name = malloc(1); }
MPI_Comm_connect(port_name, MPI_INFO_NULL, mall->root, comm, child);
for(i=0; i<spawn_data.total_spawns; i++) {
MPI_Comm_disconnect(&intercomms[i]);
}
free(port_name);
}
void multiple_strat_children(MPI_Comm *parents) {
int i, start, total_spawns, new_root;
int rootBcast = MPI_PROC_NULL;
char *port_name, aux;
MPI_Status stat;
MPI_Comm newintracomm, intercomm, parents_comm;
new_root = 0;
parents_comm = *parents;
if(mall->myId == mall->root) {
port_name = (char *) malloc(MPI_MAX_PORT_NAME * sizeof(char));
MPI_Probe(MPI_ANY_SOURCE, MPI_ANY_TAG, parents_comm, &stat);
if(stat.MPI_TAG == MAM_TAG_STRAT_MULTIPLE_FIRST) {
MPI_Recv(&total_spawns, 1, MPI_INT, stat.MPI_SOURCE, stat.MPI_TAG, parents_comm, MPI_STATUS_IGNORE);
MPI_Open_port(MPI_INFO_NULL, port_name);
MPI_Send(port_name, MPI_MAX_PORT_NAME, MPI_CHAR, stat.MPI_SOURCE, stat.MPI_TAG, parents_comm);
start = 0;
new_root = 1;
rootBcast = MPI_ROOT;
} else {
MPI_Recv(port_name, MPI_MAX_PORT_NAME, MPI_CHAR, stat.MPI_SOURCE, stat.MPI_TAG, parents_comm, &stat);
// The "+1" is because the first iteration is done before the loop
start = stat.MPI_TAG - MAM_TAG_STRAT_MULTIPLE_FIRST + 1;
}
} else { port_name = malloc(1); }
MPI_Bcast(&start, 1, MPI_INT, mall->root, mall->comm);
if(start) {
MPI_Comm_connect(port_name, MPI_INFO_NULL, mall->root, mall->comm, &intercomm);
MPI_Bcast(&total_spawns, 1, MPI_INT, mall->root, intercomm); // FIXME Seems inneficient - Should be performed by parent root?
MPI_Intercomm_merge(intercomm, 1, &newintracomm); // Get last ranks
MPI_Comm_disconnect(&intercomm);
} else {
start = 1;
MPI_Comm_dup(mall->comm, &newintracomm);
MPI_Bcast(&total_spawns, 1, MPI_INT, mall->root, mall->comm); // FIXME Seems inneficient - Should be performed by parent root?
}
for(i=start; i<total_spawns; i++) {
MPI_Comm_accept(port_name, MPI_INFO_NULL, mall->root, newintracomm, &intercomm);
MPI_Bcast(&total_spawns, 1, MPI_INT, rootBcast, intercomm); // FIXME Seems inneficient - Should be performed by parent root?
if(newintracomm != MPI_COMM_WORLD) MPI_Comm_disconnect(&newintracomm);
MPI_Intercomm_merge(intercomm, 0, &newintracomm); // Get first ranks
MPI_Comm_disconnect(&intercomm);
if(new_root) {
MPI_Send(&aux, 1, MPI_CHAR, stat.MPI_SOURCE, stat.MPI_TAG, parents_comm); // Ensures order in the created intracommunicator
}
}
// Connect with parents
MPI_Comm_accept(port_name, MPI_INFO_NULL, mall->root, newintracomm, &intercomm);
// Update communicator to expected one
MAM_comms_update(newintracomm);
MPI_Comm_rank(mall->comm, &mall->myId);
MPI_Comm_size(mall->comm, &mall->numP);
if(new_root) MPI_Close_port(port_name);
free(port_name);
MPI_Comm_disconnect(&newintracomm);
MPI_Comm_disconnect(parents);
*parents = intercomm;
}
/* /*
* Si la variable "type" es 1, la creación es con la participación de todo el grupo de padres * Si la variable "type" es 1, la creación es con la participación de todo el grupo de padres
* Si el valor es diferente, la creación es solo con la participación del proceso root * Si el valor es diferente, la creación es solo con la participación del proceso root
*/ */
int single_strat_parents(Spawn_data spawn_data, MPI_Comm *child) { void single_strat_parents(Spawn_data spawn_data, MPI_Comm *child) {
int spawn_err;
char *port_name; char *port_name;
MPI_Comm newintercomm; MPI_Comm newintercomm;
if (spawn_data.myId == spawn_data.root) { if (mall->myId == mall->root) {
spawn_err = baseline_spawn(spawn_data, MPI_COMM_SELF, child);
port_name = (char *) malloc(MPI_MAX_PORT_NAME * sizeof(char)); port_name = (char *) malloc(MPI_MAX_PORT_NAME * sizeof(char));
MPI_Recv(port_name, MPI_MAX_PORT_NAME, MPI_CHAR, spawn_data.root, 130, *child, MPI_STATUS_IGNORE); MPI_Recv(port_name, MPI_MAX_PORT_NAME, MPI_CHAR, MPI_ANY_SOURCE, MAM_TAG_STRAT_SINGLE, *child, MPI_STATUS_IGNORE);
set_spawn_state(MALL_SPAWN_SINGLE_COMPLETED, spawn_data.spawn_is_async); // Indicate other processes to join root to end spawn procedure
set_spawn_state(MAM_I_SPAWN_SINGLE_COMPLETED, spawn_data.spawn_is_async); // Indicate other processes to join root to end spawn procedure
wakeup_completion();
} else { } else {
port_name = malloc(1); port_name = malloc(1);
} }
MPI_Comm_connect(port_name, MPI_INFO_NULL, spawn_data.root, spawn_data.comm, &newintercomm); MPI_Comm_connect(port_name, MPI_INFO_NULL, mall->root, spawn_data.comm, &newintercomm);
if(spawn_data.myId == spawn_data.root) if(mall->myId == mall->root)
MPI_Comm_free(child); MPI_Comm_disconnect(child);
free(port_name); free(port_name);
*child = newintercomm; *child = newintercomm;
return spawn_err;
} }
/* /*
...@@ -93,24 +223,24 @@ int single_strat_parents(Spawn_data spawn_data, MPI_Comm *child) { ...@@ -93,24 +223,24 @@ int single_strat_parents(Spawn_data spawn_data, MPI_Comm *child) {
* Solo se utiliza cuando la creación de los procesos ha sido * Solo se utiliza cuando la creación de los procesos ha sido
* realizada por un solo proceso padre * realizada por un solo proceso padre
*/ */
void single_strat_children(int myId, int root, MPI_Comm *parents) { void single_strat_children(MPI_Comm *parents) {
char *port_name; char *port_name;
MPI_Comm newintercomm; MPI_Comm newintercomm;
if(myId == root) { if(mall->myId == mall->root) {
port_name = (char *) malloc(MPI_MAX_PORT_NAME * sizeof(char)); port_name = (char *) malloc(MPI_MAX_PORT_NAME * sizeof(char));
MPI_Open_port(MPI_INFO_NULL, port_name); MPI_Open_port(MPI_INFO_NULL, port_name);
MPI_Send(port_name, MPI_MAX_PORT_NAME, MPI_CHAR, root, 130, *parents); MPI_Send(port_name, MPI_MAX_PORT_NAME, MPI_CHAR, mall->root_parents, MAM_TAG_STRAT_SINGLE, *parents);
} else { } else {
port_name = malloc(1); port_name = malloc(1);
} }
MPI_Comm_accept(port_name, MPI_INFO_NULL, root, MPI_COMM_WORLD, &newintercomm); MPI_Comm_accept(port_name, MPI_INFO_NULL, mall->root, mall->comm, &newintercomm);
if(myId == root) { if(mall->myId == mall->root) {
MPI_Close_port(port_name); MPI_Close_port(port_name);
} }
free(port_name); free(port_name);
MPI_Comm_free(parents); MPI_Comm_disconnect(parents);
*parents = newintercomm; *parents = newintercomm;
} }
#ifndef MALLEABILITY_SPAWN_BASELINE_H #ifndef MAM_SPAWN_BASELINE_H
#define MALLEABILITY_SPAWN_BASELINE_H #define MAM_SPAWN_BASELINE_H
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <mpi.h> #include <mpi.h>
#include <string.h> #include <string.h>
#include "../malleabilityDataStructures.h" #include "Spawn_DataStructure.h"
int baseline(Spawn_data spawn_data, MPI_Comm *child); int baseline(Spawn_data spawn_data, MPI_Comm *child);
#endif #endif
#ifndef MALLEABILITY_GENERIC_SPAWN_H #ifndef MAM_GENERIC_SPAWN_H
#define MALLEABILITY_GENERIC_SPAWN_H #define MAM_GENERIC_SPAWN_H
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <mpi.h> #include <mpi.h>
#include "../malleabilityDataStructures.h" #include "Spawn_DataStructure.h"
int init_spawn(char *argv, int num_cpus, int num_nodes, char *nodelist, int myId, int initial_qty, int target_qty, int root, int type_dist, int spawn_method, int spawn_strategies, MPI_Comm comm, MPI_Comm *child); int init_spawn(MPI_Comm comm, MPI_Comm *child);
int check_spawn_state(MPI_Comm *child, MPI_Comm comm, double *real_time); int check_spawn_state(MPI_Comm *child, MPI_Comm comm, int wait_completed);
void malleability_connect_children(int myId, int numP, int root, MPI_Comm comm, int *numP_parents, int *root_parents, MPI_Comm *parents); void malleability_connect_children(MPI_Comm *parents);
void unset_spawn_postpone_flag(int outside_state); void unset_spawn_postpone_flag(int outside_state);
int malleability_spawn_contains_strat(int spawn_strategies, int strategy, int *result);
#endif #endif
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <mpi.h> #include <mpi.h>
#include "../malleabilityStates.h" #include "../MAM_Constants.h"
#include "../MAM_DataStructures.h"
#include "Merge.h" #include "Merge.h"
#include "Baseline.h" #include "Baseline.h"
...@@ -16,11 +17,11 @@ int merge(Spawn_data spawn_data, MPI_Comm *child, int data_state) { ...@@ -16,11 +17,11 @@ int merge(Spawn_data spawn_data, MPI_Comm *child, int data_state) {
int is_children_group = 1; int is_children_group = 1;
if(spawn_data.initial_qty > spawn_data.target_qty) { //Shrink if(spawn_data.initial_qty > spawn_data.target_qty) { //Shrink
if(data_state == MALL_DIST_COMPLETED) { if(data_state == MAM_I_DIST_COMPLETED) {
merge_adapt_shrink(spawn_data.target_qty, child, spawn_data.comm, spawn_data.myId); merge_adapt_shrink(spawn_data.target_qty, child, spawn_data.comm, mall->myId);
local_state = MALL_SPAWN_ADAPTED; local_state = MAM_I_SPAWN_ADAPTED;
} else { } else {
local_state = MALL_SPAWN_ADAPT_POSTPONE; local_state = MAM_I_SPAWN_ADAPT_POSTPONE;
} }
} else { //Expand } else { //Expand
MPI_Comm_get_parent(&intercomm); MPI_Comm_get_parent(&intercomm);
...@@ -29,12 +30,17 @@ int merge(Spawn_data spawn_data, MPI_Comm *child, int data_state) { ...@@ -29,12 +30,17 @@ int merge(Spawn_data spawn_data, MPI_Comm *child, int data_state) {
baseline(spawn_data, child); baseline(spawn_data, child);
merge_adapt_expand(child, is_children_group); merge_adapt_expand(child, is_children_group);
local_state = MALL_SPAWN_COMPLETED; local_state = MAM_I_SPAWN_COMPLETED;
} }
return local_state; return local_state;
} }
int intracomm_strategy(int is_children_group, MPI_Comm *child) {
merge_adapt_expand(child, is_children_group);
return MAM_I_SPAWN_COMPLETED;
}
//--------------PRIVATE MERGE TYPE FUNCTIONS---------------// //--------------PRIVATE MERGE TYPE FUNCTIONS---------------//
/* /*
...@@ -51,13 +57,8 @@ void merge_adapt_expand(MPI_Comm *child, int is_children_group) { ...@@ -51,13 +57,8 @@ void merge_adapt_expand(MPI_Comm *child, int is_children_group) {
MPI_Intercomm_merge(*child, is_children_group, &new_comm); //El que pone 0 va primero MPI_Intercomm_merge(*child, is_children_group, &new_comm); //El que pone 0 va primero
MPI_Comm_free(child); //POSIBLE ERROR? MPI_Comm_disconnect(child);
*child = new_comm; *child = new_comm;
//*numP = numC; //TODO REFACTOR Llevar a otra parte -- Hacer solo si MALL_SPAWN_ADAPTED
//if(*comm != MPI_COMM_WORLD && *comm != MPI_COMM_NULL) {
// MPI_Comm_free(comm);
//}
} }
...@@ -71,6 +72,7 @@ void merge_adapt_expand(MPI_Comm *child, int is_children_group) { ...@@ -71,6 +72,7 @@ void merge_adapt_expand(MPI_Comm *child, int is_children_group) {
void merge_adapt_shrink(int numC, MPI_Comm *child, MPI_Comm comm, int myId) { void merge_adapt_shrink(int numC, MPI_Comm *child, MPI_Comm comm, int myId) {
int color = MPI_UNDEFINED; int color = MPI_UNDEFINED;
if(*child != MPI_COMM_NULL && *child != MPI_COMM_WORLD) MPI_Comm_disconnect(child);
if(myId < numC) { if(myId < numC) {
color = 1; color = 1;
} }
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
File mode changed from 100644 to 100755
This diff is collapsed.
This diff is collapsed.
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment