Changeset 37595

Timestamp:

Nov 13, 2014, 4:32:59 PM (12 years ago)

Author:

eugene

Message:

update relphot to launch only remote host groups (at least for updating the objects)

Location:

branches/eam_branches/ipp-20140904/Ohana/src

Files:

• libdvo/include/dvo.h (modified) (1 diff)
• libdvo/src/HostTable.c (modified) (3 diffs)
• relphot/src/reload_catalogs.c (modified) (5 diffs)

branches/eam_branches/ipp-20140904/Ohana/src/libdvo/include/dvo.h

@@ -273 +273 @@
   short *index;
 } HostTable;
+
+typedef struct {
+  int Nhosts;
+  HostInfo **hosts;
+} HostTableGroup;
 
 // A RegionHost processes data for some region in parallel with other regions

branches/eam_branches/ipp-20140904/Ohana/src/libdvo/src/HostTable.c

@@ -40 +40 @@
 }
 
+// a HostTableGroup is a pointer set to one or more HostTable entries
+void FreeTableGroup (HostTableGroup *table) {
+
+  if (!table) return;
+  if (table->hosts) {
+    free (table->hosts);
+  }
+  if (table->index) free (table->index);
+  free (table);
+  return;
+}
+
 HostTable *HostTableLoad (char *catdir, char *rootname) {
 
@@ -136 +148 @@
   fclose (f);
   return table;
+}
+
+// split a host table into Ngroups, each with a unique set of hosts
+HostTableGroup *HostTableGroups (HostTable *table, int *ngroups) {
+
+  // identify the unique host names and the number of times they each exist
+  // (Host tables are not very large; I can afford to do this in an inefficient way)
+
+  int Nunique = 0;
+  char **unique_hosts = NULL; // an array of pointers to a set of the names
+  int  *Nunique_hosts = NULL; // an array of pointers to the count of each
+  ALLOCATE ( unique_hosts, char *, table->Nhosts);
+  ALLOCATE (Nunique_hosts, int, table->Nhosts);
+  for (i = 0; i < table->Nhosts; i++) Nunique_hosts[i] = 0;
+
+  for (i = 0; i < table->Nhosts; i++) {
+    found = FALSE;
+    for (j = 0; !found && (j < Nunique); j++) {
+      if (!strcmp(table->hosts[i].hostname, unique_hosts[j])) {
+        Nunique_hosts[j] ++;
+        found = TRUE;
+      }
+    }
+    if (!found) {
+      unique_hosts[Nunique] = table->hosts[i].hostname;
+      Nunique_hosts[Nunique] = 1;
+      Nunique ++;
+    }
+  }
+
+  int Ngroups = 0;
+  for (i = 0; i < Nunique; i++) {
+    Ngroups = MAX (Nunique_hosts[i], Ngroups);
+  }
+
+  HostTableGroup *groups = NULL;
+  int *foundHost = NULL;
+  
+  ALLOCATE (foundHost, int, table->Nhosts);
+  ALLOCATE (groups, HostTableGroup, Ngroups);
+
+  for (i = 0; i < table->Nhosts; i++) foundHost[i] = FALSE;
+
+  // in each group, attempt to add one of each unique host
+  for (i = 0; i < Ngroups; i++) {
+    groups[i].Nhosts = 0;
+    ALLOCATE (groups[i].hosts, HostTable *, Nunique);
+    for (j = 0; j < Nunique; j++) {
+      found = FALSE;
+      for (k = 0; !found && (k < table->Nhosts); k++) {
+        if (foundHost[k]) continue;
+        if (strcmp(unique_hosts[j], table->hosts[k].hostname));
+        groups[i].hosts[groups[i].Nhosts] = &table->hosts[k];
+        groups[i].Nhosts ++;
+        found = TRUE;
+        foundHost[k] = TRUE;
+      }
+    }
+  }
+  *ngroups = Ngroups;
+  return groups;
 }
 
@@ -389 +462 @@
 }
 
