Changeset 19125

Timestamp:

Aug 19, 2008, 12:29:10 PM (18 years ago)

Author:

eugene

Message:

adding host tracking and restart

Location:

trunk/Ohana/src/opihi

Files:

• include/pantasks.h (modified) (2 diffs)
• pantasks/ControllerOps.c (modified) (6 diffs)
• pantasks/controller_host.c (modified) (3 diffs)
• pantasks/test/threads.sh (added)

trunk/Ohana/src/opihi/include/pantasks.h

@@ -58 +58 @@
   int Nrun;
 } TimeRange;
+
+/* data to define a host machine */
+typedef struct {
+  char       *hostname;
+  int         max_threads;
+} Host;
 
 /* a task is a description of the wrapping of a job */
@@ -233 +239 @@
 int QuitController ();
 int StopController ();
+int RestartController ();
 int VerboseMode ();
 int KillLocalJob (Job *job);

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

@@ -10 +10 @@
 static IOBuffer stderr_buffer;
 static int ControllerPID = 0;
+
+/* local static variables to track the controller host properties */
+static Host *hosts = NULL;
+static int Nhosts = 0;
+static int NHOSTS = 0;
+
+int AddHost (char *hostname, int max_threads) {
+
+  int N;
+
+  N = Nhosts;
+  Nhosts ++;
+
+  CHECK_REALLOCATE (hosts, Host, NHOSTS, Nhosts, 16);
+
+  hosts[N].hostname = strcreate (hostname);
+  hosts[N].max_threads = max_threads;
+  
+  return (TRUE);
+}
+
+int DeleteHost (char *hostname) {
+
+  int i, j;
+
+  for (i = 0; i < Nhosts; i++) {
+    if (strcmp (hosts[i].hostname, hostname)) continue;
+    
+    // delete this one
+    free (hosts[i].hostname);
+    for (j = i; j < Nhosts - 1; j++) {
+      hosts[j].hostname = hosts[j+1].hostname;
+      hosts[j].max_threads = hosts[j+1].max_threads;
+    }
+    Nhosts --;
+    return (TRUE);
+  }
+  return (FALSE);
+}
 
 /* test if the controller is running */
@@ -146 +185 @@
 
   /* is pipe still open? */
-  if ((status == -1) && (errno == EPIPE)) return (CONTROLLER_DOWN);
+  // XXX call StopController() here?
+  if ((status == -1) && (errno == EPIPE)) {
+    StopController ();
+    return (CONTROLLER_DOWN);
+  }
 
   /* read at least Nbytes, then watch for CONTROLLER_PROMPT */
@@ -158 +201 @@
     if (status == -1) nanosleep (&request, &remain);
   }
-  if (status ==  0) return (CONTROLLER_DOWN);
+  if (status ==  0) {
+    StopController ();
+    return (CONTROLLER_DOWN);
+  }
   if (status == -1) return (CONTROLLER_HUNG);
 
@@ -245 +291 @@
   if (VarConfig ("CONTROLLER", "%s", cmd) == NULL) strcpy (cmd, "pcontrol");
 
+  if (hosts == NULL) {
+    NHOSTS = 16;
+    ALLOCATE (hosts, Host, NHOSTS);
+  }
+
   bzero (stdin_fd,  2*sizeof(int));
   bzero (stdout_fd, 2*sizeof(int));
@@ -360 +411 @@
   if ((status == -1) && (errno == EPIPE)) {
     StopController ();
-    gprint (GP_ERR, "controller is down\n");
+    gprint (GP_ERR, "controller is down, restarting\n");
+    if (!RestartController ()) {
+      return (FALSE);
+    }
     return (FALSE);
   }
@@ -509 +563 @@
   return (TRUE);
 }
+
+int RestartController () {
+
+  int i, status;
+  char command[256];
+  IOBuffer buffer;
+
+  gprint (GP_ERR, "controller is down, restarting\n");
+  if (!StartController ()) {
+    gprint (GP_ERR, "failure to re-start pcontrol\n");
+    return (FALSE);
+  }
+
+  InitIOBuffer (&buffer, 0x100);
+
+  // XXX lock the host table? SerialThreadLock ();
+  fprintf (stderr, "controller is down, restarting\n");
+  for (i = 0; i < Nhosts; i++) {
+    fprintf (stderr, "controller host add %s -threads %d\n", hosts[i].hostname, hosts[i].max_threads);
+    snprintf (command, 256, "host add %s -threads %d\n", hosts[i].hostname, hosts[i].max_threads);
+    status = ControllerCommand (command, CONTROLLER_PROMPT, &buffer);
+  }
+  // SerialThreadUnlock ();
+
+  if (status) gwrite (buffer.buffer, 1, buffer.Nbuffer, GP_LOG);
+
+  FreeIOBuffer (&buffer);
+
+  return (TRUE);
+}

trunk/Ohana/src/opihi/pantasks/controller_host.c

@@ -3 +3 @@
 int controller_host (int argc, char **argv) {
 
-  int status;
+  int N, status, max_threads;
   char command[1024];
   IOBuffer buffer;
 
-  if (argc != 3) {
-    gprint (GP_ERR, "USAGE: controller host (command) (hostname)\n");
-    return (FALSE);
+  max_threads = 0;
+  if ((N = get_argument (argc, argv, "-threads"))) {
+    remove_argument (N, &argc, argv);
+    max_threads = atoi(argv[N]);
+    remove_argument (N, &argc, argv);
   }
+
+  if (argc != 3) goto usage;
 
   /* start controller connection (if needed) */
@@ -18 +22 @@
   }
 
-  sprintf (command, "host %s %s", argv[1], argv[2]);
+  // the user may issue any of these commands:
+  // ADD, ON, RETRY, CHECK, OFF, DELETE
+  // we need to catch ADD and DELETE and modify our host table accordingly
+  if (!strcasecmp (argv[1], "ADD")) {
+    AddHost (argv[2], max_threads);
+  } else {
+    if (max_threads) goto usage;
+  }
+
+  if (!strcasecmp (argv[1], "DELETE")) {
+    DeleteHost (argv[2]);
+  }
+
+  if (max_threads) {
+    sprintf (command, "host %s %s -threads %d", argv[1], argv[2], max_threads);
+  } else {
+    sprintf (command, "host %s %s", argv[1], argv[2]);
+  }
   InitIOBuffer (&buffer, 0x100);
 
@@ -29 +50 @@
   FreeIOBuffer (&buffer);
   return (TRUE);
+  
+usage:
+  gprint (GP_LOG, "USAGE: controller host (command) (hostname)\n");
+  gprint (GP_ERR, "  valid commands: add, on, retry, check, off, delete\n");
+  gprint (GP_ERR, "  -threads Nthreads is optional for 'add'\n");
+  return (FALSE);
 }