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

Timestamp:

Feb 20, 2012, 3:07:22 PM (14 years ago)

Author:

eugene

Message:

fixed up mextract -parallel to work correctly; parallel ops now fairly complete

File:

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

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

@@ -2 +2 @@
 
 # define DEBUG 1
-# define PARALLEL_MANUAL 1
+# define PARALLEL_MANUAL 0
 # define PARALLEL_SERIAL 0
 # define MAX_PATH_LENGTH 1024
@@ -8 +8 @@
 int HostTableLaunchJobs (HostTable *table, char *basecmd);
 Vector **MergeVectors (Vector **vec, int *Nvec, Vector **invec, int Ninvec);
+int HostTableParallelOps (int argc, char **argv, char *ResultFile);
 
 int mextract (int argc, char **argv) {
@@ -56 +57 @@
   }
 
-  // XXX not yet thought through (where are these set?)
+  // init here so free in 'escape' block does not crash
+  dvo_catalog_init (&catalog, TRUE);
+
+  // parse skyregion options.  NOTE: this is stripped off in parallel operation and always
+  // defined for the client via the -skyregion option.  The dvo_client parses this
+  // argument in the main program, before it is passed to the command (like mextract)
+  if ((selection = SetRegionSelection (&argc, argv)) == NULL) goto escape;
+
+  // this does all the work of re-packaging the command, calling it on the remote machines, then loading in the results
   if (PARALLEL && !HOST_ID) {
-    // 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
-    int    Nvec = 0;
-    Vector **vec = NULL;
-    for (i = 0; i < table->Nhosts; i++) {
-      char resultFile[MAX_PATH_LENGTH];
-      snprintf (resultFile, MAX_PATH_LENGTH, "%s/dvo.results.fits", table->hosts[i].pathname);
-        
-      int    Ninvec = 0;
-      Vector **invec = ReadVectorTableFITS (resultFile, "MEXTRACT", &Ninvec);
-
-      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);
-
-    return TRUE;
-  }
-
-  dvo_catalog_init (&catalog, TRUE);
+    // not all parallel ops need to set the region selection...
+    if (!SetSkyRegions (selection)) goto escape;
+
+    int status = HostTableParallelOps (argc, argv, ResultFile);
+    return status;
+  }
 
   /* load photcode information */
@@ -127 +80 @@
   // init locally static variables (time refs)
   dbExtractMeasuresInit();
-
-  // parse skyregion options
-  if ((selection = SetRegionSelection (&argc, argv)) == NULL) goto escape;
 
   // command-line is of the form: avextract field,field, field [where (field op value)...]
@@ -174 +124 @@
   for (i = 0; !loadImages && (i < Nfields); i++) {
     if (!MEASURE_HAS_XCCD) {
-      // I'm keeping this code because it gives a way of handing dvo dbs that don't have
+      // I'm keeping this code because it gives a way of handling dvo dbs that don't have
       // measure.xccd if we need it
       if (fields[i].ID == MEAS_XCCD) loadImages = TRUE;
@@ -279 +229 @@
 
   for (n = 0; n < Nreturn; n++) {
-    ResetVector (vec[n], fields[n].type, MAX(0,Npts));
-  }
-
+    ResetVector (vec[n], fields[n].type, Npts);
+  }
+
+  // write vectors to a table (this is used by parallel dvo operations, but can be used elsewhere)
   if (ResultFile) {
-    // write vectors to a table
-    // char *resultFile?
-    int status = WriteVectorTableFITS (ResultFile, "MEXTRACT", vec, Nreturn, FALSE, NULL);
+    int status = WriteVectorTableFITS (ResultFile, "RESULT", vec, Nreturn, FALSE, NULL);
     if (!status) goto escape;
   }
@@ -443 +392 @@
 
     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);
+    snprintf (command, 1024, "dvo_client %s -result %s -hostID %d -hostdir %s", basecmd, resultFile, table->hosts[i].hostID, table->hosts[i].pathname);
+
+    if (DEBUG || PARALLEL_MANUAL) fprintf (stderr, "command: %s\n", command);
 
     if (PARALLEL_MANUAL) {
@@ -488 +437 @@
     myAssert (vec[i]->type == invec[i]->type, "programming error (2)");
 
+    int N = vec[i]->Nelements;
     if (vec[i]->type == OPIHI_FLT) {
       REALLOCATE (vec[i]->elements.Flt, opihi_flt, vec[i]->Nelements + invec[i]->Nelements);
       for (j = 0; j < invec[i]->Nelements; j++) {
-        vec[i]->elements.Flt[i+j] = vec[i]->elements.Flt[j];
+        vec[i]->elements.Flt[N+j] = invec[i]->elements.Flt[j];
       }
     } else {
       REALLOCATE (vec[i]->elements.Int, opihi_int, vec[i]->Nelements + invec[i]->Nelements);
       for (j = 0; j < invec[i]->Nelements; j++) {
-        vec[i]->elements.Int[i+j] = vec[i]->elements.Int[j];
+        vec[i]->elements.Int[N+j] = invec[i]->elements.Int[j];
       }
     }
@@ -503 +453 @@
   return vec;
 }
+
+int HostTableParallelOps (int argc, char **argv, char *ResultFile) {
+
+  int i;
+
+  // load the list of hosts
+  SkyTable *sky = GetSkyTable();
+  if (!sky) return FALSE;
+
+  char *tmppath = GetCATDIR ();
+  if (!tmppath) return FALSE;
+
+  char *CATDIR = abspath (tmppath, MAX_PATH_LENGTH);
+  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);
+
+  // determine the sky region 
+  double Rmin, Rmax, Dmin, Dmax;
+  get_skyregion (&Rmin, &Rmax, &Dmin, &Dmax);
+
+  // determine time reference and format
+  char *TimeRef = get_variable ("TIMEREF");
+  if (!TimeRef) return FALSE;
+
+  char *TimeFormat = get_variable ("TIMEFORMAT");
+  if (!TimeFormat) return FALSE;
+
+  char tmp;
+  char *command = NULL;
+  int length = snprintf (&tmp, 0, "%s -D CATDIR %s -time %s %s -skyregion %f %f %f %f", basecmd, CATDIR, TimeRef, TimeFormat, Rmin, Rmax, Dmin, Dmax);
+
+  ALLOCATE (command, char, length);
+  snprintf (command, length, "%s -D CATDIR %s -time %s %s -skyregion %f %f %f %f", basecmd, CATDIR, TimeRef, TimeFormat, 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
+  int    Nvec = 0;
+  Vector **vec = NULL;
+  for (i = 0; i < table->Nhosts; i++) {
+    char resultFile[MAX_PATH_LENGTH];
+    snprintf (resultFile, MAX_PATH_LENGTH, "%s/dvo.results.fits", table->hosts[i].pathname);
+        
+    int    Ninvec = 0;
+    Vector **invec = ReadVectorTableFITS (resultFile, "RESULT", &Ninvec);
+
+    vec = MergeVectors (vec, &Nvec, invec, Ninvec);
+    if (vec != invec) {
+      FreeVectorArray (invec, Ninvec);
+    }
+  }
+
+  // write vectors to a table (this is used by parallel dvo operations, but can be used elsewhere)
+  if (ResultFile) {
+    int status = WriteVectorTableFITS (ResultFile, "RESULT", vec, Nvec, FALSE, NULL);
+    if (!status) return FALSE;
+  }
+
+  for (i = 0; i < Nvec; i++) {
+    AssignVector (vec[i], vec[i]->name, ANYVECTOR, TRUE);
+  }
+  free (vec);
+
+  return TRUE;
+}