Changeset 35699

Timestamp:

Jun 21, 2013, 3:21:44 PM (13 years ago)

Author:

eugene

Message:

enable dvomerge to read from a parallel dvo db (regardless of output parallel state)

Location:

branches/eam_branches/ipp-20130509/Ohana/src/dvomerge

Files:

• include/dvomerge.h (modified) (1 diff)
• src/args.c (modified) (3 diffs)
• src/dvomergeUpdate_catalogs.c (modified) (6 diffs)
• test/dvomerge.parallel.sh (modified) (1 diff)

branches/eam_branches/ipp-20130509/Ohana/src/dvomerge/include/dvomerge.h

@@ -23 +23 @@
 int    HOST_ID;
 char  *HOSTDIR;
+
+int    PARALLEL_INPUT;
 
 int    VERBOSE;

branches/eam_branches/ipp-20130509/Ohana/src/dvomerge/src/args.c

@@ -76 +76 @@
   }
 
+  // is the input database a parallel db?
+  PARALLEL_INPUT = FALSE;
+  if ((N = get_argument (*argc, argv, "-parallel-input"))) {
+    PARALLEL_INPUT = TRUE;
+    remove_argument (N, argc, argv);
+  }
+
+  // is the output database a parallel db?
   PARALLEL = FALSE;
   if ((N = get_argument (*argc, argv, "-parallel"))) {
@@ -117 +125 @@
   IMAGES_ONLY = FALSE;
 
+  // is the input database a parallel db?
+  PARALLEL_INPUT = FALSE;
+  if ((N = get_argument (*argc, argv, "-parallel-input"))) {
+    PARALLEL_INPUT = TRUE;
+    remove_argument (N, argc, argv);
+  }
+
   HOST_ID = 0;
   if ((N = get_argument (*argc, argv, "-hostID"))) {
@@ -246 +261 @@
   } 
 
+  // is the output database a parallel db?
   PARALLEL = FALSE;
   if ((N = get_argument (*argc, argv, "-parallel"))) {

branches/eam_branches/ipp-20130509/Ohana/src/dvomerge/src/dvomergeUpdate_catalogs.c

@@ -19 +19 @@
     int status = dvomergeUpdate_parallel (input, output, outsky, IDmap);
     return status;
+  }
+
+  // load the list of hosts
+  HostTable *table_input = NULL;
+  if (PARALLEL_INPUT) {
+    table_input = HostTableLoad (input, inlist->hosts);
+    if (!table_input) {
+      fprintf (stderr, "ERROR: failure reading Host Table %s for database %s\n", inlist->hosts, input);
+      exit (1);
+    }    
+
+    // ensure we have absolute paths for hostdirs
+    for (i = 0; i < table_input->Nhosts; i++) {
+      char *tmppath = abspath (table_input->hosts[i].pathname, DVO_MAX_PATH);
+      free (table_input->hosts[i].pathname);
+      table_input->hosts[i].pathname = tmppath;
+    }
   }
 
@@ -25 +42 @@
     if (!inlist[0].regions[i][0].table) continue;
     if (VERBOSE) fprintf (stderr, "input: %s\n", inlist[0].regions[i][0].name);
+
+    // in a parallel context, we need to re-map the input filename
+    char *filename_input = strcreate (inlist[0].filename[i]);
+    if (PARALLEL_INPUT) {
+
+      int hostID_input = inlist[0].regions[i]->hostID;
+      int table_index = table_input->index[hostID_input];
+      char *hostdir_input = table_input->hosts[table_index].pathname;
+
+      // set the parameters which guide catalog open/load/create
+      char hostfile_input[DVO_MAX_PATH];
+      snprintf (hostfile_input, DVO_MAX_PATH, "%s/%s.cpt", hostdir_input, inlist[0].regions[i]->name);
+      free (filename_input);
+      filename_input = strcreate (hostfile_input);
+    }
 
     // SkyListByBounds will return neighbor catalogs if the boundaries exactly match (due to rounding).  Since the regions are not infinitely small, 
@@ -49 +81 @@
 
     // get stats for history check, skip input catalog if file not found (NULL inStats)
-    dmhObjectStats *inStats = dmhObjectStatsRead (inlist[0].filename[i]);
+    dmhObjectStats *inStats = dmhObjectStatsRead (filename_input);
     if (!inStats) {
-      if (VERBOSE) fprintf (stderr, "skipping %s, empty \n", inlist[0].filename[i]);
+      if (VERBOSE) fprintf (stderr, "skipping %s, empty \n", filename_input);
       OutputStatusFree (outstat, outlist->Nregions);
       SkyListFree (outlist); 
@@ -76 +108 @@
     }
     if (!missed) {
-      if (VERBOSE || VERIFY) fprintf (stderr, "skipping %s, already merged\n", inlist[0].filename[i]);
+      if (VERBOSE || VERIFY) fprintf (stderr, "skipping %s, already merged\n", filename_input);
       OutputStatusFree (outstat, outlist->Nregions);
       dmhObjectStatsFree (inStats);
@@ -83 +115 @@
     }
     if (VERIFY) { 
-      fprintf (stderr, "%s NOT merged\n", inlist[0].filename[i]);
+      fprintf (stderr, "%s NOT merged\n", filename_input);
       continue;
     }
 
     // read the input catalog
-    LoadCatalog (&incatalog, &inlist[0].regions[i][0], inlist[0].filename[i], "r", NsecfiltInput);
+    LoadCatalog (&incatalog, &inlist[0].regions[i][0], filename_input, "r", NsecfiltInput);
+    free (filename_input);
 
     // skip empty input catalogs
@@ -198 +231 @@
 
     char tmpline[DVO_MAX_PATH];
-    if (VERBOSE)             { snprintf (tmpline, DVO_MAX_PATH, "%s -v",       command); strcpy (command, tmpline); }
-    if (VERIFY)              { snprintf (tmpline, DVO_MAX_PATH, "%s -verify",  command); strcpy (command, tmpline); }
+    if (VERBOSE)             { snprintf (tmpline, DVO_MAX_PATH, "%s -v",       command);          strcpy (command, tmpline); }
+    if (VERIFY)              { snprintf (tmpline, DVO_MAX_PATH, "%s -verify",  command);          strcpy (command, tmpline); }
     if (VERIFY_CATALOG_ONLY) { snprintf (tmpline, DVO_MAX_PATH, "%s -verify-catalogs",  command); strcpy (command, tmpline); }
-    if (REPLACE_BY_PHOTCODE) { snprintf (tmpline, DVO_MAX_PATH, "%s -replace", command); strcpy (command, tmpline); }
+    if (REPLACE_BY_PHOTCODE) { snprintf (tmpline, DVO_MAX_PATH, "%s -replace", command);          strcpy (command, tmpline); }
+    if (PARALLEL_INPUT)      { snprintf (tmpline, DVO_MAX_PATH, "%s -parallel-input",  command);  strcpy (command, tmpline); }
 
     fprintf (stderr, "command: %s\n", command);

branches/eam_branches/ipp-20130509/Ohana/src/dvomerge/test/dvomerge.parallel.sh

@@ -1 +1 @@
 #!/bin/csh -f
 
- rm -rf catdir.merge
- rsync -auv catdir.2mass/ catdir.merge/
+ rm -rf catdir.merge catdir.merge.p?
+ rsync --exclude=.svn -auv catdir.2mass/ catdir.merge/
  cp HostTable.dat catdir.merge/
  dvodist -out catdir.merge
+ rm -rf catdir.merge/n????
+
  dvomerge -parallel catdir.grizy into catdir.merge
+ dvomerge -parallel catdir.grizy into catdir.merge -verify