Changeset 33308 for branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/mextract.c

Timestamp:

Feb 18, 2012, 8:44:13 PM (14 years ago)

Author:

eugene

Message:

some support functions for dvo_client; mextract example nearly complete

File:

• branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/mextract.c (modified) (6 diffs)

branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/mextract.c

@@ -1 +1 @@
 # include "dvoshell.h"
+
+# define DEBUG 1
+# define PARALLEL_MANUAL 1
+# define PARALLEL_SERIAL 0
+# define MAX_PATH_LENGTH 1024
 
 int mextract (int argc, char **argv) {
@@ -34 +39 @@
   }
 
+  int PARALLEL = FALSE;
+  if ((N = get_argument (argc, argv, "-parallel"))) {
+    remove_argument (N, &argc, argv);
+    PARALLEL = TRUE;
+  }
+
+  // dump results directly to fits file (esp for parallel dvo)
+  char *ResultFile = NULL;
+  if ((N = get_argument (argc, argv, "-result"))) {
+    remove_argument (N, &argc, argv);
+    ResultFile = strcreate(argv[N]);
+    remove_argument (N, &argc, argv);
+  }
+
   // XXX not yet thought through (where are these set?)
   if (PARALLEL && !HOST_ID) {
-    HostTableLaunchJobs (CATDIR, argc, argv);
-    HostTableWaitJobs (table, __FILE__, __LINE__);
+    // load the list of hosts
+    SkyTable *sky = GetSkyTable();
+    if (!sky) return FALSE;
+
+    char *CATDIR = GetCATDIR ();
+    if (!CATDIR) return FALSE;
+
+    HostTable *table = HostTableLoad (CATDIR, sky->hosts);
+    if (!table) {
+      fprintf (stderr, "ERROR: failure reading Host Table %s for database %s\n", sky->hosts, CATDIR);
+      return FALSE;
+    }    
+
+    // other things I need to append?
+    char *basecmd = paste_args (argc, argv);
+    double Rmin, Rmax, Dmin, Dmax;
+    get_skyregion (&Rmin, &Rmax, &Dmin, &Dmax);
+
+    char tmp;
+    char *command = NULL;
+    int length = snprintf (&tmp, 0, "%s -D CATDIR %s -skyregion %f %f %f %f", basecmd, CATDIR, Rmin, Rmax, Dmin, Dmax);
+
+    ALLOCATE (command, char, length);
+    snprintf (command, length, "%s -D CATDIR %s -skyregion %f %f %f %f", basecmd, CATDIR, Rmin, Rmax, Dmin, Dmax);
+
+    // launch this command remotely
+    HostTableLaunchJobs (table, command);
+    free (command);
+
+    if (PARALLEL_MANUAL) {
+      fprintf (stderr, "run the relphot_client commands above.  when these are done, hit return\n");
+      getchar();
+    }
+    if (!PARALLEL_MANUAL && !PARALLEL_SERIAL) {
+      HostTableWaitJobs (table, __FILE__, __LINE__);
+    }
 
     // load fields from file
-    ReadVectorTables
+    // ReadVectorTables
     return TRUE;
   }
@@ -132 +185 @@
   for (i = 0; (i < skylist[0].Nregions) && !interrupt; i++) {
 
-    // XXX parallel case, wrong client
+    // parallel case, wrong client
     if (HOST_ID && (HOST_ID != skylist[0].regions[i]->hostID)) continue;
 
     /* lock, load, unlock catalog */
-    catalog.filename = skylist[0].filename[i];
+    char hostfile[1024];
+    snprintf (hostfile, 1024, "%s/%s.cpt", HOSTDIR, skylist[0].regions[i]->name);
+    catalog.filename = HOST_ID ? hostfile : skylist[0].filename[i];
     catalog.catflags = LOAD_AVES | LOAD_MEAS | LOAD_SECF;
     catalog.Nsecfilt = Nsecfilt;
@@ -205 +260 @@
   }
 
-  if (HOST_ID) {
-    // write table
-    WriteVectorTable (vec);
-    return TRUE;
+  if (ResultFile) {
+    // write vectors to a table
+    // char *resultFile?
+    int status = WriteVectorTableFITS (ResultFile, "MEXTRACT", vec, Nreturn, FALSE, NULL);
+    if (!status) goto escape;
   }
 
@@ -222 +278 @@
 
 escape:
+  if (ResultFile) free (ResultFile);
   if (vec) free (vec);
   free (values);
@@ -349 +406 @@
   return (FALSE);
 }
+
+int HostTableLaunchJobs (HostTable *table, char *basecmd) {
+
+  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);
+    free (table->hosts[i].pathname);
+    table->hosts[i].pathname = tmppath;
+
+    char resultFile[MAX_PATH_LENGTH];
+    snprintf (resultFile, MAX_PATH_LENGTH, "%s/dvo.results.fits", table->hosts[i].pathname);
+
+    char command[1024];
+    snprintf (command, 1024, "%s -result %s -hostID %d -hostdir %s", basecmd, resultFile, table->hosts[i].hostID, table->hosts[i].pathname);
+
+    fprintf (stderr, "command: %s\n", command);
+
+    if (PARALLEL_MANUAL) {
+      continue;
+    }
+
+    if (PARALLEL_SERIAL) {
+      int status = system (command);
+      if (status) {
+        fprintf (stderr, "ERROR running relphot_client\n");
+        exit (2);
+      }
+    } else {
+      // 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);
+      if (!pid) {
+        if (DEBUG) fprintf (stderr, "failure to start %s (error %d)\n", table->hosts[i].hostname, errorInfo);
+        exit (1);
+      }
+      table->hosts[i].pid = pid; // save for future reference
+    }
+  }
+  return TRUE;
+}
+