Changeset 33875 for branches/eam_branches/ipp-20120405/Ohana

Timestamp:

May 16, 2012, 4:20:32 PM (14 years ago)

Author:

eugene

Message:

add failsafes to dvomerge (write size & date/time of input catalog into output and do not merge if it already exists; make backup files (move target to new name) and delete backups only on success

Location:

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

Files:

• doc/failsafe.txt (added)
• include/dvomerge.h (modified) (1 diff)
• src/dvomergeUpdate_catalogs.c (modified) (8 diffs)

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

@@ -17 +17 @@
 # include <glob.h>
 
-# define MAX_PATH_LENGTH 1024
+# define DVO_MAX_PATH 1024
 
 int    PARALLEL;

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

@@ -36 +36 @@
       found = (found || HostTableTestHost(outlist[0].regions[j], HOST_ID));
     }
+
     // skip this input table for if no output files are on this machine
     if (!found) {
@@ -42 +43 @@
     }
 
-    // load / create output catalog (if catalog does not exist, it will be created)
+    // get the stats on this input file (for comparison with the output headers)
+    struct stat instats;
+    int stat_result = stat (inlist[0].filename[i], &instats);
+    if (stat_result) {
+      if (errno == ENOENT) continue;
+      fprintf (stderr, "cannot read stats on input file %s\n", inlist[0].filename[i]);
+      perror ("stats error message:");
+      exit (2);
+    }
+
+    // instats.st_size & instats.st_mtime
+
+    // check if any of the output files have NOT yet received data from this input file
+
+    int missed = FALSE;
+    for (j = 0; !missed && (j < outlist[0].Nregions); j++) {
+      if (!HostTableTestHost(outlist[0].regions[j], HOST_ID)) continue;
+
+      // set the parameters which guide catalog open/load/create
+      char hostfile[DVO_MAX_PATH];
+      snprintf (hostfile, DVO_MAX_PATH, "%s/%s.cpt", HOSTDIR, outlist[0].regions[j]->name);
+      char *filename = HOST_ID ? hostfile : outlist[0].filename[j];
+
+      // get the stats on this input file (for comparison with the output headers)
+      struct stat outstats;
+      stat_result = stat (filename, &outstats);
+      if (stat_result) {
+        if (errno == ENOENT) {
+          missed = TRUE;
+          break;
+        }
+        fprintf (stderr, "cannot read stats on output file %s\n", filename);
+        perror ("stats error message:");
+        exit (2);
+      }
+
+      FILE *fout = fopen (filename, "r");
+      if (!fout) {
+        fprintf (stderr, "problem opening output file to read header %s\n", filename);
+        perror ("stats error message:");
+        exit (2);
+      }
+
+      Header outheader;
+      if (!gfits_fread_header (fout, &outheader)) {
+        fprintf (stderr, "problem reading header for output file %s\n", filename);
+        exit (2);
+      }
+      
+      // update header of output catalog
+      long long last_size;
+      char last_moddate[80];
+      gfits_scan (&outheader, "LMRG_SZ", "%lld", 1, &last_size);      
+      gfits_scan (&outheader, "LMRG_DT", "%s", 1, last_moddate);      
+
+      time_t last_mod = ohana_date_to_sec (last_moddate);
+
+      if (last_size != instats.st_size) {
+        missed = TRUE;
+        break;
+      }
+      
+      if (last_mod != instats.st_mtime) {
+        missed = TRUE;
+        break;
+      }
+    }
+    if (!missed) {
+      fprintf (stderr, "skipping %s, already merged\n", inlist[0].filename[i]);
+      continue;
+    }
+
+    // read the input catalog
     LoadCatalog (&incatalog, &inlist[0].regions[i][0], inlist[0].filename[i], "r", NsecfiltInput);
+
     // skip empty input catalogs
     if (!incatalog.Naves_disk) {
@@ -52 +126 @@
     }
 
