ProcessDist.c 12 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
#include <stdio.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <pthread.h>
#include <mpi.h>
#include <string.h>
#include <slurm/slurm.h>
#include "ProcessDist.h"

#define ROOT 0

int commSlurm = COMM_UNRESERVED;
struct Slurm_data *slurm_data;  
pthread_t slurm_thread;
16
MPI_Comm *returned_comm;
17
18
19
20
21
22
23
24
25
26

struct Slurm_data {
  char *cmd; // Executable name
  int qty_procs;
  MPI_Info info;
  int type_creation;
};

struct Creation_data {
  char **argv;
27
28
  int numP_childs, myId, root, type_dist;
  MPI_Comm comm;
29
30
};

31

32
33
34
35
36
37
38
39
40
41
42
//--------------PRIVATE SPAWN TYPE DECLARATIONS---------------//
void* thread_work(void* creation_data_arg);

//--------------PRIVATE DECLARATIONS---------------//

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);
iker_martin's avatar
iker_martin committed
43
void fill_hostfile(slurm_job_info_t job_record, int ptr, int *qty, int used_nodes);
44

45
46
47
48
49
//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);
//

50
51
52
53
void print_Info(MPI_Info info);

//--------------PUBLIC FUNCTIONS---------------//

54
55
56
57
58
59
60
61
62
63
64
65
/*
 * Se solicita la creacion de un nuevo grupo de "numP" procesos con una distribucion
 * fisica "type_dist".
 *
 * Se puede solicitar en primer plano, encargandose por tanto el proceso que llama a esta funcion,
 * o en segundo plano, donde un hilo se encarga de configurar esta creacion.
 *
 * Si se pide en primer plano, al terminarla es posible llamar a "check_slurm_comm()" para crear
 * los procesos.
 *
 * Si se pide en segundo plano, llamar a "check_slurm_comm()" comprobara si la configuracion para
 * crearlos esta lista, y si es asi, los crea.
66
67
68
 *
 * Devuelve el estado de el procedimiento. Si no devuelve "COMM_FINISHED", es necesario llamar a
 * "check_slurm_comm()".
69
 */
70
int init_slurm_comm(char **argv, int myId, int numP, int root, int type_dist, int type_creation, MPI_Comm comm, MPI_Comm *child) {
71
72
73

  slurm_data = malloc(sizeof(struct Slurm_data));

74
75
  slurm_data->type_creation = type_creation;
  if(type_creation == COMM_SPAWN_SERIAL) {
76

77
    if(myId == root) {
78
      processes_dist(argv, numP, type_dist);
79
80
81
82
    }
    create_processes(myId, root, child, comm);
    free(slurm_data);
    commSlurm = COMM_FINISHED;
83

84
85
  } else if(type_creation == COMM_SPAWN_PTHREAD) {
    commSlurm = COMM_IN_PROGRESS;
86

87
88
89
90
91
92
93
    struct Creation_data *creation_data = malloc(sizeof(struct Creation_Data*));
    creation_data->argv = argv;
    creation_data->numP_childs = numP;
    creation_data->myId = myId;
    creation_data->root = root;
    creation_data->type_dist = type_dist;
    creation_data->comm = comm;
94

95
96
97
98
    if(pthread_create(&slurm_thread, NULL, thread_work, creation_data)) {
      printf("Error al crear el hilo de contacto con SLURM\n");
      MPI_Abort(MPI_COMM_WORLD, -1);
      return -1;
99
100
101
    }
  }
    
102
  return commSlurm;
103
104
}

105
106
/*
 * Comprueba si una configuracion para crear un nuevo grupo de procesos esta lista,
107
 * y en caso de que lo este, se devuelve el communicador a estos nuevos procesos.
108
 */
109
110
int check_slurm_comm(int myId, int root, MPI_Comm *child) {
  int state;
111

112
113
114
115
116
117
  if(slurm_data->type_creation == COMM_SPAWN_PTHREAD) {
    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 

  } else { 
    return commSlurm;
118
119
  }

120
121
122
123
124
  if(pthread_join(slurm_thread, NULL)) {
    printf("Error al esperar al hilo\n");
    MPI_Abort(MPI_COMM_WORLD, -1);
    return -10;
  }  
125

126
127
  commSlurm = COMM_FINISHED;
  *child = *returned_comm;
128

129
130
131
  free(slurm_data);

  return commSlurm;
132
133
134
}

