Changeset 18098 for trunk/Ohana

Timestamp:

Jun 12, 2008, 12:52:06 PM (18 years ago)

Author:

eugene

Message:

use virtual stacks for host and job queries by user thread

Location:

trunk/Ohana/src/opihi/pcontrol

Files:

• CheckHost.c (modified) (1 diff)
• CheckIdleHost.c (modified) (1 diff)
• CheckSystem.c (modified) (2 diffs)
• HostOps.c (modified) (8 diffs)
• IDops.c (modified) (2 diffs)
• JobID.c (modified) (1 diff)
• JobOps.c (modified) (9 diffs)
• StopHosts.c (modified) (2 diffs)
• check.c (modified) (4 diffs)
• delete.c (modified) (1 diff)
• host.c (modified) (3 diffs)
• pcontrol.c.in (modified) (1 diff)
• status.c (modified) (3 diffs)

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

@@ -15 +15 @@
     host[0].markoff = FALSE;
     StopHost (host);
-    OffHost (host);
     return (TRUE);
   }

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

@@ -14 +14 @@
     host[0].markoff = FALSE;
     StopHost (host);
-    OffHost (host);
     return (TRUE);
   }

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

@@ -346 +346 @@
     if (host == NULL) break;
     if (host[0].markoff) {
+      // DOWN -> OFF
       host[0].markoff = FALSE;
       OffHost (host);
@@ -354 +355 @@
       PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM);
     } else {
+      // DOWN -> IDLE (maybe)
+      // this is a race condition with "host retry", but the only 
+      // consequence is that both StartHost and reset set the times to 0.0
       StartHost (host);
     }

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

@@ -1 +1 @@
 # include "pcontrol.h"
+
+Stack *HostPool_AllHosts;  // virtual pool for user status queries
 
 Stack *HostPool_Idle; // these hosts are waiting for something to do
