scheduling package

Submodules

scheduling.fidelity module

scheduling.fidelity.fidelity_bounds_and_paths(nodes: list[str], fidelities: dict[tuple[str, str], float], K: int = 8) → Tuple[Dict, Dict][source]

Compute E2E fidelity bounds and simple paths between end nodes based on the given edge fidelities. The function performs a DFS from each end node to find all simple paths up to K hops and calculates the corresponding E2E fidelities.

Parameters:

  • nodes (list[str]) – List of nodes for which to compute fidelity bounds

  • paths. (and)

  • fidelities (dict[tuple[str, str], float]) – Mapping of directed edges

  • (u

  • fidelities. (v) to their)

  • K (int, optional) – Maximum number of hops.

Returns:

A tuple containing:

  • bounds: A dictionary mapping pairs of end nodes to their minimum

and maximum fidelity bounds. - paths: A dictionary mapping pairs of end nodes to a list of tuples where each tuple contains the fidelity and the corresponding simple path between the nodes.

Return type:

Tuple[Dict, Dict]

scheduling.fidelity.werner_adj_list(fidelities: Dict[Tuple[str, str], float]) → Dict[str, List[Tuple[str, float]]][source]

Convert directed edge fidelities to an undirected adjacency list with Werner parameters as weights.

Parameters:

  • fidelities (Dict[Tuple[str, str], float]) – Mapping of directed edges

  • (u

  • fidelities. (v) to their)

Returns:

Adjacency list where each key is a node and the value is a list of tuples (neighbor, weight) representing the undirected edges and their corresponding Werner parameters.

Return type:

Dict[str, List[Tuple[str, float]]]

scheduling.main module

QNScheduling

Overview:

This script simulates the scheduling of quantum network applications. It generates a set of applications based on a given network configuration, computes their durations using Packet Generation Attempt (PGA). There are two scheduling strategies: static scheduling (deprecated) using Earliest Deadline First (EDF) that precomputes a schedule, and dynamic scheduling that makes decisions online based on application arrivals. Application releases are drawn from a Poisson process with rate --arrival-rate over the observation horizon. The simulation runs for a fixed time horizon; any pending releases past the horizon are dropped. Metrics collected over the window [warmup, horizon].

scheduling.main.main()[source]
scheduling.main.run_simulation(config: str, arrival_rate: float, inst_range: int, epr_range: tuple[int, int], deadline_range: tuple[float, float], p_packet: float, memory: float, p_swap: float, time_slot_duration: float, seed: int, output_dir: str, instance_arrival_rate: float = 10.0, routing: str = 'shortest', save_csv: bool = True, verbose: bool = True, graph: str | None = None, provisioning: bool = True, full_dynamic: bool = True, static_routing_mode: bool = False, windows: tuple[float, float] | None = None)[source]

Run the quantum network scheduling simulation.

Parameters:

  • config (yaml or gml) – Configuration file path in YAML or GML format.

  • arrival_rate (float) – Mean rate lambda for the Poisson arrival process. The number of applications is drawn from this rate over the observation horizon.

  • inst_range (tuple[int, int]) – Range (min, max) for the number of instances per application.

  • epr_range (tuple[int, int]) – Range (min, max) for the number of EPR pairs to generate per application.

  • deadline_range (tuple[float, float]) – Range (min, max) for the relative deadline budget of each application (deadline = release + deadline_budget).

  • p_packet (float) – Probability of a packet being generated.

  • memory (float) – Memory: number of independent link-generation trials per slot.

  • p_swap (float) – Probability of swapping an EPR pair in a single trial.

  • time_slot_duration (float) – Duration of a time slot in seconds.

  • seed (int) – Random seed for reproducibility of the simulation.

  • output_dir (str) – Directory where the results will be saved.

  • windows (tuple[float, float] | None) – Post-warm-up observation window as (min_time, max_time). In dynamic mode, max_time is used as the simulation horizon.

Returns:

A tuple containing: - bool: whether the schedule is feasible - dict: summary metrics dictionary

Return type:

tuple[bool, dict]

scheduling.pga module

Packet Generation Attempt

This module provides functions to calculate the end-to-end probability of generating EPR pairs in a quantum network, check if the probability of generating a certain number of EPR pairs exceeds a given threshold, and compute the duration of a Packet Generation Attempt (PGA) based on these probabilities.

