k4bench.analysis.dataflow

k4bench.analysis.dataflow

What each algorithm of a k4run job read and wrote: {label}_dataflow.json.

The k4run benchmark builds the file after each job (:func:build_dataflow) from what the job recorded, and :func:load_dataflow reads it back as a :class:DataFlow, whose :meth:DataFlow.graph joins the algorithms by the collections one writes and a later one reads.

Where each algorithm's collections come from

• A Gaudi algorithm, k4FWCore's included: the keys of the data handles it declared, which the k4BenchAuditor records (inputs and outputs in _components.json). These are exact. The k4FWCore Reader declares one output per collection of its input file.
• A Marlin processor run by a MarlinProcessorWrapper: its LCIO collection parameters. Marlin marks a parameter registered as an input or output collection with its LCIO type, and Marlin -x lists those for every processor type; the collection names are the parameter's value in the job options, or its default where the job leaves it unset. The wrapper's EDM4hep-to-LCIO and LCIO-to-EDM4hep converters come from the job options.
• The k4FWCore Writer, which reads whatever its keep/drop rules (IOSvc.outputCommands) select from the event store.

Limitations

• A Marlin processor that takes a collection name as a plain string parameter, rather than registering it as a collection, declares nothing: what it reads and writes is missing, and a collection only it writes has no producer.
• A declared collection need not be one the algorithm uses: a Marlin input whose default the job never set is listed all the same.
• The converters are modelled, not observed: one converting everything (convertAll) is taken to convert every collection the other store holds at that point, under its collNameMapping name, unless the target store already holds one of that name.
• Gaudi handles carry a key, not a type; only LCIO collections have a type.
• Algorithms are in the order the auditor first saw them, which is the order they were initialized and, in a sequential job, the order they run.
• A run whose auditor predates recording data handles has none: its Gaudi algorithms' reads and writes are unknown (None).

Collection dataclass

Collection(name: str, store: str, type: str | None = None, parameter: str | None = None, explicit: bool | None = None)

One collection an algorithm reads or writes.

Converter dataclass

Converter(direction: str, convert_all: bool, mapping: Mapping[str, str])

A k4MarlinWrapper converter run before (EDM4hep to LCIO) or after (LCIO to EDM4hep) its processor.

Algorithm dataclass

Algorithm(name: str, type: str, impl: str, framework: str, inputs: tuple[Collection, ...] | None, outputs: tuple[Collection, ...] | None, converters: tuple[Converter, ...] = (), output_commands: tuple[str, ...] | None = None)

One algorithm, with what it read and wrote; None where unknown.

Link dataclass

Link(name: str, store: str, type: str | None, source: str, source_store: str)

One collection passed from a producer to a consumer.

Graph dataclass

Graph(edges: Mapping[tuple[str, str], tuple[Link, ...]], unresolved: tuple[tuple[str, Collection], ...])

Algorithms joined by the collections one writes and another reads.

DataFlow dataclass

DataFlow(algorithms: tuple[Algorithm, ...], handles_recorded: bool, marlin_recorded: bool)

One label's _dataflow.json.

graph

graph() -> Graph

Replay the job in order: each read is linked to the latest earlier write of that collection in that store, a converter carrying collections from one store to the other, and the Writer reading what its rules keep. A read with no earlier write falls back to any algorithm that writes that collection directly.

Source code in k4bench/analysis/dataflow.py

def graph(self) -> Graph:
    """Replay the job in order: each read is linked to the latest earlier
    write of that collection in that store, a converter carrying
    collections from one store to the other, and the Writer reading what
    its rules keep. A read with no earlier write falls back to any
    algorithm that writes that collection directly."""
    # name -> (producer, name it was written under, store it was written to)
    stores: dict[str, dict[str, tuple[str, str, str]]] = {GAUDI: {}, LCIO: {}}
    writers = {
        (out.store, out.name): algorithm.name
        for algorithm in self.algorithms for out in algorithm.outputs or ()
    }
    edges: dict[tuple[str, str], list[Link]] = {}
    unresolved: list[tuple[str, Collection]] = []
    for algorithm in self.algorithms:
        for converter in algorithm.converters:
            if converter.direction == "edm4hep_to_lcio":
                _convert(stores[GAUDI], stores[LCIO], converter)
        if algorithm.output_commands is not None:
            for name, origin in stores[GAUDI].items():
                if _kept(name, algorithm.output_commands):
                    _link(edges, origin, algorithm.name, Collection(name, GAUDI))
        for read in algorithm.inputs or ():
            origin = stores[read.store].get(read.name)
            if origin is None and (writer := writers.get((read.store, read.name))):
                origin = (writer, read.name, read.store)
            if origin is None or origin[0] == algorithm.name:
                unresolved.append((algorithm.name, read))
            else:
                _link(edges, origin, algorithm.name, read)
        for out in algorithm.outputs or ():
            stores[out.store][out.name] = (algorithm.name, out.name, out.store)
        for converter in algorithm.converters:
            if converter.direction == "lcio_to_edm4hep":
                _convert(stores[LCIO], stores[GAUDI], converter)
    return Graph({pair: tuple(links) for pair, links in edges.items()}, tuple(unresolved))

parse_joboptions

parse_joboptions(text: str) -> dict[str, dict[str, object]]

{owner: {property: value}} of a JobOptionsSvc dump. Values that are Python literals are parsed; others, such as tool references, kept as text.

Source code in k4bench/analysis/dataflow.py

