ProcessDist.c 21 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 commState = MAL_NOT_STARTED;
12
struct Slurm_data *slurm_data;  
13
14
pthread_t spawn_thread;
pthread_mutex_t spawn_mutex;
15
MPI_Comm *returned_comm;
16

17
18
double end_time; //FIXME REFACTOR

19
20
struct Slurm_data {
  char *cmd; // Executable name
21
22
  char *nodelist;
  int num_cpus, num_nodes;
23
  int qty_procs, result_procs;
24
25
  MPI_Info info;
  int type_creation;
26
  int spawn_is_single;
27
28
};

29
typedef struct {
30
  char *argv;
31
32
  int numP_childs, myId, root, already_created;
  int type_dist;
33
34
  int spawn_is_single;
  int spawn_method;
35
  MPI_Comm comm;
36
}Creation_data;
37

38

39
40
41
42
//--------------PRIVATE SPAWN TYPE DECLARATIONS---------------//
void* thread_work(void* creation_data_arg);

//--------------PRIVATE DECLARATIONS---------------//
43
void processes_dist(char *argv, int numP_childs, int already_created, int type_dist);
44
45
46

void generic_spawn(int myId, int root, int is_single, MPI_Comm *child, MPI_Comm comm);
void single_spawn_connection(int myId, int root, MPI_Comm comm, MPI_Comm *child);
47
int create_processes(int myId, int root, MPI_Comm *child, MPI_Comm comm);
48

49
void node_dist(int type, int total_procs, int already_created, int **qty, int *used_nodes);
50

51
void fill_str_hostfile(int *qty, int used_nodes, char **hostfile_str);
52
53
int write_str_node(char **hostfile_str, int len_og, int qty, char *node_name);

54
55
56
57
//@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);
58
59
60

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

61
62
63
64
65
66
67
68
69
70
71
72
/*
 * 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.
73
74
75
 *
 * Devuelve el estado de el procedimiento. Si no devuelve "COMM_FINISHED", es necesario llamar a
 * "check_slurm_comm()".
76
 */
77
int init_slurm_comm(char *argv, int num_cpus, int num_nodes, char *nodelist, int myId, int numP, int numC, int root, int type_dist, int type_creation, int spawn_is_single, MPI_Comm comm, MPI_Comm *child) {
78
  int spawn_qty, already_created = 0;
79
80
  slurm_data = malloc(sizeof(struct Slurm_data));

81
  spawn_thread = pthread_self();
82
  slurm_data->type_creation = type_creation;
83
  slurm_data->spawn_is_single = spawn_is_single;
84
  slurm_data->result_procs = numC;
85
86
87
  slurm_data->num_cpus = num_cpus;
  slurm_data->num_nodes = num_nodes;
  slurm_data->nodelist = nodelist;
88
89
90
91
  spawn_qty = numC;
  if(type_creation == COMM_SPAWN_MERGE || type_creation == COMM_SPAWN_MERGE_PTHREAD) {
    if (numP < slurm_data->result_procs) {
      spawn_qty = slurm_data->result_procs - numP;
92
      already_created = numP;
93
94
    }
  }
95
  pthread_mutex_init(&spawn_mutex,NULL);
96

97
  if(type_creation == COMM_SPAWN_SERIAL || slurm_data->type_creation == COMM_SPAWN_MERGE) {
98

99
    if(myId == root) {
100
      processes_dist(argv, spawn_qty, already_created, type_dist);
101
102
103
    } else {
      slurm_data->cmd = malloc(1 * sizeof(char));
      slurm_data->info = MPI_INFO_NULL;
104
    }
105
106
107
108

    // WORK
    generic_spawn(myId, root, slurm_data->spawn_is_single, child, comm);
    // END WORK
109
110
111
112

    if(myId == root && slurm_data->info != MPI_INFO_NULL) {
      MPI_Info_free(&(slurm_data->info));
    }
113
    pthread_mutex_destroy(&spawn_mutex);
114
    free(slurm_data->cmd);
115
    free(slurm_data);
116

117
  } else if(type_creation == COMM_SPAWN_PTHREAD || slurm_data->type_creation == COMM_SPAWN_MERGE_PTHREAD) {
118
    commState = MAL_SPAWN_PENDING;
119
    
120
    if((spawn_is_single && myId == root) || !spawn_is_single || (slurm_data->type_creation == COMM_SPAWN_MERGE_PTHREAD && numP > slurm_data->result_procs)) {
121
122
      Creation_data *creation_data = (Creation_data *) malloc(sizeof(Creation_data));
      creation_data->argv = argv;
123
      creation_data->numP_childs = spawn_qty;
124
      creation_data->already_created = already_created;
125
126
127
128
129
      creation_data->myId = myId;
      creation_data->root = root;
      creation_data->type_dist = type_dist;
      creation_data->comm = comm;

130
      if(pthread_create(&spawn_thread, NULL, thread_work, (void *)creation_data)) {
131
132
133
134
        printf("Error al crear el hilo de contacto con SLURM\n");
        MPI_Abort(MPI_COMM_WORLD, -1);
        return -1;
      }
135
136
137
    }
  }
    
138
  return commState;
139
140
}

