Functions
const char *	aggregate_basename ()
	Basename of the aggregate file ("ZPENodes_agg.txt"). It is in the MPI merge's append-files list, so rank copies concatenate rather than overwrite. Note that the glob `ZPENode*` also matches this name. More...

void	append_record (const std::string &pe_dir, int branch_id, const std::string &content)
	Append one node record to `<pe_dir>ZPENodes_agg.txt`. More...

void	parse_aggregate (const fs::path &path, std::vector< std::pair< int, std::string > > &out)
	Parse one aggregate file, appending (id, content) to `out` in file order, duplicates and negative ids included. More...

std::pair< size_t, size_t >	compact_aggregate (const fs::path &agg_path)
	Rewrite one rank's aggregate file last-wins per id, sorted by id. More...

std::vector< std::pair< int, std::string > >	collect (const std::string &pe_dir)
	Collect every PE node under `pe_dir` (recursively), from either layout, reading all node content into memory. More...

size_t	rewrite_sequential (const std::string &pe_dir, std::vector< std::pair< int, std::string > > entries, bool aggregate)
	Replace the contents of `pe_dir` with `entries` renumbered 0..N-1 in ascending id order (the content path of the renumber). More...

size_t	finalize_sequential (const std::string &pe_dir, bool aggregate)
	End-of-run finalize: renumber every PE node 0..N-1 in id order, ending in the same layout `rewrite_sequential` produces (a single `PEFolder_0/`, no negative-id sentinels, no other buckets). More...

Function Documentation

◆ aggregate_basename()

const char * adsidac::pe::aggregate_basename ( )

inline

Basename of the aggregate file ("ZPENodes_agg.txt"). It is in the MPI merge's append-files list, so rank copies concatenate rather than overwrite. Note that the glob ZPENode* also matches this name.

Definition at line 59 of file PEOutput.h.

Referenced by adsidac::merge::append_basenames(), append_record(), collect(), adsidac::merge::concat_appends_from_ranks(), finalize_sequential(), adsidac::merge::looks_finalized(), Server::merge_outputs(), rewrite_sequential(), and Client::self_merge_outputs().

Here is the caller graph for this function:

◆ append_record()

void adsidac::pe::append_record	(	const std::string &	pe_dir,
		int	branch_id,
		const std::string &	content
	)

inline

Append one node record to <pe_dir>ZPENodes_agg.txt.

Creates pe_dir if needed and opens the file in append mode for this one record. If the file cannot be opened the record is dropped silently, and the write itself is not checked.

Parameters

pe_dir	PE directory of this process, with trailing '/'.
branch_id	Node id written in the record header (may be negative, and may repeat an id written earlier; the later record wins).
content	Node body, stored byte for byte.

Definition at line 70 of file PEOutput.h.

References aggregate_basename().

Here is the call graph for this function:

◆ collect()

std::vector< std::pair< int, std::string > > adsidac::pe::collect ( const std::string & pe_dir )

inline

Collect every PE node under pe_dir (recursively), from either layout, reading all node content into memory.

Negative ids (sentinels such as -999) are skipped, matching the per-file merge. When an id appears more than once the LAST occurrence wins: within an aggregate that is file order; across files it is directory iteration order, which the filesystem does not define.

Returns: (id, content) pairs in ascending id order; empty if pe_dir does not exist.

Definition at line 156 of file PEOutput.h.

References aggregate_basename(), and parse_aggregate().

Referenced by finalize_sequential(), and Server::merge_outputs().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ compact_aggregate()

std::pair< size_t, size_t > adsidac::pe::compact_aggregate ( const fs::path & agg_path )

inline

Rewrite one rank's aggregate file last-wins per id, sorted by id.

A client re-saves a whole lineage after every sequence, so a rank's aggregate carries every content revision of every node it ever held, several records per unique id. The merge's collect() applies last-wins anyway, so compacting the rank file first changes nothing downstream while cutting the merge's read volume. Within a rank, last-wins here is exactly what collect() chooses from this rank's stream; cross-rank resolution still rides on rank-ordered concatenation. Negative sentinel ids collapse like any other (collect() drops them, and the intermediate-state readers test existence, not multiplicity). The rewrite goes to <agg_path>.compact and is renamed into place, so an interruption leaves either the uncompacted or the compacted file, both valid aggregates. A file with no duplicate ids is left untouched, which makes a second call a no-op. If the temp file cannot be written or renamed the input file is left as it is.

