Changeset 33314 for branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/avextract.c

Timestamp:

Feb 20, 2012, 3:18:25 PM (14 years ago)

Author:

eugene

Message:

convert avextract to parallel form

File:

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

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

@@ -1 +1 @@
 # include "dvoshell.h"
+
+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 avextract (int argc, char **argv) {
@@ -37 +41 @@
   }
 
+  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);
+  }
+
   dvo_catalog_init (&catalog, TRUE);
 
@@ -43 +61 @@
   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; 
+  }
+
+  // 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) {
+    // not all parallel ops need to set the region selection...
+    if (!SetSkyRegions (selection)) goto escape;
+
+    int status = HostTableParallelOps (argc, argv, ResultFile);
+    return status;
   }
 
@@ -114 +143 @@
   interrupt = FALSE;
   for (i = 0; (i < skylist[0].Nregions) && !interrupt; i++) {
+    // 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_SECF;
     if (needMeasures) {
@@ -169 +203 @@
   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 (ResultFile) {
+    int status = WriteVectorTableFITS (ResultFile, "RESULT", vec, Nreturn, FALSE, NULL);
+    if (!status) goto escape;
   }