141

142
143
/*
 * Comprueba si una configuracion para crear un nuevo grupo de procesos esta lista,
144
 * y en caso de que lo este, se devuelve el communicador a estos nuevos procesos.
145
 */
146
147
int check_slurm_comm(int myId, int root, int numP, MPI_Comm *child, MPI_Comm comm, MPI_Comm comm_thread, double *real_time) { 

148
  if(slurm_data->type_creation == COMM_SPAWN_PTHREAD || slurm_data->type_creation == COMM_SPAWN_MERGE_PTHREAD) {
149
150
151
152
153
154
155
156

    if (slurm_data->type_creation == COMM_SPAWN_MERGE_PTHREAD && numP > slurm_data->result_procs) { //TODO REFACTOR
      printf("Error Check spawn: Configuracion invalida\nSe intenta usar el método Spawn junto a un Shrink merge\n");
      MPI_Abort(MPI_COMM_WORLD, -1);
      return -10;
    }

    if(!slurm_data->spawn_is_single || commState == MAL_SPAWN_SINGLE_PENDING || commState == MAL_SPAWN_COMPLETED) {
157
      int state=-10;
158
159

      //printf("[%d][3] Test min\n", myId); fflush(stdout);
160
      //pthread_mutex_lock(&spawn_mutex); // TODO Descomentar
161
      MPI_Allreduce(&commState, &state, 1, MPI_INT, MPI_MIN, comm);
162
163
164
      //pthread_mutex_unlock(&spawn_mutex);

      if(state != MAL_SPAWN_COMPLETED) return state; // Continue only if asynchronous process creation has ended 
165

166
167
      //printf("[%d][5] Test Passed-----------\n", myId); fflush(stdout);
      if(pthread_join(spawn_thread, NULL)) {
168
169
170
171
172
        printf("Error al esperar al hilo\n");
        MPI_Abort(MPI_COMM_WORLD, -1);
        return -10;
      }  
      *child = *returned_comm;
173
174

    } else if (slurm_data->spawn_is_single) {
175

176
      //pthread_mutex_lock(&spawn_mutex); // TODO Descomentar
177
      MPI_Bcast(&commState, 1, MPI_INT, root, comm);
178
179
      //pthread_mutex_unlock(&spawn_mutex);
      int threads_not_spawned = pthread_equal(pthread_self(), spawn_thread);
180
181
182

      // Non-root processes join root to finalize the spawn
      // They also must join if the application has ended its work
183
      if(commState == MAL_SPAWN_SINGLE_START) { 
184
        commState = MAL_SPAWN_SINGLE_PENDING;
185
186

        if(myId != root && threads_not_spawned) {
187
188
189
190
191
192
193
194
195
          Creation_data *creation_data = (Creation_data *) malloc(sizeof(Creation_data));
          creation_data->argv = NULL;
          creation_data->numP_childs = -1;
          creation_data->already_created = -1;
          creation_data->myId = myId;
          creation_data->root = root;
          creation_data->type_dist = -1;
          creation_data->comm = comm_thread;

196
          if(pthread_create(&spawn_thread, NULL, thread_work, (void *)creation_data)) {
197
198
199
200
201
            printf("Error al crear el hilo de apoyo\n");
            MPI_Abort(MPI_COMM_WORLD, -1);
            return -1;
          }
	}
202
203
      }

204
205
      // Continue only if asynchronous process creation has ended or application does not have more work
      if(commState != MAL_SPAWN_COMPLETED) return commState; 
206
207
      
      //printf("[%d][4] Test Passed-----------\n", myId); fflush(stdout);
208
      //Asegurar que los hilos han terminado
209
      if(pthread_join(spawn_thread, NULL)) { 
210
211
212
213
214
215
        printf("Error al esperar al hilo\n");
        MPI_Abort(MPI_COMM_WORLD, -1);
        return -10;
      }
      *child = *returned_comm;

216
    } else {
217
218
219
      printf("Error Check spawn: Configuracion invalida\n");
      MPI_Abort(MPI_COMM_WORLD, -1);
      return -10;
220
221
    }
  } else {  
222
    return commState;
223
  }
224

225
  //Free memory
226
  if(myId == root && slurm_data->info != MPI_INFO_NULL) {
227
228
229
    MPI_Info_free(&(slurm_data->info));
  }
  free(slurm_data->cmd);
230
  free(slurm_data);
231
232
233
234
  pthread_mutex_destroy(&spawn_mutex);
  spawn_thread = pthread_self();

  *real_time=end_time;
235

236
  return commState;
237
238
}

