scope_profiler.h5writer

Writing a merged HDF5 profiling file.

The exact inverse of :mod:scope_profiler.h5reader: schemas 2 and 3 store region names once, one compact row per rank/region pair, and all timing events in shared typed columns. Rank groups remain only for auxiliary LIKWID and line-profiler records.

Keeping it here rather than in profile_manager means the layout has exactly two files that must change together, and lets the writer be tested without MPI or a full profiling run.

Attributes

Name	Description
AUTO_COMPRESSION	No description.
AUTO_COMPRESSION_MIN_EVENTS	No description.
COMPACT_MAX_BYTES	No description.
REPACK_MAX_FILE_BYTES	No description.

Classes

Name	Description
ColumnarIndex	The bookkeeping :func:append_columnar_rank needs, held outside the file.
ProfilingWriter	The merged output file, open across the rank-by-rank write.

Functions

Name	Description
append_aggregate_rank	Append one rank of aggregate-only statistics.
append_columnar_rank	Append one rank's region arrays to the schema-2 shared columns.
atomic_publish	Durably replace final_path with a completed sibling file.
compression_filter_available	Whether the active HDF5 library can encode the requested filter.
dataset_storage_options	Build h5py keyword arguments for one one-dimensional event dataset.
decode_start_deltas	Rebuild absolute start timestamps from :func:encode_start_deltas.
encode_durations	Encode one region's end timestamps as durations.
encode_start_deltas	Encode one region's start timestamps as first-absolute-then-gaps.
initialize_columnar_layout	Create the schema-2 region dictionary, pair index, and event columns.
parallel_hdf5_available	Whether this h5py build has the MPI-IO driver enabled.
payload_layout	Return the small, array-free schema needed for collective creation.
publish_file	Rewrite a finished profile with its final storage layout.
rank_group_name	Name of one rank's group. h5reader parses the rank back out of it.
repack_file	Copy an HDF5 file object by object, storing small datasets contiguously.
write_lane_tables	Store one rank's thread and task tables, if the run recorded any.
write_line_profile	Write copied line-profiler records for one rank.
write_metadata	Create the top-level metadata group from a run's metadata dict.
write_parallel_payload	Collectively create the current schema, then write each rank's slices.
write_rank_payload	Write one rank's payload into rank<N>.
write_regions	Write one rank's recorded timestamps under <group>/regions.

AUTO_COMPRESSIONattributemodule attribute#

View source

AUTO_COMPRESSION = ('gzip', 4)

AUTO_COMPRESSION_MIN_EVENTSattributemodule attribute#

View source

AUTO_COMPRESSION_MIN_EVENTS = 1 << 14

COMPACT_MAX_BYTESattributemodule attribute#

View source

COMPACT_MAX_BYTES = 64 * 1024

REPACK_MAX_FILE_BYTESattributemodule attribute#

View source

REPACK_MAX_FILE_BYTES = 16 * 1024 * 1024

ColumnarIndexclass#

View source

class ColumnarIndex(names=(), ranks=())

The bookkeeping :func:append_columnar_rank needs, held outside the file.

Appending a rank has to know which region names already have an id and which ranks have already been written. Both are recoverable from the file, but reading them back costs a scan of columns that grow with every rank, which makes writing a job quadratic in its rank count. A writer therefore keeps one of these for its lifetime, and the MPI relay (ProfileManager._write_payload_direct) passes it along with the write token instead of re-deriving it per rank.

name_to_idattributeinstance attribute#

View source

name_to_id = {name: index for index, name in enumerate(self.names)}

namesattributeinstance attribute#

View source

names = list(names)

ranksattributeinstance attribute#

View source

ranks = {int(rank) for rank in ranks}

from_filemethodclassmethod#

View source

def from_file(h5file) -> ColumnarIndex

Recover the index by reading a partially written file.

registermethod#

View source

def register(names) -> list

Assign ids to any unseen region names; return the new names in order.

statemethod#

View source

def state() -> dict

A picklable snapshot, for handing to the next rank in the relay.

ProfilingWriterclass#

View source

class ProfilingWriter(file_path, metadata: dict | None = None, mode: str = 'w', *, atomic: bool | None = None, index_state: ColumnarIndex | None = None, compression: str | None = None, compression_level: int | None = None, chunk_size: int | None = None, repack: bool = False)

The merged output file, open across the rank-by-rank write.

