Multi-tenant contention: the cross-policy ranking-inversion result (Deliverable 4).
The earlier phases run one application at a time, so the arbiter never has to
choose — there is only ever one pending demand. Scheduling only matters under
contention: several tenants competing for a link whose entanglement-generation
capacity is below aggregate demand. This module runs that experiment.
A shared link produces one pair per service tick (capacity pairs/second). Tenants
— parameterized by real applications' demand contracts — issue requests over time.
At each tick the chosen Policy orders the pending queue and one request is served.
A served request's delivered fidelity decays with how long it waited in queue (the
memory holding its pair decoheres), so a high-min_fidelity demand must be served
promptly, while a deadline demand must be served before its deadline. Those two
pressures pull different ways, so the policy that wins depends on the workload mix —
and the ranking inverts between a deadline-heavy and a fidelity-heavy mix. That
inversion is the proof that single-workload evaluation produces unreliable rankings.
module-attribute
module-attribute
module-attribute
module-attribute
dataclass
Tenant(
app: str,
min_fidelity: float,
budget: float | None,
n_requests: int,
interval: float,
pattern: tuple[float, ...] = (),
phase: float = 0.0,
)
A demand stream: an application's contract, issued n_requests times at a
fixed interval.
instance-attribute
instance-attribute
instance-attribute
instance-attribute
class-attribute
instance-attribute
pattern: tuple[float, ...] = ()
class-attribute
instance-attribute
dataclass
ContentionResult(
policy: str,
aggregate_utility: float = 0.0,
violation_rate: float = 0.0,
per_app: dict[str, float] = dict(),
)
instance-attribute
class-attribute
instance-attribute
aggregate_utility: float = 0.0
class-attribute
instance-attribute
violation_rate: float = 0.0
class-attribute
instance-attribute
per_app: dict[str, float] = field(default_factory=dict)
app_profile(
app: str,
n_requests: int,
interval: float,
*,
trace_seed: int = 0,
phase: float = 0.0,
) -> Tenant
Build a tenant from an application's real demand contract and real arrival
pattern, both read from an actual run.
trace_seed selects which run the arrival pattern is read from, and phase
offsets the tenant's first request. Two tenants of the same application built
with the same seed and phase are indistinguishable — same contract, same
arrival instants — so distinct seeds are what make a mix genuinely
multi-tenant rather than one stream counted several times.
Source code in qnetbench/contention.py
| def app_profile(
app: str,
n_requests: int,
interval: float,
*,
trace_seed: int = 0,
phase: float = 0.0,
) -> Tenant:
"""Build a tenant from an application's real demand contract and real arrival
pattern, both read from an actual run.
`trace_seed` selects which run the arrival pattern is read from, and `phase`
offsets the tenant's first request. Two tenants of the same application built
with the same seed and phase are indistinguishable — same contract, same
arrival instants — so distinct seeds are what make a mix genuinely
multi-tenant rather than one stream counted several times.
"""
events = run_once(app, seed=trace_seed)
pattern = _arrival_pattern(events)
for ev in events:
if isinstance(ev, EntanglementRequested):
d = ev.demand
if d.latency_budget is not None:
budget: float | None = d.latency_budget
elif d.deadline is not None:
budget = d.deadline - ev.t
else:
budget = None
return Tenant(
app, d.min_fidelity, budget, n_requests, interval, pattern, phase
)
raise ValueError(f"application {app!r} issued no entanglement requests")
|
Discrete-event contention simulation over one shared link.
Source code in qnetbench/contention.py
| def simulate(
tenants: list[Tenant],
policy: Policy,
*,
capacity: float,
link_fidelity: float,
coherence_time: float,
) -> ContentionResult:
"""Discrete-event contention simulation over one shared link."""
requests: list[_Req] = []
for tenant in tenants:
arrival = tenant.phase
for k in range(tenant.n_requests):
if k:
# Replay the application's own gap sequence, scaled to `interval`.
gap = tenant.pattern[(k - 1) % len(tenant.pattern)] if tenant.pattern else 1.0
arrival += gap * tenant.interval
deadline = arrival + tenant.budget if tenant.budget is not None else math.inf
requests.append(_Req(tenant.app, arrival, tenant.min_fidelity, deadline))
requests.sort(key=lambda r: r.arrival)
dt = 1.0 / capacity
horizon = max((r.arrival for r in requests), default=0.0) + dt
n_ticks = int(horizon / dt) + len(requests) + 1 # enough ticks to drain the queue
pending: list[_Req] = []
idx = 0
for tick in range(n_ticks):
t = tick * dt
while idx < len(requests) and requests[idx].arrival <= t:
pending.append(requests[idx])
idx += 1
pending = [r for r in pending if r.deadline >= t] # expired = violation (met stays False)
if not pending:
if idx >= len(requests):
break
continue
# Ask the policy to rank the queue; each request's req_id is its index in
# `pending`, so order[0] is the position of the request to serve this tick.
order = policy.order([_to_pending(i, r, t) for i, r in enumerate(pending)], now=t)
chosen = pending[order[0]]
# Its delivered fidelity degrades with how long it waited (memory decoheres);
# the contract is met only if fidelity clears the bar and service beat the deadline.
wait = t - chosen.arrival
delivered = _age_fidelity(link_fidelity, wait, coherence_time)
chosen.met = delivered >= chosen.min_fidelity and t <= chosen.deadline
pending.remove(chosen)
met = sum(1 for r in requests if r.met)
total = len(requests)
result = ContentionResult(policy=policy.name)
result.aggregate_utility = met / total if total else 0.0
result.violation_rate = 1.0 - result.aggregate_utility
for app in {r.app for r in requests}:
app_reqs = [r for r in requests if r.app == app]
result.per_app[app] = sum(1 for r in app_reqs if r.met) / len(app_reqs)
return result
|
ranking_experiment(
mixes: dict[str, list[Tenant]],
*,
policies: tuple[str, ...] = (
"fifo",
"fidelity_first",
"edf",
),
capacity: float,
link_fidelity: float,
coherence_time: float,
) -> dict[str, dict[str, ContentionResult]]
Run every mix under every policy. Returns results[mix][policy].
Source code in qnetbench/contention.py
| def ranking_experiment(
mixes: dict[str, list[Tenant]],
*,
policies: tuple[str, ...] = ("fifo", "fidelity_first", "edf"),
capacity: float,
link_fidelity: float,
coherence_time: float,
) -> dict[str, dict[str, ContentionResult]]:
"""Run every mix under every policy. Returns `results[mix][policy]`."""
out: dict[str, dict[str, ContentionResult]] = {}
for mix_name, tenants in mixes.items():
out[mix_name] = {
name: simulate(
tenants,
get_policy(name),
capacity=capacity,
link_fidelity=link_fidelity,
coherence_time=coherence_time,
)
for name in policies
}
return out
|
The policy with the highest aggregate utility for one mix.
Resolves ties by dict order, so prefer winning_policies wherever a tie
would be reported as a result: several of this experiment's operating points
have two or three policies on exactly the same score, and an argmax silently
turns that into a winner.
Source code in qnetbench/contention.py
| def best_policy(results: dict[str, ContentionResult]) -> str:
"""The policy with the highest aggregate utility for one mix.
Resolves ties by dict order, so prefer `winning_policies` wherever a tie
would be reported as a result: several of this experiment's operating points
have two or three policies on exactly the same score, and an argmax silently
turns that into a winner.
"""
return max(results, key=lambda p: results[p].aggregate_utility)
|
Every policy tied for the highest aggregate utility.
Source code in qnetbench/contention.py
| def winning_policies(results: dict[str, ContentionResult]) -> frozenset[str]:
"""Every policy tied for the highest aggregate utility."""
top = max(r.aggregate_utility for r in results.values())
return frozenset(p for p, r in results.items() if r.aggregate_utility == top)
|
True if the winning policy differs across mixes (a ranking inversion).
Source code in qnetbench/contention.py
| def has_inversion(experiment: dict[str, dict[str, ContentionResult]]) -> bool:
"""True if the winning policy differs across mixes (a ranking inversion)."""
winners = {best_policy(res) for res in experiment.values()}
return len(winners) > 1
|
default_mixes(
n_requests: int = 12,
interval: float = 0.03,
*,
seed: int | None = None,
) -> dict[str, list[Tenant]]
Two workload classes: one dominated by deadline-critical demand (distributed
gates), one by fidelity-thresholded demand (BQC + CHSH + QKD).
seed=None keeps the synchronised model: every tenant of an application
replays the same measured arrival sequence starting at t=0, so four
distributed_gate tenants are one stream counted four times. An integer seed
gives each tenant its own arrival realisation and start phase, which is what
"five tenants" ought to mean, and varying it turns a single deterministic
realisation into a distribution over arrival patterns.
Source code in qnetbench/contention.py
| def default_mixes(
n_requests: int = 12, interval: float = 0.03, *, seed: int | None = None
) -> dict[str, list[Tenant]]:
"""Two workload classes: one dominated by deadline-critical demand (distributed
gates), one by fidelity-thresholded demand (BQC + CHSH + QKD).
`seed=None` keeps the synchronised model: every tenant of an application
replays the same measured arrival sequence starting at t=0, so four
`distributed_gate` tenants are one stream counted four times. An integer seed
gives each tenant its own arrival realisation and start phase, which is what
"five tenants" ought to mean, and varying it turns a single deterministic
realisation into a distribution over arrival patterns.
"""
rng = random.Random(seed) if seed is not None else None
def group(app: str, count: int) -> list[Tenant]:
out: list[Tenant] = []
for _ in range(count):
if rng is None:
out.append(app_profile(app, n_requests, interval))
else:
out.append(
app_profile(
app,
n_requests,
interval,
trace_seed=rng.randrange(1 << 30),
phase=rng.uniform(0.0, interval),
)
)
return out
return {
"deadline_heavy": group("distributed_gate", 4) + group("qkd", 1),
"fidelity_heavy": group("bqc", 2) + group("chsh", 2) + group("qkd", 1),
}
|
burstiness_mixes(
app: str = "heralded_teleport",
count: int = 5,
n_requests: int = 12,
interval: float = 0.03,
*,
seed: int | None = None,
) -> dict[str, list[Tenant]]
Two mixes that differ only in burstiness.
Both use the same application, so contracts, tenant count and mean arrival rate
are identical; the only difference is whether each tenant replays its measured
inter-arrival pattern or issues on a perfectly regular cadence. This isolates
the burstiness axis of Section VI, which the fixed-cadence arrival model could
not express: under it every tenant was smooth regardless of the workload.
Source code in qnetbench/contention.py
| def burstiness_mixes(
app: str = "heralded_teleport",
count: int = 5,
n_requests: int = 12,
interval: float = 0.03,
*,
seed: int | None = None,
) -> dict[str, list[Tenant]]:
"""Two mixes that differ *only* in burstiness.
Both use the same application, so contracts, tenant count and mean arrival rate
are identical; the only difference is whether each tenant replays its measured
inter-arrival pattern or issues on a perfectly regular cadence. This isolates
the burstiness axis of Section VI, which the fixed-cadence arrival model could
not express: under it every tenant was smooth regardless of the workload.
"""
rng = random.Random(seed) if seed is not None else None
bursty: list[Tenant] = []
smooth: list[Tenant] = []
for _ in range(count):
if rng is None:
t = app_profile(app, n_requests, interval)
else:
t = app_profile(
app,
n_requests,
interval,
trace_seed=rng.randrange(1 << 30),
phase=rng.uniform(0.0, interval),
)
bursty.append(t)
# Same tenant, same phase, arrivals evenly spaced instead of replayed:
# the comparison isolates arrival *shape* and nothing else.
smooth.append(replace(t, pattern=()))
return {"bursty": bursty, "smooth": smooth}
|
The headline cross-policy evaluation (Deliverable 4).
Source code in qnetbench/contention.py
| def default_experiment() -> dict[str, dict[str, ContentionResult]]:
"""The headline cross-policy evaluation (Deliverable 4)."""
return ranking_experiment(
default_mixes(),
capacity=CAPACITY,
link_fidelity=LINK_FIDELITY,
coherence_time=COHERENCE_TIME,
)
|
A policy × workload-mix utility matrix, marking each mix's winner (*).
Source code in qnetbench/contention.py
| def render_experiment(experiment: dict[str, dict[str, ContentionResult]]) -> str:
"""A policy × workload-mix utility matrix, marking each mix's winner (*)."""
mixes = list(experiment)
policies = list(next(iter(experiment.values())))
width = max(len(m) for m in mixes) + 4
lines = [f"{'policy':16}" + "".join(f"{m:>{width}}" for m in mixes)]
lines.append("-" * len(lines[0]))
winners = {m: best_policy(experiment[m]) for m in mixes}
for policy in policies:
cells = []
for m in mixes:
util = experiment[m][policy].aggregate_utility
mark = "*" if winners[m] == policy else " "
cells.append(f"{util:>{width - 2}.3f}{mark} ")
lines.append(f"{policy:16}" + "".join(cells))
lines.append(f"{'winner':16}" + "".join(f"{winners[m]:>{width}}" for m in mixes))
if has_inversion(experiment):
verdict = " → ".join(f"{m}: {winners[m]}" for m in mixes)
lines.append("")
lines.append(f"RANKING INVERSION — the best policy flips across workloads ({verdict}).")
lines.append(
"Single-workload evaluation would have crowned one policy and been wrong on the other."
)
return "\n".join(lines)
|