+// wait for all children to complete, report output to stdout
+int HostTableGroupWaitJobsGetIO (HostTableGroup *table, char *file, int lineno, int VERBOSE) {
+
+  // we have launched table->Nhosts jobs; wait for all of them to complete...
+  // if one (N) failed to launch, we will get an ECHILD error from the last (N) calls
+
+  // we need to read any data waiting on stderr or stdout from these jobs, or the overfull
+  // buffers can cause a problem.  we alternate between 'select' and 'waitpid' calls with
+  // timeouts for both
+
+  // add all hosts' sockets to the fd_sets
+  fd_set rdSet, wtSet;
+  FD_ZERO (&rdSet);
+  FD_ZERO (&wtSet);
+
+  // XXX can I set the fd_sets once, since I am not actually closing the fd's?
+
+  int globalStatus = TRUE;
+
+  int i;
+  int Nmax = 0;
+  for (i = 0; i < table->Nhosts; i++) {
+    if (!table->hosts[i][0].pid) continue; // any unconnected hosts should be skipped
+    FD_SET (table->hosts[i][0].stdio[HOST_STDIN], &wtSet);
+    Nmax = MAX (Nmax, table->hosts[i][0].stdio[HOST_STDIN]);
+    FD_SET (table->hosts[i][0].stdio[HOST_STDOUT], &rdSet);
+    Nmax = MAX (Nmax, table->hosts[i][0].stdio[HOST_STDOUT]);
+    FD_SET (table->hosts[i][0].stdio[HOST_STDERR], &rdSet);
+    Nmax = MAX (Nmax, table->hosts[i][0].stdio[HOST_STDERR]);
+  }    
+  Nmax ++;
+
+  // need the list of connected hosts for exit test below
+  int Nrunning = 0;
+  for (i = 0; i < table->Nhosts; i++) {
+    if (!table->hosts[i][0].pid) continue; // any unconnected hosts should be skipped
+    Nrunning ++;
+  }
+
+  int Nfound = 0;
+
+  // this loop has 2 chunks: (a) check for I/O + (b) check for jobs done
+  while (1) {
+
+    // Wait up to 0.5 second for host to provide I/O
+    // timeout gets mucked: need to reset before each select
+    struct timeval timeout;
+    timeout.tv_sec = 10;
+    timeout.tv_usec = 500000;
+
+    int status = select (Nmax, NULL, &wtSet, NULL, &timeout);
+    if (status == -1) {
+      perror("select()");
+      exit (2);
+    }
+
+    // we have some sockets to check, check sockets for all hosts
+    for (i = 0; (status > 0) && (i < table->Nhosts); i++) {
+      if (!table->hosts[i][0].pid) continue; // any unconnected hosts should be skipped
+
+      if (FALSE && FD_ISSET (table->hosts[i][0].stdio[HOST_STDIN], &wtSet)) {
+        // this host is waiting for input : this is an error, so exit
+        fprintf (stderr, "host %s is waiting for input\n", table->hosts[i][0].hostname);
+        abort();
+      }
+      
+      if ((table->hosts[i][0].stdio[HOST_STDOUT] > 0) && FD_ISSET (table->hosts[i][0].stdio[HOST_STDOUT], &rdSet)) {
+        // this host has waiting output : read to buffer, and dump if necessary
+        ReadtoIOBuffer (&table->hosts[i][0].stdout, table->hosts[i][0].stdio[HOST_STDOUT]);
+        // if (table->hosts[i][0].stdout.Nbuffer > 0x10000) {
+        if (table->hosts[i][0].stdout.Nbuffer > 0x1000) {
+          int printHead = VERBOSE || (table->hosts[i][0].stdout.Nbuffer > 0);
+          if (printHead) fprintf (stdout, "--- stdout from %s --- (%d bytes, v1)\n", table->hosts[i][0].hostname, table->hosts[i][0].stdout.Nbuffer);
+          int Nout = write (STDOUT_FILENO, table->hosts[i][0].stdout.buffer, table->hosts[i][0].stdout.Nbuffer);
+          if (Nout != table->hosts[i][0].stdout.Nbuffer) { fprintf (stderr, "(error writing log?)\n"); }
+          FlushIOBuffer (&table->hosts[i][0].stdout);
+          if (printHead) fprintf (stdout, "\n");
+        }
+      }
+
+      if ((table->hosts[i][0].stdio[HOST_STDERR] > 0) && FD_ISSET (table->hosts[i][0].stdio[HOST_STDERR], &rdSet)) {
+        // this host has waiting output : read to buffer, and dump if necessary
+        ReadtoIOBuffer (&table->hosts[i][0].stderr, table->hosts[i][0].stdio[HOST_STDERR]);
+        // if (table->hosts[i][0].stderr.Nbuffer > 0x10000) {
+        if (table->hosts[i][0].stderr.Nbuffer > 0x1000) {
+          int printHead = VERBOSE || (table->hosts[i][0].stderr.Nbuffer > 0);
+          if (printHead) fprintf (stdout, "--- stderr from %s --- (%d bytes, v1)\n", table->hosts[i][0].hostname, table->hosts[i][0].stderr.Nbuffer);
+          int Nout = write (STDOUT_FILENO, table->hosts[i][0].stderr.buffer, table->hosts[i][0].stderr.Nbuffer);
+          if (Nout != table->hosts[i][0].stderr.Nbuffer) { fprintf (stderr, "(error writing log?)\n"); }
+          FlushIOBuffer (&table->hosts[i][0].stderr);
+          if (printHead) fprintf (stdout, "\n");
+        }
+      }
+    }
+
+    // now check if any children have finished...
+    while (TRUE) {
+      int status = 0;
+      int pid = waitpid (-1, &status, WNOHANG);
+      if (!pid) {
+        // fprintf (stderr, "no hosts to harvest\n");
+        usleep (500000);
+        break; // no outstanding jobs have finished 
+      }
+      if ((pid == -1) && (errno == ECHILD)) goto escape; // no more jobs on which to wait
+      if ((pid == -1) && (errno != ECHILD)) {
+        fprintf (stderr, "programming error (2)? %s %d", file, lineno);
+        exit (2);
+      }
+
+      // find the host which has finished
+      int found = FALSE;
+      for (i = 0; (i < table->Nhosts) && !found; i++) {
+        if (table->hosts[i][0].pid != pid) continue;
+        found = TRUE;
+
+        HostInfo *host = table->hosts[i];
+
+        // check on the status of this and report any output?
+        if (VERBOSE) fprintf (stdout, "job finished for %s (%d)\n", host->hostname, pid);
+
+        // read stdout
+        int printHead;
+        printHead = VERBOSE || (host->stdout.Nbuffer > 0);
+        EmptyIOBuffer (&host->stdout, 100, host->stdio[HOST_STDOUT]);
+        if (printHead) fprintf (stdout, "--- stdout from %s --- (%d bytes, v2)\n", host->hostname, host->stdout.Nbuffer);
+        int Nout = write (STDOUT_FILENO, host->stdout.buffer, host->stdout.Nbuffer);
+        if (Nout != host->stdout.Nbuffer) { fprintf (stderr, "(error writing log?)\n"); }
+        FlushIOBuffer (&host->stdout);
+        if (printHead) fprintf (stdout, "\n");
+            
+        // read stderr
+        printHead = VERBOSE || (host->stderr.Nbuffer > 0);
+        EmptyIOBuffer (&host->stderr, 100, host->stdio[HOST_STDERR]);
+        if (printHead) fprintf (stdout, "--- stderr from %s --- (%d bytes, v2)\n", host->hostname, host->stderr.Nbuffer);
+        Nout = write (STDOUT_FILENO, host->stderr.buffer, host->stderr.Nbuffer);
+        if (Nout != host->stderr.Nbuffer) { fprintf (stderr, "(error writing log?)\n"); }
+        FlushIOBuffer (&host->stderr);
+        if (printHead) fprintf (stdout, "\n");
+
+        if (WIFEXITED(status)) {
+          if (VERBOSE) fprintf (stdout, "normal completion, exit status is %d\n", WEXITSTATUS(status));
+          host->status = WEXITSTATUS(status);
+          if (host->status) {
+            fprintf (stdout, "job failed on %s\n", host->hostname);
+            globalStatus = FALSE;
+          }
+        } else {
+          host->status = -1;
+          fprintf (stdout, "job exited abnormally on %s\n", host->hostname);
+          globalStatus = FALSE;
+          continue;
+        }
+      }
+      if (!found) {
+        fprintf (stderr, "Programming error: failed to matched finished job to known host!\n");
+        exit (2);
+      }
+      Nfound ++;
+      if (Nfound == Nrunning) goto escape; // we've harvested all jobs
+    }
+  }
+
+escape:
+
+  // close all opened connections
+  for (i = 0; i < table->Nhosts; i++) {
+    if (!table->hosts[i][0].pid) continue; // any unconnected hosts should be skipped
+    close (table->hosts[i][0].stdio[HOST_STDIN]);
+    close (table->hosts[i][0].stdio[HOST_STDOUT]);
+    close (table->hosts[i][0].stdio[HOST_STDERR]);
+  }
+
+  return globalStatus;
+}
+
 int HostTableTestHost (SkyRegion *region, int hostID) {
 

branches/eam_branches/ipp-20140904/Ohana/src/relphot/src/reload_catalogs.c

@@ -167 +167 @@
   }
 