239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
/*
 * Conectar grupo de hijos con grupo de padres
 * Devuelve un intercomunicador para hablar con los padres
 *
 * Solo se utiliza cuando la creación de los procesos ha sido
 * realizada por un solo proceso padre
 */
void malleability_establish_connection(int myId, int root, MPI_Comm *intercomm) {
  char *port_name;
  MPI_Comm newintercomm;

  if(myId == root) {
    port_name = (char *) malloc(MPI_MAX_PORT_NAME * sizeof(char));
    MPI_Open_port(MPI_INFO_NULL, port_name);
    MPI_Send(port_name, MPI_MAX_PORT_NAME, MPI_CHAR, root, 130, *intercomm);
  } else {
    port_name = malloc(1);
  }

  MPI_Comm_accept(port_name, MPI_INFO_NULL, root, MPI_COMM_WORLD, &newintercomm);

  if(myId == root) {
    MPI_Close_port(port_name);
  }
  free(port_name);
  MPI_Comm_free(intercomm);
  *intercomm = newintercomm;
}

268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311

//--------------PRIVATE THREAD FUNCTIONS---------------//

/*
 * 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.
 */
void* thread_work(void* creation_data_arg) {
  Creation_data *creation_data = (Creation_data*) creation_data_arg;
  returned_comm = (MPI_Comm *) malloc(sizeof(MPI_Comm));
 
  if(creation_data->myId == creation_data->root) {
    processes_dist(creation_data->argv, creation_data->numP_childs, creation_data->already_created, creation_data->type_dist);
  } else {
    slurm_data->cmd = malloc(1 * sizeof(char));
    slurm_data->info = MPI_INFO_NULL;
  }
  generic_spawn(creation_data->myId, creation_data->root, slurm_data->spawn_is_single, returned_comm, creation_data->comm);

  free(creation_data);
  pthread_exit(NULL);
}

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


/*
 * Funcion generica para la creacion de procesos. Obtiene la configuracion
 * y segun esta, elige como deberian crearse los procesos.
 *
 * Cuando termina, modifica la variable global para indicar este cambio
 */
void generic_spawn(int myId, int root, int spawn_is_single, MPI_Comm *child, MPI_Comm comm) {
  if(spawn_is_single) {
    single_spawn_connection(myId, root, comm, child);
  } else {
    int rootBcast = MPI_PROC_NULL;
    if(myId == root) rootBcast = MPI_ROOT;
    create_processes(myId, root, child, comm);
    MPI_Bcast(&spawn_is_single, 1, MPI_INT, rootBcast, *child);
  }
312
  pthread_mutex_lock(&spawn_mutex);
313
  commState = MAL_SPAWN_COMPLETED; 
314
315
    end_time = MPI_Wtime();
  pthread_mutex_unlock(&spawn_mutex);
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
}

