Skip to content

qnetbench.harness

One run of one application, on one backend, under one arbitration mode, over one topology. Everything in the suite goes through run_once.

Narrative guide: Running benchmarks and Command line.

qnetbench.harness.runner

Run one application on one backend under one arbitration mode, over a topology.

This is the single entry point the CLI, the tests, and (later) the cross-policy sweep all go through.

run_once 

run_once(
    app_name: str,
    *,
    seed: int = 0,
    backend: str = "reference",
    arbitration: str = "native",
    topology: Topology | None = None,
    cfg: dict[str, object] | None = None,
    pair_age: float = 0.0,
    coherence_time: float = inf,
) -> list[Event]

Execute one run and return its trace as a list of events.

pair_age/coherence_time model staleness (used by characterization); they are only supported on the reference backend.

Source code in qnetbench/harness/runner.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def run_once(
    app_name: str,
    *,
    seed: int = 0,
    backend: str = "reference",
    arbitration: str = "native",
    topology: Topology | None = None,
    cfg: dict[str, object] | None = None,
    pair_age: float = 0.0,
    coherence_time: float = math.inf,
) -> list[Event]:
    """Execute one run and return its trace as a list of events.

    `pair_age`/`coherence_time` model staleness (used by characterization); they
    are only supported on the reference backend."""
    app = get_app(app_name)
    topo = topology or _default_topology(app.roles())
    roles_to_nodes = {role: role for role in app.roles()}
    missing = [n for n in roles_to_nodes.values() if n not in topo.nodes]
    if missing:
        raise ValueError(f"topology {topo.name!r} is missing nodes for roles {missing}")

    arb_label, policy = _parse_arbitration(arbitration)

    if backend == "reference":
        ref = ReferenceBackend(
            topo,
            seed=seed,
            arbitration=arb_label,
            policy=policy,  # type: ignore[arg-type]
            pair_age=pair_age,
            coherence_time=coherence_time,
        )
        return ref.run(app, cfg or {}, roles_to_nodes)

    if (pair_age, coherence_time) != (0.0, math.inf):
        raise NotImplementedError(
            f"pair aging (staleness) is only modelled on the reference backend, not {backend!r}"
        )

    if backend == "sequence":
        try:
            from qnetbench.backends.sequence import SequenceBackend
        except ImportError as exc:  # pragma: no cover - depends on optional extra
            raise RuntimeError(
                "the 'sequence' backend needs the optional SeQUeNCe dependency; "
                "install it with: pip install qnetbench[sequence]"
            ) from exc
        seq = SequenceBackend(topo, seed=seed, arbitration=arb_label, policy=policy)  # type: ignore[arg-type]
        return seq.run(app, cfg or {}, roles_to_nodes)

    if backend == "netsquid":
        try:
            from qnetbench.backends.netsquid import NetSquidBackend
        except ImportError as exc:  # pragma: no cover - depends on optional extra
            raise RuntimeError(
                "the 'netsquid' backend needs the optional NetSquid dependency; "
                "register at netsquid.org and install with: "
                "pip install --extra-index-url https://pypi.netsquid.org qnetbench[netsquid]"
            ) from exc
        nsq = NetSquidBackend(topo, seed=seed, arbitration=arb_label, policy=policy)  # type: ignore[arg-type]
        return nsq.run(app, cfg or {}, roles_to_nodes)

    raise NotImplementedError(
        f"unknown backend {backend!r}; available: 'reference', 'sequence', 'netsquid'."
    )

The command-line entry point

qnetbench.harness.cli

qnetbench command-line entry point.

main 

main(argv: list[str] | None = None) -> int
Source code in qnetbench/harness/cli.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def main(argv: list[str] | None = None) -> int:
    args = _build_parser().parse_args(argv)

    if args.command == "list":
        apps = catalog_apps() if args.all else available_apps()
        label = f"catalog ({len(apps)})" if args.all else "core"
        print(f"apps [{label}]: " + ", ".join(apps))
        print("policies:     " + ", ".join(available_policies()))
        return 0

    if args.command == "characterize":
        return _characterize(args.app, args.seeds, args.out, args.latex)

    if args.command == "spec":
        from qnetbench.spec import SPEC_VERSION, write_specs

        paths = write_specs(args.out)
        print(f"spec v{SPEC_VERSION} written:")
        for path in paths:
            print(f"  {path}")
        return 0

    if args.command == "corpus":
        from qnetbench.spec import generate_reference_corpus

        manifest = generate_reference_corpus(args.out, seed=args.seed)
        print(f"reference corpus v{manifest['spec_version']} written to {args.out}/ "
              f"({len(manifest['traces'])} traces)")
        return 0

    if args.command == "contention":
        from qnetbench.contention import default_experiment, render_experiment

        print(render_experiment(default_experiment()))
        return 0

    try:
        events = run_once(
            args.app, seed=args.seed, backend=args.backend, arbitration=args.arbitration
        )
    except KeyError as exc:
        print(str(exc).strip('"'), file=sys.stderr)
        return 2
    if args.out:
        write_trace(args.out, events)
    report = compute_report(events)
    if args.json:
        print(report.model_dump_json(indent=2))
    else:
        print(render(report))
        if args.out:
            print(f"\ntrace written to {args.out}")
    return 0