IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Changeset 3203


Ignore:
Timestamp:
Feb 12, 2005, 9:34:34 AM (21 years ago)
Author:
eugene
Message:

pcontrol dev

Location:
trunk/Ohana/src/opihi
Files:
11 added
1 deleted
15 edited

Legend:

Unmodified
Added
Removed

  • trunk/Ohana/src/opihi/include/pcontrol.h

    r3187 r3203  
    77# include <sys/wait.h>
    88
    9 /** pcontrol global data **/
     9typedef struct timeval Ptime;
     10typedef unsigned long long IDtype;
    1011
    1112/** job status values **/
    12 enum {
     13typedef enum {
    1314  PCONTROL_JOB_PENDING,
    1415  PCONTROL_JOB_BUSY, 
    1516  PCONTROL_JOB_EXIT,
    1617  PCONTROL_JOB_CRASH,
     18  PCONTROL_JOB_HUNG, 
     19  PCONTROL_JOB_DONE, 
    1720} JobStat;
    1821
    1922/** job status values **/
    20 enum {
     23typedef enum {
    2124  PCONTROL_JOB_ANYHOST,
    2225  PCONTROL_JOB_WANTHOST,
    23   PCONTROL_JOB_NEEDHOST, 
     26  PCONTROL_JOB_NEEDHOST,
    2427} JobMode;
    2528
    2629/** host status values **/
    27 enum {
     30typedef enum {
    2831  PCONTROL_HOST_IDLE,
    2932  PCONTROL_HOST_BUSY, 
    3033  PCONTROL_HOST_DOWN,
     34  PCONTROL_HOST_DONE,
    3135  PCONTROL_HOST_OFF,
    3236} HostStat;
     37
     38/* stack special positions */
     39typedef enum {
     40  STACK_TOP = 0,
     41  STACK_BOTTOM = -1,
     42} StackWhere;
     43
     44typedef enum {
     45  PCLIENT_HUNG = -1,
     46  PCLIENT_DOWN = 0,
     47  PCLIENT_GOOD = 1,
     48} PclientStat;
    3349
    3450typedef struct {
    … …  
    4965} Fifo;
    5066
    51 typedef struct timeval Ptime;
    52 
    5367/* data to define a job */
    5468typedef struct {
    55   int        argc;
    56   char     **argv;
    57   char      *hostname;
    58   int        exitstatus;
    59   JobMode    mode;
    60   JobStat    status;
    61   IOBuffer   stdin;
    62   IOBuffer   stdout;
    63   IOBuffer   stderr;
    64   Ptime      start;
    65   Ptime      accum;
    66   Ptime      timer;
     69  int          argc;
     70  char       **argv;
     71  char        *hostname;
     72  int          exit_status;
     73  int          Reset;
     74  JobMode      mode;
     75  JobStat      state;
     76  JobStat      stack;
     77  IOBuffer     stdout;
     78  IOBuffer     stderr;
     79  Ptime        start;
     80  Ptime        accum;
     81  Ptime        timer;
    6782  struct Host *host;
    68   int        pid;
    69   IDtype     JobID;
     83  int          pid;
     84  IDtype       JobID;
    7085} Job;
    7186
    7287/* data to define a host machine */
    7388typedef struct {
    74   char      *hostname;
    75   int        stdin;
    76   int        stdout;
    77   int        stderr;
    78   int        markoff;
    79   HostStat   status;
    80   Ptime      start;
    81   Ptime      accum;
    82   Ptime      timer;
     89  char       *hostname;
     90  int         stdin;
     91  int         stdout;
     92  int         stderr;
     93  int         markoff;
     94  int         pid;
     95  HostStat    stack;
     96  Ptime       start;
     97  Ptime       accum;
     98  Ptime       timer;
     99  IDtype      HostID;
    83100  struct Job *job;
    84101} Host;
    … …  
    88105  int    Nobject;
    89106  int    NOBJECT;
    90 } Queue;
     107} Stack;
    91108
     109/* these should be loaded from ConfigInit or from opihi variables */
    92110static char *CONNECT;  /* connection method (ssh, rsh, etc) */
    93111static char *PCLIENT;  /* shell on remote host (eg, pclient) */
     112
     113# define FREE(X) if (X != NULL) { free (X); }
     114# define CLOSE(FD) { if (FD) close (FD); FD = 0; }
    94115
    95116/** IOBuffers allow the following operations (see pclient.h) **/
    … …  
    99120void FreeIOBuffer (IOBuffer *buffer);
    100121
    101 # define FREE(X) if (X != NULL) { free (X); }
    102 
    103 enum {
    104   QUEUE_TOP = 0,
    105   QUEUE_BOTTOM = -1,
    106 } QueueWhere;
    107 
     122void *GetStack (Stack *stack, int where);
     123int PutStack (Stack *stack, int where, void *object);
     124Stack *InitStack ();
     125Stack *GetHostStack (int StackID);
     126int PutHost (Host *host, int StackID, int where);
     127Host *GetHost (int StackID, int where);
     128int FindHost (IDtype HostID, int StackID);
     129int FindNamedHost (char *name, int StackID);
     130void AddHost (char *hostname);
     131void DownHost (Host *host);
     132void DelHost (Host *host);
     133int CheckSystem ();
     134int CheckBusyJobs ();
     135int CheckDoneJobs ();
     136int CheckDoneHosts ();
     137int CheckLiveHosts ();
     138int CheckDownHosts ();
     139int CheckIdleHosts ();
     140int CheckIdleHost (Host *host);
     141void InitIDs ();
     142IDtype NextJobID ();
     143IDtype NextHostID ();
     144void PrintID (FILE *f, IDtype ID);
     145int CheckBusyJob (Job *job);
     146int StartHost (Host *host);
     147int CheckDoneHost (Host *host);
     148int CheckDoneJob (Job *job);
     149int GetJobOutput (Job *job, char *channel);
     150int ResetJob (Job *job);
     151int PclientCommand (Host *host, char *command, IOBuffer *buffer);
     152int rconnect (char *command, char *hostname, char *shell, int *stdio);
     153int CheckHost (Host *host);
     154char *memstr (char *m1, char *m2, int n);
     155Stack *GetJobStack (int StackID);
     156int PutJob (Job *job, int StackID, int where);
     157Job *GetJob (int StackID, int where);
     158int FindJob (IDtype JobID, int StackID);
     159IDtype AddJob (char *hostname, JobMode mode, int timeout, int argc, char **argv);
     160void KillJob (Job *job);
     161void DelJob (Job *job);
     162Host *UnlinkJobAndHost (Job *job);
     163void LinkJobAndHost (Job *job, Host *host);

  • trunk/Ohana/src/opihi/pclient/reset.c

    r3187 r3203  
    77  if (argc != 1) {
    88    fprintf (stderr, "USAGE: reset\n");
    9     fprintf (stdout, "STATUS %d\n", -2);
     9    fprintf (stdout, "STATUS -1\n");
    1010    return (FALSE);
    1111  }

  • trunk/Ohana/src/opihi/pcontrol/CheckHost.c

    r3187 r3203  
    33int CheckHost (Host *host) {
    44 
    5   int i, status;
     5  int i, N, status;
    66  IOBuffer buffer;
    77  Job *job;
    … …  
    1515
    1616      /* if host has a job, job is dead, push to Pending */
    17       if (host[0].status == PCONTROL_HOST_BUSY) {
     17      if (host[0].stack == PCONTROL_HOST_BUSY) {
    1818        job = host[0].job;
    1919        N = FindJob (job[0].JobID, PCONTROL_JOB_BUSY);
    … …  
    2424        job[0].host = NULL; /* unlink host & job */
    2525        job = GetJob (PCONTROL_JOB_BUSY, N);
    26         PutJob (job, PCONTROL_JOB_PENDING, QUEUE_BOTTOM);
     26        PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM);
    2727      }
    2828      host[0].job = NULL;
    29       PutHost (host, PCONTROL_HOST_DOWN, QUEUE_BOTTOM);
    30       FreeBuffer (&buffer);
     29      PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM);
     30      FreeIOBuffer (&buffer);
    3131      return (FALSE);
    3232     
    … …  
    3434      fprintf (stderr, "host %s is not responding\n", host[0].hostname);
    3535      /*** do we mark this in some way (HUNG) ? ***/
    36       PutHost (host, host[0].status, QUEUE_BOTTOM);
    37       FreeBuffer (&buffer);
     36      PutHost (host, host[0].stack, STACK_BOTTOM);
     37      FreeIOBuffer (&buffer);
    3838      return (FALSE);
    3939
    4040    default:
    4141      fprintf (stderr, "host %s is alive\n", host[0].hostname);
    42       PutHost (host, host[0].status, QUEUE_BOTTOM);
    43       FreeBuffer (&buffer);
     42      PutHost (host, host[0].stack, STACK_BOTTOM);
     43      FreeIOBuffer (&buffer);
    4444      return (TRUE);
    4545  }

  • trunk/Ohana/src/opihi/pcontrol/CheckSystem.c

    r3187 r3203  
    44
    55  CheckBusyJobs();
     6  CheckDoneJobs();
     7  CheckDoneHosts();
     8  CheckIdleHosts();
     9  CheckDownHosts();
     10
     11  /** add in different timescales for heartbeat? **/
    612  CheckLiveHosts();
    7   CheckDownHosts();
    8   CheckIdleHosts();
    9 
    1013}
    1114
    12 CheckBusyJobs () {
     15int CheckBusyJobs () {
    1316
    14   queue = GetJobQueue (PCONTROL_JOB_BUSY);
    15   Nobjects = queue[0].Nobjects;
     17  int i, Nobject;
     18  Stack *stack;
     19  Job   *job;
    1620
    17   for (i = 0; i < Nobjects; i++) {
    18     job = GetObject (queue, QUEUE_TOP);
    19     CheckJob (job);
     21  stack = GetJobStack (PCONTROL_JOB_BUSY);
     22  Nobject = stack[0].Nobject;
     23
     24  for (i = 0; i < Nobject; i++) {
     25    job = GetStack (stack, STACK_TOP);
     26    CheckBusyJob (job);
    2027  }
    2128}
    2229
    23 CheckLiveHosts () {
     30int CheckDoneJobs () {
    2431
    25   queue = GetHostQueue (PCONTROL_HOST_IDLE);
    26   Nobjects = queue[0].Nobjects;
     32  int i, Nobject;
     33  Stack *stack;
     34  Job   *job;
    2735
    28   for (i = 0; i < Nobjects; i++) {
    29     host = GetObject (queue, QUEUE_TOP);
     36  stack = GetJobStack (PCONTROL_JOB_DONE);
     37  Nobject = stack[0].Nobject;
     38
     39  for (i = 0; i < Nobject; i++) {
     40    job = GetStack (stack, STACK_TOP);
     41    CheckDoneJob (job);
     42  }
     43}
     44
     45int CheckDoneHosts () {
     46
     47  int i, Nobject;
     48  Stack *stack;
     49  Host  *host;
     50
     51  stack = GetHostStack (PCONTROL_HOST_DONE);
     52  Nobject = stack[0].Nobject;
     53
     54  for (i = 0; i < Nobject; i++) {
     55    host = GetStack (stack, STACK_TOP);
     56    StartHost (host);
     57  }
     58}
     59
     60/* this is just a heartbeat check */
     61int CheckLiveHosts () {
     62
     63  int i, Nobject;
     64  Stack *stack;
     65  Host  *host;
     66
     67  stack = GetHostStack (PCONTROL_HOST_IDLE);
     68  Nobject = stack[0].Nobject;
     69
     70  for (i = 0; i < Nobject; i++) {
     71    host = GetStack (stack, STACK_TOP);
    3072    CheckHost (host);
    3173  }
    3274
    33   queue = GetHostQueue (PCONTROL_HOST_BUSY);
    34   Nobjects = queue[0].Nobjects;
     75  stack = GetHostStack (PCONTROL_HOST_BUSY);
     76  Nobject = stack[0].Nobject;
    3577
    36   for (i = 0; i < Nobjects; i++) {
    37     host = GetObject (queue, QUEUE_TOP);
     78  for (i = 0; i < Nobject; i++) {
     79    host = GetStack (stack, STACK_TOP);
    3880    CheckHost (host);
    3981  }
    4082}
    4183
    42 CheckDownHosts () {
     84int CheckDownHosts () {
    4385
    44   queue = GetHostQueue (PCONTROL_HOST_DOWN);
    45   Nobjects = queue[0].Nobjects;
     86  int i, Nobject;
     87  Stack *stack;
     88  Host  *host;
    4689
    47   for (i = 0; i < Nobjects; i++) {
    48     host = GetObject (queue, QUEUE_TOP);
     90  stack = GetHostStack (PCONTROL_HOST_DOWN);
     91  Nobject = stack[0].Nobject;
     92
     93  for (i = 0; i < Nobject; i++) {
     94    host = GetStack (stack, STACK_TOP);
    4995    StartHost (host);
    5096  }
    5197}
    5298
    53 CheckIdleHosts () {
     99int CheckIdleHosts () {
    54100
    55   hqueue = HostPool_Idle;
    56   jqueue = JobPool_Pending;
     101  int i, Nobject;
     102  Stack *stack;
     103  Host  *host;
    57104
    58   for (i = 0; i < hqueue[0].Nobjects; i++) {
    59     host = (Host *) hqueue[0].object[i];
    60     /** need to add a timeout here **/
    61     for (j = 0; j < jqueue[0].Nobjects; j++) {
    62       job = (Job *) jqueue[0].object[j];
    63       if ((job[0].mode == PCONTROL_JOB_ANYHOST) ||
    64           (!strcasecmp (job[0].hostname, host[0].hostname))
     105  /* check if there are any pending jobs, otherwise skip step */
     106  stack = GetJobStack (PCONTROL_JOB_PENDING);
     107  if (!stack[0].Nobject) return;
    65108
    66     StartHost (host);
     109  stack = GetHostStack (PCONTROL_HOST_IDLE);
     110  Nobject = stack[0].Nobject;
     111
     112  for (i = 0; i < Nobject; i++) {
     113    host = GetStack (stack, STACK_TOP);
     114    CheckIdleHost (host);
    67115  }
    68116}
    69 
    70 /*** this section is tricky: need to have a plan on making WANTHOST likely
    71      but not required  (on average, every host is currently in use.  some possible tricks:
    72 
    73      - search through PENDING for N entries looking for WANTHOST / matching host
    74      - keep job through N tries waiting for WANTHOST
    75 ***/

  • trunk/Ohana/src/opihi/pcontrol/HostOps.c

    r3189 r3203  
    11# include "pcontrol.h"
    22
    3 Queue *HostPool_Idle;
    4 Queue *HostPool_Busy;
    5 Queue *HostPool_Down;
    6 Queue *HostPool_Off;
     3Stack *HostPool_Idle;
     4Stack *HostPool_Busy;
     5Stack *HostPool_Down;
     6Stack *HostPool_Off;
    77
    8 Queue *GetHostQueue (int QueueID) {
    9   switch (QueueID) {
     8void InitHostStacks () {
     9  HostPool_Idle = InitStack ();
     10  HostPool_Busy = InitStack ();
     11  HostPool_Down = InitStack ();
     12  HostPool_Off = InitStack ();
     13}
     14
     15Stack *GetHostStack (int StackID) {
     16  switch (StackID) {
    1017    case PCONTROL_HOST_IDLE:
    1118      return (HostPool_Idle);
    … …  
    1724      return (HostPool_Off);
    1825    default:
    19       fprintf (stderr, "error: unknown host queue\n");
     26      fprintf (stderr, "error: unknown host stack\n");
    2027      return (NULL);
    2128  }
    … …  
    2330}
    2431
    25 int PutHost (Host *host, int QueueID, int where) {
     32int PutHost (Host *host, int StackID, int where) {
    2633
    2734  int status;
    28   Queue *queue;
     35  Stack *stack;
    2936
    30   queue = GetHostQueue (QueueID);
    31   if (queue == NULL) return (FALSE);
     37  stack = GetHostStack (StackID);
     38  if (stack == NULL) return (FALSE);
    3239
    33   host[0].status = QueueID;
    34   status = PutObject (queue, where, host);
     40  host[0].stack = StackID;
     41  status = PutStack (stack, where, host);
    3542  return (status);
    3643}
    3744 
    38 Host *GetHost (int QueueID, int where) {
     45Host *GetHost (int StackID, int where) {
    3946
    4047  Host *host;
    41   Queue *queue;
     48  Stack *stack;
    4249
    43   queue = GetHostQueue (QueueID);
    44   if (queue == NULL) return (NULL);
     50  stack = GetHostStack (StackID);
     51  if (stack == NULL) return (NULL);
    4552
    46   host = GetObject (queue, where);
     53  host = GetStack (stack, where);
    4754  return (host);
    4855}
    4956
    50 int FindHost (IDtype HostID, int QueueID) {
     57int FindHost (IDtype HostID, int StackID) {
    5158
    52   Queue *queue;
     59  int i;
     60  Host *host;
     61  Stack *stack;
    5362
    54   queue = GetHostQueue (QueueID);
    55   if (queue == NULL) return (-2);
     63  stack = GetHostStack (StackID);
     64  if (stack == NULL) return (-2);
    5665
    57   for (i = 0; i < queue[0].Nobject; i++) {
    58     host = (Host *) queue[0].object;
     66  for (i = 0; i < stack[0].Nobject; i++) {
     67    host = (Host *) stack[0].object;
    5968    if (host[0].HostID == HostID) {
    6069      return (i);
    … …  
    6473}
    6574
    66 int FindNamedHost (char *name, int QueueID) {
     75int FindNamedHost (char *name, int StackID) {
    6776
    68   Queue *queue;
     77  int i;
     78  Host *host;
     79  Stack *stack;
    6980
    70   queue = GetHostQueue (QueueID);
    71   if (queue == NULL) return (-2);
     81  stack = GetHostStack (StackID);
     82  if (stack == NULL) return (-2);
    7283
    73   for (i = 0; i < queue[0].Nobject; i++) {
    74     host = (Host *) queue[0].object;
    75     if (!strcasecmp (host[0].hostname, host)) {
     84  for (i = 0; i < stack[0].Nobject; i++) {
     85    host = (Host *) stack[0].object;
     86    if (!strcasecmp (host[0].hostname, name)) {
    7687      return (i);
    7788    }
    … …  
    8091}
    8192
    82 AddHost (char *hostname) {
     93void AddHost (char *hostname) {
    8394
    8495  Host *host;
    … …  
    94105  host[0].markoff  = FALSE;
    95106  host[0].job      = NULL;
    96   DownHost (host);
     107  PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM);
    97108}
    98109
    99 DelHost (Host *host) {
     110void DownHost (Host *host) {
     111
     112  CLOSE (host[0].stdin);
     113  CLOSE (host[0].stdout);
     114  CLOSE (host[0].stderr);
     115  host[0].job = NULL;
     116  PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM);
     117}
     118
     119void DelHost (Host *host) {
    100120  FREE (host[0].hostname);
    101121  FREE (host[0].job);
    … …  
    103123}
    104124
    105 MarkOffHost (Host *host) {
    106   if (host[0].status != PCONTROL_HOST_BUSY) {
    107     return (FALSE);
    108   }
    109   host[0].markoff  = TRUE;
    110   PutObject (HostPool_Busy, host);  /**** where are we removing hosts? **/
    111 }
    112 

  • trunk/Ohana/src/opihi/pcontrol/JobID.c

    r3187 r3203  
    11# include "pcontrol.h"
    2 
    3 typedef unsigned long long IDtype;
    42
    53static IDtype CurrentJobID  = 0;
    … …  
    75
    86/* for now, no persistence : we could use the date/time to seed the upper byte(s) if needed */
    9 InitIDs () {
     7void InitIDs () {
    108  CurrentJobID = 0;
    119  CurrentHostID = 0;
    … …  
    3230void PrintID (FILE *f, IDtype ID) {
    3331
     32  unsigned short int word0, word1, word2, word3;
     33
    3434  word0 = 0xffff & ID;
    3535  word1 = 0xffff & (ID >> 16);

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

    r3189 r3203  
    11# include "pcontrol.h"
    22
    3 Queue *JobPool_Pending;
    4 Queue *JobPool_Busy;
    5 Queue *JobPool_Exit;
    6 Queue *JobPool_Crash;
     3Stack *JobPool_Pending;
     4Stack *JobPool_Busy;
     5Stack *JobPool_Exit;
     6Stack *JobPool_Crash;
    77
    8 Queue *GetJobQueue (int QueueID) {
    9   switch (QueueID) {
     8void InitJobStacks () {
     9  JobPool_Pending = InitStack ();
     10  JobPool_Busy = InitStack ();
     11  JobPool_Exit = InitStack ();
     12  JobPool_Crash = InitStack ();
     13}
     14
     15Stack *GetJobStack (int StackID) {
     16  switch (StackID) {
    1017    case PCONTROL_JOB_PENDING:
    1118      return (JobPool_Pending);
    … …  
    1421    case PCONTROL_JOB_EXIT:
    1522      return (JobPool_Exit);
    16     case PCONTROL_JOB_Crash:
     23    case PCONTROL_JOB_CRASH:
    1724      return (JobPool_Crash);
    1825    default:
    19       fprintf (stderr, "error: unknown job queue\n");
     26      fprintf (stderr, "error: unknown job stack\n");
    2027      return (NULL);
    2128  }
    … …  
    2330}
    2431
    25 int PutJob (Job *job, int QueueID, int where) {
     32int PutJob (Job *job, int StackID, int where) {
    2633
    2734  int status;
    28   Queue *queue;
     35  Stack *stack;
    2936
    30   queue = GetJobQueue (QueueID);
    31   if (queue == NULL) return (FALSE);
     37  stack = GetJobStack (StackID);
     38  if (stack == NULL) return (FALSE);
    3239
    33   job[0].status = QueueID;
    34   status = PutObject (queue, where, job);
     40  /* be default, these are both the same - to override, set state after PutJob */
     41  job[0].state = StackID;
     42  job[0].stack = StackID;
     43  status = PutStack (stack, where, job);
    3544  return (status);
    3645}
    3746 
    38 Job *GetJob (int QueueID, int where) {
     47Job *GetJob (int StackID, int where) {
    3948
    4049  Job *job;
    41   Queue *queue;
     50  Stack *stack;
    4251
    43   queue = GetJobQueue (QueueID);
    44   if (queue == NULL) return (NULL);
     52  stack = GetJobStack (StackID);
     53  if (stack == NULL) return (NULL);
    4554
    46   job = GetObject (queue, where);
     55  job = GetStack (stack, where);
    4756  return (job);
    4857}
    4958 
    50 int FindJob (IDtype JobID, int QueueID) {
     59int FindJob (IDtype JobID, int StackID) {
    5160
     61  int i;
    5262  Job *job;
    53   Queue *queue;
     63  Stack *stack;
    5464
    55   queue = GetJobQueue (QueueID);
    56   if (queue == NULL) return (-2);
     65  stack = GetJobStack (StackID);
     66  if (stack == NULL) return (-2);
    5767
    58   for (i = 0; i < queue[0].Nobject; i++) {
    59     job = (Job *) queue[0].object;
     68  for (i = 0; i < stack[0].Nobject; i++) {
     69    job = (Job *) stack[0].object;
    6070    if (job[0].JobID == JobID) {
    6171      return (i);
    … …  
    6575}
    6676
    67 AddJob (char *hostname, JobMode mode, int timeout, int argc, char **argv) {
     77IDtype AddJob (char *hostname, JobMode mode, int timeout, int argc, char **argv) {
    6878
    6979  Job *job;
    … …  
    7686  job[0].mode     = mode;
    7787  job[0].host     = NULL;
    78   job[0].ID       = NextJobID();
     88  job[0].JobID    = NextJobID();
     89  job[0].Reset    = FALSE;
    7990
    80   InitIOBuffer (&job[0].stdin,  0x1000);
     91  /* do this step on start? */
    8192  InitIOBuffer (&job[0].stdout, 0x1000);
    8293  InitIOBuffer (&job[0].stderr, 0x1000);
    8394
    84   PutJob (job, PCONTROL_JOB_PENDING, QUEUE_BOTTOM);
     95  PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM);
     96  return (job[0].JobID);
    8597}
    8698
    87 DelJob (Job *job) {
     99void DelJob (Job *job) {
     100
     101  int i;
    88102
    89103  FREE (job[0].hostname);
    … …  
    93107  FREE (job[0].argv);
    94108
    95   FreeIOBuffer (&job[0].stdin);
    96109  FreeIOBuffer (&job[0].stdout);
    97110  FreeIOBuffer (&job[0].stderr);
    … …  
    100113}
    101114
    102 KillJob (Job *job) {
     115# if (0)
     116void KillJob (Job *job) {
    103117
     118  int status;
    104119  char line[64];
     120  Host *host;
    105121
    106122  sprintf (line, "reset\n");
    … …  
    110126  if ((status == -1) && (errno == EPIPE)) {
    111127    fprintf (stderr, "host %s is down\n", host[0].hostname);
    112     PutHost (host, PCONTROL_HOST_DOWN, QUEUE_BOTTOM);
    113     PutJob (job, PCONTROL_JOB_PENDING, QUEUE_BOTTOM);
     128    PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM);
     129    PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM);
    114130    return (FALSE);
    115131  }
    116132  /* check for a response */
    117133
     134  /** this needs to be cleaned up to handle the slow response case **/
    118135}
     136# endif
     137
     138Host *UnlinkJobAndHost (Job *job) {
     139
     140  int N;
     141  Host *host;
     142
     143  host = job[0].host;
     144
     145  /* unlink host & job */
     146  job[0].host = NULL;
     147  host[0].job = NULL;
     148 
     149  /*** need to pop host off of correct stack XXX ***/
     150  N = FindHost (host[0].HostID, host[0].stack);
     151  if (N < 0) {
     152    fprintf (stderr, "programming error: host is not found in current stack\n");
     153    exit (2);
     154  }
     155  host = GetHost (host[0].stack, N);
     156  return (host);
     157}
     158
     159void LinkJobAndHost (Job *job, Host *host) {
     160  job[0].host = host;
     161  host[0].job = job;
     162}

  • trunk/Ohana/src/opihi/pcontrol/QueueOps.c

    r3187 r3203  
    22
    33/* get object from point in stack (negative == distance from end) */
    4 void *GetQueue (Queue *queue, int where) {
     4void *GetStack (Stack *stack, int where) {
    55
    66  int i;
    77  void *object;
    88 
    9   /* QUEUE_TOP == 0, QUEUE_BOTTOM == -1 */
     9  /* STACK_TOP == 0, STACK_BOTTOM == -1 */
    1010  /* this code correctly handles the negative 'where' and Nobject == 0 */
    11   if (where < 0) where += queue[0].Nobject;
     11  if (where < 0) where += stack[0].Nobject;
    1212  if (where < 0) return (NULL);
    13   if (where >= queue[0].Nobject) return (NULL);
     13  if (where >= stack[0].Nobject) return (NULL);
    1414
    15   object = queue[0].object[where];
    16   queue[0].Nobject --;
    17   for (i = where; i < queue[0].Nobject; i++) {
    18     queue[0].object[i] = queue[0].object[i+1];
     15  object = stack[0].object[where];
     16  stack[0].Nobject --;
     17  for (i = where; i < stack[0].Nobject; i++) {
     18    stack[0].object[i] = stack[0].object[i+1];
    1919  }
    2020  return (object);
    … …  
    2222
    2323/* push object on top of stack */
    24 int PutQueue (Queue *queue, int where, void *object) {
     24int PutStack (Stack *stack, int where, void *object) {
    2525
    2626  int i;
    2727
    28   /* QUEUE_TOP == 0, QUEUE_BOTTOM == -1 */
     28  /* STACK_TOP == 0, STACK_BOTTOM == -1 */
    2929  /* this code correctly handles the negative 'where' and Nobject == 0 */
    30   if (where < 0) where += queue[0].Nobject + 1;
     30  if (where < 0) where += stack[0].Nobject + 1;
    3131  if (where < 0) return (FALSE);
    32   if (where > queue[0].Nobject) return (FALSE);
     32  if (where > stack[0].Nobject) return (FALSE);
    3333
    34   /* extend queue as needed */
    35   if (queue[0].Nobject >= queue[0].NOBJECT) {
    36     queue[0].NOBJECT += 100;
    37     REALLOCATE (queue[0].object, void *, queue[0].NOBJECT);
     34  /* extend stack as needed */
     35  if (stack[0].Nobject >= stack[0].NOBJECT) {
     36    stack[0].NOBJECT += 100;
     37    REALLOCATE (stack[0].object, void *, stack[0].NOBJECT);
    3838  }
    3939
    40   for (i = queue[0].Nobject; i > where; i--) {
    41     queue[0].object[i] = queue[0].object[i-1];
     40  for (i = stack[0].Nobject; i > where; i--) {
     41    stack[0].object[i] = stack[0].object[i-1];
    4242  }
    43   queue[0].object[where] = object;
    44   queue[0].Nobject ++;
     43  stack[0].object[where] = object;
     44  stack[0].Nobject ++;
    4545  return (TRUE);
    4646}
    4747
    48 /* allocate queue, setup with default values, allocate data */
    49 Queue *InitQueue () {
     48/* allocate stack, setup with default values, allocate data */
     49Stack *InitStack () {
    5050
    51   Queue *queue;
     51  Stack *stack;
    5252
    53   ALLOCATE (queue, Queue, 1);
     53  ALLOCATE (stack, Stack, 1);
    5454
    55   queue[0].Nobject = 0;
    56   queue[0].NOBJECT = 50;
    57   ALLOCATE (queue[0].object, void *, queue[0].NOBJECT);
    58   return (queue);
     55  stack[0].Nobject = 0;
     56  stack[0].NOBJECT = 50;
     57  ALLOCATE (stack[0].object, void *, stack[0].NOBJECT);
     58  return (stack);
    5959}
    6060
    61 /* these queues are not super efficient, and should probably be replaced with linked lists,
     61/* these stacks are not super efficient, and should probably be replaced with linked lists,
    6262   but I find these easier to get my brain around
    6363*/

  • trunk/Ohana/src/opihi/pcontrol/StartHost.c

    r3187 r3203  
    33int StartHost (Host *host) {
    44
     5  int pid;
    56  int stdio[3];
    67
    7   pid = rconnect (CONNECT, host[0].hostname, PCLIENT, stdio);
     8  /* pid = rconnect (CONNECT, host[0].hostname, PCLIENT, stdio); */
     9
     10  pid = rconnect ("ssh", host[0].hostname, "pclient", stdio);
    811  if (!pid) {     /** failure to start **/
    912    fprintf (stderr, "failure to start %s\n", host[0].hostname);
    10     PutHost (host, PCONTROL_HOST_DOWN, QUEUE_BOTTOM);
     13    PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM);
    1114    return (FALSE);
    1215  }
    … …  
    1518  host[0].stderr = stdio[2];
    1619  host[0].pid    = pid;
    17   PutHost (host, PCONTROL_HOST_IDLE, QUEUE_BOTTOM);
     20  PutHost (host, PCONTROL_HOST_IDLE, STACK_BOTTOM);
    1821  return (TRUE);
    1922}

  • trunk/Ohana/src/opihi/pcontrol/StartJob.c

    r3187 r3203  
    33int StartJob (Job *job) {
    44
    5   int  Nline;
    6   char *line;
     5  int  i, Nline, status;
     6  char *line, *p;
     7  Host *host;
    78  IOBuffer buffer;
    89
    9   InitIOBuffer (&buffer);
     10  InitIOBuffer (&buffer, 0x100);
    1011
    1112  /* job must have assigned host */
    … …  
    1718
    1819  /* construct command line : job arg0 arg1 ... argN\n */
    19   Nline = 10 + argc;
    20   for (i = 0; i < argc; i++) {
     20  Nline = 10 + job[0].argc;
     21  for (i = 0; i < job[0].argc; i++) {
    2122    Nline += strlen (job[0].argv[i]);
    2223  }
    … …  
    2425  bzero (line, Nline);
    2526  strcpy (line, "job");
    26   for (i = 0; i < argc; i++) {
     27  for (i = 0; i < job[0].argc; i++) {
    2728    strcat (line, " ");
    2829    strcat (line, job[0].argv[i]);
    2930  }
    30   strcat (line, "\n");
    3131
    3232  status = PclientCommand (host, line, &buffer);
    … …  
    3535  /* check on success of pclient command */
    3636  switch (status) {
    37     case DOWN:
     37    case PCLIENT_DOWN:
    3838      fprintf (stderr, "host %s is down\n", host[0].hostname);
    39       goto escape;
    40       /*** different behavior for ANYHOST, WANTHOST, NEEDHOST ***/
     39      goto failure;
    4140
    42     case HUNG:
     41    case PCLIENT_HUNG:
    4342      fprintf (stderr, "host %s is not responding\n", host[0].hostname);
    44       goto escape;
    45       /*** should we consider a HUNG host DOWN? ***/
     43      goto failure;
    4644
    47     case TRUE:
     45    case PCLIENT_GOOD:
    4846      fprintf (stderr, "message received\n"); 
    4947      break;
    … …  
    6058    exit (1);
    6159  }
    62   fscan (p, "%*s %d", &status);
     60  sscanf (p, "%*s %d", &status);
    6361  switch (status) {
    6462    case -1:
    6563      fprintf (stderr, "error in pclient child\n");
    66       goto escape;
    67       /*** should we consider this host DOWN? ***/
     64      goto failure;
    6865
    6966    case -2:
    … …  
    7774    default:
    7875      job[0].pid = status;
    79       PutHost (host, PCONTROL_HOST_BUSY, QUEUE_BOTTOM);
    80       PutJob (job, PCONTROL_JOB_BUSY, QUEUE_BOTTOM);
     76      PutHost (host, PCONTROL_HOST_BUSY, STACK_BOTTOM);
     77      PutJob (job, PCONTROL_JOB_BUSY, STACK_BOTTOM);
     78      FreeIOBuffer (&buffer);
     79      /** need to start timers here **/
    8180      return (TRUE);
    8281  }
    … …  
    8584  exit (1);
    8685
    87 escape:
    88   job[0].host = NULL; /* unlink host & job */
     86failure:
     87  /* unlink host & job */
     88  job[0].host = NULL;
    8989  host[0].job = NULL;
    90   /*** need to pop host from corresponding queue (BUSY) ***/
    91   PutHost (host, PCONTROL_HOST_DOWN, QUEUE_BOTTOM);
    92   PutJob (job, PCONTROL_JOB_PENDING, QUEUE_BOTTOM);
    93   FreeBuffer (&buffer);
     90  PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM);
     91  PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM);
     92  FreeIOBuffer (&buffer);
    9493  return (FALSE);
    9594}
    9695
    97 /*
    98    this function sends a new command to the pclient.
    99    before calling this function:
    100      - grab the job from its queue (remove it from that queue with GetQueue)
    101      - identify the target host (MUST already be IDLE)
    102      - grab the host from its queue (remove it from that queue with GetQueue)
    103      - link job and host (job[0].host = host, host[0].job = job)
    104      - job must also have a valid command (argc, argv)
    105 */
     96/** note : host and job popped off stacks : can't use UnlinkJobAndHost **/

  • trunk/Ohana/src/opihi/pcontrol/host.c

    r3189 r3203  
    33int host (int argc, char **argv) {
    44
    5   int pid, status;
     5  int N, Delete, Off, On;
     6  Host *host;
    67
    7   /* this section needs some help: find the specified host in the queues */
     8  /* this section needs some help: find the specified host in the stacks */
    89  Delete = FALSE;
    910  if ((N = get_argument (argc, argv, "-delete"))) {
    … …  
    1213  }
    1314 
    14   /* this section needs some help: find the specified host in the queues */
     15  /* this section needs some help: find the specified host in the stacks */
    1516  Off = FALSE;
    1617  if ((N = get_argument (argc, argv, "-off"))) {
    … …  
    2930 
    3031  if (Delete) {
    31     host = FindHost (argv[1], PCONTROL_HOST_OFF);
    32     if (host == NULL) {
     32    N = FindNamedHost (argv[1], PCONTROL_HOST_OFF);
     33    if (N < 0) {
    3334      fprintf (stderr, "host %s is not OFF\n", argv[1]);
    3435      return (FALSE);
    3536    }
     37    host = GetHost (PCONTROL_HOST_OFF, N);
    3638    DelHost (host);
    3739    return (TRUE);
    … …  
    4547    }
    4648    host = GetHost (PCONTROL_HOST_OFF, N);
    47     PutHost (host, PCONTROL_HOST_DOWN, QUEUE_BOTTOM);
     49    DownHost (host);
    4850    return (TRUE);
    4951  }
    … …  
    5254    N = FindNamedHost (argv[1], PCONTROL_HOST_IDLE);
    5355    if (N >= 0) {
    54       host = GetHost (PCONTROL_HOST_OFF, N);
    55       PutHost (host, PCONTROL_HOST_OFF, QUEUE_BOTTOM);
     56      host = GetHost (PCONTROL_HOST_IDLE, N);
     57      CLOSE (host[0].stdin);
     58      CLOSE (host[0].stdout);
     59      CLOSE (host[0].stderr);
     60      PutHost (host, PCONTROL_HOST_OFF, STACK_BOTTOM);
    5661      return (TRUE);
    5762    }
    5863    N = FindNamedHost (argv[1], PCONTROL_HOST_DOWN);
    5964    if (N >= 0) {
    60       host = GetHost (PCONTROL_HOST_OFF, N);
    61       PutHost (host, PCONTROL_HOST_OFF, QUEUE_BOTTOM);
     65      host = GetHost (PCONTROL_HOST_DOWN, N);
     66      PutHost (host, PCONTROL_HOST_OFF, STACK_BOTTOM);
    6267      return (TRUE);
    6368    }
    6469    N = FindNamedHost (argv[1], PCONTROL_HOST_BUSY);
    6570    if (N >= 0) {
    66       host = GetHost (PCONTROL_HOST_OFF, N);
    67       PutHost (host, PCONTROL_HOST_OFF, QUEUE_BOTTOM);
     71      host = GetHost (PCONTROL_HOST_BUSY, N);
     72      host[0].markoff  = TRUE;
     73      PutHost (host, PCONTROL_HOST_BUSY, STACK_BOTTOM);
    6874      return (TRUE);
    6975    }

  • trunk/Ohana/src/opihi/pcontrol/job.c

    r3187 r3203  
    33int job (int argc, char **argv) {
    44
    5   int pid, status;
     5  char *Host, **targv;
     6  int i, N, Mode, targc, Timeout;
     7  IDtype JobID;
    68
    79  Host = NULL;
    … …  
    1618    if (Mode == PCONTROL_JOB_WANTHOST) {
    1719      fprintf (stderr, "ERROR: -host and +host are incompatible\n");
    18       free (Host);
     20      FREE (Host);
    1921      return (FALSE);
    2022    }
    … …  
    3436  if (argc < 2) {
    3537    fprintf (stderr, "USAGE: job [options] (arg0) (arg1) ... (argN)\n");
     38    FREE (Host);
    3639    return (FALSE);
    3740  }
    … …  
    4649  FREE (Host);
    4750
    48   fprintf (stdout, "STATUS %d\n", ChildPID);
     51  fprintf (stdout, "STATUS %d\n", JobID);
    4952  return (TRUE);
    5053

  • trunk/Ohana/src/opihi/pcontrol/pclient.c

    r3187 r3203  
    44# define PCLIENT_PROMPT "pclient:"
    55
    6 # define DOWN  0
    7 # define HUNG -1
    8 
    96int PclientCommand (Host *host, char *command, IOBuffer *buffer) {
    107
    11   /* send command to client */
    12   status = write (host[0].stdin, command, strlen(command));
    13   if ((status == -1) && (errno == EPIPE)) return (DOWN);
     8  int i;
     9  int status;
     10  char *line;
     11 
     12  /* send command to client (adding on \n) */
     13  ALLOCATE (line, char, MAX (1, strlen(command)));
     14  sprintf (line, "%s\n", command);
     15  status = write (host[0].stdin, line, strlen(line));
     16  free (line);
     17
     18  /* is pipe still open? */
     19  if ((status == -1) && (errno == EPIPE)) return (PCLIENT_DOWN);
    1420 
    1521  /* watch for response - wait up to 1 second */
    16   p = NULL;
     22  line = NULL;
    1723  status = -1;
    18   for (i = 0; (i < PCLIENT_TIMEOUT) && (status == -1) && (p == NULL); i++) {
     24  for (i = 0; (i < PCLIENT_TIMEOUT) && (status == -1) && (line == NULL); i++) {
    1925    status = ReadtoIOBuffer (buffer, host[0].stdout);
    20     p = memstr (buffer[0].buffer, PCLIENT_PROMPT, buffer[0].Nbuffer);
     26    line = memstr (buffer[0].buffer, PCLIENT_PROMPT, buffer[0].Nbuffer);
    2127  }
    22   if (status ==  0) return (DOWN);
    23   if (status == -1) return (HUNG);
    24   return (TRUE);
     28  if (status ==  0) return (PCLIENT_DOWN);
     29  if (status == -1) return (PCLIENT_HUNG);
     30  return (PCLIENT_GOOD);
    2531}
     32
     33/* memstr returns a view, not an allocated string : don't free */

  • trunk/Ohana/src/opihi/pcontrol/pcontrol.c

    r3187 r3203  
    4141  InitOutfile ();
    4242  InitPcontrol ();
    43   InitChild ();
    4443
    4544  rl_readline_name = opihi_name;

  • trunk/Ohana/src/opihi/pcontrol/rconnect.c

    r3187 r3203  
    1010  int i, stdin_fd[2], stdout_fd[2], stderr_fd[2], status;
    1111  pid_t pid;
    12   char command[0x100];
    1312  char *p;
     13  char **argv;
    1414  IOBuffer buffer;
    1515
    … …  
    7676  status = -1;
    7777  for (i = 0; (i < CONNECT_TIMEOUT) && (status == -1) && (p == NULL); i++) {
    78     status = ReadtoIOBuffer (buffer, stdout_fd[0]);
    79     p = memstr (buffer[0].buffer, "CONNECTED", buffer[0].Nbuffer);
     78    status = ReadtoIOBuffer (&buffer, stdout_fd[0]);
     79    p = memstr (buffer.buffer, "CONNECTED", buffer.Nbuffer);
    8080  }
    8181  if (status == 0) goto pipe_error;
Note: See TracChangeset for help on using the changeset viewer.