@@ -9 +11 @@
 
 void InitHostStacks () {
+  HostPool_AllHosts = InitStack ();
+
   HostPool_Idle = InitStack ();
   HostPool_Busy = InitStack ();
@@ -26 +30 @@
 
 void FreeHostStacks () {
+  // AllHosts is a virtual stack : all hosts are references
+  FreeStack (HostPool_AllHosts);
+
   FreeHostStack (HostPool_Idle);
   FreeHostStack (HostPool_Busy);
@@ -36 +43 @@
 char *GetHostStackName (int StackID) {
   switch (StackID) {
+    case PCONTROL_HOST_ALLHOSTS: return ("ALLHOSTS");
     case PCONTROL_HOST_IDLE: return ("IDLE");
     case PCONTROL_HOST_DOWN: return ("DOWN");
@@ -50 +58 @@
 Stack *GetHostStack (int StackID) {
   switch (StackID) {
+    case PCONTROL_HOST_ALLHOSTS: return (HostPool_AllHosts);
     case PCONTROL_HOST_IDLE: return (HostPool_Idle);
     case PCONTROL_HOST_DOWN: return (HostPool_Down);
@@ -63 +72 @@
 
 Stack *GetHostStackByName (char *name) {
+  if (!strcasecmp (name, "all")) return (HostPool_AllHosts);
   if (!strcasecmp (name, "idle")) return (HostPool_Idle);
   if (!strcasecmp (name, "down")) return (HostPool_Down);
@@ -202 +212 @@
   host[0].markoff  = FALSE;
   host[0].job      = NULL;
+
+  PutHost (host, PCONTROL_HOST_ALLHOSTS, STACK_BOTTOM);
   PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM);
   return (host[0].HostID);
@@ -207 +219 @@
 
 void DelHost (Host *host) {
+
+  Host *copy;
+
+  copy = PullStackByID (HostPool_AllHosts, host[0].HostID);
+  ASSERT (copy == host, "programming error: ALLHOSTS entry does not match");
+
   FreeIOBuffer (&host[0].comms_buffer);
   FREE (host[0].hostname);

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

@@ -1 +1 @@
 # include "pcontrol.h"
 
-static IDtype CurrentJobID  = 0;
-static IDtype CurrentHostID = 0;
+static IDtype CurrentJobID  = 1;
+static IDtype CurrentHostID = 1;
 
-/* for now, no persistence : we could use the date/time to seed the upper byte(s) if needed */
+/* for now, no persistence between sessions : we could use the date/time to seed the upper
+ * byte(s) if needed */
 void InitIDs () {
-  CurrentJobID = 0;
-  CurrentHostID = 0;
+  CurrentJobID = 1;
+  CurrentHostID = 1;
 }
 
@@ -41 +42 @@
 }
 
+IDtype GetID (char *IDword) {
+
+  int Nargs;
+  IDtype ID;
+  unsigned int word0, word1, word2, word3;
+  char *endptr;
+
+  Nargs = sscanf (IDword, "%x.%x.%x.%x", &word3, &word2, &word1, &word0);
+  if (Nargs == 4) {
+      IDtype tmp;
+    ID = 0;
+    ID |= (word0 << 0);
+    ID |= (word1 << 16);
+    tmp = word2;
+    ID |= (tmp << 32);
+    tmp = word3;
+    ID |= (tmp << 48);
+    return ID;
+  } 
+    
+  ID = strtoll (IDword, &endptr, 10);
+  if (*endptr == 0) {
+    return ID;
+  }
+
+  return 0;
+}

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

@@ -40 +40 @@
 }
 
+IDtype GetID (char *IDword) {
+
+  int Nargs;
+  IDtype ID;
+  unsigned short int word0, word1, word2, word3;
+
+  Nargs = sscanf (IDword, "%x.%x.%x.%x", &word0, &word1, &word2, &word3);
+  if (Nargs == 4) {
+    ID = 0;
+    ID |= (word0 << 0);
+    ID |= (word1 << 16);
+    ID |= (word2 << 32);
+    ID |= (word3 << 48);
+    return ID;
+  } 
+    
+  ID = strtoll (IDword, &endptr, 10);
+  if (*endptr == 0) {
+    return ID;
+  }
+
+  return 0;
+}

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

@@ -1 +1 @@
 # include "pcontrol.h"
 
-// Stack *JobPool_AllJobs;
+Stack *JobPool_AllJobs;  // virtual pool for user status queries
 
 Stack *JobPool_Pending;
@@ -12 +12 @@
 
 void InitJobStacks () {
-  // JobPool_AllJobs = InitStack ();
+  JobPool_AllJobs = InitStack ();
 
   JobPool_Pending = InitStack ();
@@ -32 +32 @@
 
 void FreeJobStacks () {
-
-  // AllJobs is a virtual stack : all jobs are references
-  // FreeStack (JobPool_AllJobs);
-
   FreeJobStack (JobPool_Pending);
   FreeJobStack (JobPool_Busy   );
@@ -43 +39 @@
   FreeJobStack (JobPool_Exit   );
   FreeJobStack (JobPool_Crash  );
+
+  // AllJobs is a virtual stack : all jobs are references
+  FreeStack (JobPool_AllJobs);
 }
 
 char *GetJobStackName (int StackID) {
   switch (StackID) {
-    // case PCONTROL_JOB_ALLJOBS: return ("ALLJOBS");
+    case PCONTROL_JOB_ALLJOBS: return ("ALLJOBS");
 
     case PCONTROL_JOB_PENDING: return ("PENDING");
@@ -64 +63 @@
 Stack *GetJobStack (int StackID) {
   switch (StackID) {
-    // case PCONTROL_JOB_ALLJOBS: return (JobPool_AllJobs);
+    case PCONTROL_JOB_ALLJOBS: return (JobPool_AllJobs);
 
     case PCONTROL_JOB_PENDING: return (JobPool_Pending);
@@ -81 +80 @@
 Stack *GetJobStackByName (char *name) {
 
-  // if (!strcasecmp (name, "all"))     return (JobPool_AllJobs);
+  if (!strcasecmp (name, "all"))     return (JobPool_AllJobs);
 
   if (!strcasecmp (name, "pending")) return (JobPool_Pending);
@@ -186 +185 @@
   job[0].argv     = argv;
   job[0].hostname = hostname;
+  job[0].realhost = NULL;
+
+  job[0].exit_status = 0;
+  job[0].Reset    = FALSE;
+  job[0].stdout_size = 0;
+  job[0].stderr_size = 0;
+
   job[0].mode     = mode;
-  job[0].host     = NULL;
-  job[0].JobID    = NextJobID();
-  job[0].Reset    = FALSE;
-  job[0].realhost = NULL;
+
+  job[0].state = 0;
+  job[0].stack = 0;
 
   /* do this step on start? */
@@ -196 +201 @@
   InitIOBuffer (&job[0].stderr_buff, 0x1000);
 
+  job[0].dtime = 0.0;
+  job[0].pid = 0;
+
+  job[0].JobID    = NextJobID();
+  job[0].host     = NULL;
+
   JobID = job[0].JobID;
 
+  // Put a copy of all created jobs on the ALLJOBS stack
+  // This is a virtual stack: do not free the job from this stack
+  PutJob (job, PCONTROL_JOB_ALLJOBS, STACK_BOTTOM);
   PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM);
-  // PutJob (job, PCONTROL_JOB_ALLJOBS, STACK_BOTTOM);
 
   if (VerboseMode()) gprint (GP_ERR, "added new job\n");
@@ -208 +221 @@
 
   int i;
+
+  Job *copy;
+
+  copy = PullStackByID (JobPool_AllJobs, job[0].JobID);
+  ASSERT (copy == job, "programming error: ALLJOBS entry does not match");
 
   FREE (job[0].hostname);

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

@@ -95 +95 @@
 int StopHostResponse (Host *host) {
 
+  OffHost (host);
   HarvestHost (host[0].pid);
   return (TRUE);
@@ -104 +105 @@
   int i, result, waitstatus;
 
-  gprint (GP_ERR, "harvesting within thread %d\n", pthread_self());
-  gprint (GP_ERR, "child process %d is down, wait for exit status\n", pid);
+  if (VerboseMode()) gprint (GP_ERR, "harvesting within thread %p\n", pthread_self());
+  if (VerboseMode()) gprint (GP_ERR, "child process %d is down, wait for exit status\n", pid);
   
   // Loop a few times waiting for child to exit

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

@@ -3 +3 @@
 int check (int argc, char **argv) {
 
-  Job *job;
-  Host *host;
-  int JobID, HostID, StackID;
+  int JobID, HostID;
+
+  Stack *stack = NULL;
+  Job *job = NULL;
+  Host *host = NULL;
 
   if (argc != 3) {
@@ -14 +16 @@
 
   if (!strcasecmp (argv[1], "JOB")) {
-    JobID = atoi (argv[2]);
+    JobID = GetID (argv[2]);
+    if (!JobID) {
+      gprint (GP_ERR, "invalid job id %s\n", argv[2]);
+      return (FALSE);
+    }
 
-    SetCheckPoint ();  // ensure the JOB is on one of the stacks
-    job = PullJobByID (JobID, &StackID);
+    stack = GetJobStack (PCONTROL_JOB_ALLJOBS);
+    job = PullStackByID (stack, JobID);
     if (job == NULL) {
       gprint (GP_LOG, "job not found\n");
-      ClearCheckPoint ();
       return (FALSE);
     }
-    gprint (GP_LOG, "STATUS %s\n", GetJobStackName(StackID));
+
+    gprint (GP_LOG, "STATUS %s\n", GetJobStackName(job[0].stack));
     gprint (GP_LOG, "EXITST %d\n", job[0].exit_status);
     gprint (GP_LOG, "STDOUT %d\n", job[0].stdout_size);
@@ -33 +39 @@
         gprint (GP_LOG, "HOSTNAME NONE\n");
     }
-    PutJob (job, StackID, STACK_BOTTOM);
-    ClearCheckPoint ();
+    PushStack (stack, STACK_BOTTOM, job, job[0].JobID, job[0].argv[0]);
     return (TRUE);
   }
@@ -41 +46 @@
     HostID = atoi (argv[2]);
 
-    SetCheckPoint ();  // ensure the HOST is on one of the stacks
-    host = PullHostByID (HostID, &StackID);
+    stack = GetHostStack (PCONTROL_HOST_ALLHOSTS);
+    host = PullStackByID (stack, HostID);
     if (host == NULL) {
       gprint (GP_LOG, "host not found\n");
-      ClearCheckPoint ();
       return (FALSE);
     }
-    gprint (GP_LOG, "host %s\n", GetHostStackName(StackID));
-    PutHost (host, StackID, STACK_BOTTOM);
-    ClearCheckPoint ();
+    gprint (GP_LOG, "host %s\n", GetHostStackName(host[0].stack));
+    PushStack (stack, STACK_BOTTOM, host, host[0].HostID, host[0].hostname);
     return (TRUE);
   }

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

@@ -10 +10 @@
     return (FALSE);
   }
-  JobID = atoi (argv[1]);
+  JobID = GetID (argv[1]);
+  if (!JobID) {
+    gprint (GP_ERR, "invalid job id %s\n", argv[1]);
+    return (FALSE);
+  }
+      
   /* use a string interp to convert JobIDs to ints ? */
 

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

@@ -1 +1 @@
 # include "pcontrol.h"
 
-// we use CheckPoints in this function to prevent objects in flight from being missing.
 int host (int argc, char **argv) {
 
-  int StackID;
   IDtype HostID;
   Host *host;
+  Stack *AllHosts;
 
   if (argc != 3) goto usage;
+
+  AllHosts = GetHostStack (PCONTROL_HOST_ALLHOSTS);
 
   if (!strcasecmp (argv[1], "ADD")) {
@@ -15 +16 @@
     return (TRUE);
   }
+
+  // this one is safe from in-flight entries: no one else pulls from OFF
   if (!strcasecmp (argv[1], "ON")) {
     host = PullHostFromStackByName (PCONTROL_HOST_OFF, argv[2]);
@@ -25 +28 @@
     return (TRUE);
   }
+
+  // this is a race condition with "CheckDownHosts", but the only 
+  // consequence is that both StartHost and reset set the times to 0.0
   if (!strcasecmp (argv[1], "RETRY")) {
     // no need to use a check point [thief: CheckDownHost (DOWN->IDLE)]
-    host = PullHostFromStackByName (PCONTROL_HOST_DOWN, argv[2]);
+    host = PullHostFromStackByName (PCONTROL_HOST_ALLHOSTS, argv[2]);
     if (!host) {
+      gprint (GP_LOG, "host %s not found\n", argv[2]);
+      return (FALSE);
+    }
+    if (host[0].stack != PCONTROL_HOST_DOWN) {
       gprint (GP_LOG, "host %s is not DOWN\n", argv[2]);
       return (FALSE);
     }
-    /* reset time, place back on DOWN stack */
+    /* reset time, place back on ALLHOSTS stack */
     host[0].nexttry.tv_sec  = 0;
     host[0].nexttry.tv_usec = 0;
     host[0].lasttry.tv_sec  = 0;
     host[0].lasttry.tv_usec = 0;
-    PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM);
+    PushStack (AllHosts, STACK_BOTTOM, host, host[0].HostID, host[0].hostname);
     return (TRUE);
   }
+
   if (!strcasecmp (argv[1], "CHECK")) {
-    SetCheckPoint ();  // ensure the host is on one of the stacks
-    host = PullHostByName (argv[2], &StackID);
+    host = PullHostFromStackByName (PCONTROL_HOST_ALLHOSTS, argv[2]);
     if (host == NULL) {
       gprint (GP_LOG, "host %s not found\n", argv[2]);
-      ClearCheckPoint ();
       return (FALSE);
     }
-    PutHost (host, StackID, STACK_BOTTOM);
-    ClearCheckPoint ();
-
-    gprint (GP_LOG, "host %s is %s\n", argv[2], GetHostStackName (StackID));
+    gprint (GP_LOG, "host %s is %s\n", argv[2], GetHostStackName (host[0].stack));
+    PushStack (AllHosts, STACK_BOTTOM, host, host[0].HostID, host[0].hostname);
     return (TRUE);
   }
+
   if (!strcasecmp (argv[1], "OFF")) {
-    SetCheckPoint (); // ensure we can find the specified host
-    host = PullHostByName (argv[2], &StackID);
+    host = PullHostFromStackByName (PCONTROL_HOST_ALLHOSTS, argv[2]);
     if (host == NULL) {
       gprint (GP_LOG, "host %s not found\n", argv[2]);
-      ClearCheckPoint ();
       return (FALSE);
     }
     host[0].markoff = TRUE;
-    PutHost (host, StackID, STACK_BOTTOM);
-    ClearCheckPoint ();
+    PushStack (AllHosts, STACK_BOTTOM, host, host[0].HostID, host[0].hostname);
     return (TRUE);
   }
 
+  // this one is safe from in-flight entries: no one else pulls from OFF
   if (!strcasecmp (argv[1], "DELETE")) {
-    // a check point is not required: no possible thief
     host = PullHostFromStackByName (PCONTROL_HOST_OFF, argv[2]);
     if (!host) {

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

@@ -68 +68 @@
 /* add program-dependent exit functions here */
 void cleanup () {
+  // stop checking on the jobs
+  SetRunLevel (PCONTROL_RUN_HOSTS); 
   DownHosts ();
   ConfigFree ();

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

@@ -6 +6 @@
 int status (int argc, char **argv) {
 
-  SetCheckPoint ();
-  PrintJobStack (PCONTROL_JOB_PENDING);
-  PrintJobStack (PCONTROL_JOB_BUSY);
-  PrintJobStack (PCONTROL_JOB_DONE);
-  PrintJobStack (PCONTROL_JOB_KILL);
-  PrintJobStack (PCONTROL_JOB_EXIT);
-  PrintJobStack (PCONTROL_JOB_CRASH);
+  PrintJobStack (PCONTROL_JOB_ALLJOBS);
+  PrintHostStack (PCONTROL_HOST_ALLHOSTS);
 
-  PrintHostStack (PCONTROL_HOST_OFF);
-  PrintHostStack (PCONTROL_HOST_DOWN);
-  PrintHostStack (PCONTROL_HOST_IDLE);
-  PrintHostStack (PCONTROL_HOST_BUSY);
-  PrintHostStack (PCONTROL_HOST_DONE);
-  ClearCheckPoint ();
   return (TRUE);
 }
@@ -39 +28 @@
     job = stack[0].object[i];
     ASSERT (job != NULL, "programming error");
-    gprint (GP_LOG, "%d  %s  %d  ", i, job[0].hostname, job[0].argc);
+    if (job[0].realhost == NULL) {
+        gprint (GP_LOG, "%3d %9s ", i, job[0].hostname);
+    } else {
+        gprint (GP_LOG, "%3d %9s ", i, job[0].realhost);
+    }
+    gprint (GP_LOG, "%7s  ", GetJobStackName (job[0].state));
     for (j = 0; j < job[0].argc; j++) {
       gprint (GP_LOG, "%s ", job[0].argv[j]);
@@ -66 +60 @@
     host = stack[0].object[i];
     gprint (GP_LOG, "%d  %s  ", i, host[0].hostname);
+    gprint (GP_LOG, "%5s  ", GetHostStackName (host[0].stack));
     PrintID (GP_LOG, host[0].HostID);
     gprint (GP_LOG, "\n");