/*
 * Crea un grupo de procesos segun la configuracion indicada por la funcion
 * "processes_dist()".
 */
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;
}

/* 
 * 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
 */
void single_spawn_connection(int myId, int root, MPI_Comm comm, MPI_Comm *child){
  char *port_name;
  int auxiliar_conf = COMM_SPAWN_SINGLE;
  MPI_Comm newintercomm;

  if (myId == root) {
    create_processes(myId, root, child, MPI_COMM_SELF);

    MPI_Bcast(&auxiliar_conf, 1, MPI_INT, MPI_ROOT, *child);

    port_name = (char *) malloc(MPI_MAX_PORT_NAME * sizeof(char));
    MPI_Recv(port_name, MPI_MAX_PORT_NAME, MPI_CHAR, root, 130, *child, MPI_STATUS_IGNORE);
    commState = MAL_SPAWN_SINGLE_START; // Indicate other processes to join root to end spawn procedure
  } else {
350
    port_name = malloc(1);
351
352
353
354
355
356
357
358
359
360
361
  }

  MPI_Comm_connect(port_name, MPI_INFO_NULL, root, comm, &newintercomm);

  if(myId == root)
    MPI_Comm_free(child);
  free(port_name);
  *child = newintercomm;
}

//--------------PRIVATE MERGE TYPE FUNCTIONS---------------//
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380

/*
 * Se encarga de que el grupo de procesos resultante se 
 * encuentren todos en un intra comunicador, uniendo a
 * padres e hijos en un solo comunicador.
 *
 * Se llama antes de la redistribución de datos.
 *
 * TODO REFACTOR
 */
void proc_adapt_expand(int *numP, int numC, MPI_Comm intercomm, MPI_Comm *comm, int is_children_group) {
  MPI_Comm new_comm = MPI_COMM_NULL;

  MPI_Intercomm_merge(intercomm, is_children_group, &new_comm); //El que pone 0 va primero
  //MPI_Comm_free(intercomm); TODO Nueva redistribucion para estos casos y liberar aqui
  // *intercomm = MPI_COMM_NULL;

  *numP = numC;
  if(*comm != MPI_COMM_WORLD && *comm != MPI_COMM_NULL) {
381
    MPI_Comm_free(comm);
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
  }
  *comm=new_comm;
}


/*
 * Se encarga de que el grupo de procesos resultante se
 * eliminen aquellos procesos que ya no son necesarios.
 * Los procesos eliminados se quedaran como zombies.
 *
 * Se llama una vez ha terminado la redistribución de datos.
 */
void proc_adapt_shrink(int numC, MPI_Comm *comm, int myId) {
  int color = MPI_UNDEFINED;
  MPI_Comm new_comm = MPI_COMM_NULL;

  if(myId < numC) {
      color = 1;  
  }
  MPI_Comm_split(*comm, color, myId, &new_comm);

  if(*comm != MPI_COMM_WORLD && *comm != MPI_COMM_NULL)
    //MPI_Comm_free(comm); FIXME
  *comm=new_comm;
}

408
409
410
411
412
/*
 * 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.
 */