+    char *moddate = ohana_sec_to_date (instats.st_mtime);
+
     // merge input into the appropriate output tables
     for (j = 0; j < outlist[0].Nregions; j++) {
@@ -60 +136 @@
 
       // set the parameters which guide catalog open/load/create
-      char hostfile[MAX_PATH_LENGTH];
-      snprintf (hostfile, MAX_PATH_LENGTH, "%s/%s.cpt", HOSTDIR, outlist[0].regions[j]->name);
+      char hostfile[DVO_MAX_PATH];
+      snprintf (hostfile, DVO_MAX_PATH, "%s/%s.cpt", HOSTDIR, outlist[0].regions[j]->name);
       outcatalog.filename  = HOST_ID ? hostfile : outlist[0].filename[j];
 
@@ -71 +147 @@
 
       outcatalog.catflags = LOAD_AVES | LOAD_MEAS | LOAD_MISS | LOAD_SECF;
+
+      // update header of output catalog
+      gfits_modify (&outcatalog.header, "LMRG_SZ", "%lld", 1, (long long) instats.st_size);      
+      gfits_modify (&outcatalog.header, "LMRG_DT", "%s", 1, moddate);      
+
+      if (!dvo_catalog_backup (&outcatalog, TRUE)) {
+        fprintf (stderr, "ERROR: failed to make backup for catalog %s\n", outlist[0].filename[j]);
+        exit (1);
+      }
 
       // if we receive a signal which would cause us to exit, wait until the full catalog is written
@@ -80 +165 @@
       SetProtect (FALSE);
 
+      if (!dvo_catalog_unlink_backup (&outcatalog, TRUE)) {
+        fprintf (stderr, "WARNING: failed to remove backup for catalog %s\n", outlist[0].filename[j]);
+      }
+
       dvo_catalog_unlock (&outcatalog);
       dvo_catalog_free (&outcatalog);
@@ -86 +175 @@
     }
     SkyListFree (outlist);
+    free (moddate);
 
     dvo_catalog_unlock (&incatalog);
@@ -108 +198 @@
 
     // ensure that the paths are absolute path names
-    char *tmppath = abspath (table->hosts[i].pathname, MAX_PATH_LENGTH);
+    char *tmppath = abspath (table->hosts[i].pathname, DVO_MAX_PATH);
     free (table->hosts[i].pathname);
     table->hosts[i].pathname = tmppath;
 
     // ensure that the paths are absolute path names
-    char *absinput  = abspath (input,  MAX_PATH_LENGTH);
-    char *absoutput = abspath (output, MAX_PATH_LENGTH);
+    char *absinput  = abspath (input,  DVO_MAX_PATH);
+    char *absoutput = abspath (output, DVO_MAX_PATH);
 
     // options / arguments that can affect relastro_client -update-objects:
-    char command[MAX_PATH_LENGTH];
-    snprintf (command, MAX_PATH_LENGTH, "dvomerge_client %s into %s -hostID %d -hostdir %s -region %f %f %f %f -D ADDSTAR_RADIUS %f", 
+    char command[DVO_MAX_PATH];
+    snprintf (command, DVO_MAX_PATH, "dvomerge_client %s into %s -hostID %d -hostdir %s -region %f %f %f %f -D ADDSTAR_RADIUS %f", 
               absinput, absoutput, table->hosts[i].hostID, table->hosts[i].pathname, 
               UserPatch.Rmin, UserPatch.Rmax, UserPatch.Dmin, UserPatch.Dmax, RADIUS
       );
 
-    char tmpline[MAX_PATH_LENGTH];
-    if (VERBOSE)            { snprintf (tmpline, MAX_PATH_LENGTH, "%s -v",                command);            strcpy (command, tmpline); }
+    char tmpline[DVO_MAX_PATH];
+    if (VERBOSE)            { snprintf (tmpline, DVO_MAX_PATH, "%s -v",                command);            strcpy (command, tmpline); }
 
     // XXX these do not make sense for dvomerge client, where the output is required to exist
-    // if (CATFORMAT)          { snprintf (tmpline, MAX_PATH_LENGTH, "%s -D CATFORMAT %s",   command, CATFORMAT); strcpy (command, tmpline); }
-    // if (CATMODE)            { snprintf (tmpline, MAX_PATH_LENGTH, "%s -D CATMODE %s",     command, CATMODE);   strcpy (command, tmpline); }
-    // if (SKY_DEPTH)          { snprintf (tmpline, MAX_PATH_LENGTH, "%s -D SKY_DEPTH %d",   command, SKY_DEPTH); strcpy (command, tmpline); }
+    // if (CATFORMAT)          { snprintf (tmpline, DVO_MAX_PATH, "%s -D CATFORMAT %s",   command, CATFORMAT); strcpy (command, tmpline); }
+    // if (CATMODE)            { snprintf (tmpline, DVO_MAX_PATH, "%s -D CATMODE %s",     command, CATMODE);   strcpy (command, tmpline); }
+    // if (SKY_DEPTH)          { snprintf (tmpline, DVO_MAX_PATH, "%s -D SKY_DEPTH %d",   command, SKY_DEPTH); strcpy (command, tmpline); }
 
     fprintf (stderr, "command: %s\n", command);