Used as a context manager by finalize(), which writes rank 0’s own payload and then each incoming one as it arrives::

with ProfilingWriter(path, metadata) as writer:
    writer.write_rank(0, own_payload)
    writer.write_rank(source, received_payload)

index_stateproperty#

View source

index_state: ColumnarIndex

This file’s region-name ids and written ranks, for the next writer.

closemethod#

View source

def close(commit: bool = True) -> None

Close the file and publish it, or discard an unsuccessful write.

open_existingmethodclassmethod#

View source

def open_existing(file_path, *, index_state: ColumnarIndex | None = None, compression: str | None = None, compression_level: int | None = None, chunk_size: int | None = None) -> ProfilingWriter

Open a completed prefix of an MPI output file for one more rank.

write_rankmethod#

View source

def write_rank(rank: int, payload) -> bool

Write one rank’s payload; see :func:write_rank_payload.

append_aggregate_rankfunction#

View source

def append_aggregate_rank(h5file, rank, payload, *, index_state=None) -> bool

Append one rank of aggregate-only statistics.

append_columnar_rankfunction#

View source

def append_columnar_rank(h5file, rank: int, payload, *, index_state: ColumnarIndex | None = None, compression=None, compression_level=None, chunk_size=None) -> bool

Append one rank’s region arrays to the schema-2 shared columns.

Everything this rank contributes is appended one column at a time, not one region at a time: a resize plus a write per region and per column turned a 128-rank, 40-region file into ~41,000 separate dataset operations (3.2s, against 0.2s batched).

Parameters

index_state : ColumnarIndex, optional Region-name ids and already-written ranks, carried by the caller across ranks. Recovered from the file when omitted, which costs a read of two columns that grow with every rank written.

atomic_publishfunction#

View source

def atomic_publish(temporary_path, final_path, *, repack: bool = False, compression=None, compression_level=None, chunk_size=None) -> None

Durably replace final_path with a completed sibling file.

repack runs :func:publish_file first, giving the profile its final storage layout. Pass it only where the file is genuinely finished: the direct MPI writer publishes an intermediate file that the next rank reopens and appends to, and a repacked dataset is contiguous and can no longer grow.

compression_filter_availablefunction#

View source

def compression_filter_available(compression: str | None) -> bool

Whether the active HDF5 library can encode the requested filter.

dataset_storage_optionsfunction#

View source

def dataset_storage_options(length: int, compression: str | None = None, compression_level: int | None = None, chunk_size: int | None = None) -> dict

Build h5py keyword arguments for one one-dimensional event dataset.

compression="auto" contributes no filter here: the automatic policy needs the run’s total size, which is only known once every rank has been written, so it is applied at publication by :func:publish_file.

decode_start_deltasfunction#

View source

def decode_start_deltas(deltas) -> np.ndarray

Rebuild absolute start timestamps from :func:encode_start_deltas.

encode_durationsfunction#

View source

def encode_durations(starts, ends) -> np.ndarray

Encode one region’s end timestamps as durations.

A duration is the same information as an absolute end timestamp given the start, and is small where the timestamp is large, so it compresses in the same way :func:encode_start_deltas describes.

encode_start_deltasfunction#

View source

def encode_start_deltas(starts) -> np.ndarray

Encode one region’s start timestamps as first-absolute-then-gaps.

Schema 3 stores the gap between consecutive calls of a region rather than the absolute timestamp of each. The information is identical - the first element is the run’s absolute start, so :func:decode_start_deltas is an exact cumsum - but the magnitudes collapse: measured gaps between consecutive calls run to a few hundred nanoseconds, about 15 bits, against the ~60 bits an absolute nanosecond timestamp needs. Compressed, that is the difference between 290 KiB and 74 KiB on a 100k-event profile.

The encoding is per run, never across the whole column, so each writer encodes the events it owns without needing any other rank’s data.

initialize_columnar_layoutfunction#

View source

def initialize_columnar_layout(h5file, *, compression=None, compression_level=None, chunk_size=None) -> None

Create the schema-2 region dictionary, pair index, and event columns.

parallel_hdf5_availablefunction#

View source

def parallel_hdf5_available() -> bool

Whether this h5py build has the MPI-IO driver enabled.

payload_layoutfunction#

View source

def payload_layout(payload) -> dict

Return the small, array-free schema needed for collective creation.

publish_filefunction#

View source

def publish_file(path, *, compression=None, compression_level=None, chunk_size=None) -> bool