def parse_joboptions(text: str) -> dict[str, dict[str, object]]:
    """``{owner: {property: value}}`` of a JobOptionsSvc dump. Values that are
    Python literals are parsed; others, such as tool references, kept as text."""
    options: dict[str, dict[str, object]] = {}
    for line in text.splitlines():
        if not (match := _OPTION_RE.match(line.strip())):
            continue
        raw = match["value"].strip()
        try:
            value = ast.literal_eval(raw.replace("set()", "[]"))
        except (ValueError, SyntaxError):
            value = raw
        options.setdefault(match["owner"], {})[match["prop"]] = value
    return options

marlin_collection_parameters

marlin_collection_parameters(xml: str) -> dict[str, dict[str, dict]]

{processor type: {parameter: {direction, type, default}}} for the LCIO collection parameters Marlin -x lists in xml.

Source code in k4bench/analysis/dataflow.py

def marlin_collection_parameters(xml: str) -> dict[str, dict[str, dict]]:
    """``{processor type: {parameter: {direction, type, default}}}`` for the
    LCIO collection parameters ``Marlin -x`` lists in *xml*."""
    root = ET.fromstring(xml[xml.index("<marlin"):])
    out: dict[str, dict[str, dict]] = {}
    for processor in root.iter("processor"):
        parameters = {}
        for parameter in processor.iter("parameter"):
            if lcio := parameter.get("lcioInType"):
                direction = "input"
            elif lcio := parameter.get("lcioOutType"):
                direction = "output"
            else:
                continue
            parameters[parameter.get("name")] = {
                "direction": direction, "type": lcio,
                "default": (parameter.text or "").split(),
            }
        out.setdefault(processor.get("type"), parameters)
    return out

build_dataflow

build_dataflow(components: Mapping, joboptions: str, marlin_xml: str | None, *, reader_collections: Iterable[str] | None = None) -> dict

The _dataflow.json document of one job.

components is the job's parsed _components.json, joboptions its resolved job options and marlin_xml Marlin -x of its stack (None when the job ran no Marlin processor, and an unreadable one counts as none). reader_collections stands in for the Reader's outputs where the auditor recorded no data handles.

Source code in k4bench/analysis/dataflow.py

def build_dataflow(
    components: Mapping,
    joboptions: str,
    marlin_xml: str | None,
    *,
    reader_collections: Iterable[str] | None = None,
) -> dict:
    """The ``_dataflow.json`` document of one job.

    *components* is the job's parsed ``_components.json``, *joboptions* its
    resolved job options and *marlin_xml* ``Marlin -x`` of its stack (``None``
    when the job ran no Marlin processor, and an unreadable one counts as
    none). *reader_collections* stands in for
    the Reader's outputs where the auditor recorded no data handles.
    """
    options = parse_joboptions(joboptions)
    try:
        marlin = marlin_collection_parameters(marlin_xml) if marlin_xml else None
    except (ValueError, ET.ParseError):
        # An unreadable dump costs the Marlin side only.
        marlin = None
    entries = components["components"]
    parents = {entry.get("parent") for entry in entries if entry.get("parent") is not None}
    algorithms = [
        entry for index, entry in enumerate(entries)
        if entry.get("category") == "algorithm" and index not in parents
    ]
    handles = any("inputs" in entry and entry["inputs"] is not None for entry in algorithms)
    out = []
    for entry in algorithms:
        record = {"name": entry["name"], "type": entry["type"], "impl": entry.get("impl") or entry["type"]}
        if entry["type"] == "MarlinProcessorWrapper":
            record.update(_marlin_algorithm(entry["name"], record["impl"], options, marlin))
        else:
            record["framework"] = "gaudi"
            for direction in ("inputs", "outputs"):
                keys = entry.get(direction)
                record[direction] = None if keys is None else [
                    {"name": _gaudi_key(key), "store": GAUDI} for key in keys
                ]
            if entry["type"] == "Reader" and record["outputs"] is None and reader_collections is not None:
                record["inputs"] = []
                record["outputs"] = [{"name": n, "store": GAUDI} for n in reader_collections]
            if entry["type"] == "Writer":
                commands = options.get("IOSvc", {}).get("outputCommands")
                record["output_commands"] = list(commands) if commands else list(_DEFAULT_OUTPUT_COMMANDS)
        out.append(record)
    return {
        "schema_version": DATAFLOW_SCHEMA_VERSION,
        "handles_recorded": handles,
        "marlin_recorded": marlin is not None,
        "algorithms": out,
    }

load_dataflow

load_dataflow(run_dir: str | Path) -> dict[str, DataFlow]

Every label's data flow in run_dir; empty when it has none.

Source code in k4bench/analysis/dataflow.py

def load_dataflow(run_dir: str | Path) -> dict[str, DataFlow]:
    """Every label's data flow in *run_dir*; empty when it has none."""
    flows = {}
    for path, label in labelled_files(Path(run_dir), DATAFLOW_SUFFIX):
        flows[label] = DataFlow.from_json(json.loads(path.read_text()), source=path)
    return flows

read_marlin_processors

read_marlin_processors(run_dir: str | Path) -> str | None

The run's Marlin -x dump, or None when it has none.

Source code in k4bench/analysis/dataflow.py

def read_marlin_processors(run_dir: str | Path) -> str | None:
    """The run's ``Marlin -x`` dump, or ``None`` when it has none."""
    path = Path(run_dir) / MARLIN_PROCESSORS
    return gzip.decompress(path.read_bytes()).decode() if path.exists() else None