IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Ignore:
Timestamp:
Jul 4, 2005, 5:35:47 PM (21 years ago)
Author:
eugene
Message:

substantial dev work on scheduler/pcontrol/pclient

Location:
trunk/Ohana/src/opihi/pantasks
Files:
7 added
9 edited

Legend:

Unmodified
Added
Removed

  • trunk/Ohana/src/opihi/pantasks/JobOps.c

    r3140 r4450  
    1 # include "opihi.h"
    21# include "scheduler.h"
    32
    … …  
    98
    109static Job *jobs;
    11 static int   Njobs;
    12 static int   NJOBS;
     10static int  Njobs;
     11static int  NJOBS;
    1312
    1413static char *JobName = dot;
    … …  
    6564
    6665  for (i = 0; i < Njobs; i++) {
    67     fprintf (stderr, "%d: %-15s %5d %20s (%x)\n", Njobs, jobs[i].task.name, jobs[i].JobID, jobs[i].task.argv[0], jobs[i].task.argv);
     66    fprintf (stderr, "%d: %-15s %5d %20s (%x)\n", Njobs, jobs[i].task[0].name, jobs[i].JobID, jobs[i].argv[0], jobs[i].argv);
    6867  }
    6968  return;
    … …  
    7574  int i;
    7675
    77   /* try for an exact match first */
     76  /* return job with matching JobID */
    7877  for (i = 0; i < Njobs; i++) {
    7978    if (jobs[i].JobID == JobID) {
    … …  
    9089
    9190  jobs[Njobs].JobID = NextJobID ();
    92   jobs[Njobs].PID = 0;
    93   jobs[Njobs].task = task[0];
    94 
    95   /* we need our own copy of task[0].argv */
    96   ALLOCATE (jobs[Njobs].task.argv, char *, MAX (jobs[Njobs].task.argc, 1));
    97   for (i = 0; i < jobs[Njobs].task.argc; i++) {
    98     jobs[Njobs].task.argv[i] = strcreate (task[0].argv[i]);
    99   }
    100   jobs[Njobs].task.host = strcreate (task[0].host);
     91  jobs[Njobs].pid = 0;
     92  jobs[Njobs].mode = JOB_LOCAL;
     93  if (task[0].host != NULL) {
     94    jobs[Njobs].mode = JOB_CONTROLLER;
     95  }
     96
     97  /* we need our own copy of task[0].argv
     98   *  argc is the number of valid args, like the usual command line.
     99   *  we allocate one extra element, with value 0 to be passed to execvp
     100   */
     101  jobs[Njobs].argc = task[0].argc;
     102  ALLOCATE (jobs[Njobs].argv, char *, MAX (task[0].argc + 1, 1));
     103  for (i = 0; i < task[0].argc; i++) {
     104    jobs[Njobs].argv[i] = strcreate (task[0].argv[i]);
     105  }
     106  jobs[Njobs].argv[i] = 0;
     107
     108  /* other data from the task is needed by the job
     109     we carry a pointer back to the task.  this means we
     110     cannot modify the task once a job is created, or the changes will
     111     be applied to the existing jobs */
     112
     113  jobs[Njobs].task = task;
     114 
     115  /* if we decide we need to be able to dynamically set task qualities
     116     (like host, timeouts, etc), the we will need to have matched
     117     entries to these quantites in the job structure */
    101118
    102119  Njobs ++;
    … …  
    108125}
    109126
     127void FreeJob (Job *job) {
     128 
     129  int i;
     130
     131  if (job == NULL) return;
     132
     133  if ((job[0].JobID >= 0) || (job[0].JobID < MAX_N_JOBS)) {
     134    JobIDList[job[0].JobID] = FALSE;
     135  }
     136
     137  for (i = 0; i < job[0].argc; i++) {
     138    free (job[0].argv[i]);
     139  }
     140  free (job[0].argv);
     141  return;
     142}
     143
    110144void SetCurrentJob (char *name) {
    111145  JobName = name;
    … …  
    114148char *GetCurrentJob () {
    115149  return (JobName);
    116 }
    117 
    118 void FreeJob (Job *job) {
    119  
    120   int i;
    121 
    122   if (job == NULL) return;
    123 
    124   if ((job[0].JobID >= 0) || (job[0].JobID < MAX_N_JOBS)) {
    125     JobIDList[job[0].JobID] = FALSE;
    126   }
    127 
    128   for (i = 0; i < job[0].task.argc; i++) {
    129     free (job[0].task.argv[i]);
    130   }
    131   free (job[0].task.argv);
    132   free (job[0].task.host);
    133   return;
    134150}
    135151
    … …  
    161177}
    162178
    163 /* this needs to:
    164    1) distinguish local from controller jobs
    165    2) fork the local jobs in the background
    166    3) send the controller jobs to the controller */
    167 
    168179int SubmitJob (Job *job) {
    169180
    170   int i, Nchar;
    171   char *string;
    172 
    173   Nchar = 0;
    174   for (i = 0; i < job[0].task.argc; i++) {
    175     Nchar += strlen (job[0].task.argv[i]) + 1;
    176   }
    177   ALLOCATE (string, char, Nchar);
    178   bzero (string, Nchar);
    179 
    180   strcat (string, job[0].task.argv[0]);
    181   for (i = 1; i < job[0].task.argc; i++) {
    182     strcat (string, " ");
    183     strcat (string, job[0].task.argv[i]);
    184   }
    185  
    186   fprintf (stderr, "executing job %d ...%s...\n", job[0].JobID, string);
    187   free (string);
     181  if (job[0].mode == JOB_LOCAL) {
     182    SubmitLocalJob (job);
     183  } else {
     184    SubmitControllerJob (job);
     185  }
    188186
    189187  /* reset clock for start and poll-test */
    … …  
    196194int CheckJob (Job *job) {
    197195
    198   float f;
    199   f = drand48 ();
    200 
    201   if ((0.00 < f) && (f < 0.25)) {
    202     job[0].state = JOB_BUSY;
    203     job[0].exit_status = -1;
    204     return (JOB_BUSY);
    205   }
    206   if ((0.25 < f) && (f < 0.50)) {
    207     job[0].state = JOB_BUSY;
    208     job[0].exit_status = -1;
    209     return (JOB_CRASH);
    210   }
    211   if ((0.50 < f) && (f < 0.75)) {
    212     job[0].state = JOB_BUSY;
    213     job[0].exit_status = 0;
    214     return (JOB_EXIT);
    215   }
    216   if ((0.75 < f) && (f < 1.00)) {
    217     job[0].state = JOB_BUSY;
    218     job[0].exit_status = 1;
    219     return (JOB_EXIT);
    220   }
    221 }
     196  /* add checks for timeouts */
     197
     198  if (job[0].mode == JOB_LOCAL) {
     199    CheckLocalJob (job);
     200  } else {
     201    CheckControllerJob (job);
     202  }
     203  return (job[0].state);
     204}

  • trunk/Ohana/src/opihi/pantasks/LocalJob.c

    r3392 r4450  
     1# include "scheduler.h"
    12
    2 /* local jobs are forked in the background */
     3/* this could be written a just a one-way pipe */
     4int SubmitLocalJob (Job *job) {
    35
    4 SubmitLocalJob (Job *job) {
     6  int status, pid;
     7  int stdout_fd[2], stderr_fd[2];
    58
    6   /*
    7      construct the command line
    8      fork the command, get back the PID
    9      increment local job counter
    10   */
     9  bzero (stdout_fd, 2*sizeof(int));
     10  bzero (stderr_fd, 2*sizeof(int));
    1111
     12  if (pipe (stdout_fd) < 0) goto pipe_error;
     13  if (pipe (stderr_fd) < 0) goto pipe_error;
     14
     15  pid = fork ();
     16  if (!pid) { /* must be child process */
     17    fprintf (stderr, "starting controller connection\n");
     18
     19    /* close the other ends of the pipes */
     20    close (stdout_fd[0]);
     21    close (stderr_fd[0]);
     22
     23    /* tie our ends of the pipes to stdin, stdout, stderr */
     24    dup2 (stdout_fd[1], STDOUT_FILENO);
     25    dup2 (stderr_fd[1], STDERR_FILENO);
     26
     27    /* set all three unblocking */
     28    setvbuf (stdout, (char *) NULL, _IONBF, BUFSIZ);
     29    setvbuf (stderr, (char *) NULL, _IONBF, BUFSIZ);
     30
     31    status = execvp (job[0].argv[0], job[0].argv);
     32    exit (1);
     33  }
     34
     35  /* close the other ends of the pipes */
     36  close (stdout_fd[1]); stdout_fd[1] = 0;
     37  close (stderr_fd[1]); stderr_fd[1] = 0;
     38
     39  /* make the pipes non-blocking */
     40  fcntl (stdout_fd[0], F_SETFL, O_NONBLOCK);
     41  fcntl (stderr_fd[0], F_SETFL, O_NONBLOCK);
     42
     43  job[0].stdout_fd = stdout_fd[0];
     44  job[0].stderr_fd = stderr_fd[0];
     45  job[0].pid = pid;
     46
     47  return (TRUE);
     48
     49pipe_error:
     50  perror ("pipe error:");
     51  if (stdout_fd[0] != 0) close (stdout_fd[0]);
     52  if (stdout_fd[1] != 0) close (stdout_fd[1]);
     53  if (stderr_fd[0] != 0) close (stderr_fd[0]);
     54  if (stderr_fd[1] != 0) close (stderr_fd[1]);
     55  return (FALSE);
    1256}
    1357
    14 CheckLocalJob (Job *job) {
     58/* update current state, drain stdout/stderr buffers */
     59int CheckLocalJob (Job *job) {
    1560
    16   /*
    17      
     61  int Nread;
     62
     63  // XXX do something useful with exit status?
     64  CheckLocalJobStatus (job);
     65
     66  /* read stdout buffer */
     67  Nread = ReadtoIOBuffer (&job[0].stdout, job[0].stdout_fd);
     68  switch (Nread) {
     69    case -2:  /* error in read (programming error?  system level error?) */
     70      fprintf (stderr, "serious IO error\n");
     71      exit (2);
     72    case -1:  /* no data in pipe */
     73      break;
     74    case 0:   /* pipe is closed */
     75      /** change child state? **/
     76      break;
     77    default:  /* data in pipe */
     78      break;
     79  }
     80 
     81  /* read stderr buffer */
     82  Nread = ReadtoIOBuffer (&job[0].stderr, job[0].stderr_fd);
     83  switch (Nread) {
     84    case -2:  /* error in read (programming error?  system level error?) */
     85      fprintf (stderr, "serious IO error\n");
     86      exit (2);
     87    case -1:  /* no data in pipe */
     88      break;
     89    case 0:   /* pipe is closed */
     90      /** change child state? **/
     91      break;
     92    default:  /* data in pipe */
     93      break;
     94  }
     95  return (TRUE);
     96}
     97
     98int CheckLocalJobStatus (Job *job) {
     99
     100  int result, waitstatus;
     101
     102  /* check local job status */
     103  result = waitpid (job[0].pid, &waitstatus, WNOHANG);
     104  switch (result) {
     105    case -1:  /* error with waitpid */
     106      switch (errno) {
     107        case ECHILD:
     108          fprintf (stderr, "unknown PID, not a child proc\n");
     109          fprintf (stderr, "did process already exit?  programming error?\n");
     110          job[0].state = JOB_NONE;
     111          job[0].exit_status = 0;
     112          return (FALSE);
     113        case EINVAL:
     114          fprintf (stderr, "error EINVAL (waitpid): programming error\n");
     115          exit (1);
     116        case EINTR:
     117          fprintf (stderr, "error EINTR (waitpid): programming error\n");
     118          exit (1);
     119        default:
     120          fprintf (stderr, "unknown error for waitpid (%d): programming error\n", errno);
     121          exit (1);
     122      }
     123      break;
     124     
     125    case 0:  /* process not exited */
     126      job[0].state = JOB_BUSY;
     127      job[0].exit_status = 0;
     128      return (TRUE);
     129
     130    default:
     131      if (result != job[0].pid) {
     132        fprintf (stderr, "waitpid error: mis-matched PID (%d vs %d).  programming error\n", result, job[0].pid);
     133        exit (1);
     134      }
     135     
     136      if (WIFEXITED(waitstatus)) {
     137        job[0].state = JOB_EXIT;
     138        job[0].exit_status = WEXITSTATUS(waitstatus);
     139      }
     140      if (WIFSIGNALED(waitstatus)) {
     141        job[0].state = JOB_CRASH;
     142        job[0].exit_status = WTERMSIG(waitstatus);
     143      }
     144      if (WIFSTOPPED(waitstatus)) {
     145        fprintf (stderr, "waitpid returns 'stopped': programming error\n");
     146        exit (1);
     147      }
     148  }
     149  return;
     150}

  • trunk/Ohana/src/opihi/pantasks/Makefile

    r3525 r4450  
    2525# sched user commands and support functions ########################
    2626
    27 sched = \
     27funcs = \
     28$(SDIR)/CheckJobs.$(ARCH).o \
     29$(SDIR)/CheckSystem.$(ARCH).o \
     30$(SDIR)/CheckTasks.$(ARCH).o \
     31$(SDIR)/ControllerOps.$(ARCH).o \
     32$(SDIR)/LocalJob.$(ARCH).o \
     33$(SDIR)/JobOps.$(ARCH).o \
     34$(SDIR)/TaskOps.$(ARCH).o \
     35$(SDIR)/IOBufferOps.$(ARCH).o \
     36$(SDIR)/memstr.$(ARCH).o \
     37$(SDIR)/init.$(ARCH).o
     38
     39cmds = \
     40$(SDIR)/controller.$(ARCH).o \
    2841$(SDIR)/run.$(ARCH).o \
     42$(SDIR)/scheduler.$(ARCH).o \
    2943$(SDIR)/task.$(ARCH).o \
    3044$(SDIR)/task_command.$(ARCH).o \
    3145$(SDIR)/task_host.$(ARCH).o \
    3246$(SDIR)/task_macros.$(ARCH).o \
    33 $(SDIR)/task_periods.$(ARCH).o \
    34 $(SDIR)/JobOps.$(ARCH).o \
    35 $(SDIR)/TaskOps.$(ARCH).o \
    36 $(SDIR)/init.$(ARCH).o \
    37 $(SDIR)/scheduler.$(ARCH).o
     47$(SDIR)/task_periods.$(ARCH).o
    3848
    3949libs = \
    … …  
    4959        @echo done
    5060
    51 $(BIN)/scheduler.$(ARCH) : $(sched) $(libs)
     61$(BIN)/scheduler.$(ARCH) : $(funcs) $(cmds) $(libs)
    5262
    5363install: $(DESTBIN)/scheduler

  • trunk/Ohana/src/opihi/pantasks/TaskOps.c

    r3140 r4450  
    108108
    109109  tasks[Ntasks].host = NULL;
     110  tasks[Ntasks].host_required = FALSE;
    110111
    111112  tasks[Ntasks].argc = 0;
    … …  
    124125  tasks[Ntasks].poll_period = 1.0;
    125126  tasks[Ntasks].timeout_period = 1.0;
     127
     128  /* init task timer (is reset by 'run') */ 
     129  gettimeofday (&tasks[Ntasks].last, (void *) NULL);
    126130
    127131  Ntasks ++;
    … …  
    155159  gettimeofday (timer, (void *) NULL);
    156160}
     161
     162/* start the clock for all tasks */
     163void InitTaskTimers () {
     164
     165  Task *task;
     166
     167  while ((task = NextTask ()) != NULL) {
     168    gettimeofday (&task[0].last, (void *) NULL);
     169 }
     170}

  • trunk/Ohana/src/opihi/pantasks/init.c

    r3140 r4450  
    22# include "scheduler.h"
    33
     4int controller      PROTO((int, char **));
    45int task            PROTO((int, char **));
    56int task_host       PROTO((int, char **));
    … …  
    89int task_periods    PROTO((int, char **));
    910int run             PROTO((int, char **));
     11int stop            PROTO((int, char **));
    1012
    1113static Command cmds[] = { 
    12   {"task",      task,         "define a schedulable task"},
    13   {"host",      task_host,    "define host machine for a task"},
    14   {"task.exit", task_macros,  "define exit macros for a task"},
    15   {"task.exec", task_macros,  "define pre-exec macro for a task"},
    16   {"command",   task_command, "define executed command for a task"},
    17   {"periods",   task_periods, "define time scales for a task"},
    18   {"run",       run,          "run the scheduler"},
     14  {"controller", controller,   "controller commands"},
     15  {"task",       task,         "define a schedulable task"},
     16  {"host",       task_host,    "define host machine for a task"},
     17  {"task.exit",  task_macros,  "define exit macros for a task"},
     18  {"task.exec",  task_macros,  "define pre-exec macro for a task"},
     19  {"command",    task_command, "define executed command for a task"},
     20  {"periods",    task_periods, "define time scales for a task"},
     21  {"run",        run,          "run the scheduler"},
     22  {"stop",       stop,         "stop the scheduler"},
    1923};
    2024

  • trunk/Ohana/src/opihi/pantasks/run.c

    r3140 r4450  
    1 # include "basic.h"
    21# include "scheduler.h"
    32
    43int run (int argc, char **argv) {
    5 
    6   Job *job;
    7   Task *task;
    8   Macro *macro;
    9   int i, found, status;
    10   int Ntest;
    114
    125  if (argc != 1) {
    … …  
    158  }
    169
    17   /* start the clock for all tasks */
    18   while ((task = NextTask ()) != NULL) {
    19     gettimeofday (&task[0].last, (void *) NULL);
     10  InitTaskTimers ();
     11  rl_event_hook = CheckSystem;
     12  rl_set_keyboard_input_timeout (1000000);
     13
     14  return (TRUE);
     15}
     16
     17int stop (int argc, char **argv) {
     18
     19  if (argc != 1) {
     20    fprintf (stderr, "USAGE: stop\n");
     21    return (FALSE);
    2022  }
    2123
    22   Ntest = 0;
     24  rl_event_hook = NULL;
     25  rl_set_keyboard_input_timeout (1000000);
    2326
    24   /* loop forever, checking for completed jobs and ready tasks */
    25   while (1) {
    26     if (Ntest > 5) {
    27       ListJobs ();
    28       Ntest = 0;
    29     }
    30     usleep (10000);
    31     Ntest ++;
    32 
    33     /** test all tasks: ready to test? ready to run? **/
    34     while ((task = NextTask ()) != NULL) {
    35 
    36       /* ready to test? : check exec period */
    37       if (GetTaskTimer(task[0].last) < task[0].exec_period) continue;
    38 
    39       SetCurrentTask (task[0].name);
    40       fprintf (stderr, "trying task %s\n", task[0].name);
    41 
    42       /* ready to run? : run task.exec macro */
    43       if (task[0].exec != NULL) {
    44         status = exec_loop (task[0].exec);
    45         if (!status) continue;
    46       }
    47 
    48       /* is task valid?  check state of task.(argc, argv) */
    49       /*** ADD CODE HERE ***/
    50 
    51       /* construct job from task */
    52       job = CreateJob (task);
    53 
    54       /* execute job - XXX add status test */
    55       SubmitJob (job);
    56 
    57       /* reset timer on task (don't do this if Create/Submit fails)*/
    58       gettimeofday (&task[0].last, (void *) NULL);
    59     }
    60 
    61     /** test all jobs: ready to test?  finished? **/
    62     while ((job = NextJob ()) != NULL) {
    63 
    64       /* check for timeout */
    65       if (GetTaskTimer(job[0].start) >= job[0].task.timeout_period) {
    66         fprintf (stderr, "timeout on %s\n", job[0].task.name);
    67         /* run task.timeout macro, if it exists */
    68         if (job[0].task.timeout != NULL) {
    69           exec_loop (job[0].task.timeout);
    70         }
    71         DeleteJob (job);
    72         continue;
    73       }
    74 
    75       /* check poll period (ready to run again?) */
    76       if (GetTaskTimer(job[0].last) < job[0].task.poll_period) continue;
    77 
    78       /* check current status */
    79       status = CheckJob (job);
    80       switch (status) {
    81         case JOB_BUSY:
    82           fprintf (stderr, "job %s (%d) busy\n", job[0].task.name, job[0].JobID);
    83           break;
    84 
    85         case JOB_CRASH:
    86           fprintf (stderr, "job %s (%d) crash\n", job[0].task.name, job[0].JobID);
    87           /* run task.crash macro, if it exists */
    88           if (job[0].task.crash != NULL) {
    89             exec_loop (job[0].task.crash);
    90           }
    91           DeleteJob (job);
    92           continue;
    93           break;
    94 
    95         case JOB_EXIT:
    96           fprintf (stderr, "job %s (%d) exit\n", job[0].task.name, job[0].JobID);
    97           /* run corresponding task.exit macro, if it exists */
    98           macro = job[0].task.def;
    99           for (i = 0; i < job[0].task.Nexit; i++) {
    100             if (job[0].exit_status == atoi(job[0].task.exit[i][0].name)) {
    101               macro = job[0].task.exit[i];
    102               break;
    103             }
    104           }
    105           if (macro != NULL) exec_loop (macro);
    106           DeleteJob (job);
    107           continue;
    108           break;
    109 
    110         default:
    111           fprintf (stderr, "unknown exit status\n");
    112           /** do something more useful here ?? **/
    113           break;
    114       }
    115 
    116       /* reset polling clock */
    117       SetTaskTimer (&job[0].last);
    118     }
    119   }
    12027  return (TRUE);
    12128}

  • trunk/Ohana/src/opihi/pantasks/scheduler.c

    r2598 r4450  
    4949  rl_readline_name = opihi_name;
    5050  rl_attempted_completion_function = command_completer;
     51  rl_event_hook = NULL;
     52  rl_set_keyboard_input_timeout (1000000);
    5153
    5254  set_str_variable ("HISTORY", opihi_history);

  • trunk/Ohana/src/opihi/pantasks/task_command.c

    r2598 r4450  
    4141  return (TRUE);
    4242}
    43 
    44 
    45 /**
    46     careful with this: the command is supposed to be realized
    47     for the Job, not the Task (at execution)
    48     the code is right, but who calls it when needs to be clarified.
    49  **/
    50 

  • trunk/Ohana/src/opihi/pantasks/task_host.c

    r2598 r4450  
    44int task_host (int argc, char **argv) {
    55
     6  int N, RequiredHost;
    67  Task *task;
    78  char *taskname;
    89
     10  RequiredHost = FALSE;
     11  if (N = get_argument (argc, argv, "-required")) {
     12    remove_argument (N, &argc, argv);
     13    RequiredHost = TRUE;
     14  }
     15
    916  if (argc != 2) {
    10     fprintf (stderr, "USAGE: host <name>\n");
    11     fprintf (stderr, "  (define host machine for this task (or 'none'))\n");
     17    fprintf (stderr, "USAGE: host <name> [-required]\n");
     18    fprintf (stderr, "  define host machine for this task\n");
     19    fprintf (stderr, "  -required flags indicates controller must use this host\n");
     20    fprintf (stderr, "  value of 'local' for host indicates process not using controller\n");
     21    fprintf (stderr, "  value of 'none' for host indicates controller may assign at will\n");
    1222    return (FALSE);
    1323  }
    … …  
    2333    return (FALSE);
    2434  }
     35  task[0].host_required = RequiredHost;
    2536
    2637  if (task[0].host != NULL) free (task[0].host);
    2738  task[0].host = NULL;
    2839
    29   if (!strcasecmp (argv[1], "NONE")) return (TRUE);
     40  if (!strcasecmp (argv[1], "LOCAL")) return (TRUE);
    3041
    3142  task[0].host = strcreate (argv[1]);
Note: See TracChangeset for help on using the changeset viewer.