Changeset 33244

Timestamp:

Feb 13, 2012, 11:32:14 AM (14 years ago)

Author:

eugene

Message:

parallel dvo code

Location:

branches/eam_branches/ipp-20111122/Ohana/src

Files:

• libdvo/include/dvo.h (modified) (4 diffs)
• libdvo/src/skyregion_ops.c (modified) (6 diffs)
• relphot/doc/parallel.txt (added)
• relphot/src/BrightCatalog.c (added)
• relphot/src/ImageSubset.c (added)
• relphot/src/args.c (modified) (1 diff)
• relphot/src/help.c (modified) (1 diff)
• relphot/src/initialize.c (modified) (1 diff)
• relphot/src/load_catalogs.c (modified) (3 diffs)
• relphot/src/reload_catalogs.c (modified) (1 diff)
• relphot/src/relphot.c (modified) (1 diff)
• relphot/src/relphot_client.c (added)
• uniphot/src/setphot_client.c (modified) (1 diff)

branches/eam_branches/ipp-20111122/Ohana/src/libdvo/include/dvo.h

@@ -205 +205 @@
   char **filename;
   SkyRegion **regions;
+  char hosts[80];
 } SkyList;
 
@@ -261 +262 @@
 } PhotCodeData;
 
-// a reduced-subset structure for relphot
+// a reduced-subset structure for relphot & relastro
 typedef struct {
   double         R;
@@ -268 +269 @@
   int            measureOffset;
   uint32_t       flags;
+  int            catID;
 } AverageTiny;
 
-// a reduced-subset structure for relphot
+// a reduced-subset structure for relphot & relastro
 typedef struct {
   float          dR;
@@ -285 +287 @@
   unsigned int   imageID;
   unsigned int   dbFlags;
+  int            catID; // unsigned int?
   unsigned short photcode;
 } MeasureTiny;

branches/eam_branches/ipp-20111122/Ohana/src/libdvo/src/skyregion_ops.c

@@ -20 +20 @@
   list[0].Nregions = 0;
   list[0].ownElements = FALSE; // this list is only holding a view to the elements
+  strcpy (list[0].hosts, table[0].hosts);
 
   region = table[0].regions;
@@ -67 +68 @@
   list[0].Nregions = N;
   list[0].ownElements = FALSE; // this list is only holding a view to the elements
+  strcpy (list[0].hosts, table[0].hosts);
 
   region = table[0].regions;
@@ -102 +104 @@
   list[0].Nregions = N;
   list[0].ownElements = FALSE; // this list is only holding a view to the elements
+  strcpy (list[0].hosts, table[0].hosts);
 
   region = table[0].regions;
@@ -261 +264 @@
   ALLOCATE (list[0].filename, char *, NNEW);
   list[0].ownElements = FALSE; // this list is only holding a view to the elements
+  strcpy (list[0].hosts, table[0].hosts);
 
   region = table[0].regions;
@@ -326 +330 @@
   list[0].Nregions = 0;
   list[0].ownElements = FALSE; // this list is only holding a view to the elements
+  strcpy (list[0].hosts, inList[0].hosts);
 
   region = inList[0].regions;
@@ -403 +408 @@
   ALLOCATE (list[0].filename, char *, NNEW);
   list[0].ownElements = FALSE; // this list is only holding a view to the elements
+  strcpy (list[0].hosts, table[0].hosts);
 
   region = table[0].regions;

branches/eam_branches/ipp-20111122/Ohana/src/relphot/src/args.c

@@ -265 +265 @@
   return TRUE;
 }
+
+int args_client (int argc, char **argv) {
+
+  int N;
+
+  if ((N = get_argument (argc, argv, "-v"))) {
+    VERBOSE = TRUE;
+    remove_argument (N, &argc, argv);
+  }
+
+  HOST_ID = 0;
+  if ((N = get_argument (argc, argv, "-hostID"))) {
+    remove_argument (N, &argc, argv);
+    HOST_ID = atoi (argv[N]);
+    remove_argument (N, &argc, argv);
+  }
+  if (!HOST_ID) relphot_client_usage();
+
+  HOSTDIR = NULL;
+  if ((N = get_argument (argc, argv, "-hostdir"))) {
+    remove_argument (N, &argc, argv);
+    HOSTDIR = strcreate (argv[N]);
+    remove_argument (N, &argc, argv);
+  }
+  if (!HOSTDIR) relphot_client_usage();
+
+  CATDIR[0] = 0;
+  if ((N = get_argument (argc, argv, "-catdir"))) {
+    remove_argument (N, &argc, argv);
+    strcpy (CATDIR, argv[N]);
+    remove_argument (N, &argc, argv);
+  }
+  if (!CATDIR[0]) relphot_client_usage();
+
+  IMAGES = NULL; // used in -update mode
+  BCATALOG = NULL; // used in -load mode
+  MODE = MODE_NONE;
+  if ((N = get_argument (argc, argv, "-load"))) {
+    MODE = MODE_LOAD;
+    remove_argument (N, &argc, argv);
+    BCATALOG = strcreate (argv[N]);
+    remove_argument (N, &argc, argv);
+  }
+  if ((N = get_argument (argc, argv, "-update"))) {
+    if (MODE) {
+      fprintf (stderr, "ERROR: cannot use both -load and -update options!\n");
+      usage();
+    }
+    MODE = MODE_UPDATE;
+    remove_argument (N, &argc, argv);
+    IMAGES = strcreate (argv[N]);
+    remove_argument (N, &argc, argv);
+  }
+  if (!MODE) relphot_client_usage();
+
+  if (argc != 1) relphot_client_usage ();
+
+  return TRUE;
+}