scheduling.pga.compute_durations(paths: dict[str, list[str] | None], epr_pairs: dict[str, int], p_packet: float, memory: int, p_swap: float, time_slot_duration: float, rates: dict[tuple, float]) → dict[str, float][source]

Compute the duration of each application based on the paths and link parameters.

Parameters:

  • paths (dict[str, list[str]]) – Paths for each application in the

  • network.

  • epr_pairs (dict[str, int]) – Entanglement generation pairs for each

  • application

  • generated. (indicating how many EPR pairs are to be)

  • p_packet (float) – Probability of a packet being generated.

  • memory (int) – Number of independent link-generation trials per slot.

  • p_swap (float) – Probability of swapping an EPR pair in a

  • trial. (single)

  • time_slot_duration (float) – Duration of a time slot in

  • seconds.

  • rates (dict[tuple, float]) – Per-link p_gen, keyed by sorted-tuple

  • its (edges. The effective p_gen for a route is the minimum across)

  • edges.

Returns:

A dictionary mapping each application to its total duration, which includes the time taken for probabilistic generation of EPR pairs and the latency based on the distance of the path.

Return type:

dict[str, float]

scheduling.pga.duration_pga(p_packet: float, epr_pairs: int, n_swap: int, memory: int = 1, p_swap: float = 0.6, p_gen: float = 0.001, time_slot_duration: float = 0.0001) → float[source]

Calculate the duration of a PGA (Packet Generation Attempt).

Parameters:

  • p_packet (float) – Probability of a packet being generated.

  • epr_pairs (int) – Number of successes (number of EPR pairs generated).

  • n_swap (int) – Number of swaps performed.

  • memory (int, optional) – Number of independent link-generation trials

  • slot. (per)

  • p_swap (float, optional) – Probability of swapping an EPR pair in a

  • trial. (single)

  • p_gen (float, optional) – Probability of generating an EPR pair in a

  • trial.

  • time_slot_duration (float, optional) – Duration of a time slot in

  • seconds.

Returns:

Duration of a PGA in seconds.

Return type:

float

scheduling.pga.exceeds_p_packet(n: int, k: int, p_e2e: float, p_packet: float) → bool[source]

Check if the probability of generating at least k EPR pairs in n trials is greater than or equal to p_packet.

Parameters:

  • n (int) – Number of trials.

  • k (int) – Number of successes (number of EPR pairs generated).

  • p_e2e (float) – Probability of generating an EPR pair end-to-end in a

  • trial. (single)

  • p_packet (float) – Probability of a packet being generated.

Returns:

True if the probability of generating at least k EPR pairs in n trials is greater than or equal to p_packet.

Return type:

bool

scheduling.pga.probability_e2e(n_swap: int, memory: int = 1, p_gen: float = 0.001, p_swap: float = 0.6) → float[source]

Calculate the end-to-end probability of generating EPR pairs in a given path.

Parameters:

  • n_swap (int) – Number of swaps performed.

  • memory (int, optional) – Number of independent link-generation trials

  • slot. (per)

  • p_gen (float, optional) – Probability of generating an EPR pair in a

  • trial. (single)

  • p_swap (float, optional) – Probability of swapping an EPR pair in a

  • trial.

Returns:

End-to-end probability of generating EPR pairs.

Return type:

float

scheduling.routing module

scheduling.routing.compute_path_durations(pga_params: Dict[str, float], rates: Dict[Tuple[str, str], float], simple_paths: Dict[Tuple[str, str], List] | None = None, provisioned_paths: List[List[str]] | None = None, src: str | None = None, dst: str | None = None) → List[Tuple][source]
scheduling.routing.dynamic_routing(candidate_paths: List[Tuple[float, Any, List[Tuple[str, str]], float]], min_fidelity: float, deadline: float, cur_t: float = 0.0, resources: Dict[Tuple[str, str], float] = None) → Tuple[Tuple | None, float | None][source]
scheduling.routing.fidelity_shortest(simple_paths: Dict[Tuple[str, str], List[List[str]]], src: str, dst: str, min_fidelity: float, rng: Generator, provisioning: bool = True) → Tuple[List[List[str]], float][source]
scheduling.routing.find_feasible_path(edges: List[Tuple[str, str]], simple_paths: Dict[Tuple[str, str], List[List[str]]], app_requests: Dict[str, Dict[str, Any]], fidelities: Dict[Tuple[str, str], float] | None, pga_rel_times: Dict[str, float] | None = None, routing_mode: str = 'shortest', p_packet: float | None = None, memory: int = 1, p_swap: float = 0.6, rates: Dict[Tuple[str, str], float] | None = None, time_slot_duration: float = 0.0001, rng: Generator | None = None, provisioning: bool = True) → Dict[str, List[List[str]]][source]

