Index: /branches/eam_branches/ipp-20120405/Ohana/src/dvomerge/doc/failsafe.txt =================================================================== --- /branches/eam_branches/ipp-20120405/Ohana/src/dvomerge/doc/failsafe.txt (revision 33875) +++ /branches/eam_branches/ipp-20120405/Ohana/src/dvomerge/doc/failsafe.txt (revision 33875) @@ -0,0 +1,84 @@ + +we have a problem with dvomerge and flaky hosts. the problem is that +an I/O glitch can prevent a table from being written. although we can +(and do?) catch the failure, we do not have a good recovery method. +recovery would involve 2 things: + +1) we should revert the failed table to the original state +2) we should be able to re-run the merge correctly + +point (1) implies that we need a solid copy of the original file before +writing the new output tables. + +point (2) implies that we need to know which tables have been +successfully merged. + +For point (1), since we are writing out the entire table, we can +*move* the original file to a temporary name, and unlink it when the +output is successful. + +For point (2), we should write an ID to the output header that comes +from the input catalog file. then a second attempt will skip any +tables for which the IDs already match. + +Problems & Questions for (2): + +* do we need to have the ID in each input db table? +* we do not have an ID already in existing tables -- I'll need a tool + to write the ID +OR +* can we use something like the name & date of the file to check the +* identity? + +(name, date, size can be determined without reading the file, so that +is quite tempting...) + +I'll still need to do 2 passes, on where I determine which output +tables match the given input table, then check those output tables for +the merge data. + +--- + +Proposal: + +1) Add the following data to the output catalogs: + +OBJID = 675 / +LMRG_DB = 'catdir.2mass' / name of last merged database +LMRG_ST = 80640 / size of last merged cpt file +LMRG_SS = 118080 / size of last merged cps file +LMRG_SM = 331200 / size of last merged cpm file +LMRG_DT = '2012-04-27 16:42:45' / mod-date of last merged cpm file + +This allows us to tell if the current catalog has been merged in yet, +but it does not require us to track any database IDs. If a file has +been merged, we skip it; otherwise, we merge it. + +The proposal may do the wrong thing in the following case: + +dvomerge db1 into db2 +(failure part way through) + +addstar file -D CATDIR db1 + +dvomerge db1 into db2 + +If 'file' overlaps the already-merged sections of db1, they will be +re-ingested. In fact, the first pass need not have failed -- it will +still double merge some data. + +However, it does mean the following does the right thing: + +dvomerge db1 into db2 + +dvomerge db1 into db2 + +(no merge should happen on the second pass) + +dvomerge db1 into db2 +(failure) + +dvomerge db1 into db2 + +(only unmerged entries should be merged on the second pass) + Index: /branches/eam_branches/ipp-20120405/Ohana/src/dvomerge/include/dvomerge.h =================================================================== --- /branches/eam_branches/ipp-20120405/Ohana/src/dvomerge/include/dvomerge.h (revision 33874) +++ /branches/eam_branches/ipp-20120405/Ohana/src/dvomerge/include/dvomerge.h (revision 33875) @@ -17,5 +17,5 @@ # include -# define MAX_PATH_LENGTH 1024 +# define DVO_MAX_PATH 1024 int PARALLEL; Index: /branches/eam_branches/ipp-20120405/Ohana/src/dvomerge/src/dvomergeUpdate_catalogs.c =================================================================== --- /branches/eam_branches/ipp-20120405/Ohana/src/dvomerge/src/dvomergeUpdate_catalogs.c (revision 33874) +++ /branches/eam_branches/ipp-20120405/Ohana/src/dvomerge/src/dvomergeUpdate_catalogs.c (revision 33875) @@ -36,4 +36,5 @@ 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,6 +43,79 @@ } - // 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,4 +126,6 @@ } + 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,6 +136,6 @@ // 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,4 +147,13 @@ 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,4 +165,8 @@ 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,4 +175,5 @@ } SkyListFree (outlist); + free (moddate); dvo_catalog_unlock (&incatalog); @@ -108,26 +198,26 @@ // 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);