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

Timestamp:

Mar 16, 2012, 7:51:21 AM (14 years ago)

Author:

eugene

Message:

parallelize gstar, update avmatch to work with merged vectors by index; remove elements from dvoshell.h which have moved to libdvo

File:

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

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

@@ -1 +1 @@
 # include "dvoshell.h"
-
-# define DEBUG 0
-# define PARALLEL_MANUAL 0
-# define PARALLEL_SERIAL 0
-# define MAX_PATH_LENGTH 1024
-
-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) {
@@ -77 +68 @@
     if (!SetSkyRegions (selection)) goto escape;
 
-    int status = HostTableParallelOps (argc, argv, ResultFile);
+    int status = HostTableParallelOps (argc, argv, ResultFile, 0);
     return status;
   }
@@ -161 +152 @@
 
     // does this host ID match the desired location for the table?
-    if (!HostTableTestHost(skylist[0].regions[i], HOST_ID)) continue;
+    if (PARALLEL && !HostTableTestHost(skylist[0].regions[i], HOST_ID)) continue;
 
     /* lock, load, unlock catalog */
@@ -380 +371 @@
   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, "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) {
-      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;
-}
-
-// take two arrays of vectors and merge equal named vectors.
-// for ease, require that the order of the names match & number of vectors match
-Vector **MergeVectors (Vector **vec, int *Nvec, Vector **invec, int Ninvec) {
-
-  int i, j;
-
-  if (vec == NULL) {
-    *Nvec = Ninvec;
-    return invec;
-  }
-
-  myAssert (*Nvec == Ninvec, "programming error (1)");
-
-  for (i = 0; i < Ninvec; i++) {
-    myAssert (!strcmp(vec[i]->name, invec[i]->name), "programming error (2)");
-    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[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[N+j] = invec[i]->elements.Int[j];
-      }
-    }
-    vec[i]->Nelements += invec[i]->Nelements;
-  }
-  return vec;
-}
-
-int HostTableParallelOps (int argc, char **argv, char *ResultFile) {
-
-  int i;
-
-  // load the list of hosts
-  SkyTable *sky = GetSkyTable();
-  if (!sky) {
-    gprint (GP_ERR, "failed to load sky table for database\n");
-    return FALSE;
-  }
-
-  char *tmppath = GetCATDIR ();
-  if (!tmppath) {
-    gprint (GP_ERR, "failed to get CATDIR for database\n");
-    return FALSE;
-  }
-
-  char *CATDIR = abspath (tmppath, MAX_PATH_LENGTH);
-  if (!CATDIR) {
-    gprint (GP_ERR, "failed to make an absolute path from %s (too long)\n", tmppath);
-    return FALSE;
-  }
-
-  HostTable *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;
-  }    
-
-  // 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) {
-    gprint (GP_ERR, "failed to find TIMEREF variable\n");
-    return FALSE;
-  }
-
-  char *TimeFormat = get_variable ("TIMEFORMAT");
-  if (!TimeFormat) {
-    gprint (GP_ERR, "failed to find TIMEFORMAT variable\n");
-    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) {
-    gprint (GP_ERR, "run the relphot_client commands above.  when these are done, hit return\n");
-    getchar();
-  }
-  if (!PARALLEL_MANUAL && !PARALLEL_SERIAL) {
-    HostTableWaitJobsGetIO (table, __FILE__, __LINE__);
-  }
-
-  // load fields from file
-  int    Nvec = 0;
-  Vector **vec = NULL;
-  for (i = 0; i < table->Nhosts; i++) {
-    if (table->hosts[i].status) continue; 
-
-    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) {
-      gprint (GP_ERR, "failed to write result file %s\n", ResultFile);
-      return FALSE;
-    }
-  }
-
-  for (i = 0; i < Nvec; i++) {
-    AssignVector (vec[i], vec[i]->name, ANYVECTOR, TRUE);
-  }
-  free (vec);
-
-  return TRUE;
-}