Find feasible paths for each application request based on the specified routing and the fidelity threshold.

Parameters:

  • edges (List[Tuple[str, str]]) – List of edges in the quantum network, where each edge is a tuple of nodes (src, dst).

  • app_requests (Dict[str, Dict[str, Any]]) – A dictionary where keys are application names and values are dictionaries containing source and destination nodes (src, dst) and minimum fidelity (min_fidelity) for the application.

  • fidelities (Dict[Tuple[str, str], float]) – Per-edge fidelities in the quantum network, where keys are directed edges (src, dst) and values are the fidelities of those edges.

  • routing_mode (str, optional) – Routing mode, either “shortest”, “random” , “degree”, or “capacity”. In “capacity” mode, edges that have reached the capacity threshold are excluded from path selection. In “random” mode, a random path is selected among all shortest paths that meet the fidelity requirement. In “degree” mode, the path with the lowest maximum degree of internal nodes is selected among all shortest paths that meet the fidelity requirement.

  • p_packet (float, optional) – Probability of a packet being generated.

  • memory (int, optional) – Number of independent link-generation trials per slot.

  • p_swap (float, optional) – Probability of swapping an EPR pair in a single trial.

  • rates (Dict[Tuple[str, str], float], optional) – Per-link p_gen.

  • time_slot_duration (float, optional) – Duration of a time slot in seconds.

  • provisioning (bool, optional) – Whether to enable provisioning for routing.

Returns:

Tuple[

Dict[str, List[List[str]]], Dict[str, float],

]: A tuple of:

  • paths: dict mapping application names to lists of feasible paths fidelity of the selected path, or nan if none was found.

  • e2e_fids: dict mapping application names to the end-to-end fidelity of the selected path, or nan if none was found.

scheduling.routing.highest_fidelity(simple_paths: Dict[Tuple[str, str], List[List[str]]], src: str, dst: str, min_fidelity: float, rng: Generator, provisioning: bool = True) → Tuple[List[List[str]], float][source]

Select the path with the highest E2E fidelity among all paths that meet the minimum fidelity requirement. Ties are broken randomly.

Parameters:

  • simple_paths – Pre-computed (fidelity, path) pairs keyed by node pair.

  • src – Source node.

  • dst – Destination node.

  • min_fidelity – Minimum acceptable E2E fidelity.

  • rng – Random number generator for tie-breaking.

  • provisioning – If True, return selected path plus all tied candidates; otherwise return only the selected path.

Returns:

Tuple of (list of paths with selected first, e2e fidelity or nan).

scheduling.routing.least_capacity(simple_paths: Dict[Tuple[str, str], List[List[str]]], src: str, dst: str, req: Dict[str, Any], cap: Dict[Tuple[str, str], float], p_packet: float | None, memory: int, p_swap: float, rates: Dict[Tuple[str, str], float], time_slot_duration: float, rng: Generator | None = None, provisioning: bool = True) → Tuple[List[List[str]], float, float][source]

Select the path with the least total capacity utilization among all paths that meet the fidelity requirement. The total capacity utilization of a path is defined as the sum of the capacity utilizations of all edges along the path after accounting for the additional load (delta) introduced by the new request.

scheduling.routing.rerouting(precomputed: Dict[str, List[Tuple]], deadline: float, cur_t: float, app: str, resources: Dict[Tuple[str, str], float] | None = None) → Tuple[List[str], List[Tuple[str, str]], float, float, float] | None[source]
scheduling.routing.shortest_paths(edges: List[Tuple[str, str]], app_requests: Dict[str, Dict[str, Any]]) → Dict[str, List[str]][source]

Find shortest paths for each applications in a quantum network graph represented by edges.

Parameters:

  • edges (List[Tuple[str, str]]) – List of edges in the quantum network, where each edge is a tuple of nodes (src, dst).

  • app_requests (Dict[str, Dict[str, Any]]) –

    A dictionary where keys are application names and values are dictionaries containing source and destination nodes (src, dst) for the application. For example:

    {

    ‘A’: {‘src’: ‘Alice’, ‘dst’: ‘Bob’}, ‘B’: {‘src’: ‘Alice’, ‘dst’: ‘Charlie’}, ‘C’: {‘src’: ‘Charlie’, ‘dst’: ‘David’}, ‘D’: {‘src’: ‘Bob’, ‘dst’: ‘David’}

    }