branches/eam_branches/ipp-20111122/Ohana/src/relphot/src/help.c

@@ -44 +44 @@
 }
 
+void relphot_client_usage (void) {
+  fprintf (stderr, "ERROR: USAGE: relphot -load -bcatalog (filename) -hostID (hostID) -hostdir (hostdir) -catdir (catdir)\n");
+  fprintf (stderr, "       or:    relphot -update -images (filename) -hostID (hostID) -hostdir (hostdir) -catdir (catdir)\n");
+  fprintf (stderr, "  use -h for more usage information\n");
+  exit (2);
+} 
+
+void relphot_client_help (int argc, char **argv) {
+
+  /* check for help request */
+  if (get_argument (argc, argv, "-help")) goto show_help;
+  if (get_argument (argc, argv, "-h"))    goto show_help;
+  if (argc == 1) relphot_client_usage();
+  return;
+
+show_help:
+  fprintf (stderr, "USAGE: relphot_client [-load / -update] (db info)\n\n");
+  fprintf (stderr, "       relphot_client -load (bcatalog) : extract the bright catalog subset from client's tables\n");
+  fprintf (stderr, "                            (bcatalog) : location where the bright subset is saved\n");
+  fprintf (stderr, "       relphot_client -update (images) : apply calculated zero points to the client's tables\n\n");
+  fprintf (stderr, "                              (images) : location of the table with the image zero points\n");
+  fprintf (stderr, "       db info : -hostID (hostID) -hostdir (hostdir) -catdir (catdir)\n");
+  fprintf (stderr, "other options:\n");
+  fprintf (stderr, "  -v : verbose output\n");
+  fprintf (stderr, "  \n");
+  exit (2);
+}
+

branches/eam_branches/ipp-20111122/Ohana/src/relphot/src/initialize.c

@@ -95 +95 @@
   srand48(B);
 }
+
+void initialize_relphot_client (int argc, char **argv) {
+
+  int N;
+
+  relphot_client_help (argc, argv);
+  // ConfigInit (&argc, argv); XXX not sure if we need anything in here (maybe photcode table)
+  args_client (argc, argv);
+}

branches/eam_branches/ipp-20111122/Ohana/src/relphot/src/load_catalogs.c

@@ -1 +1 @@
 # include "relphot.h"
 