+  int Ngroups;
+  HostTable *groups = HostTableGroups (table, &Ngroups);
+  // split the table into Ngroups, each with a unique set of machines (this avoids overloading the remote host machines)
+
+  for (i = 0; i < Ngroups; i++) {
+    // update only a group of unique machines at a time
+    reload_catalog_parallel_group (&groups[i], sky, imageFile);
+  }
+  return TRUE;
+}
+
+int reload_catalog_parallel_group (HostTableGroup *group, SkyList *sky, char *imageFile) {
+
   int i, j;
-  for (i = 0; i < table->Nhosts; i++) {
-
-    if (sky->Nregions < table->Nhosts) {
+  for (i = 0; i < group->Nhosts; i++) {
+
+    if (sky->Nregions < group->Nhosts) {
       // do any of the regions want this host?
       int wantThisHost = FALSE;
       for (j = 0; j < sky->Nregions; j++) {
-        if (HostTableTestHost (sky->regions[j], table->hosts[i].hostID)) {
+        if (HostTableTestHost (sky->regions[j], group->hosts[i][0].hostID)) {
           wantThisHost = TRUE;
           break;
@@ -180 +193 @@
       }
       if (!wantThisHost) {
-        // fprintf (stderr, "skip host %s\n", table->hosts[i].hostname);
+        // fprintf (stderr, "skip host %s\n", group->hosts[i][0].hostname);
         continue;
       }
@@ -186 +199 @@
 
     // ensure that the paths are absolute path names
-    char *tmppath = abspath (table->hosts[i].pathname, DVO_MAX_PATH);
-    free (table->hosts[i].pathname);
-    table->hosts[i].pathname = tmppath;
+    char *tmppath = abspath (group->hosts[i][0].pathname, DVO_MAX_PATH);
+    free (group->hosts[i][0].pathname);
+    group->hosts[i][0].pathname = tmppath;
 
     char command[1024];
     snprintf (command, 1024, "relphot_client %s -update-catalogs %s -hostID %d -D CATDIR %s -hostdir %s -region %f %f %f %f -statmode %s -D CAMERA %s -D STAR_TOOFEW %d -minerror %f", 
-              PhotcodeList, imageFile, table->hosts[i].hostID, CATDIR, table->hosts[i].pathname, UserPatch.Rmin, UserPatch.Rmax, UserPatch.Dmin, UserPatch.Dmax, STATMODE, CAMERA, STAR_TOOFEW, MIN_ERROR);
+              PhotcodeList, imageFile, group->hosts[i][0].hostID, CATDIR, group->hosts[i][0].pathname, UserPatch.Rmin, UserPatch.Rmax, UserPatch.Dmin, UserPatch.Dmax, STATMODE, CAMERA, STAR_TOOFEW, MIN_ERROR);
 
     // options & configs which affect relphot_client -update-catalogs
@@ -237 +250 @@
       // launch the job on the remote machine (no handshake)
       int errorInfo = 0;
-      int pid = rconnect ("ssh", table->hosts[i].hostname, command, table->hosts[i].stdio, &errorInfo, FALSE);
+      int pid = rconnect ("ssh", group->hosts[i][0].hostname, command, group->hosts[i][0].stdio, &errorInfo, FALSE);
       if (!pid) {
-        if (DEBUG) fprintf (stderr, "failure to start %s (error %d)\n", table->hosts[i].hostname, errorInfo);
+        if (DEBUG) fprintf (stderr, "failure to start %s (error %d)\n", group->hosts[i][0].hostname, errorInfo);
         exit (1);
       }
-      table->hosts[i].pid = pid; // save for future reference
+      group->hosts[i][0].pid = pid; // save for future reference
     }
   }
@@ -251 +264 @@
   } 
   if (!PARALLEL_MANUAL && !PARALLEL_SERIAL) {
-    int status = HostTableWaitJobsGetIO (table, __FILE__, __LINE__, VERBOSE);
+    int status = HostTableGroupWaitJobsGetIO (group, __FILE__, __LINE__, VERBOSE);
     if (!status) {
       fprintf (stderr, "at least one remote client job failed to load data, exiting\n");