Returns:

A dictionary where keys are application names and values are lists of nodes representing the shortest path from source to destination for that application.

Return type:

Dict[str, List[str]]

scheduling.routing.smallest_bottleneck(simple_paths: Dict[Tuple[str, str], List[List[str]]], src: str, dst: str, req: Dict[str, Any], cap: Dict[Tuple[str, str], float], p_packet: float | None, memory: int, p_swap: float, rates: Dict[Tuple[str, str], float], time_slot_duration: float, k: int | None = None) → Tuple[List[List[str]], float, float][source]
scheduling.routing.static_routing(app_requests: Dict[str, Dict[str, Any]], simple_paths: Dict[Tuple[str, str], List[List[str]]]) → Tuple[Dict[str, List[List[str]]], Dict[str, float]][source]

scheduling.simulation module

Simulation Of PGAs Scheduling

This module provides classes and functions to simulate the scheduling of Packet Generation Attempts (PGAs) in a quantum network. Each PGA tries to generate entangled EPR pairs over a specified route within a defined time window, considering resource availability and link busy times. The function, simulate_static, simulates a static schedule of PGAs and returns performance metrics, link utilizations, and other relevant data. While the function simulate_dynamic implement a dynamic scheduling approach.

class scheduling.simulation.PGA(name: str, arrival: float, start: float, end: float, route: List[str], resources: Dict[Tuple[str, str], float], link_busy: Dict[Tuple[str, str], float], link_p_gens: List[float] | ndarray, epr_pairs: int, slot_duration: float, rng: Generator, log: List[Dict[str, Any]], policy: str, p_swap: float, memory: int, deadline: float | None = None, route_links: List[Tuple[str, str]] | None = None)[source]

Bases: object

run() → Dict[str, Any][source]
scheduling.simulation.simulate_dynamic(app_specs: Dict[str, Dict[str, Any]], durations: Dict[str, float], pga_parameters: Dict[str, Dict[str, float]], pga_rel_times: Dict[str, float], pga_network_paths: Dict[str, List[List[str]]], rng: Generator, full_dynamic: bool = True, rerouting_mode: bool = False, all_links: List[Tuple[str, str]] | None = None, simple_paths: Dict[str, List[List[str]]] | None = None, static_routing_mode: bool = False, horizon_time: float | None = None, warmup_time: float = 0.0, rng_routing: Generator | None = None, rng_arrivals: Dict[str, Generator] | None = None, instance_arrival_rate: float = 10.0, rates: Dict[Tuple[str, str], float] = None)[source]

scheduling.static module

Static Scheduling

This module implements a static scheduling algorithm for Packet Generation Attempts (PGAs) in a quantum network. It uses an Earliest Deadline First (EDF) approach to schedule PGAs while considering their parallelization capabilities. The schedule is constructed over a specified number of hyperperiod cycles that ensure PGAs are executed within their deadlines if feasible.

scheduling.static.edf_parallel_static(pga_rel_times: Dict[str, float], pga_periods: Dict[str, float], durations: Dict[str, float], parallel_apps: Dict[str, Set[str]], horizon_cycles: int) → Tuple[bool, List[Tuple[str, float, float, float]] | str][source]

EDF scheduling with parallelization capabilities. The static schedule is constructed over a given number of hyperperiod cycles. It checks if the set of PGAs is feasible and returns the schedule if so.

Parameters:

  • pga_rel_times (Dict[str, float]) – PGA relative start times.

  • pga_periods (Dict[str, float]) – PGA periods.

  • durations (Dict[str, float]) – PGA execution durations.

  • parallel_apps (Dict[str, Set[str]]) – Dictionary mapping each PGA to a

  • it. (set of PGAs that can run in parallel with)

  • horizon_cycles (int) – Number of hyperperiod cycles to consider.

Returns:

A tuple where the first element indicates if the schedule is feasible, and the second element is either the schedule (pga_name, start, end, deadline) or an error message (if not).

Return type:

Tuple[bool, List[Tuple[str, float, float, float]] | str]

scheduling.static.hyperperiod(periods: dict[str, float]) → float[source]

Compute the hyperperiod of a set of periods.

Parameters:

  • periods (dict[str, float]) – A dictionary mapping PGA names to their

  • periods.

Returns:

The hyperperiod of the given periods.

Return type:

float

Module contents