Changeset 34041 for branches/meh_branches/ppstack_test/Ohana/src/opihi/dvo/avextract.c

Timestamp:

Jun 19, 2012, 5:24:19 PM (14 years ago)

Author:

mhuber

Message:

merging latest r34040 trunk changes to branch

Location:

branches/meh_branches/ppstack_test

Files:

• . (modified) (1 prop)
• Ohana/src/opihi/dvo/avextract.c (modified) (8 diffs)

branches/meh_branches/ppstack_test/Ohana/src/opihi/dvo/avextract.c

@@ -37 +37 @@
   }
 
+  int VERBOSE2 = FALSE;
+  if ((N = get_argument (argc, argv, "-vv"))) {
+    remove_argument (N, &argc, argv);
+    VERBOSE = TRUE;
+    VERBOSE2 = TRUE;
+  }
+
+  int PARALLEL = FALSE;
+  if ((N = get_argument (argc, argv, "-parallel"))) {
+    remove_argument (N, &argc, argv);
+    PARALLEL = TRUE;
+  }
+
+  // use remote tables, but not dvo_client..
+  int PARALLEL_LOCAL = FALSE;
+  HostTable *table = NULL;
+  if ((N = get_argument (argc, argv, "-parallel-local"))) {
+    remove_argument (N, &argc, argv);
+    PARALLEL_LOCAL = TRUE;
+
+    char *CATDIR = GetCATDIR();
+    if (!CATDIR) {
+      gprint (GP_ERR, "CATDIR is not set\n");
+      return FALSE;
+    }
+    SkyTable *sky = GetSkyTable();
+    if (!sky) {
+      gprint (GP_ERR, "failed to load sky table for database\n");
+      return FALSE;
+    }
+    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;
+    }    
+  }
+
+  // this is used to NOT save the results in the results file
+  // use this option when mextract is used in a script which does its
+  // own job of packaging the results
+  int SKIP_RESULTS = FALSE;
+  if ((N = get_argument (argc, argv, "-skip-results"))) {
+    remove_argument (N, &argc, argv);
+    SKIP_RESULTS = TRUE;
+  }
+
   dvo_catalog_init (&catalog, TRUE);
 
@@ -43 +89 @@
   Nsecfilt = GetPhotcodeNsecfilt ();
 
-  // parse skyregion options
+  // 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) {
     print_error(); 
     goto escape; 
   }
+
+  // init locally static variables (time refs)
+  dbExtractAveragesInit (); 
 
   // command-line is of the form: avextract field,field, field [where (field op value)...]
@@ -85 +136 @@
   /* load region corresponding to selection above */
   if ((skylist = SelectRegions (selection)) == 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) {
+    int status = HostTableParallelOps (argc, argv, RESULT_FILE, 0, VERBOSE);
+
+    dbFreeFields (fields, Nfields);
+    dbFreeStack (stack, Nstack);
+    free (stack);
+    FreeSkyRegionSelection (selection);
+    dvo_catalog_free (&catalog);
+
+    return status;
+  }
 
   /* create output storage vectors */
@@ -114 +178 @@
   interrupt = FALSE;
   for (i = 0; (i < skylist[0].Nregions) && !interrupt; i++) {
+
+    // does this host ID match the desired location for the table?
+    if (!HostTableTestHost(skylist[0].regions[i], HOST_ID)) continue;
+
     /* lock, load, unlock catalog */
-    catalog.filename = skylist[0].filename[i];
+    char hostfile[1024];
+    if (PARALLEL_LOCAL) {
+      int hostID = (skylist[0].regions[i]->hostFlags & DATA_USE_BCK) ? skylist[0].regions[i]->backupID : skylist[0].regions[i]->hostID;
+      int seq = table->index[hostID];
+      HOSTDIR = table->hosts[seq].pathname;
+    }
+    snprintf (hostfile, 1024, "%s/%s.cpt", HOSTDIR, skylist[0].regions[i]->name);
+    catalog.filename = (HOST_ID || PARALLEL_LOCAL) ? hostfile : skylist[0].filename[i];
     catalog.catflags = LOAD_AVES | LOAD_SECF;
-    if (needMeasures) {
-      catalog.catflags |= LOAD_MEAS;
-    }
+    catalog.catflags |= needMeasures ? LOAD_MEAS : SKIP_MEAS;
     catalog.Nsecfilt = 0;
 
@@ -125 +198 @@
       
     // an error exit status here is a significant error
-    if (!dvo_catalog_open (&catalog, NULL, FALSE, "r")) {
+    if (!dvo_catalog_open (&catalog, NULL, VERBOSE2, "r")) {
       gprint (GP_ERR, "ERROR: failure to open catalog file %s\n", catalog.filename);
       exit (2);
@@ -137 +210 @@
       m = catalog.average[j].measureOffset;
 
-      // reset counters for saved fields, extract fields
-      dbExtractAveragesInit (); 
+      dbExtractAveragesInitAve ();  // reset counters for saved fields (costs very little)
+
       for (n = 0; n < Nfields; n++) {
         values[n] = dbExtractAverages (&catalog.average[j], &catalog.secfilt[j*Nsecfilt], &catalog.measure[m], &fields[n]);
@@ -169 +242 @@
   interrupt = FALSE;
   for (n = 0; n < Nreturn; n++) {
-    vec[n][0].Nelements = Npts;
-    if (vec[n][0].type == OPIHI_FLT) {
-      REALLOCATE (vec[n][0].elements.Flt, opihi_flt, MAX(1,Npts));
-    } else {
-      REALLOCATE (vec[n][0].elements.Int, opihi_int, MAX(1,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 (RESULT_FILE && !SKIP_RESULTS) {
+    int status = WriteVectorTableFITS (RESULT_FILE, "RESULT", vec, Nreturn, FALSE, NULL);
+    if (!status) {
+      goto escape;
+    }
+  }
+
+  if (table) free (table);
   if (vec) free (vec);
   if (values) free (values);
@@ -187 +264 @@
 
  escape:
+  if (table) free (table);
   if (vec) free (vec);
   if (values) free (values);