diff --git a/proteus/Archiver.py b/proteus/Archiver.py index 8af27bb831..5c94ce2bec 100644 --- a/proteus/Archiver.py +++ b/proteus/Archiver.py @@ -175,11 +175,23 @@ class AR_base(object): or write the ``.ymf`` only at close, from the HDF5 metadata. ``driver`` how the ``.h5`` is used: ``"mpio"`` (the default), every rank - writing its part of one file through parallel HDF5; or - ``"serial"``, one file that master alone opens and writes, the - other ranks sending it their arrays. The serial driver needs no - parallel HDF5 and gives the same file; it costs the gather. A - serial-driver archive cannot be hot started on several ranks. + writing its part of one file through parallel HDF5; ``"serial"``, + one file that master alone opens and writes, the other ranks + sending it their arrays (no parallel HDF5 needed, the same file, + at the cost of the gather; such an archive cannot be hot started + on several ranks); or ``"per-rank"``, every rank writing its own + ``_p.h5`` with plain HDF5 and no communication at + all. Per-rank files hold per-rank datasets, so this driver sets + ``global_sync`` off. + ``document`` + with per-rank files, where the grids are described: ``"global"`` + (the default), one ``.ymf`` whose every step is a spatial + collection of the ranks' grids, each referencing its rank's file + (the roster's layout B: one thing to open, built by master from + the gathered grids); or ``"per-rank"``, one ``_p.ymf`` + per rank, each step a single grid, each rank writing its own with + no collective step and no single point of failure (layout C). The + single-file drivers take ``"global"`` only. ``includeLog`` carry the run's log, and proteus's version and command line, in the ``.ymf`` under ``extra: proteus``. @@ -188,14 +200,14 @@ class AR_base(object): (``values`` on every data item), which is then self-contained. For problems small enough to want that. - ====================== =========== ====== ========== ====== - mode incremental driver includeLog inline - ====================== =========== ====== ========== ====== - ``"without-a-net"`` off mpio off off - ``"split"`` (default) on mpio off off - ``"kitchen-sink"`` on mpio on off - ``"ymf-only"`` on -- off on - ====================== =========== ====== ========== ====== + ====================== =========== ====== ======== ========== ====== + mode incremental driver document includeLog inline + ====================== =========== ====== ======== ========== ====== + ``"without-a-net"`` off mpio global off off + ``"split"`` (default) on mpio global off off + ``"kitchen-sink"`` on mpio global on off + ``"ymf-only"`` on -- global off on + ====================== =========== ====== ======== ========== ====== *Without a net* is the ``.h5`` alone during the run, the ``.ymf`` derived from it at close -- and derivable again by ``gatherTimes``, @@ -203,7 +215,10 @@ class AR_base(object): current at every step; the problem's own ``.ymf`` specification, the third file of that mode, is the ``pro`` command's. *Kitchen sink* adds the log and the run's provenance to the document. *Ymf only* is one - file. + file. The driver and document switches combine with any mode: + ``driver="per-rank"`` with ``document="global"`` or ``"per-rank"`` are + the two per-rank-file layouts, and ``driver="serial"`` the one for a + build without parallel HDF5. """ #: Names for combinations of the switches; see the class docstring. MODES = { @@ -213,7 +228,12 @@ class AR_base(object): "ymf-only": {"inline": True}, } #: The HDF5 drivers an archive can be opened with. - DRIVERS = ("mpio", "serial") + DRIVERS = ("mpio", "serial", "per-rank") + #: Where the grids of per-rank files are described. + DOCUMENTS = ("global", "per-rank") + #: HDF5 root attribute recording the document layout (:data:`DOCUMENTS`), so + #: a reader of a per-rank file knows whether its steps are single grids. + DOCUMENT_ATTR = "ymf_archive_document" def __init__(self,dataDir,filename, useTextArchive=False, @@ -225,6 +245,7 @@ def __init__(self,dataDir,filename, mode=None, incremental=None, driver=None, + document=None, includeLog=None, inline=None): comm=Comm.get() @@ -241,10 +262,12 @@ def __init__(self,dataDir,filename, self._liveYmfWarned = False self.global_sync = global_sync switches = self.resolve_mode(mode, incremental=incremental, driver=driver, - includeLog=includeLog, inline=inline) + document=document, includeLog=includeLog, + inline=inline) self.mode = switches["mode"] self.writeIncrementalYmf = switches["incremental"] self.driver = switches["driver"] + self.document = switches["document"] self.includeLog = switches["includeLog"] self.inline = switches["inline"] #the arrays of an inline archive, by dataset name, on master; and @@ -259,6 +282,12 @@ def __init__(self,dataDir,filename, raise ValueError( "a serial-driver archive is written by master alone, so the other " "ranks cannot read it: hot start or read it with the mpio driver") + if self.driver == "per-rank" and self.global_sync: + #a global array cannot span files: per-rank files hold per-rank + #datasets, which is what global_sync off means + logEvent("Archive %s: per-rank files hold per-rank datasets; " + "global_sync is off" % (filename,), level=2) + self.global_sync = False if self.inline: #no file: the "reference" every data item carries is the dataset #name, which allGatherIncremental replaces with the values @@ -272,11 +301,14 @@ def __init__(self,dataDir,filename, mode = "r" else: mode = "w" - self.hdfFilename=filename+".h5" + if self.driver == "per-rank": + self.hdfFilename = "%s_p%d.h5" % (filename, self.rank) + else: + self.hdfFilename=filename+".h5" path = os.path.join(self.dataDir,self.hdfFilename) if self.driver == "mpio": self.hdfFile=h5py.File(path, mode, driver="mpio", comm = comm_world) - elif self.comm.isMaster(): + elif self.driver == "per-rank" or self.comm.isMaster(): self.hdfFile=h5py.File(path, mode) else: self.hdfFile=None @@ -311,8 +343,8 @@ def resolve_mode(cls, mode=None, **switches): if mode not in cls.MODES: raise ValueError("archive mode %r is not one of %s" % (mode, ", ".join(sorted(cls.MODES)))) - resolved = {"incremental": True, "driver": "mpio", "includeLog": False, - "inline": False} + resolved = {"incremental": True, "driver": "mpio", "document": "global", + "includeLog": False, "inline": False} resolved.update(cls.MODES[mode]) for name, value in switches.items(): if name not in resolved: @@ -322,9 +354,34 @@ def resolve_mode(cls, mode=None, **switches): if resolved["driver"] not in cls.DRIVERS: raise ValueError("archive driver %r is not one of %s" % (resolved["driver"], ", ".join(cls.DRIVERS))) + if resolved["document"] not in cls.DOCUMENTS: + raise ValueError("archive document %r is not one of %s" + % (resolved["document"], ", ".join(cls.DOCUMENTS))) + if resolved["document"] == "per-rank" and resolved["driver"] != "per-rank": + raise ValueError( + "a per-rank document describes a per-rank file: document='per-rank' " + "needs driver='per-rank', not %r" % (resolved["driver"],)) + if resolved["inline"] and resolved["driver"] != "mpio": + raise ValueError("an inline (ymf-only) archive has no .h5, so it takes no driver") resolved["mode"] = mode return resolved + @property + def writesDocument(self): + """Whether this rank writes a ``.ymf``: master, or every rank for per-rank documents.""" + return self.document == "per-rank" or self.comm.isMaster() + + @property + def uniformSteps(self): + """Whether a step of this rank's document is one grid (not a spatial collection).""" + return self.global_sync or self.document == "per-rank" + + def documentPath(self): + """This rank's ``.ymf``: ``.ymf``, or ``_p.ymf`` per rank.""" + if self.document == "per-rank": + return os.path.join(self.dataDir, "%s_p%d.ymf" % (self.filename, self.rank)) + return os.path.join(self.dataDir, self.filename + ".ymf") + @property def writesDatasets(self): """Whether the writers hand their arrays to the archive as datasets. @@ -347,7 +404,10 @@ def _document_extra(self, with_log): only complete at close). """ import sys - provenance = {"ranks": self.size, "archive_mode": self.mode} + provenance = {"ranks": self.size, "archive_mode": self.mode, + "archive_driver": self.driver} + if self.document == "per-rank": + provenance["rank"] = self.rank try: from proteus import __version__ provenance["version"] = str(__version__) @@ -547,6 +607,13 @@ def assemble_domain(self, n_steps=None): #records nothing, so there it *is* inferred -- see _v1_grids. recorded = self.hdfFile.attrs.get(self.GLOBAL_SYNC_ATTR) global_sync = self.global_sync if recorded is None else bool(int(recorded)) + #a per-rank document's steps are single grids whatever the dataset + #layout: its file describes one subdomain + document = self.hdfFile.attrs.get(self.DOCUMENT_ATTR) + if isinstance(document, bytes): + document = document.decode("utf-8") + if document == "per-rank": + global_sync = True #Time outside, space inside: walk the steps outermost, and let each #collection's grid join the instant at its time (add_*_step with a @@ -754,7 +821,7 @@ def _appendLiveStep(self, live_steps): 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") + path = self.documentPath() try: from ymf.archive import append_step, new_domain, write_ymf if not self._liveYmfStarted: @@ -767,7 +834,7 @@ def _appendLiveStep(self, live_steps): #run killed here has the whole instant or none of it. instants = [] for _, name, t, grids in live_steps: - if self.global_sync: + if self.uniformSteps: g = grids[0] space = {"topology": g["topology"], "geometry": g["geometry"], @@ -815,14 +882,14 @@ def gatherAndWriteTimes(self): logEvent("Text archive %s: no grid metadata to write" % (self.filename,)) return - ymf_path = os.path.join(self.dataDir, self.filename + ".ymf") + ymf_path = self.documentPath() write_ymf(domain, ymf_path, extra=self._document_extra(with_log=True)) 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: + if self.writesDocument and not self.readOnly: self.gatherAndWriteTimes() if self.hdfFile is not None: self.hdfFile.close() @@ -875,7 +942,15 @@ def allGatherIncremental(self): c.name for c in Domain) self.hdfFile.attrs[self.GLOBAL_SYNC_ATTR] = \ 1 if self.global_sync else 0 - + self.hdfFile.attrs[self.DOCUMENT_ATTR] = self.document + + #who holds this step's grids, and writes them: with one file, master + #gathers the ranks' grids (or holds the one global grid); with + #per-rank files every rank records its own grid in its own file, and + #for a global document master gathers them as well, to describe the + #spatial collection the ranks' files make up + gather = (not self.global_sync) and ( + self.driver != "per-rank" or self.document == "global") live_steps = [] for i, TemporalGridCollection in enumerate(Domain): local_grid = TemporalGridCollection.grid @@ -892,13 +967,17 @@ def allGatherIncremental(self): t = TemporalGridCollection.time TimeAttrib = repr(t) - #the step's grids, on master: one per rank, or the one global grid - if not self.global_sync: - grid_dicts = comm_world.gather(local_grid) - n_rows = self.size + #the grids this rank records: the step's, on master (one per + #rank, or the one global grid); its own, on a rank with its own + #file + if gather: + gathered = comm_world.gather(local_grid) else: - grid_dicts = [local_grid] if self.comm.isMaster() else None - n_rows = 1 + gathered = None + if self.comm.isMaster(): + grid_dicts = gathered if gather else [local_grid] + else: + grid_dicts = [local_grid] if self.driver == "per-rank" else None if self.inline: #no file: master replaces every dataset reference with the #values it was handed and keeps the step in the document @@ -909,10 +988,11 @@ def allGatherIncremental(self): t, grid_dicts)) continue payloads = ([dump_grid(g).encode("utf-8") for g in grid_dicts] - if self.comm.isMaster() else []) + if grid_dicts is not None else []) #every rank must create the dataset with the same width, so agree #on it collectively. Each rank contributes the length it knows. + #(Per-rank files need no agreement; the allreduce is harmless.) local_width = max((len(pl) for pl in payloads), default=0) width = comm_world.allreduce(local_width, op=MPI.MAX) width = max(int(width), 1) @@ -920,15 +1000,16 @@ def allGatherIncremental(self): dataset_name = self._metadata_dataset_name( TemporalGridCollection.name, self.n_datasets) if self.hdfFile is not None: + n_rows = len(payloads) if self.driver == "per-rank" else ( + self.size if gather else 1) grid_data = self._dataset(dataset_name, (n_rows,), '|S'+str(width)) grid_data.attrs['Time'] = TimeAttrib - 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 + for j, payload in enumerate(payloads): + grid_data[j] = payload + if self.writesDocument and self.hdfFile is not None: + #the writer of a document already holds its step's grids, so + #the running document costs no extra read and no extra parse live_steps.append((i, TemporalGridCollection.name, t, grid_dicts)) self.n_datasets += 1 @@ -957,6 +1038,9 @@ def create_dataset_async(self,name,data): the arrays are gathered to master, which writes (or keeps) them. """ comm_world = self.comm.comm.tompi4py() + if self.driver == "per-rank": + self._store(name, numpy.asarray(data)) + return if self.driver == "mpio" and not self.inline: metadata = comm_world.allgather((name,data.shape,data.dtype)) for i,m in enumerate(metadata): @@ -973,8 +1057,15 @@ def create_dataset_sync(self,name,offsets,data): With the mpio driver the dataset is created collectively and every rank writes its slice. With the serial driver, or inline, the - slices are gathered to master and assembled there. + slices are gathered to master and assembled there. Per-rank files + cannot hold a global array, which is why that driver turns + global_sync off. """ + if self.driver == "per-rank": + raise RuntimeError( + "a global array (%s) cannot be written across per-rank files; " + "the per-rank driver turns global_sync off, so this is a writer " + "that ignored ar.global_sync" % (name,)) if self.driver == "mpio" and not self.inline: dataset = self._dataset(name, tuple([offsets[-1]]+list(data.shape[1:])), data.dtype) @@ -1277,8 +1368,13 @@ def readArchiveDomain(filename, dataDir='.'): import h5py base = os.path.join(dataDir, filename) + hdfFilename = filename + ".h5" + if not os.path.exists(base + ".h5") and os.path.exists(base + "_p0.h5"): + #per-rank files with a global document: master's file holds the + #gathered grids of every rank + base, hdfFilename = base + "_p0", filename + "_p0.h5" archive = AR_base.__new__(AR_base) - archive.hdfFilename = filename + ".h5" + archive.hdfFilename = hdfFilename archive.global_sync = True # overridden by the recorded value archive.archived_domain = None archive._liveYmfStarted = False diff --git a/proteus/NumericalSolution.py b/proteus/NumericalSolution.py index 4c48e81eb9..2d32535170 100644 --- a/proteus/NumericalSolution.py +++ b/proteus/NumericalSolution.py @@ -156,6 +156,7 @@ def __init__(self,so,pList,nList,sList,opts,simFlagsList=None,TwoPhaseFlow=False tmp = Archiver.XdmfArchive(opts.dataDir,so.name,useTextArchive=opts.useTextArchive, gatherAtClose=opts.gatherArchive,hotStart=opts.hotStart, mode=getattr(opts,'archiveMode',None),driver=getattr(opts,'archiveDriver',None), + document=getattr(opts,'archiveDocument',None), useGlobalXMF=(not opts.subdomainArchives), global_sync=opts.global_sync) if self.fastArchive==True: @@ -165,11 +166,13 @@ def __init__(self,so,pList,nList,sList,opts,simFlagsList=None,TwoPhaseFlow=False elif len(self.pList) == 1: self.ar = {0:Archiver.XdmfArchive(opts.dataDir,so.name,useTextArchive=opts.useTextArchive, gatherAtClose=opts.gatherArchive,hotStart=opts.hotStart, - mode=getattr(opts,'archiveMode',None),driver=getattr(opts,'archiveDriver',None))} #reuse so.name if possible + mode=getattr(opts,'archiveMode',None),driver=getattr(opts,'archiveDriver',None), + document=getattr(opts,'archiveDocument',None))} #reuse so.name if possible else: self.ar = dict([(i,Archiver.XdmfArchive(opts.dataDir,p.name,useTextArchive=opts.useTextArchive, gatherAtClose=opts.gatherArchive,hotStart=opts.hotStart, - mode=getattr(opts,'archiveMode',None),driver=getattr(opts,'archiveDriver',None))) for i,p in enumerate(self.pList)]) + mode=getattr(opts,'archiveMode',None),driver=getattr(opts,'archiveDriver',None), + document=getattr(opts,'archiveDocument',None))) for i,p in enumerate(self.pList)]) #by default do not save quadrature point info self.archive_q = dict([(i,False) for i in range(len(self.pList))]); self.archive_ebq_global = dict([(i,False) for i in range(len(self.pList))]); diff --git a/proteus/cli/run.py b/proteus/cli/run.py index 841ef33f7d..36ff49f1b4 100644 --- a/proteus/cli/run.py +++ b/proteus/cli/run.py @@ -79,6 +79,20 @@ def add_parser(subparsers): help="cache archive writes") g.add_argument("-n", "--no-global-sync", action="store_true", help="do not synchronize archive metadata globally") + g.add_argument("--archive-mode", choices=["without-a-net", "split", "kitchen-sink", "ymf-only"], + help="what the archive writes: without-a-net (the .h5 alone during the " + "run, the .ymf at close), split (the .h5 and a .ymf kept current at " + "every step; the default), kitchen-sink (split, plus the log and the " + "run's provenance in the .ymf), ymf-only (no .h5: the arrays go in " + "the .ymf)") + g.add_argument("--archive-driver", choices=["mpio", "serial", "per-rank"], + help="how the .h5 is written: mpio (every rank, through parallel HDF5; " + "the default), serial (one file, master alone; needs no parallel " + "HDF5) or per-rank (each rank its own _p.h5; implies -n)") + g.add_argument("--archive-document", choices=["global", "per-rank"], + help="with per-rank files, where the grids are described: global (one " + ".ymf whose steps are spatial collections of the ranks' grids; the " + "default) or per-rank (one .ymf per rank, each step a single grid)") g.add_argument("-S", "--save-dof", action="store_true", help="save the degrees of freedom") g.add_argument("-B", "--write-vpp", action="store_true", @@ -147,6 +161,9 @@ def _options(args): gatherArchive=args.gather_archive, subdomainArchives=args.subdomain_archives, global_sync=not args.no_global_sync, + archiveMode=args.archive_mode, + archiveDriver=args.archive_driver, + archiveDocument=args.archive_document, hotStart=args.hot_start, hotStartMode=args.hot_start_mode, hotStartTime=args.hot_start_time, diff --git a/proteus/runtime.py b/proteus/runtime.py index dd7dd66491..5d59f738ef 100644 --- a/proteus/runtime.py +++ b/proteus/runtime.py @@ -49,6 +49,11 @@ class RunOptions: subdomainArchives: bool = False #: synchronize archive metadata globally (parun -n clears it) global_sync: bool = True + #: the archive's mode, driver and document layout (--archive-mode, + #: --archive-driver, --archive-document); None is Archiver's default + archiveMode: str = None + archiveDriver: str = None + archiveDocument: str = None #: restart from the archive (-H) hotStart: bool = False #: what a hot start continues (--hot-start-mode): "continue" resumes the diff --git a/scripts/parun b/scripts/parun index 5254f7b5fa..9876ca8ee1 100755 --- a/scripts/parun +++ b/scripts/parun @@ -169,9 +169,15 @@ parser.add_option("--archive-mode", parser.add_option("--archive-driver", dest="archiveDriver", type="choice", - choices=["mpio", "serial"], + choices=["mpio", "serial", "per-rank"], default=None, - help="""how the .h5 is written: mpio (every rank, through parallel HDF5; the default) or serial (master alone, the other ranks sending it their arrays; needs no parallel HDF5)""") + help="""how the .h5 is written: mpio (every rank, through parallel HDF5; the default), serial (one file, master alone, the other ranks sending it their arrays; needs no parallel HDF5) or per-rank (each rank its own _p.h5, no communication; implies -n)""") +parser.add_option("--archive-document", + dest="archiveDocument", + type="choice", + choices=["global", "per-rank"], + default=None, + help="""with per-rank files, where the grids are described: global (one .ymf whose steps are spatial collections of the ranks' grids; the default) or per-rank (one _p.ymf per rank, each step a single grid)""") parser.add_option("-H","--hotStart", dest="hotStart", action="store_true", diff --git a/test/test_archive_parallel_metadata.py b/test/test_archive_parallel_metadata.py index 86db8e7cd5..5dc86d8998 100644 --- a/test/test_archive_parallel_metadata.py +++ b/test/test_archive_parallel_metadata.py @@ -80,9 +80,12 @@ def _find_mpiexec(): ar = Archiver.XdmfArchive(datadir, "mpitest", global_sync=global_sync, hotStart=hot, mode=os.environ.get("MODE") or None, - driver=os.environ.get("DRIVER") or None) + driver=os.environ.get("DRIVER") or None, + document=os.environ.get("DOCUMENT") or None) if os.environ.get("NO_INCREMENTAL") == "1": ar.writeIncrementalYmf = False + # the per-rank driver turns global_sync off: write what the archive is + global_sync = ar.global_sync if hot: # what NumericalSolution does: continue after the archive's last step # (or, for HOT_START=back, from the first step of two: the second @@ -167,19 +170,20 @@ def _find_mpiexec(): def run_ranks(tmp_path, n_ranks, global_sync, hot_start=False, skip_close=False, incremental=True, mode=None, driver=None, - with_log=False, name="archive"): + document=None, with_log=False, name="archive"): """Run the rank script under mpiexec (directly only if there is none). ``hot_start`` continues the archive an earlier call wrote, for one - more step, the way a hot-started run does. ``mode`` and ``driver`` are - the archive's; ``with_log`` opens a proteus log first, as a run does. + more step, the way a hot-started run does. ``mode``, ``driver`` and + ``document`` are the archive's; ``with_log`` opens a proteus log + first, as a run does. """ datadir = str(tmp_path / name) env = dict(os.environ, GLOBAL_SYNC="1" if global_sync else "0", DATADIR=datadir, HOT_START=("back" if hot_start == "back" else "1") if hot_start else "0", SKIP_CLOSE="1" if skip_close else "0", NO_INCREMENTAL="0" if incremental else "1", - MODE=mode or "", DRIVER=driver or "", + MODE=mode or "", DRIVER=driver or "", DOCUMENT=document or "", WITH_LOG="1" if with_log else "0") # 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, @@ -810,8 +814,10 @@ def test_an_ymf_without_append_step_costs_the_document_not_the_run(tmp_path): import ymf.archive live = types.SimpleNamespace(dataDir=str(tmp_path), filename="skew", - global_sync=True, _liveYmfStarted=False, - _liveYmfWarned=False) + global_sync=True, uniformSteps=True, + documentPath=lambda: str(tmp_path / "skew.ymf"), + _document_extra=lambda with_log: None, + _liveYmfStarted=False, _liveYmfWarned=False) step = [(0, "Grid", 0.0, [{"topology": {}, "geometry": {}}])] saved = ymf.archive.append_step @@ -973,3 +979,114 @@ def test_ymf_only_streams_its_steps_too(tmp_path): domain, _ = read_ymf(os.path.join(datadir, "mpitest.ymf")) assert len(archive_collections(domain)[0]["data"]) == 2 assert "values" in archive_collections(domain)[0]["data"][1]["geometry"]["data_item"] + + +# -------------------------------------------------------------------------- +# every combination of the switches: the archive is whole +# -------------------------------------------------------------------------- + + +def check_archive(datadir, n_ranks, global_sync, driver, document): + """Every document the run left describes arrays that are there. + + The integrity check the combinations share: each document validates, + has every collection's two steps in the expected shape (one grid, or + one per rank), and every data item names a dataset that exists in the + file it says, with the dimensions it declares; and the archive can be + rebuilt from the HDF5 alone. + """ + from ymf.archive import YmfArchiveError, load_array, read_ymf, validate_domain + from proteus.Archiver import readArchiveDomain + + if document == "per-rank": + documents = ["mpitest_p%d.ymf" % r for r in range(n_ranks)] + files = ["mpitest_p%d.h5" % r for r in range(n_ranks)] + else: + documents = ["mpitest.ymf"] + files = (["mpitest_p%d.h5" % r for r in range(n_ranks)] if driver == "per-rank" + else ["mpitest.h5"]) + assert sorted(os.listdir(datadir)) == sorted(documents + files + ["mpitest.log"] * 0) + grids_per_step = (1 if global_sync or document == "per-rank" else n_ranks) + referenced = set() + for doc in documents: + domain, extra = read_ymf(os.path.join(datadir, doc)) + validate_domain(domain) + (collection,) = archive_collections(domain) + assert len(collection["data"]) == 2, doc + for step in collection["data"]: + grids = step.get("spatial_collection", [step]) + assert len(grids) == grids_per_step, (doc, step.get("time")) + for grid in grids: + for owner in [grid["topology"], grid["geometry"]] + grid["attributes"]: + item = owner["data_item"] + assert item["format"] == "HDF", item + file_name, dataset = item["data"].split(":/") + assert file_name in files, item["data"] + referenced.add(file_name) + array = load_array(item, datadir) # the dataset exists + assert array.size == np.prod(item["dimensions"]), item["data"] + assert extra["proteus"]["archive_driver"] == driver + if driver == "per-rank" and document == "global": + # the ranks' grids reference every rank's file, and the document + # is derivable from master's file alone + assert referenced == set(files) + rebuilt = readArchiveDomain("mpitest", datadir) + assert len(archive_collections(rebuilt)[0]["data"]) == 2 + (grids,) = [s["spatial_collection"] for s in archive_collections(rebuilt)[0]["data"][:1]] + assert len(grids) == n_ranks + + +SINGLE_FILE = [(d, g) for d in ("mpio", "serial") for g in (True, False)] + + +@pytest.mark.parametrize("n_ranks", [1, 2, 3]) +@pytest.mark.parametrize("driver,global_sync", SINGLE_FILE) +def test_every_single_file_layout_leaves_a_whole_archive(tmp_path, n_ranks, driver, global_sync): + datadir = run_ranks(tmp_path, n_ranks, global_sync, driver=driver) + check_archive(datadir, n_ranks, global_sync, driver, "global") + + +@pytest.mark.parametrize("n_ranks", [1, 2, 3]) +@pytest.mark.parametrize("document", ["global", "per-rank"]) +def test_every_per_rank_file_layout_leaves_a_whole_archive(tmp_path, n_ranks, document): + """Layouts B (one global document) and C (one document per rank).""" + datadir = run_ranks(tmp_path, n_ranks, True, driver="per-rank", document=document) + check_archive(datadir, n_ranks, False, "per-rank", document) + + +def test_per_rank_files_hold_per_rank_datasets(): + """The per-rank driver turns global_sync off: a global array cannot span files.""" + from proteus.Archiver import AR_base + + assert AR_base.resolve_mode(driver="per-rank")["document"] == "global" + with pytest.raises(ValueError, match="document='per-rank' needs driver='per-rank'"): + AR_base.resolve_mode(driver="mpio", document="per-rank") + with pytest.raises(ValueError, match="archive document 'spread'"): + AR_base.resolve_mode(document="spread") + with pytest.raises(ValueError, match="takes no driver"): + AR_base.resolve_mode("ymf-only", driver="serial") + + +def test_a_per_rank_document_is_the_running_document_of_its_rank(tmp_path): + """Layout C's documents are appended a step at a time like the global one.""" + from ymf.archive import read_ymf + + datadir = run_ranks(tmp_path, 2, True, driver="per-rank", document="per-rank", + skip_close=True) + for r in range(2): + domain, extra = read_ymf(os.path.join(datadir, "mpitest_p%d.ymf" % r)) + assert len(archive_collections(domain)[0]["data"]) == 2 + assert extra["proteus"]["rank"] == r + assert "spatial_collection" not in archive_collections(domain)[0]["data"][0] + + +def test_per_rank_files_hot_start(tmp_path): + """Each rank continues its own file; no collective read is needed.""" + from ymf.archive import read_ymf + + run_ranks(tmp_path, 2, True, driver="per-rank", document="global") + datadir = run_ranks(tmp_path, 2, True, driver="per-rank", document="global", hot_start=True) + domain, _ = read_ymf(os.path.join(datadir, "mpitest.ymf")) + assert [s["time"] for s in archive_collections(domain)[0]["data"]] == [0.0, 0.5, 1.0] + check_steps = archive_collections(domain)[0]["data"] + assert all(len(s["spatial_collection"]) == 2 for s in check_steps) diff --git a/test/test_pro_cli.py b/test/test_pro_cli.py index f3d5645a67..5b0a587196 100644 --- a/test/test_pro_cli.py +++ b/test/test_pro_cli.py @@ -1267,3 +1267,26 @@ def test_a_missing_spec_is_an_argument_error(tmp_path): out = pro("run", "nothere.ymf", cwd=tmp_path) assert out.returncode == 2 and "nothere.ymf: no such file" in out.stderr assert "Traceback" not in out.stderr + + +@pytest.mark.skipif(shutil.which("parun") is None, reason="parun not on PATH") +def test_run_archive_options_reach_the_archive(tmp_path): + """--archive-mode, --archive-driver and --archive-document are pro run's too.""" + from ymf.archive import read_ymf + sources = [os.path.join(HERE, "ci", f) + for f in ("adr.py", "ladr_2d_p.py", "ladr_2d_n.py")] + for s in sources: + shutil.copy(s, tmp_path) + ran = pro("run", "--archive-mode", "kitchen-sink", "--archive-driver", "serial", + "ladr_2d_p.py", "ladr_2d_n.py", cwd=tmp_path) + assert ran.returncode == 0, ran.stderr[-2000:] + _, extra = read_ymf(tmp_path / "ladr_2d.ymf") + provenance = extra["proteus"] + assert provenance["archive_mode"] == "kitchen-sink" + assert provenance["archive_driver"] == "serial" + assert "log" in provenance and "command" in provenance + # the document layout is refused where it makes no sense, by name + ran = pro("run", "--archive-document", "per-rank", "ladr_2d_p.py", "ladr_2d_n.py", + cwd=tmp_path) + assert ran.returncode != 0 + assert "document='per-rank' needs driver='per-rank'" in ran.stderr