//--------------PRIVATE SPAWN TYPE FUNCTIONS---------------//
135
136
137
138
139
140
141
142

/*
 * Funcion llamada por un hilo para que este se encarge
 * de configurar la creacion de un nuevo grupo de procesos.
 *
 * Una vez esta lista la configuracion y es posible crear los procesos
 * se avisa al hilo maestro.
 */
143
144
void* thread_work(void* creation_data_arg) {
  struct Creation_data *creation_data = (struct Creation_data*) creation_data_arg;
145
  returned_comm = (MPI_Comm *) malloc(sizeof(MPI_Comm));
146
 
147
148
149
150
151
152
  if(creation_data->myId == creation_data->root) {
    //if(creation_data->myId == creation_data->root) { printf("WORKD SPAWN 1\n");} fflush(stdout);
    processes_dist(creation_data->argv, creation_data->numP_childs, creation_data->type_dist);
  }

  create_processes(creation_data->myId, creation_data->root, returned_comm, creation_data->comm);
153
154
  commSlurm = COMM_FINISHED;

155
  //free(creation_data); //FIXME No se libera bien
156
157
158
159
160
  pthread_exit(NULL);
}

//--------------PRIVATE SPAWN CREATION FUNCTIONS---------------//

161
162
163
164
165
/*
 * Configura la creacion de un nuevo grupo de procesos, reservando la memoria
 * para una llamada a MPI_Comm_spawn, obteniendo una distribucion fisica
 * para los procesos y creando un fichero hostfile.
 */
166
167
168
169
170
171
172
173
void processes_dist(char *argv[], int numP_childs, int type) {
    int jobId, ptr;
    char *tmp;
    job_info_msg_t *j_info;
    slurm_job_info_t last_record;

    int used_nodes=0;
    int *procs_array;
174
    char *hostfile;
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190

    // Get Slurm job info
    tmp = getenv("SLURM_JOB_ID");
    jobId = atoi(tmp);
    slurm_load_job(&j_info, jobId, 1);
    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]);

    // GET NEW DISTRIBUTION 
    node_dist(last_record, type, numP_childs, &procs_array, &used_nodes);
    slurm_data->qty_procs = numP_childs;

    // CREATE/UPDATE HOSTFILE
191
192
    
    ptr = create_hostfile(tmp, &hostfile);
193
    MPI_Info_create(&(slurm_data->info));
194
195
    MPI_Info_set(slurm_data->info, "hostfile", hostfile);
    free(hostfile);
196
197

    // SET NEW DISTRIBUTION 
iker_martin's avatar
iker_martin committed
198
    fill_hostfile(last_record, ptr, procs_array, used_nodes);
199
    close(ptr);
200
201
202
203
204
205
206
207
    

    // TEST
    /*
    fill_str_hostfile(last_record, procs_array, used_nodes, &hostfile);
    MPI_Info_create(&(slurm_data->info));
    MPI_Info_set(slurm_data->info, "hosts", hostfile);
    */
208
209
210
211
212

    // Free JOB INFO
    slurm_free_job_info_msg(j_info); 
}

213
214
215
216
/*
 * Crea un grupo de procesos segun la configuracion indicada por la funcion
 * "processes_dist()".
 */
217
int create_processes(int myId, int root, MPI_Comm *child, MPI_Comm comm) {
218
    //if(myId == root) { printf("WORKD SPAWN 2.1 cmd=%s pr=%d\n", slurm_data->cmd, slurm_data->qty_procs);} fflush(stdout);
219
220
221
222
223
224
225
226
227
228
229
230
231
232
  int spawn_err = MPI_Comm_spawn(slurm_data->cmd, MPI_ARGV_NULL, slurm_data->qty_procs, slurm_data->info, root, comm, child, MPI_ERRCODES_IGNORE); 

  if(spawn_err != MPI_SUCCESS) {
    printf("Error creating new set of %d procs.\n", slurm_data->qty_procs);
  }

  if(myId == root) {
    MPI_Info_free(&(slurm_data->info));
    free(slurm_data->cmd);
  }

  return spawn_err;
}

