ProcessDist.c 20.6 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
  int qty_procs, result_procs;
22
23
  MPI_Info info;
  int type_creation;
24
  int spawn_is_single;
25
26
};

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

36

37
38
39
40
//--------------PRIVATE SPAWN TYPE DECLARATIONS---------------//
void* thread_work(void* creation_data_arg);

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

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);
45
int create_processes(int myId, int root, MPI_Comm *child, MPI_Comm comm);
46

47
void node_dist(slurm_job_info_t job_record, int type, int total_procs, int already_created, int **qty, int *used_nodes);
48

49
50
51
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);

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

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

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

79
  spawn_thread = pthread_self();
80
  slurm_data->type_creation = type_creation;
81
  slurm_data->spawn_is_single = spawn_is_single;
82
83
84
85
86
  slurm_data->result_procs = numC;
  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;
87
      already_created = numP;
88
89
    }
  }
90
  pthread_mutex_init(&spawn_mutex,NULL);
91

92
  if(type_creation == COMM_SPAWN_SERIAL || slurm_data->type_creation == COMM_SPAWN_MERGE) {
93

94
    if(myId == root) {
95
      processes_dist(argv, spawn_qty, already_created, type_dist);
96
97
98
    } else {
      slurm_data->cmd = malloc(1 * sizeof(char));
      slurm_data->info = MPI_INFO_NULL;
99
    }
100
101
102
103

    // WORK
    generic_spawn(myId, root, slurm_data->spawn_is_single, child, comm);
    // END WORK
104
105
106
107

    if(myId == root && slurm_data->info != MPI_INFO_NULL) {
      MPI_Info_free(&(slurm_data->info));
    }
108
    pthread_mutex_destroy(&spawn_mutex);
109
    free(slurm_data->cmd);
110
    free(slurm_data);
111

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

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

136

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

143
  if(slurm_data->type_creation == COMM_SPAWN_PTHREAD || slurm_data->type_creation == COMM_SPAWN_MERGE_PTHREAD) {
144
145
146
147
148
149
150
151

    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) {
152
      int state=-10;
153
154
155

      //printf("[%d][3] Test min\n", myId); fflush(stdout);
      //pthread_mutex_lock(&spawn_mutex);
156
      MPI_Allreduce(&commState, &state, 1, MPI_INT, MPI_MIN, comm);
157
158
159
      //pthread_mutex_unlock(&spawn_mutex);

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

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

    } else if (slurm_data->spawn_is_single) {
170
171

      //pthread_mutex_lock(&spawn_mutex);
172
      MPI_Bcast(&commState, 1, MPI_INT, root, comm);
173
174
      //pthread_mutex_unlock(&spawn_mutex);
      int threads_not_spawned = pthread_equal(pthread_self(), spawn_thread);
175
176
177

      // Non-root processes join root to finalize the spawn
      // They also must join if the application has ended its work
178
      if(commState == MAL_SPAWN_SINGLE_START) { 
179
        commState = MAL_SPAWN_SINGLE_PENDING;
180
181

        if(myId != root && threads_not_spawned) {
182
183
184
185
186
187
188
189
190
          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;

191
          if(pthread_create(&spawn_thread, NULL, thread_work, (void *)creation_data)) {
192
193
194
195
196
            printf("Error al crear el hilo de apoyo\n");
            MPI_Abort(MPI_COMM_WORLD, -1);
            return -1;
          }
	}
197
198
      }

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

211
    } else {
212
213
214
      printf("Error Check spawn: Configuracion invalida\n");
      MPI_Abort(MPI_COMM_WORLD, -1);
      return -10;
215
216
    }
  } else {  
217
    return commState;
218
  }
219

220
  //Free memory
221
  if(myId == root && slurm_data->info != MPI_INFO_NULL) {
222
223
224
    MPI_Info_free(&(slurm_data->info));
  }
  free(slurm_data->cmd);
225
  free(slurm_data);
226
227
228
229
  pthread_mutex_destroy(&spawn_mutex);
  spawn_thread = pthread_self();

  *real_time=end_time;
230

231
  return commState;
232
233
}

234
235
236
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
/*
 * 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;
}

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
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306

//--------------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);
  }
307
308
309
310
  pthread_mutex_lock(&spawn_mutex);
    commState = MAL_SPAWN_COMPLETED; 
    end_time = MPI_Wtime();
  pthread_mutex_unlock(&spawn_mutex);
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
}

/*
 * 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 {
345
    port_name = malloc(1);
346
347
348
349
350
351
352
353
354
355
356
  }

  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---------------//
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375

/*
 * 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) {
376
    MPI_Comm_free(comm);
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
  }
  *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;
}

403
404
405
406
407
/*
 * 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.
 */
408
void processes_dist(char *argv, int numP_childs, int already_created, int type) {
409
    int jobId;
410
411
412
413
414
415
    char *tmp;
    job_info_msg_t *j_info;
    slurm_job_info_t last_record;

    int used_nodes=0;
    int *procs_array;
416
    char *hostfile;
417
418
419
420
421
422
423
424

    // 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
425
426
    slurm_data->cmd = malloc(strlen(argv) * sizeof(char));
    strcpy(slurm_data->cmd, argv);
427
428

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

432
433
    /*
    // CREATE/UPDATE HOSTFILE 
434
    int ptr;
435
    ptr = create_hostfile(tmp, &hostfile);
436
    MPI_Info_create(&(slurm_data->info));
437
438
    MPI_Info_set(slurm_data->info, "hostfile", hostfile);
    free(hostfile);
439
440

    // SET NEW DISTRIBUTION 
iker_martin's avatar
iker_martin committed
441
    fill_hostfile(last_record, ptr, procs_array, used_nodes);
442
    close(ptr);
443
    */
444
    
445
    // TEST 
446
447
448
    fill_str_hostfile(last_record, procs_array, used_nodes, &hostfile);
    MPI_Info_create(&(slurm_data->info));
    MPI_Info_set(slurm_data->info, "hosts", hostfile);
449
    free(hostfile);
450
    free(procs_array);
451
   
452
453
454
455
456

    // Free JOB INFO
    slurm_free_job_info_msg(j_info); 
}

457
458
459
460
461
462
463
464
465
466
467
468
/*
 * 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.
 */
469
void node_dist(slurm_job_info_t job_record, int type, int total_procs, int already_created, int **qty, int *used_nodes) {
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
  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;
490
491
    i = already_created / tamBl;
    *used_nodes = already_created / tamBl;
492
493
494
495
496
497
498

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

500
501
502
503
    if(asigCores < total_procs) {
      procs[i] += total_procs - asigCores;
      (*used_nodes)++;
    }
504
    if(*used_nodes > job_record.num_nodes) *used_nodes = job_record.num_nodes;  //FIXME Si ocurre esto no es un error?
505
506
507
508
509
510
511
  }

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

513
514
}

515
/*
516
517
518
519
520
521
522
523
524
525
 * 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) {
526
527
528
    if(qty[i] != 0) {
      len = write_str_node(hostfile_str, len, qty[i], host);
    }
529
530
531
532
533
534
535
536
537
538
539
540
541
542
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
    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
582
583
584
585
586
587
588
589
590
 * 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.
 */
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
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
}

608
/*
609
 * @deprecated
610
611
612
613
 * 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
614
void fill_hostfile(slurm_job_info_t job_record, int ptr, int *qty, int used_nodes) {
615
616
617
618
619
620
621
622
623
624
625
626
627
  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);
}

628
/*
629
 * @deprecated
630
631
632
633
634
 * 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.
 */
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
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;
}