Returns: (records read, unique ids); (0, 0) for a missing or empty file.

Definition at line 123 of file PEOutput.h.

References parse_aggregate().

Referenced by adsidac::merge::concat_appends_from_ranks(), and Client::self_merge_outputs().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ finalize_sequential()

size_t adsidac::pe::finalize_sequential	(	const std::string &	pe_dir,
		bool	aggregate
	)

inline

End-of-run finalize: renumber every PE node 0..N-1 in id order, ending in the same layout rewrite_sequential produces (a single PEFolder_0/, no negative-id sentinels, no other buckets).

With aggregate true it is collect() + rewrite_sequential(). Per-file layout fast path: the renumber is FILENAME-ONLY (node bodies keep their rank-offset ids, as they do on the content path), so it is done as a journaled two-pass rename with no content read:

///     pass 1  ZPENode_<old>.txt            -> PEFolder_0/tmp_<new>
///     pass 2  PEFolder_0/tmp_<new>         -> PEFolder_0/ZPENode_<new>.txt
///

The mapping from each current id to its sequential id is written to <pe_dir>/renumber.map before pass 1. An interrupted finalize is completed by calling the function again: a journal that is present is resumed, never recomputed, and both passes skip entries that are already moved. The tmp hop exists because a final name (ZPENode_5.txt) can equal a not-yet-moved current name. Afterwards negative-id sentinel files, emptied bucket directories and the journal are removed. A directory that is already finalized (dense ids, all in PEFolder_0) is left as is apart from that cleanup.

Falls back to collect() + rewrite_sequential (content path, per-file output) when the input holds an aggregate file, duplicate ids, a journal with a bad header or no entries, or when the journal cannot be written.

Parameters

pe_dir	PE directory (trailing '/' required by the content path).
aggregate	Layout to produce: true = aggregate, false = per-file.

Returns: Number of nodes in the finalized layout; 0 if pe_dir does not exist or holds no nodes.

Definition at line 254 of file PEOutput.h.

References aggregate_basename(), collect(), and rewrite_sequential().

Referenced by main(), Server::merge_outputs(), and adsidac::merge::merge_project().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ parse_aggregate()

void adsidac::pe::parse_aggregate	(	const fs::path &	path,
		std::vector< std::pair< int, std::string > > &	out
	)

inline

Parse one aggregate file, appending (id, content) to out in file order, duplicates and negative ids included.

Lines that do not start with "#PE " between records are skipped. Parsing stops, without throwing, at a malformed header or a truncated final record: a run killed mid-write leaves a partial trailing record, and the records before it are still good. A file that cannot be opened adds nothing.

Definition at line 88 of file PEOutput.h.

Referenced by collect(), compact_aggregate(), and adsidac::merge::looks_finalized().

Here is the caller graph for this function:

◆ rewrite_sequential()

size_t adsidac::pe::rewrite_sequential	(	const std::string &	pe_dir,
		std::vector< std::pair< int, std::string > >	entries,
		bool	aggregate
	)

inline

Replace the contents of pe_dir with entries renumbered 0..N-1 in ascending id order (the content path of the renumber).

DELETES pe_dir recursively first, then writes everything into a single <pe_dir>PEFolder_0/: one ZPENode_<n>.txt per node, or one aggregate file whose record headers carry the new ids. Node bodies are written unchanged (they keep the ids they were saved with). Not crash-safe: the input exists only in entries between the delete and the last write.

Parameters

pe_dir	PE directory, with trailing '/'.
entries	(id, content) pairs, e.g. from collect().
aggregate	true = aggregate layout, false = one file per node.

Returns: Number of nodes written (entries.size()).

Definition at line 199 of file PEOutput.h.

References aggregate_basename().

Referenced by finalize_sequential(), and Server::merge_outputs().

Here is the call graph for this function:

Here is the caller graph for this function:

Functions

Function Documentation

◆ aggregate_basename()

◆ append_record()

◆ collect()

◆ compact_aggregate()

◆ finalize_sequential()

◆ parse_aggregate()

◆ rewrite_sequential()