diff --git a/proteus/Archiver.py b/proteus/Archiver.py index ced92758b0..1b0187f18a 100644 --- a/proteus/Archiver.py +++ b/proteus/Archiver.py @@ -70,6 +70,9 @@ def __init__(self,dataDir,filename, self.readOnly = readOnly self.n_datasets = 0 self.archived_domain = None + #the running document: appended to a step at a time, never rewritten + self._liveYmfStarted = False + self._liveYmfWarned = False self.global_sync = global_sync comm_world = self.comm.comm.tompi4py() self.tree=ElementTree( @@ -427,6 +430,59 @@ def _metadata_dataset_name(self, collection_name, index): """Name of the HDF5 dataset holding one collection's step metadata.""" return (collection_name + "_" + str(index)).replace(" ", "_") + #: Whether each step is appended to the ``.ymf`` as it is produced, so + #: that the archive of record is current during the run rather than only + #: after a clean close. Appending is constant work per step -- see + #: :meth:`_appendLiveStep` -- so this is on by default. Set False to + #: restore the write-at-close-only behaviour. + writeIncrementalYmf = True + + def _appendLiveStep(self, live_steps): + """Append this timestep to the running document. + + One record per collection, appended and flushed. Master already + holds the grids, so this costs no HDF5 read and no YAML parse, and + because it appends rather than rewrites it costs the same whether + the archive holds ten steps or ten thousand. + + The first call writes the header document -- the archive version and + an empty domain -- which the records are then appended to. The + header is written to a temporary file and renamed so that a reader + arriving mid-write never sees half of it; the records after it need + no such care, since each ends with the document-end marker that + tells a reader it is complete. + + Failures are logged and swallowed, the import among them: an ymf + too old to have ``append_step`` should cost the run its live + document, not the run. The authoritative document is still written + at close from the HDF5 file, and the steps already in the archive + matter more than this one. + """ + path = os.path.join(self.dataDir, self.filename + ".ymf") + try: + from ymf.archive import append_step, new_domain, write_ymf + if not self._liveYmfStarted: + tmp = path + ".writing" + write_ymf(new_domain(live_steps[0][1]), tmp) + os.replace(tmp, path) + self._liveYmfStarted = True + for _, name, t, grids in live_steps: + if self.global_sync: + g = grids[0] + step = {"time": t, "topology": g["topology"], + "geometry": g["geometry"], + "attributes": g.get("attributes", [])} + else: + step = {"time": t, "spatial_collection": grids} + append_step(path, name, step) + except Exception as error: + if not self._liveYmfWarned: + self._liveYmfWarned = True + logEvent("Could not update %s during the run (%s); the " + "archive is unaffected and the document will be " + "written at close" + % (os.path.basename(path), error), level=1) + def gatherAndWriteTimes(self): """Assemble the whole archive document and write it out. @@ -452,6 +508,7 @@ def gatherAndWriteTimes(self): write_ymf(domain, ymf_path) logEvent("Wrote YMF archive %s (ymf2xmf %s writes an .xmf for viewers)" % (ymf_path, os.path.basename(ymf_path))) + def close(self): logEvent("Closing Archive") if self.comm.isMaster() and not self.readOnly: @@ -509,6 +566,7 @@ def allGatherIncremental(self): self.hdfFile.attrs[self.GLOBAL_SYNC_ATTR] = \ 1 if self.global_sync else 0 + live_steps = [] for i, TemporalGridCollection in enumerate(Domain): GridLocal = TemporalGridCollection[-1] time_elem = GridLocal.find("Time") @@ -545,7 +603,16 @@ def allGatherIncremental(self): if self.comm.isMaster(): for j, payload in enumerate(payloads): grid_data[j] = payload + if self.comm.isMaster() and self.hdfFile is not None: + #master already holds this step's grids, so the running + #document costs no extra read and no extra parse + live_grids = (grid_dicts if not self.global_sync + else [local_grid]) + live_steps.append((i, TemporalGridCollection.attrib['Name'], + float(TimeAttrib), live_grids)) self.n_datasets += 1 + if live_steps and self.writeIncrementalYmf: + self._appendLiveStep(live_steps) logEvent("Done Gathering Archive Time Step") def sync(self): logEvent("Syncing Archive",level=3) @@ -789,6 +856,8 @@ def readArchiveDomain(filename, dataDir='.'): archive.hdfFilename = filename + ".h5" archive.global_sync = True # overridden by the recorded value archive.archived_domain = None + archive._liveYmfStarted = False + archive._liveYmfWarned = False with h5py.File(base + ".h5", "r") as hdfFile: archive.hdfFile = hdfFile domain = archive.assemble_domain() diff --git a/test/test_archive_parallel_metadata.py b/test/test_archive_parallel_metadata.py index de6abb4330..e0d5a24fde 100644 --- a/test/test_archive_parallel_metadata.py +++ b/test/test_archive_parallel_metadata.py @@ -74,6 +74,8 @@ def _find_mpiexec(): hot = os.environ.get("HOT_START") == "1" ar = Archiver.XdmfArchive(datadir, "mpitest", global_sync=global_sync, hotStart=hot) + if os.environ.get("NO_INCREMENTAL") == "1": + ar.writeIncrementalYmf = False if hot: # what NumericalSolution does: continue after the archive's last step ar.n_datasets = len(ar.archived_times) @@ -127,7 +129,13 @@ def _find_mpiexec(): dimensions=[n_nodes]) ar.sync() - ar.close() + if os.environ.get("SKIP_CLOSE") == "1": + # what a killed run leaves behind: the steps are synced, close() + # never happens. Flush so the HDF5 is as it would be, then stop. + if ar.hdfFile is not None: + ar.hdfFile.flush() + else: + ar.close() comm.barrier() if rank == 0: print("OBSERVED_SIZE=%d" % size) @@ -135,7 +143,8 @@ def _find_mpiexec(): ) -def run_ranks(tmp_path, n_ranks, global_sync, hot_start=False): +def run_ranks(tmp_path, n_ranks, global_sync, hot_start=False, + skip_close=False, incremental=True): """Run the rank script under mpiexec (or directly for one rank). ``hot_start`` continues the archive an earlier call wrote, for one @@ -143,7 +152,9 @@ def run_ranks(tmp_path, n_ranks, global_sync, hot_start=False): """ datadir = str(tmp_path / "archive") env = dict(os.environ, GLOBAL_SYNC="1" if global_sync else "0", - DATADIR=datadir, HOT_START="1" if hot_start else "0") + DATADIR=datadir, HOT_START="1" if hot_start else "0", + SKIP_CLOSE="1" if skip_close else "0", + NO_INCREMENTAL="0" if incremental else "1") # More ranks than cores (3 on a 2-core CI runner): Open MPI refuses # unless told it may oversubscribe -- OMPI_MCA_* for Open MPI 4, # PRTE_MCA_* for Open MPI 5's PRRTE launcher. MPICH oversubscribes by @@ -637,3 +648,131 @@ def test_a_parallel_hot_start_continues_the_archive(tmp_path): run_ranks(tmp_path, 2, global_sync=False, hot_start=True) (collection,) = read_domain(datadir)["time_collections"] assert [step["time"] for step in collection["data"]] == [0.0, 0.5, 1.0] + + +# -------------------------------------------------------------------------- +# the document is current during the run, not only after a clean close +# -------------------------------------------------------------------------- + + +def test_an_interrupted_run_still_leaves_a_readable_document(tmp_path): + """The point of the incremental write. + + A run that is killed, times out on a scheduler or dies on one rank never + reaches close(), and used to leave an archive with no .ymf at all: the + metadata was in the HDF5 file but the document that describes it was + assembled only at close. Recovering it meant a separate pass. Now the + document is already there and already correct. + """ + datadir = run_ranks(tmp_path, 1, global_sync=False, skip_close=True) + (collection,) = read_domain(datadir)["time_collections"] + assert [s["time"] for s in collection["data"]] == [0.0, 0.5] + + +def test_what_it_looked_like_before(tmp_path): + """The same interrupted run with the incremental write switched off. + + Kept as the control: without it there is no document, which is what + makes the test above worth having rather than a tautology. + """ + datadir = run_ranks(tmp_path, 1, global_sync=False, skip_close=True, + incremental=False) + assert not os.path.exists(os.path.join(datadir, "mpitest.ymf")) + assert os.path.exists(os.path.join(datadir, "mpitest.h5")) + + +def test_the_running_document_agrees_with_the_one_written_at_close(tmp_path): + """Two independent paths to the same document, so they can be compared. + + The running document is appended to in memory from the grids master + already holds; the one at close is re-assembled from the YAML in the + HDF5 file. They are built by different code from different sources, and + they have to agree -- if they ever diverge, the archive of record + depends on whether the run finished, which is exactly the property this + campaign exists to remove. + """ + running = read_domain(run_ranks(tmp_path / "a", 1, global_sync=False, + skip_close=True)) + closed = read_domain(run_ranks(tmp_path / "b", 1, global_sync=False)) + assert running == closed + + +@pytest.mark.parametrize("global_sync", [False, True]) +def test_the_running_document_agrees_for_both_layouts(tmp_path, global_sync): + running = read_domain(run_ranks(tmp_path / "a", 1, global_sync=global_sync, + skip_close=True)) + closed = read_domain(run_ranks(tmp_path / "b", 1, global_sync=global_sync)) + assert running == closed + + +@pytest.mark.skipif(MPIEXEC is None, reason="no MPI launcher available") +@pytest.mark.parametrize("n_ranks", [2, 3]) +def test_the_running_document_holds_every_ranks_grid(tmp_path, n_ranks): + """Only master writes it, but it must describe the whole decomposition.""" + datadir = run_ranks(tmp_path, n_ranks, global_sync=False, skip_close=True) + (collection,) = read_domain(datadir)["time_collections"] + assert len(collection["data"]) == 2 + for step in collection["data"]: + subs = step["spatial_collection"] + assert len(subs) == n_ranks + assert [g["name"] for g in subs] == [ + "Grid_p%d" % r for r in range(n_ranks)] + + +def test_no_leftover_temporary_file(tmp_path): + """The write is rename-based, so nothing may be left beside the target.""" + datadir = run_ranks(tmp_path, 1, global_sync=False, skip_close=True) + leftovers = [f for f in os.listdir(datadir) if f.endswith(".writing")] + assert leftovers == [] + + +def test_the_running_document_is_appended_not_rewritten(tmp_path): + """The file is a stream mid-run, which is what makes the write O(1). + + Rewriting the whole document to record a step is O(steps) per step and + so quadratic over a run -- 35 ms a step at 500 steps, minutes over a + long one. Appending writes one record whatever the archive holds. The + observable difference is the shape of the file: a stream of documents + while the run is going, collapsed to a single document at close. + """ + datadir = run_ranks(tmp_path, 1, global_sync=False, skip_close=True) + text = open(os.path.join(datadir, "mpitest.ymf"), encoding="utf-8").read() + assert text.count("\n---\n") == 2 # one record per step + assert text.count("\n...\n") == 2 # each one completed + + +def test_close_leaves_a_single_document(tmp_path): + """A finished run is in the canonical shape, so nothing downstream has + to care that the live form exists.""" + datadir = run_ranks(tmp_path, 1, global_sync=False) + text = open(os.path.join(datadir, "mpitest.ymf"), encoding="utf-8").read() + assert "\n---\n" not in text + + +def test_an_ymf_without_append_step_costs_the_document_not_the_run(tmp_path): + """proteus and ymf move independently, so the skew has to be survivable. + + ``append_step`` arrived in ymf after this writer was written, and the + import of it used to sit outside the guard -- which turned "your live + document is stale" into an exception out of the middle of a timestep. + CI caught it honestly: three legs of one run resolved ``ymf@main`` five + minutes before the merge that added the function. Any user whose ymf + predates it would have hit the same thing. + """ + import types + import proteus.Archiver as AR + import ymf.archive + + live = types.SimpleNamespace(dataDir=str(tmp_path), filename="skew", + global_sync=True, _liveYmfStarted=False, + _liveYmfWarned=False) + step = [(0, "Grid", 0.0, [{"topology": {}, "geometry": {}}])] + + saved = ymf.archive.append_step + del ymf.archive.append_step + try: + AR.AR_base._appendLiveStep(live, step) # must not raise + finally: + ymf.archive.append_step = saved + + assert live._liveYmfWarned, "the failure has to be reported, not hidden"