-Catalog *load_catalogs (SkyList *skylist, int *Ncatalog) {
+// this function loops over the catalogs, loads the data, and extracts a bright subset
+// the returned array (catalog, Ncatalog) has the same layout as the full database, but
+// only a subset of the detetions & objects
+
+// if this function is called in parallel mode, it in turn calls load_catalogs_parallel,
+// which distributes the work to the remote hosts and loads their results
+
+// if this function is called with a specified hostID, then only the fraction of the
+// database hosted by that hostID is loaded
+Catalog *load_catalogs (SkyList *skylist, int *Ncatalog, int hostID, char *hostpath) {
 
   off_t i, Nmeas, Nstar, Nmeas_total, Nstar_total;
   Catalog *catalog, tcatalog;
 
+  // XXX need to decide how to determine PARALLEL mode...
+  if (PARALLEL & !hostID) {
+    catalog = load_catalogs_parallel (skylist, Ncatalog);
+    return catalog;
+  }
+
   if (VERBOSE2) fprintf (stderr, "loading catalog data\n");
 
+  // a bit of an over-alloc, since we don't load all catalogs for a region 
   ALLOCATE (catalog, Catalog, skylist[0].Nregions);
 
@@ -16 +32 @@
     dvo_catalog_init (&catalog[i], TRUE);
 
+    if (hostID && (skylist[0].regions[i]->hostID != hostID)) continue;
+
     // set up the basic catalog info
-    tcatalog.filename = skylist[0].filename[i];
+    char hostfile[1024];
+    snprintf (hostfile, 1024, "%s/%s.cpt", hostpath, skylist[0].regions[i]->name);
+    tcatalog.filename = hostID ? hostfile : skylist[0].filename[i];
+
     tcatalog.catformat = dvo_catalog_catformat (CATFORMAT);    // set the default catformat from config data
     tcatalog.catmode   = dvo_catalog_catmode (CATMODE);        // set the default catmode from config data
@@ -67 +88 @@
    need to use an XCLD lock here.  
 */
+
+// CATDIR is supplied globally
+Catalog *load_catalogs_parallel (SkyTable *sky, int *Ncatalog) {
+
+  // launch the setphot_client jobs to the parallel hosts
+
+  // load the list of hosts
+  HostTable *table = HostTableLoad (CATDIR, sky->hosts);
+
+  int i;
+  for (i = 0; i < table->Nhosts; i++) {
+
+    char catalogFile[512];
+    snprintf (catalogFile, 512, "%s/relphot.catalog.subset.dat", table->hosts[i].pathname);
+
+    char command[1024];
+    snprintf (command, 1024, "relphot_client -test -load %s -hostID %d -catdir %s -hostdir %s", 
+              catalogFile, table->hosts[i].hostID, CATDIR, table->hosts[i].pathname);
+
+    fprintf (stderr, "command: %s\n", command);
+
+    // 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
+
+    // check that all hosts started OK?
+  }
+
+  // watch for stdout / stderr from those jobs...
+  // wait for all of them to complete...
+
+  // each host generates a BrightCatalog structure, with the measure, average, etc value loaded into a single 
+  // set of arrays (of MeasureTiny, AverageTiny, Secfilt).  I need to split out the
+  // per-catalog measurements into separate catalog entries.
+
+  // we don't know how many catalogs we actually need.  allocate a single entry, then
+  // below we will allocate the max number needed after the BrightCatalog is loaded
+  int Ncatalog = 0;
+  int NCATALOG = 1;
+  ALLOCATE (catalog, Catalog, NCATALOG);
+
+  int *NAVERAGE;
+  int *NMEASURE;
+  ALLOCATE (NAVERAGE, int, NCATALOG);
+  ALLOCATE (NMEASURE, int, NCATALOG);
+  for (i = 0; i < NCATALOG; i++) {
+    NAVERAGE[i] = 100;
+    NMEASURE[i] = 100;
+    catalog[i].Naverage = 0;
+    catalog[i].Nmeasure = 0;
+    ALLOCATE (catalog[i].averageT, AverageTiny, NAVERAGE[i]);
+    ALLOCATE (catalog[i].measureT, MeasureTiny, NMEASURE[i]);
+    ALLOCATE (catalog[i].secfilt, SecFilt, Nsecfilt*NAVERAGE[i]);
+  }
+
+  for (i = 0; i < table->Nhosts; i++) {
+
+    // XXX save this name in table->hosts[]?  it is created above as well...
+    char catalogFile[512];
+    snprintf (catalogFile, 512, "%s/relphot.catalog.subset.dat", table->hosts[i].pathname);
+
+    BrightCatalog *bcatalog = BrightCatalogLoad (catalogFile);
+    assert (bcatalog);
+    
+    // XXX put the code below into a function equivalent to BrightCatalogMerge?
+
+    // find the max value of catID and realloc to match that (check measure & average or
+    // just average?)
+    int catIDmax = 0;
+    for (j = 0; j < bcatalog->Naverage; j++) {
+      catIDmax = MAX(catIDmax, bcatalog->average[j].catID);
+    }
+    // XXX validate the measure ID range here
+    
+    int NCATALOG_OLD = NCATALOG;
+    NCATALOG = MAX (NCATALOG, catIDmax);
+    REALLOCATE (catalog, Catalog, NCATALOG);
+    
+    REALLOCATE (NAVERAGE, int, NCATALOG);
+    REALLOCATE (NMEASURE, int, NCATALOG);
+    for (i = NCATALOG_OLD; i < NCATALOG; i++) {
+      NAVERAGE[i] = 100;
+      NMEASURE[i] = 100;
+      catalog[i].Naverage = 0;
+      catalog[i].Nmeasure = 0;
+      ALLOCATE (catalog[i].averageT, AverageTiny, NAVERAGE[i]);
+      ALLOCATE (catalog[i].measureT, MeasureTiny, NMEASURE[i]);
+      ALLOCATE (catalog[i].secfilt, SecFilt, Nsecfilt*NAVERAGE[i]);
+    }
+
+    // each bcatalog (from each host) has a different set of catID ranges
+    // they also (probably) have contiguous ranges.  But, it is a bit tricky to be sure I
+    // can use that info, so perhaps ignore it
+
+    // assign the averages to the corresponding catalog
+    for (j = 0; j < bcatalog->Naverage; j++) {
+      Nc = bcatalog->average[j].catID;
+      assert (Nc < NCATALOG);
+      Na = catalog[Nc].Naverage;
+      catalog[Nc].averageT[Na] = bcatalog->average[j];
+
+      // secfilt entries are grouped in blocks for each average
+      for (k = 0; k < Nsecfilt; k++) {
+        catalog[Nc].secfilt[Na*Nsecfilt + k] = bcatalog->secfilt[j*Nsecfilt + k];
+      }      
+
+      catalog[Nc].Naverage ++;
+      if (catalog[Nc].Naverage >= NAVERAGE[N]) {
+        NAVERAGE[Nc] += 100;
+        REALLOCATE (catalog[Nc].averageT, AverageTiny, NAVERAGE[Nc]);
+        REALLOCATE (catalog[Nc].secfilt, SecFilt, Nsecfilt*NAVERAGE[Nc]);
+      }        
+    }
+
+    // assign the measures to the corresponding catalog
+    // XXX what about averef and related links?  Do I need them?
+    for (j = 0; j < bcatalog->Nmeasure; j++) {
+      Nc = bcatalog->measure[j].catID;
+      assert (Nc < NCATALOG);
+      Na = catalog[Nc].Nmeasure;
+      catalog[Nc].measureT[Na] = bcatalog->measure[j];
+      catalog[Nc].Nmeasure ++;
+      if (catalog[Nc].Nmeasure >= NMEASURE[N]) {
+        NMEASURE[Nc] += 100;
+        REALLOCATE (catalog[Nc].measureT, MeasureTiny, NMEASURE[Nc]);
+      }        
+    }
+    // XXX free the input bcatalog entries
+  }
+  *Ncatalog = NCATALOG;
+  return (catalog);
+}      

branches/eam_branches/ipp-20111122/Ohana/src/relphot/src/reload_catalogs.c

@@ -94 +94 @@
   fprintf (stderr, "time7  %f : free catalog\n", time7);
 }
+
+int reload_catalogs_parallel (SkyTable *sky, Image *image, off_t Nimage) {
+
+  // write out the subset table of image information
+  char imageFile[512];
+  snprintf (imageFile, 512, "%s/Images.subset.dat", CATDIR);
+  if (!ImageSubsetSave (imageFile, image, Nimage)) {
+    fprintf (stderr, "failed to write image subset\n");
+    exit (1);
+  }
+
+  // now launch the relphot_client jobs to the parallel hosts
+
+  // load the list of hosts
+  HostTable *table = HostTableLoad (CATDIR, sky->hosts);
+
+  int i;
+  for (i = 0; i < table->Nhosts; i++) {
+
+    char command[1024];
+    snprintf (command, 1024, "relphot_client -reload -hostname %s -catdir %s -hostdir %s -images %s", 
+              table->hosts[i].hostname, CATDIR, table->hosts[i].pathname, imageFile);
+
+    fprintf (stderr, "command: %s\n", command);
+
+    // 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
+
+    // check that all hosts started OK?
+  }
+
+  // watch for stdout / stderr from those jobs...
+  // wait for all of them to complete...
+
+  return (TRUE);
+}      

branches/eam_branches/ipp-20111122/Ohana/src/relphot/src/relphot.c

@@ -241 +241 @@
 
   /* load catalog data from region files, update Mrel include all data */
-  reload_catalogs (skylist, flatcorr);
+  if (PARALLEL) {
+    reload_catalogs_parallel ();
+  } else {
+    reload_catalogs (skylist, flatcorr);
+  }
   MARKTIME("-- updated all catalogs: %f sec\n", dtime);
 

branches/eam_branches/ipp-20111122/Ohana/src/uniphot/src/setphot_client.c

@@ -29 +29 @@
   exit (0);
 }
-  
-
-/* setphot : set the zero points for images in the db (perhaps based on external information)
-   setphot (zpt_table)
-
-   setphot has 2 modes : basic & ubercal
-
-   * outline of basic mode:
-
-   ** load text table of zpts, time (photcode?)
-   ** load images
-   ** match images to zpts
-   ** set zpts
-   ** load catalogs
-   ** update detection (& averages?)
-
-   * outline of ubercal mode:
-
-   ** load fits table of zpts & time + table of flat-field corrections
-   ** load images
-   ** match images to zpts
-   ** set zpts
-   ** load catalogs
-   ** update detection (& averages?)
-
- */
-