ProcessDist.c 12.8 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
#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"

11
int commSlurm = MAL_NOT_STARTED;
12
13
struct Slurm_data *slurm_data;  
pthread_t slurm_thread;
14
MPI_Comm *returned_comm;
15
16
17
18
19
20
21
22

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

23
typedef struct {
24
  char *argv;
25
26
  int numP_childs, myId, root, type_dist;
  MPI_Comm comm;
27
}Creation_data;
28

29

30
31
32
33
//--------------PRIVATE SPAWN TYPE DECLARATIONS---------------//
void* thread_work(void* creation_data_arg);

//--------------PRIVATE DECLARATIONS---------------//
34
void processes_dist(char *argv, int numP_childs, int type_dist);
35
36
37
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);

38
39
40
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);

41
42
43
44
//@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);
45
46
47

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

48
49
50
51
52
53
54
55
56
57
58
59
/*
 * 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.
60
61
62
 *
 * Devuelve el estado de el procedimiento. Si no devuelve "COMM_FINISHED", es necesario llamar a
 * "check_slurm_comm()".
63
 */
64
int init_slurm_comm(char *argv, int myId, int numP, int root, int type_dist, int type_creation, MPI_Comm comm, MPI_Comm *child) {
65
66
67

  slurm_data = malloc(sizeof(struct Slurm_data));

68
69
  slurm_data->type_creation = type_creation;
  if(type_creation == COMM_SPAWN_SERIAL) {
70

71
    if(myId == root) {
72
      processes_dist(argv, numP, type_dist);
73
74
75
    } else {
      slurm_data->cmd = malloc(1 * sizeof(char));
      slurm_data->info = MPI_INFO_NULL;
76
77
    }
    create_processes(myId, root, child, comm);
78
79
80
81
82

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

85
    commSlurm = MAL_SPAWN_COMPLETED;
86

87
  } else if(type_creation == COMM_SPAWN_PTHREAD) {
88
    commSlurm = MAL_SPAWN_PENDING;
89

90
    Creation_data *creation_data = (Creation_data *) malloc(sizeof(Creation_data));
91
92
93
94
95
96
    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;
97

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

108
109
/*
 * Comprueba si una configuracion para crear un nuevo grupo de procesos esta lista,
110
 * y en caso de que lo este, se devuelve el communicador a estos nuevos procesos.
111
 */
112
int check_slurm_comm(int myId, int root, int numP, MPI_Comm *child) { // TODO Borrar numP si no se usa
113
  int state=-10;
114

115
  if(slurm_data->type_creation == COMM_SPAWN_PTHREAD) {
116

117
    MPI_Allreduce(&commSlurm, &state, 1, MPI_INT, MPI_MIN, MPI_COMM_WORLD);
118

119
    if(state != MAL_SPAWN_COMPLETED) return state; // Continue only if asynchronous process creation has ended 
120
121
122

  } else { 
    return commSlurm;
123
124
  }

125
126
127
128
129
  if(pthread_join(slurm_thread, NULL)) {
    printf("Error al esperar al hilo\n");
    MPI_Abort(MPI_COMM_WORLD, -1);
    return -10;
  }  
130

131
  *child = *returned_comm;
132

133
  if(myId == root && slurm_data->info != MPI_INFO_NULL) {
134
135
136
    MPI_Info_free(&(slurm_data->info));
  }
  free(slurm_data->cmd);
137
138
139
  free(slurm_data);

  return commSlurm;
140
141
142
}

//--------------PRIVATE SPAWN TYPE FUNCTIONS---------------//
143
144
145
146
147
148
149
150

/*
 * 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.
 */
151
void* thread_work(void* creation_data_arg) {
152
  Creation_data *creation_data = (Creation_data*) creation_data_arg;
153
  returned_comm = (MPI_Comm *) malloc(sizeof(MPI_Comm));
154
 
155
156
  if(creation_data->myId == creation_data->root) {
    processes_dist(creation_data->argv, creation_data->numP_childs, creation_data->type_dist);
157
158
159
  } else {
    slurm_data->cmd = malloc(1 * sizeof(char));
    slurm_data->info = MPI_INFO_NULL;
160
161
162
  }

  create_processes(creation_data->myId, creation_data->root, returned_comm, creation_data->comm);
163
  commSlurm = MAL_SPAWN_COMPLETED;
164

165
  free(creation_data);
166
167
168
169
170
  pthread_exit(NULL);
}

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

171
172
173
174
175
/*
 * 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.
 */
176
177
void processes_dist(char *argv, int numP_childs, int type) {
    int jobId;
178
179
180
181
182
183
    char *tmp;
    job_info_msg_t *j_info;
    slurm_job_info_t last_record;

    int used_nodes=0;
    int *procs_array;
184
    char *hostfile;
185
186
187
188
189
190
191
192

    // 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
193
194
    slurm_data->cmd = malloc(strlen(argv) * sizeof(char));
    strcpy(slurm_data->cmd, argv);
195
196
197
198
199

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

200
201
    /*
    // CREATE/UPDATE HOSTFILE 
202
    int ptr;
203
    ptr = create_hostfile(tmp, &hostfile);
204
    MPI_Info_create(&(slurm_data->info));
205
206
    MPI_Info_set(slurm_data->info, "hostfile", hostfile);
    free(hostfile);
207
208

    // SET NEW DISTRIBUTION 
iker_martin's avatar
iker_martin committed
209
    fill_hostfile(last_record, ptr, procs_array, used_nodes);
210
    close(ptr);
211
    */
212
    
213
    // TEST 
214
215
216
    fill_str_hostfile(last_record, procs_array, used_nodes, &hostfile);
    MPI_Info_create(&(slurm_data->info));
    MPI_Info_set(slurm_data->info, "hosts", hostfile);
217
218
    free(hostfile);
   
219
220
221
222
223

    // Free JOB INFO
    slurm_free_job_info_msg(j_info); 
}

224
225
226
227
/*
 * Crea un grupo de procesos segun la configuracion indicada por la funcion
 * "processes_dist()".
 */
228
229
230
231
232
233
234
235
236
237
int create_processes(int myId, int root, MPI_Comm *child, MPI_Comm comm) {
  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);
  }

  return spawn_err;
}

238
239
240
241
242
243
244
245
246
247
248
249
/*
 * 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.
 */
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
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)++;
    }
284
    if(*used_nodes > job_record.num_nodes) *used_nodes = job_record.num_nodes;  //FIXME Si ocurre esto no es un error?
285
286
287
288
289
290
291
292
293
  }

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

294
/*
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
 * 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
359
360
361
362
363
364
365
366
367
 * 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.
 */
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
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
}

385
/*
386
 * @deprecated
387
388
389
390
 * 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
391
void fill_hostfile(slurm_job_info_t job_record, int ptr, int *qty, int used_nodes) {
392
393
394
395
396
397
398
399
400
401
402
403
404
  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);
}

405
/*
406
 * @deprecated
407
408
409
410
411
 * 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.
 */
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
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;
}