IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Ignore:
Timestamp:
Aug 2, 2005, 9:40:50 AM (21 years ago)
Author:
eugene
Message:

psched bugfixes, moved IOBuffers and queues

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

Legend:

Unmodified
Added
Removed

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

    r4693 r4697  
    2929        /* XXX this will break on 0 values in output streams */
    3030        if (VerboseMode()) fprintf (stderr, "job %s (%d) crash\n", job[0].task[0].name, job[0].JobID);
    31         set_str_variable ("stdout", job[0].stdout.buffer);
    32         set_str_variable ("stderr", job[0].stderr.buffer);
     31        PushNamedQueue ("stdout", job[0].stdout.buffer);
     32        PushNamedQueue ("stderr", job[0].stderr.buffer);
    3333        if (job[0].task[0].crash != NULL) {
    3434          exec_loop (job[0].task[0].crash);
    … …  
    4040      case JOB_EXIT:
    4141        if (VerboseMode()) fprintf (stderr, "job %s (%d) exit\n", job[0].task[0].name, job[0].JobID);
    42         set_str_variable ("stdout", job[0].stdout.buffer);
    43         set_str_variable ("stderr", job[0].stderr.buffer);
     42        PushNamedQueue ("stdout", job[0].stdout.buffer);
     43        PushNamedQueue ("stderr", job[0].stderr.buffer);
    4444        /* run corresponding task[0].exit macro, if it exists */
    4545        macro = job[0].task[0].def;

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

    r4693 r4697  
    2222      if (!status) continue;
    2323    }
     24    if (!ValidateTask (task, TRUE)) continue;
    2425
    2526    /* construct job from task */

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

    r4693 r4697  
    33# define CONNECT_TIMEOUT 300
    44
     5/* local static variables to hold the connection to the controller */
    56static int status = FALSE;
    67static int stdin_cntl, stdout_cntl, stderr_cntl;
    … …  
    89static IOBuffer stderr_buffer;
    910
     11/* test if the controller is running */
    1012int CheckControllerStatus () {
    1113  return (status);
    1214}
    1315
     16/* check job / get output if done */
    1417int CheckControllerJob (Job *job) {
    1518
    … …  
    2326}
    2427
     28/* ask controller about job status */
    2529int CheckControllerJobStatus (Job *job) {
    2630

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

    r4693 r4697  
    11# include "psched.h"
     2
     3/* local jobs are forked in the background
     4   we might need to limit the maximum number of local jobs.
     5   should we have a queue/stack of pending local jobs, much
     6   like controller? */
     7
     8/* update current state, drain stdout/stderr buffers */
     9int CheckLocalJob (Job *job) {
     10
     11  int Nread;
     12
     13  // XXX do something useful with exit status?
     14  CheckLocalJobStatus (job);
     15
     16  if ((job[0].state == JOB_EXIT) || (job[0].state == JOB_CRASH)) {
     17    EmptyIOBuffer (&job[0].stdout, 10, job[0].stdout_fd);
     18    EmptyIOBuffer (&job[0].stderr, 10, job[0].stderr_fd);
     19    close (job[0].stdout_fd);
     20    close (job[0].stderr_fd);
     21  } else {
     22    /* read stdout buffer */
     23    Nread = ReadtoIOBuffer (&job[0].stdout, job[0].stdout_fd);
     24    switch (Nread) {
     25      case -2:  /* error in read (programming error?  system level error?) */
     26        fprintf (stderr, "serious IO error\n");
     27        exit (2);
     28      case -1:  /* no data in pipe */
     29      case 0:   /* pipe is closed, change child state? **/
     30      default:  /* data in pipe */
     31        break;
     32    }
     33 
     34    /* read stderr buffer */
     35    Nread = ReadtoIOBuffer (&job[0].stderr, job[0].stderr_fd);
     36    switch (Nread) {
     37      case -2:  /* error in read (programming error?  system level error?) */
     38        fprintf (stderr, "serious IO error\n");
     39        exit (2);
     40      case -1:  /* no data in pipe */
     41      case 0:   /* pipe is closed, change child state? **/
     42      default:  /* data in pipe */
     43        break;
     44    }
     45  }
     46  return (TRUE);
     47}
     48
     49int CheckLocalJobStatus (Job *job) {
     50
     51  int result, waitstatus;
     52
     53  /* check local job status */
     54  result = waitpid (job[0].pid, &waitstatus, WNOHANG);
     55  switch (result) {
     56    case -1:  /* error with waitpid */
     57      switch (errno) {
     58        case ECHILD:
     59          fprintf (stderr, "unknown PID, not a child proc\n");
     60          fprintf (stderr, "did process already exit?  programming error?\n");
     61          job[0].state = JOB_NONE;
     62          job[0].exit_status = 0;
     63          return (FALSE);
     64        case EINVAL:
     65          fprintf (stderr, "error EINVAL (waitpid): programming error\n");
     66          exit (1);
     67        case EINTR:
     68          fprintf (stderr, "error EINTR (waitpid): programming error\n");
     69          exit (1);
     70        default:
     71          fprintf (stderr, "unknown error for waitpid (%d): programming error\n", errno);
     72          exit (1);
     73      }
     74      break;
     75     
     76    case 0:  /* process not exited */
     77      job[0].state = JOB_BUSY;
     78      job[0].exit_status = 0;
     79      return (TRUE);
     80
     81    default:
     82      if (result != job[0].pid) {
     83        fprintf (stderr, "waitpid error: mis-matched PID (%d vs %d).  programming error\n", result, job[0].pid);
     84        exit (1);
     85      }
     86      if (WIFEXITED(waitstatus)) {
     87        job[0].state = JOB_EXIT;
     88        job[0].exit_status = WEXITSTATUS(waitstatus);
     89      }
     90      if (WIFSIGNALED(waitstatus)) {
     91        job[0].state = JOB_CRASH;
     92        job[0].exit_status = WTERMSIG(waitstatus);
     93      }
     94      if (WIFSTOPPED(waitstatus)) {
     95        fprintf (stderr, "waitpid returns 'stopped': programming error\n");
     96        exit (1);
     97      }
     98  }
     99  return (FALSE);
     100}
    2101
    3102/* this could be written a just a one-way pipe */
    … …  
    58157}
    59158
    60 /* update current state, drain stdout/stderr buffers */
    61 int CheckLocalJob (Job *job) {
    62 
    63   int Nread;
    64 
    65   // XXX do something useful with exit status?
    66   CheckLocalJobStatus (job);
    67 
    68   if ((job[0].state == JOB_EXIT) || (job[0].state == JOB_CRASH)) {
    69     EmptyIOBuffer (&job[0].stdout, job[0].stdout_fd);
    70     EmptyIOBuffer (&job[0].stderr, job[0].stderr_fd);
    71   } else {
    72     /* read stdout buffer */
    73     Nread = ReadtoIOBuffer (&job[0].stdout, job[0].stdout_fd);
    74     switch (Nread) {
    75       case -2:  /* error in read (programming error?  system level error?) */
    76         fprintf (stderr, "serious IO error\n");
    77         exit (2);
    78       case -1:  /* no data in pipe */
    79       case 0:   /* pipe is closed, change child state? **/
    80       default:  /* data in pipe */
    81         break;
    82     }
    83  
    84     /* read stderr buffer */
    85     Nread = ReadtoIOBuffer (&job[0].stderr, job[0].stderr_fd);
    86     switch (Nread) {
    87       case -2:  /* error in read (programming error?  system level error?) */
    88         fprintf (stderr, "serious IO error\n");
    89         exit (2);
    90       case -1:  /* no data in pipe */
    91       case 0:   /* pipe is closed, change child state? **/
    92       default:  /* data in pipe */
    93         break;
    94     }
    95   }
    96   return (TRUE);
    97 }
    98 
    99 int CheckLocalJobStatus (Job *job) {
    100 
    101   int result, waitstatus;
    102 
    103   /*** if child has exited, does the pipe hang around until it is flushed? ***/
    104   /*** who closes the child stdout/stderr fd? */
    105 
    106   /* check local job status */
    107   result = waitpid (job[0].pid, &waitstatus, WNOHANG);
    108   switch (result) {
    109     case -1:  /* error with waitpid */
    110       switch (errno) {
    111         case ECHILD:
    112           fprintf (stderr, "unknown PID, not a child proc\n");
    113           fprintf (stderr, "did process already exit?  programming error?\n");
    114           job[0].state = JOB_NONE;
    115           job[0].exit_status = 0;
    116           return (FALSE);
    117         case EINVAL:
    118           fprintf (stderr, "error EINVAL (waitpid): programming error\n");
    119           exit (1);
    120         case EINTR:
    121           fprintf (stderr, "error EINTR (waitpid): programming error\n");
    122           exit (1);
    123         default:
    124           fprintf (stderr, "unknown error for waitpid (%d): programming error\n", errno);
    125           exit (1);
    126       }
    127       break;
    128      
    129     case 0:  /* process not exited */
    130       job[0].state = JOB_BUSY;
    131       job[0].exit_status = 0;
    132       return (TRUE);
    133 
    134     default:
    135       if (result != job[0].pid) {
    136         fprintf (stderr, "waitpid error: mis-matched PID (%d vs %d).  programming error\n", result, job[0].pid);
    137         exit (1);
    138       }
    139       if (WIFEXITED(waitstatus)) {
    140         job[0].state = JOB_EXIT;
    141         job[0].exit_status = WEXITSTATUS(waitstatus);
    142       }
    143       if (WIFSIGNALED(waitstatus)) {
    144         job[0].state = JOB_CRASH;
    145         job[0].exit_status = WTERMSIG(waitstatus);
    146       }
    147       if (WIFSTOPPED(waitstatus)) {
    148         fprintf (stderr, "waitpid returns 'stopped': programming error\n");
    149         exit (1);
    150       }
    151   }
    152   return (FALSE);
    153 }
    154 
     159/* should this function close the fd's? */
    155160int KillLocalJob (Job *job) {
    156161

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

    r4693 r4697  
    6868      fprintf (stderr, "- ");
    6969    }
    70     fprintf (stderr, "%-15s %4d   %-20s\n", tasks[i][0].name, tasks[i][0].Njobs, tasks[i][0].argv[0]);
     70    if (tasks[i][0].argv == NULL) {
     71      fprintf (stderr, "%-15s %4d   %-20s\n", tasks[i][0].name, tasks[i][0].Njobs, "dynamic");
     72    } else {
     73      fprintf (stderr, "%-15s %4d   %-20s\n", tasks[i][0].name, tasks[i][0].Njobs, tasks[i][0].argv[0]);
     74    }
    7175    if (verbose) {
    7276      fprintf (stderr, "    spawn period: %f, polling period: %f, timeout period: %f\n",
    … …  
    209213}
    210214
    211 int ValidateTask (Task *task) {
    212 
    213   if (task[0].argc == 0) {
     215int ValidateTask (Task *task, int RequireStatic) {
     216
     217  int i, hash;
     218
     219  /* is a static command defined? */
     220  if (task[0].argc != 0) {
     221    if (task[0].argv == NULL) {
     222      fprintf (stderr, "task command arguments not defined (programming error)\n");
     223      return (FALSE);
     224    }
     225    return (TRUE);
     226  }
     227  if (RequireStatic) {
    214228    fprintf (stderr, "task command not defined\n");
    215229    return (FALSE);
    216230  }
    217   if (task[0].argv == NULL) {
    218     fprintf (stderr, "task command arguments not defined (programming error)\n");
    219     return (FALSE);
    220   }
    221   return (TRUE);
     231
     232  /* no static command; dynamic command? */
     233  if (task[0].exec != NULL) {
     234    for (i = 0; i < task[0].exec[0].Nlines; i++) {
     235      hash = TaskHash (task[0].exec[0].line[i]);
     236      if (hash == TASK_COMMAND) return (TRUE);
     237    }
     238  }
     239  fprintf (stderr, "task command not defined\n");
     240  return (FALSE);
    222241}
    223242
    … …  
    250269}
    251270
     271Task *GetActiveTask () {
     272  Task *task;
     273  if (ActiveTask < 0) return (NULL);
     274  task = tasks[ActiveTask];
     275  return (task);
     276}
     277
     278int TaskHash (char *input) {
     279 
     280  int hash;
     281  char *command;
     282
     283  hash = TASK_NONE;
     284
     285  command = thisword (input);
     286  if (command == NULL) return (TASK_EMPTY);
     287
     288  if (command[0] == '#')                  hash = TASK_COMMENT;
     289  if (!strcasecmp (command, "END"))       hash = TASK_END;
     290  if (!strcasecmp (command, "HOST"))      hash = TASK_HOST;
     291  if (!strcasecmp (command, "NMAX"))      hash = TASK_NMAX;
     292  if (!strcasecmp (command, "TRANGE"))    hash = TASK_TRANGE;
     293  if (!strcasecmp (command, "COMMAND"))   hash = TASK_COMMAND;
     294  if (!strcasecmp (command, "PERIODS"))   hash = TASK_PERIODS;
     295  if (!strcasecmp (command, "TASK.EXIT")) hash = TASK_EXIT;
     296  if (!strcasecmp (command, "TASK.EXEC")) hash = TASK_EXEC;
     297
     298  free (command);
     299  return (hash);
     300}
     301
    252302/*** task timer functions ***/
    253303

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

    r4693 r4697  
    22# define prompt "> "
    33
    4 enum {TASK_NONE, TASK_EMPTY, TASK_COMMENT, TASK_NMAX, TASK_TRANGE, TASK_END, TASK_HOST, TASK_COMMAND, TASK_PERIODS, TASK_EXIT, TASK_EXEC};
    5 
    64int task (int argc, char **argv) {
    75
    8   int N;
     6  int N, hash;
    97  int ThisList, status;
    108  char *input, *outline;
    … …  
    6664
    6765    stripwhite (input);
    68     switch (TaskHash (input)) {
     66    hash = TaskHash (input);
     67    switch (hash) {
    6968
    7069      case TASK_EMPTY:
    … …  
    7877        free (input);
    7978        /* validate the new task: all mandatory elements defined? */
    80         if (!ValidateTask (task)) {
     79        if (!ValidateTask (task, FALSE)) {
    8180          DeleteNewTask ();
    8281          return (FALSE);
    … …  
    107106  return (FALSE);
    108107}
    109 
    110 int TaskHash (char *input) {
    111  
    112   int hash;
    113   char *command;
    114 
    115   hash = TASK_NONE;
    116 
    117   command = thisword (input);
    118   if (command == NULL) return (TASK_EMPTY);
    119 
    120   if (command[0] == '#')                  hash = TASK_COMMENT;
    121   if (!strcasecmp (command, "END"))       hash = TASK_END;
    122   if (!strcasecmp (command, "HOST"))      hash = TASK_HOST;
    123   if (!strcasecmp (command, "NMAX"))      hash = TASK_NMAX;
    124   if (!strcasecmp (command, "TRANGE"))    hash = TASK_TRANGE;
    125   if (!strcasecmp (command, "COMMAND"))   hash = TASK_COMMAND;
    126   if (!strcasecmp (command, "PERIODS"))   hash = TASK_PERIODS;
    127   if (!strcasecmp (command, "TASK.EXIT")) hash = TASK_EXIT;
    128   if (!strcasecmp (command, "TASK.EXEC")) hash = TASK_EXEC;
    129 
    130   free (command);
    131   return (hash);
    132 }

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

    r4693 r4697  
    1414  task = GetNewTask ();
    1515  if (task == NULL) {
    16     fprintf (stderr, "ERROR: not defining or running a task\n");
    17     return (FALSE);
     16    task = GetActiveTask ();
     17    if (task == NULL) {
     18      fprintf (stderr, "ERROR: not defining or running a task\n");
     19      return (FALSE);
     20    }
    1821  }
    1922
Note: See TracChangeset for help on using the changeset viewer.