Changeset 33963 for trunk/Ohana/src/opihi/dvo/dvo_host_utils.c

Timestamp:

May 30, 2012, 1:46:12 PM (14 years ago)

Author:

eugene

Message:

merge changes from eam_branches/ipp-20120405

Location:

trunk

Files:

• . (modified) (1 prop)
• Ohana/src/opihi/dvo/dvo_host_utils.c (modified) (9 diffs)

trunk/Ohana/src/opihi/dvo/dvo_host_utils.c

@@ -4 +4 @@
 # define PARALLEL_MANUAL 0
 # define PARALLEL_SERIAL 0
-# define MAX_PATH_LENGTH 1024
-
-int HostTableLaunchJobs (HostTable *table, char *basecmd, char *options) {
+# define DVO_MAX_PATH 1024
+
+# define DIE(WHO,MSG) { perror(WHO); myAbort(MSG); }
+
+int HostTableLaunchJobs (HostTable *table, char *basecmd, char *options, int VERBOSE) {
 
   char uniquer[12];
@@ -13 +15 @@
   snprintf (uniquer, 12, "%05d.%05d", PID, TIME % 100000);
 
+  int top_status = TRUE;
   int i;
   for (i = 0; i < table->Nhosts; i++) {
 
     // ensure that the paths are absolute path names
-    char *tmppath = abspath (table->hosts[i].pathname, MAX_PATH_LENGTH);
+    char *tmppath = abspath (table->hosts[i].pathname, DVO_MAX_PATH);
     free (table->hosts[i].pathname);
     table->hosts[i].pathname = tmppath;
@@ -23 +26 @@
     // need to save the results filename with the uniquer
     // XXX a bit of a waste (but only 1024 * 60 bytes or so
-    ALLOCATE (table->hosts[i].results, char, MAX_PATH_LENGTH);
-    snprintf (table->hosts[i].results, MAX_PATH_LENGTH, "%s/dvo.results.%s.fits", table->hosts[i].pathname, uniquer);
-
-    char commandBase[MAX_PATH_LENGTH];
-    snprintf (commandBase, MAX_PATH_LENGTH, "dvo.command.%s.txt", uniquer);
-    char *commandFile = abspath(commandBase, MAX_PATH_LENGTH);
+    ALLOCATE (table->hosts[i].results, char, DVO_MAX_PATH);
+    snprintf (table->hosts[i].results, DVO_MAX_PATH, "%s/dvo.results.%s.fits", table->hosts[i].pathname, uniquer);
+
+    char commandBase[DVO_MAX_PATH];
+    snprintf (commandBase, DVO_MAX_PATH, "dvo.command.%s.txt", uniquer);
+    char *commandFile = abspath(commandBase, DVO_MAX_PATH);
 
     FILE *f = fopen (commandFile, "w");
     fprintf (f, "%s\n", basecmd);
-    fclose (f);
+    if (fflush (f)) DIE("flush", "failed to flush");
+
+    int fd = fileno (f);
+    if (fsync (fd)) DIE("fsync", "failed to fsync");
+
+    if (fclose (f)) DIE("close", "failed to close");
+
+    // force NFS to write the file to disk
+    int state;
+    f = fsetlockfile (commandFile, 0.5, LCK_XCLD, &state);
+    fclearlockfile (commandFile, f, LCK_XCLD, &state);
 
     char command[1024];
@@ -38 +51 @@
     free (commandFile);
 
-    if (DEBUG || PARALLEL_MANUAL) fprintf (stderr, "command: %s\n", command);
+    if (VERBOSE) gprint (GP_ERR, "command: %s\n", command);
 
     if (PARALLEL_MANUAL) {
@@ -47 +60 @@
       int status = system (command);
       if (status) {
-        fprintf (stderr, "ERROR running relphot_client\n");
-        exit (2);
+        gprint (GP_ERR, "ERROR running relphot_client\n");
+        top_status = FALSE;
       }
     } else {
@@ -55 +68 @@
       int pid = rconnect ("ssh", table->hosts[i].hostname, command, table->hosts[i].stdio, &errorInfo, FALSE);
       if (!pid) {
-        if (DEBUG) fprintf (stderr, "failure to start %s (error %d)\n", table->hosts[i].hostname, errorInfo);
-        exit (1);
+        gprint (GP_ERR, "failure to start %s (error %d)\n", table->hosts[i].hostname, errorInfo);
+        top_status = FALSE;
+        continue;
       }
       table->hosts[i].pid = pid; // save for future reference
     }
   }
-  return TRUE;
+  return top_status;
 }
 
@@ -86 +100 @@
   }
 
-  char *CATDIR = abspath (tmppath, MAX_PATH_LENGTH);
+  char *CATDIR = abspath (tmppath, DVO_MAX_PATH);
   if (!CATDIR) {
     gprint (GP_ERR, "failed to make an absolute path from %s (too long)\n", tmppath);
@@ -126 +140 @@
 
   // launch this command remotely
-  HostTableLaunchJobs (table, basecmd, options);
+  HostTableLaunchJobs (table, basecmd, options, VERBOSE);
   free (options);
   free (basecmd);
@@ -184 +198 @@
   return TRUE;
 }
+
+// re-gather the remote results files: this can be used in case one of the clients failed,
+// and has since been re-run
+int HostTableReloadResults (char *uniquer, int VERBOSE) {
+
+  int i;
+
+  // load the list of hosts
+  SkyTable *sky = GetSkyTable();
+  if (!sky) {
+    gprint (GP_ERR, "failed to load sky table for database\n");
+    return FALSE;
+  }
+
+  char *CATDIR = GetCATDIR ();
+  if (!CATDIR) {
+    gprint (GP_ERR, "failed to get CATDIR for database\n");
+    return FALSE;
+  }
+
+  HostTable *table = HostTableLoad (CATDIR, sky->hosts);
+  if (!table) {
+    gprint (GP_ERR, "ERROR: failure reading Host Table %s for database %s\n", sky->hosts, CATDIR);
+    return FALSE;
+  }    
+
+  // load fields from file
+  int    Nvec = 0;
+  Vector **vec = NULL;
+  for (i = 0; i < table->Nhosts; i++) {
+    // 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;
+
+    // need to save the results filename with the uniquer
+    // XXX a bit of a waste (but only 1024 * 60 bytes or so
+    ALLOCATE (table->hosts[i].results, char, DVO_MAX_PATH);
+    snprintf (table->hosts[i].results, DVO_MAX_PATH, "%s/dvo.results.%s.fits", table->hosts[i].pathname, uniquer);
+
+    int    Ninvec = 0;
+    Vector **invec = ReadVectorTableFITS (table->hosts[i].results, "RESULT", &Ninvec);
+    if (!invec) {
+      // failed to read the file, now what?
+      gprint (GP_ERR, "failed to read remote result file : %s\n", table->hosts[i].results);
+      free (table->hosts[i].results);
+      table->hosts[i].results = NULL;
+      continue;
+    }
+    free (table->hosts[i].results);
+    table->hosts[i].results = NULL;
+
+    // fprintf (stderr, "%s : %d\n", table->hosts[i].pathname, invec[0]->Nelements);
+
+    vec = MergeVectors (vec, &Nvec, invec, Ninvec);
+    if (vec != invec) {
+      FreeVectorArray (invec, Ninvec);
+    }
+  }
+
+  for (i = 0; i < Nvec; i++) {
+    AssignVector (vec[i], vec[i]->name, ANYVECTOR, TRUE);
+  }
+  free (vec);
+
+  free (table);
+  return TRUE;
+}