Rewrite a finished profile with its final storage layout.

Two things are only decidable once every rank has been written, which is why they happen here rather than at dataset creation:

• Chunk overhead. HDF5 gives every resizable dataset chunked storage, and a full chunk plus a chunk-index B-tree is allocated the moment its first element is written. On a profile with a handful of events that overhead is the whole file --- one region and one call produced 193 KiB to hold 0.4 KiB. Freed space is never returned to an HDF5 file, so only a copy into a fresh file recovers it.
• Automatic compression. compression="auto" compresses a run only once it has enough events for the saving to repay the write CPU.

An explicit chunk_size is a request for chunked storage --- it is what makes partial reads possible without compression --- so it is honoured rather than packed away, and only the compression half applies.

A file that needs neither is left exactly as it is. Returns True when the file was rewritten. The rewrite goes through a sibling temporary file and a rename, so a failure leaves the original untouched.

rank_group_namefunction#

View source

def rank_group_name(rank: int) -> str

Name of one rank’s group. h5reader parses the rank back out of it.

repack_filefunction#

View source

def repack_file(source_path, destination_path, *, contiguous_max_bytes=None, compression=None, compression_level=None) -> None

Copy an HDF5 file object by object, storing small datasets contiguously.

HDF5 gives every resizable dataset chunked storage, and the cost of that

• a full chunk plus a chunk-index B-tree - lands the moment the first element is written, whatever the chunk size. On a profile with a handful of events that overhead is the file: one region and one call produced 193 KiB holding 0.4 KiB of data. Because freed space is never returned to the file, rewriting the datasets in place recovers almost none of it; only a copy into a new file does.

Datasets that carry a compression filter keep their chunking, since a contiguous dataset cannot be filtered.

write_lane_tablesfunction#

View source

def write_lane_tables(h5file, rank: int, lanes: dict | None) -> None

Store one rank’s thread and task tables, if the run recorded any.

The tables describe what the per-call thread_ids/task_ids columns index into, so they are written from the same payload, in the same pass, as the events themselves.

write_line_profilefunction#

View source

def write_line_profile(group, records: list | None, *, compression: str | None = None, compression_level: int | None = None, chunk_size: int | None = None) -> None

Write copied line-profiler records for one rank.

write_metadatafunction#

View source

def write_metadata(h5file, metadata: dict) -> None

Create the top-level metadata group from a run’s metadata dict.

Only rank 0’s metadata is stored: it describes the run as a whole. The group holds attributes and no datasets.

Parameters

h5file : h5py.File Destination file, opened for writing. metadata : dict Environment metadata (see :mod:scope_profiler.metadata).

write_parallel_payloadfunction#

View source

def write_parallel_payload(file_path, comm, rank: int, payload, metadata: dict, *, compression: str | None = None, compression_level: int | None = None, chunk_size: int | None = None) -> None

Collectively create the current schema, then write each rank’s slices.

write_rank_payloadfunction#

View source

def write_rank_payload(h5file, rank: int, payload, *, index_state: ColumnarIndex | None = None, compression: str | None = None, compression_level: int | None = None, chunk_size: int | None = None) -> bool

Write one rank’s payload into rank<N>.

A rank that recorded nothing gets no group at all, so the file’s rank groups are exactly the ranks that have something to report.

Parameters

h5file : h5py.File Destination file, opened for writing. rank : int The rank this payload came from. payload : RankPayload The rank’s regions, LIKWID results and LIKWID environment. index_state : ColumnarIndex, optional Region-name ids and written ranks carried across calls; see :func:append_columnar_rank.

Returns

bool True if a group was created, False if the payload was empty.

write_regionsfunction#

View source

def write_regions(group, regions: dict, sources: dict | None = None, tags: dict | None = None, *, compression: str | None = None, compression_level: int | None = None, chunk_size: int | None = None) -> None

Write one rank’s recorded timestamps under <group>/regions.

The datasets are created from exactly-sized arrays and without chunking, so a sparsely-called region costs a few hundred bytes rather than a full chunk.

Parameters

group : h5py.Group The rank’s group. regions : dict Region name -> (start_times, end_times) or (start_times, end_times, gpu_durations) int64 arrays, in nanoseconds. sources : dict, optional Region name -> (source_file, source_lineno, source_text). A name missing here (or the argument itself) simply gets no source attrs, which the reader treats as “not captured”.