413
void processes_dist(char *argv, int numP_childs, int already_created, int type) {
414
415
416
417
    //int jobId;
    //char *tmp;
    //job_info_msg_t *j_info;
    //slurm_job_info_t last_record;
418
419
420

    int used_nodes=0;
    int *procs_array;
421
    char *hostfile;
422
423

    // Get Slurm job info
424
425
426
427
    //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];
428
429

    //COPY PROGRAM NAME
430
431
    slurm_data->cmd = malloc(strlen(argv) * sizeof(char));
    strcpy(slurm_data->cmd, argv);
432
433

    // GET NEW DISTRIBUTION 
434
    node_dist(type, numP_childs, already_created, &procs_array, &used_nodes);
435
436
    slurm_data->qty_procs = numP_childs;

437
438
    /*
    // CREATE/UPDATE HOSTFILE 
439
    int ptr;
440
    ptr = create_hostfile(tmp, &hostfile);
441
    MPI_Info_create(&(slurm_data->info));
442
443
    MPI_Info_set(slurm_data->info, "hostfile", hostfile);
    free(hostfile);
444
445

    // SET NEW DISTRIBUTION 
iker_martin's avatar
iker_martin committed
446
    fill_hostfile(last_record, ptr, procs_array, used_nodes);
447
    close(ptr);
448
    */
449
    
450
    // CREATE AND SET STRING HOSTFILE
451
    fill_str_hostfile(procs_array, used_nodes, &hostfile);
452
453
    MPI_Info_create(&(slurm_data->info));
    MPI_Info_set(slurm_data->info, "hosts", hostfile);
454
    free(hostfile);
455
    free(procs_array);
456
   
457
458

    // Free JOB INFO
459
    //slurm_free_job_info_msg(j_info);
460
461
}

462
463
464
465
466
467
468
469
470
471
472
473
/*
 * 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.
 */
474
void node_dist(int type, int total_procs, int already_created, int **qty, int *used_nodes) {
475
476
477
478
  int i, asigCores;
  int tamBl, remainder;
  int *procs;

479
  procs = calloc(slurm_data->num_nodes, sizeof(int)); // Numero de procesos por nodo
480
481
482

  /* GET NEW DISTRIBUTION  */
  if(type == 1) { // DIST NODES
483
484
485
    *used_nodes = slurm_data->num_nodes;
    tamBl = total_procs / slurm_data->num_nodes;
    remainder = total_procs % slurm_data->num_nodes;
486
487
488
    for(i=0; i<remainder; i++) {
      procs[i] = tamBl + 1; 
    }
489
    for(i=remainder; i<slurm_data->num_nodes; i++) {
490
491
492
      procs[i] = tamBl; 
    }
  } else if (type == 2) { // DIST CPUs
493
    tamBl = slurm_data->num_cpus / slurm_data->num_nodes;
494
    asigCores = 0;
495
496
    i = *used_nodes = already_created / tamBl;
    remainder = already_created % tamBl;
497

498
499
500
501
502
503
504
505
    //First node could already have existing procs
    if (remainder) {
      procs[i] = asigCores = tamBl - remainder;
      i = (i+1) % slurm_data->num_nodes;
      (*used_nodes)++;
    }

    //Assing tamBl to each node
506
507
508
    while(asigCores+tamBl <= total_procs) {
      asigCores += tamBl;
      procs[i] += tamBl;
509
      i = (i+1) % slurm_data->num_nodes;
510
511
      (*used_nodes)++;
    }
512

513
514
    //Last node could have less procs than tamBl
    if(asigCores < total_procs) { 
515
516
517
      procs[i] += total_procs - asigCores;
      (*used_nodes)++;
    }
518
    if(*used_nodes > slurm_data->num_nodes) *used_nodes = slurm_data->num_nodes;  //FIXME Si ocurre esto no es un error?
519
520
521
522
523
524
525
  }

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

527
528
}

529
/*
530
531
532
 * Crea y devuelve una cadena para ser utilizada por la llave "hosts"
 * al crear procesos e indicar donde tienen que ser creados.
 */
533
void fill_str_hostfile(int *qty, int used_nodes, char **hostfile_str) {
534
535
536
537
  int i=0, len=0;
  char *host;
  hostlist_t hostlist;
  
538
  hostlist = slurm_hostlist_create(slurm_data->nodelist);
539
  while ( (host = slurm_hostlist_shift(hostlist)) && i < used_nodes) {
540
541
542
    if(qty[i] != 0) {
      len = write_str_node(hostfile_str, len, qty[i], host);
    }
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
    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
595
596
597
598
599
600
601
602
603
 * 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.
 */
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
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
}

621
/*
622
 * @deprecated
623
624
625
626
 * 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
627
void fill_hostfile(slurm_job_info_t job_record, int ptr, int *qty, int used_nodes) {
628
629
630
631
632
633
634
635
636
637
638
639
640
  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);
}

641
/*
642
 * @deprecated
643
644
645
646
647
 * 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.
 */
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
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;
}