source: TOOLS/SURPACK_IPSL/glost_launch.c @ 3624

#include <mpi.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <getopt.h>
#include <signal.h>


#ifdef __HAVE_LIBCCC_USER__
#include "ccc_user.h"
#else
static int ccc_tremain(double *tremain)
{
  tremain[0] = 0. ;
  return EXIT_FAILURE;
}
#endif

#define GLOST_MAXLEN 4096
#define GLOST_MASTER 0
#define TAG_AVAIL 1
#define TAG_TSK   2

static int glost_no_new_task=0;

void read_options(int argc, char *argv[],double *user_tremain);
void read_and_exec(char *filename, double user_tremain);
void read_and_send(int slaves, char *filename, double user_tremain);
void recv_and_exec(int rank);
void set_sigaction();

static void set_glost_no_new_task(int signum)
{
  fprintf(stderr,"#signal %i received: no new task will be submitted\n",signum);
  glost_no_new_task=1;
}

static void read_next_command(FILE *fl,char *tsk, double user_tremain)
{
  int r;
  double tremain;

  tsk[0]='\0'; 

  r = ccc_tremain(&tremain);
  
  fprintf(stderr,"#time remaining = %g\n", tremain);
  fprintf(stderr,"#user_tremain = %g\n", user_tremain);
  // fprintf(stderr,"#r=%d\n", r);
                
  r = (!r && tremain < user_tremain);
  if (r) 
    fprintf(stderr,"#time remaining = %g < %g : do not launch new tasks.\n",
            tremain,user_tremain);

  if (glost_no_new_task) r = 1;   
  if (r) return;

  /* if(!fgets(tsk,GLOST_MAXLEN,fl)) tsk[0]='\0'; */
  /* if(tsk[strlen(tsk)-1]=='\n') tsk[strlen(tsk)-1]='\0'; */

  do {
    if(!fgets(tsk,GLOST_MAXLEN,fl)){
      tsk[0] = '\0';
      break;
    } 
    r = strlen(tsk);
  } while (r < 2);

  if(tsk[r-1]=='\n') tsk[r-1]='\0';

}

static void exec_command(char *tsk, int rank)
{
  double t;
  int r;

  t = MPI_Wtime();  
  r = system(tsk);
  t = MPI_Wtime() - t;
  if ( WIFEXITED(r) ) 
    {
      r = WEXITSTATUS(r); 
      fprintf(stderr,"#executed by process\t%d in %gs\twith status\t%d : %s\n",
              rank,t,r,tsk);
    }
  else { 
    if( WIFSIGNALED(r) ){
      r = WTERMSIG(r); 
      fprintf(stderr,"#executed by process\t%d in %gs\tkilled by signal %d: %s\n",
              rank,t,r,tsk);
    }
  }
}


int main(int argc,char **argv)
{
  FILE *fl;
  int rank,size;
  double user_tremain;

  MPI_Init(&argc,&argv);
  MPI_Comm_rank(MPI_COMM_WORLD,&rank);
  MPI_Comm_size(MPI_COMM_WORLD,&size);  

  set_sigaction();
  if (rank == GLOST_MASTER){
    printf("master is %d , nb slaves: %d\n", GLOST_MASTER,size-1);      
    read_options(argc,argv,&user_tremain);
  }

  if (size == 1)
    read_and_exec(argv[optind],user_tremain);
  else if(size != 1 && rank == GLOST_MASTER)
    read_and_send(size-1,argv[optind],user_tremain);
  else
    recv_and_exec(rank);

  MPI_Finalize();
}

void read_options(int argc, char *argv[],double *user_tremain )
{
  int r;
  int longoptind;
  char opt[256];

  user_tremain[0] = -1.;
  static struct option longopts[]=
    {
      {"time_remaining",required_argument,NULL,'R'},
      {0,0,0,0}
    };

  while (r = getopt_long(argc, argv,"R:",longopts,&longoptind))
    {
      if (r == -1) break; 
      if (r == 0){              /* long option */
        fprintf(stderr,"r %c\n",longopts[longoptind].val);      
        r = longopts[longoptind].val ;
      }
        
      switch(r)
        {
        case 'R': 
          *user_tremain = strtod(optarg,NULL);
#ifndef   __HAVE_LIBCCC_USER__
          fprintf(stderr,"WARNING option -R %g ignored: not compiled with libccc_user\n",*user_tremain);
#endif
          break;

        case '?':
          fprintf(stderr,"WARNING option %c ignored: unknown option\n",optopt); 
          break;

        default:
          abort();
          break;
        }
    }
}

void read_and_exec(char *filename, double user_tremain )
{
  FILE *fl;
  char tsk[GLOST_MAXLEN];
  double t;

  fl = fopen(filename,"r");
  do {
    read_next_command(fl,tsk,user_tremain);
    if(!strlen(tsk))
      break;
    exec_command(tsk,GLOST_MASTER);
  } while (strlen(tsk));
  fclose(fl);
}

void read_and_send(int slaves,char *filename, double user_tremain )
{
  int i, num, r;
  MPI_Status status;
  FILE *fl;
  char tsk[GLOST_MAXLEN];

  fl = fopen(filename,"r");
  i=0;
  do {
    read_next_command(fl,tsk,user_tremain);
    
    MPI_Recv(&num,1,MPI_INT,MPI_ANY_SOURCE,
             TAG_AVAIL,MPI_COMM_WORLD,&status);
    MPI_Send(tsk,GLOST_MAXLEN,MPI_CHAR,num,
             TAG_TSK,MPI_COMM_WORLD);
    if (!strlen(tsk)) /* terminate command sent */
      i++;
  } while (i < slaves);
  fclose(fl);
}

void recv_and_exec(int rank)
{
  char tsk[GLOST_MAXLEN];  
  MPI_Status status;  

  do {
    MPI_Send(&rank,1,MPI_INT,GLOST_MASTER,
             TAG_AVAIL,MPI_COMM_WORLD); 
    MPI_Recv(tsk,GLOST_MAXLEN,MPI_CHAR,GLOST_MASTER,
             TAG_TSK,MPI_COMM_WORLD,&status); 
    if(!strlen(tsk)) /* terminate command */
      break;
    exec_command(tsk,rank);
  } while (strlen(tsk));
}


/* signal handling */
void set_sigaction()
{
  struct sigaction new_action;

  new_action.sa_handler = set_glost_no_new_task ;
  sigemptyset(&new_action.sa_mask);
  new_action.sa_flags = SA_RESETHAND;

  if (sigaction(SIGUSR1,&new_action,NULL) < 0)
    fprintf(stderr,"#WARNING sigaction failed: no SIGQUIT support\n");
}