233
234
235
236
237
238
239
240
241
242
243
244
/*
 * Obtiene la distribucion fisica del grupo de procesos a crear, devolviendo
 * cuantos nodos se van a utilizar y la cantidad de procesos que alojara cada
 * nodo.
 *
 * Se permiten dos tipos de distribuciones fisicas segun el valor de "type":
 *
 *  COMM_PHY_NODES (1): Orientada a equilibrar el numero de procesos entre
 *                      todos los nodos disponibles.
 *  COMM_PHY_CPU   (2): Orientada a completar la capacidad de un nodo antes de
 *                      ocupar otro nodo.
 */
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
void node_dist(slurm_job_info_t job_record, int type, int total_procs, int **qty, int *used_nodes) {
  int i, asigCores;
  int tamBl, remainder;
  int *procs;

  procs = calloc(job_record.num_nodes, sizeof(int)); // Numero de procesos por nodo

  /* GET NEW DISTRIBUTION  */
  if(type == 1) { // DIST NODES
    *used_nodes = job_record.num_nodes;
    tamBl = total_procs / job_record.num_nodes;
    remainder = total_procs % job_record.num_nodes;
    for(i=0; i<remainder; i++) {
      procs[i] = tamBl + 1; 
    }
    for(i=remainder; i<job_record.num_nodes; i++) {
      procs[i] = tamBl; 
    }
  } else if (type == 2) { // DIST CPUs
    tamBl = job_record.num_cpus / job_record.num_nodes;
    asigCores = 0;
    i = 0;
    *used_nodes = 0;

    while(asigCores+tamBl <= total_procs) {
      asigCores += tamBl;
      procs[i] += tamBl;
      i = (i+1) % job_record.num_nodes;
      (*used_nodes)++;
    }
    if(asigCores < total_procs) {
      procs[i] += total_procs - asigCores;
      (*used_nodes)++;
    }
    if(*used_nodes > job_record.num_nodes) *used_nodes = job_record.num_nodes;
  }

  *qty = calloc(*used_nodes, sizeof(int)); // Numero de procesos por nodo
  for(i=0; i< *used_nodes; i++) {
    (*qty)[i] = procs[i];
  }
  free(procs);
}

289
290
291
292
293
294
295
296
297
298
/*
 * Crea un fichero que se utilizara como hostfile
 * para un nuevo grupo de procesos. 
 *
 * El nombre es devuelto en el argumento "file_name",
 * que tiene que ser un puntero vacio.
 *
 * Ademas se devuelve un descriptor de fichero para 
 * modificar el fichero.
 */
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
int create_hostfile(char *jobId, char **file_name) {
  int ptr, err, len;

  len = strlen(jobId) + 11;

  *file_name = NULL;
  *file_name = malloc( len * sizeof(char));
  if(*file_name == NULL) return -1; // No ha sido posible alojar la memoria
  err = snprintf(*file_name, len, "hostfile.o%s", jobId);
  if(err < 0) return -2; // No ha sido posible obtener el nombre de fichero

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

  return ptr; // Devolver puntero a fichero
}

316
317
318
319
320
/*
 * 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".
 */
iker_martin's avatar
iker_martin committed
321
void fill_hostfile(slurm_job_info_t job_record, int ptr, int *qty, int used_nodes) {
322
323
324
325
326
327
328
329
330
331
332
333
334
  int i=0;
  char *host;
  hostlist_t hostlist;
  
  hostlist = slurm_hostlist_create(job_record.nodes);
  while ( (host = slurm_hostlist_shift(hostlist)) && i < used_nodes) {
    write_hostfile_node(ptr, qty[i], host);
    i++;
    free(host);
  }
  slurm_hostlist_destroy(hostlist);
}

335
336
337
338
339
340
/*
 * 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
 * alojar en ese nodo.
 */
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
int write_hostfile_node(int ptr, int qty, char *node_name) {
  int err, len_node, len_int, len;
  char *line;

  len_node = strlen(node_name);
  len_int = snprintf(NULL, 0, "%d", qty);

  len = len_node + len_int + 3;
  line = malloc(len * sizeof(char));
  if(line == NULL) return -1; // No ha sido posible alojar la memoria
  err = snprintf(line, len, "%s:%d\n", node_name, qty);

  if(err < 0) return -2; // No ha sido posible escribir en el fichero

  write(ptr, line, len-1);
  free(line);

  return 0;
}
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407



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));
  } else { // Cadena ya tiene datos
    *hostfile_str = (char *) realloc(*hostfile_str, (len_og + len) * 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) {
    i++;
    strcpy(*hostfile_str, ocurrence);
  }
  for(; i<qty; i++){
    strcat(*hostfile_str, ocurrence);
  }
  
  free(ocurrence);
  return len;
}