Changeset 33662 for trunk/Ohana/src/opihi/dvo/avextract.c

Timestamp:

Apr 1, 2012, 3:24:39 PM (14 years ago)

Author:

eugene

Message:

support for parallel dvo in mextract, avextract, avmerge, gstar, gcat; remove some if-def-ed out code already moved to libdvo; list -vectors and -buffers options (store names of vectors and buffers in lists); list -copy give error if missing source; catlist, hosts & remote functions to support parallel dvos; skyregion -save option; new skycoverage modes (-min-ubercal; -min-dmag-sys; -min-mcal; -max-mcal); gstar output formatting cleanups; add more average info to gstar output; functions for spectral similarity analysis; coords returns nans for projections off the sphere; fix cumulative function; add name:type option to opihi/read function (eg read a:float 2 b:time 3 c:int 4); line -frac option; vtype function to get vector types; threshold function; write -fits option; code for unfinished mtype function (matching vtype)

File:

• trunk/Ohana/src/opihi/dvo/avextract.c (modified) (6 diffs)

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

@@ -37 +37 @@
   }
 
+  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;
+    }    
+  }
+
   dvo_catalog_init (&catalog, TRUE);
 
@@ -43 +73 @@
   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(); 
@@ -85 +117 @@
   /* 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 +159 @@
   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) {
@@ -169 +225 @@
   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) {
+    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 +245 @@
 
  escape:
+  if (table) free (table);
   if (vec) free (vec);
   if (values) free (values);