26622922786

Committed 29 May 2026 06:54AM UTC coverage: 76.297% (+0.8%) from 75.53%

Build # 26622922786

Build Type

push

github

Committed by

Quang00

Commit Message

fix tests

Coverage Stats

853 of 1118 relevant lines covered (76.3%)

2.29 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

79.59

/scheduling/simulation.py

"""
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.
"""

import heapq
import re
import time
from typing import Any, Dict, List, Tuple

import numpy as np
import pandas as pd

from scheduling.routing import (
    compute_path_durations,
    dynamic_routing,
    rerouting,
)
from utils.helper import compute_link_utilization, track_link_waiting

INIT_PGA_RE = re.compile(r"^([A-Za-z]+)(\d+)$")
EPS = 1e-12


class PGA:
    def __init__(
        self,
        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] | np.ndarray,
        epr_pairs: int,
        slot_duration: float,
        rng: np.random.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,
    ) -> None:
        """Packet Generation Attempt (PGA) simulation. A PGA tries to
        generate EPR pairs over a specified route within a defined time window,
        considering resource availability and link busy times.

        The possible outcomes for a PGA:
        - If the PGA starts but cannot generate the required E2E EPR pairs
          within its time window, it is marked as "failed".
        - If the PGA successfully generates the required E2E EPR pairs within
          its time window, it is marked as "completed".

        The conditions for EPR generation:
        - Each link attempts to generate EPR pairs in discrete time slots.
        - The number of trials needed for a successful EPR pair generation on
          each link follows a geometric distribution with success probability
          `p_gen`.
        - The first success across all links must occur within the memory
          of the first generated pair to be considered valid.
        - If there are swaps involved, each swap must also succeed based on
          the swap probability `p_swap` for the end-to-end entanglement to be
          successful.

        Args:
            name (str): PGA identifier.
            arrival (float): Arrival time of the PGA in the simulation.
            start (float): Start time of the PGA in the simulation.
            end (float): End time of the PGA in the simulation.
            route (List[str]): List of nodes in the PGA's route.
            resources (Dict[Tuple[str, str], float]): Dictionary tracking
            when undirected links become free.
            link_busy (Dict[Tuple[str, str], float]): Dictionary to track busy
            time of links.
            link_p_gens (List[float] | np.ndarray): Per-link probabilities of
            generating an EPR pair in a single trial, in route_links order.
            epr_pairs (int): Number of EPR pairs to generate for this PGA.
            slot_duration (float): Duration of a time slot for EPR generation.
            rng (np.random.Generator): Random number generator for
            probabilistic events.
            log (List[Dict[str, Any]]): Log to record PGA performance metrics.
            policy (str): Scheduling policy for the PGA, either "best_effort"
            or "deadline". If "deadline", the PGA will attempt to complete
            within the maximum burst time defined in durations.
            p_swap (float): Probability of swapping an EPR pair.
            memory (int): Memory: number of independent link-generation trials
            per slot.
            deadline (float, optional): Deadline time for the PGA. Defaults to
            None, which means no deadline.
        """
        self.name = name
        self.arrival = float(arrival)
        self.start = float(start)
        self.end = float(end)
        self.route = route
        self.resources = resources
        self.link_busy = link_busy
        self.epr_pairs = int(epr_pairs)
        self.slot_duration = float(slot_duration)
        self.rng = rng
        self.log = log
        self.policy = policy
        self.deadline = None if deadline is None else float(deadline)
        self.links = route_links
        self.n_swap = max(0, len(self.route) - 2)
        self.p_swap = float(p_swap)
        self.memory = max(0, int(memory))
        self.link_p_gens = np.asarray(link_p_gens, dtype=float)

    def _simulate_e2e_attempts(self, max_attempts: int) -> np.ndarray:
        """Single end-to-end entanglement for a batch of attempts."""
        if self.memory <= 0 or max_attempts <= 0:
            return np.zeros(max_attempts, dtype=bool)

        n_links = self.n_swap + 1
        t_mem = self.memory
        size = (max_attempts, n_links)

        if np.any(self.link_p_gens <= 0.0):
            return np.zeros(max_attempts, dtype=bool)
        starts = self.rng.geometric(self.link_p_gens, size=size) - 1
        ends = starts + (t_mem - 1)

        candidate = starts.max(axis=1)
        last_valid = ends.min(axis=1)

        succ = (candidate < t_mem) & (last_valid >= candidate)

        if self.n_swap > 0:
            if self.p_swap < 1.0:
                p_bsms = self.p_swap**self.n_swap
                swap_ok = self.rng.random(max_attempts) < p_bsms
                succ &= swap_ok

        return succ

    def _update_resources_and_links(
        self,
        completion: float,
        attempts_run: int,
    ) -> None:
        """Mark nodes and links busy through ``completion``."""
        for link in self.links:
            self.resources[link] = max(
                self.resources.get(link, 0.0), completion
            )

        if attempts_run > 0 and self.links:
            busy = attempts_run * self.slot_duration
            for link in self.links:
                self.link_busy[link] = self.link_busy.get(link, 0.0) + busy

    def run(self) -> Dict[str, Any]:
        attempts_run = 0
        pairs_generated = 0
        wait_until = 0.0
        blocking_links = []

        for link in self.links:
            lk_b_t = self.resources.get(link, 0.0)
            if lk_b_t > wait_until:
                wait_until = lk_b_t

        for link in self.links:
            lk_b_t = self.resources.get(link, 0.0)
            if abs(lk_b_t - wait_until) < EPS:
                blocking_links.append(link)

        current_time = self.start
        diff = self.end - self.start
        t_budget = diff if diff > 0.0 else 0.0
        status = "failed"

        if t_budget > EPS and self.policy == "deadline":
            max_attempts = int((t_budget + EPS) // self.slot_duration)
            succ = self._simulate_e2e_attempts(max_attempts)
            csum = (
                np.cumsum(succ, dtype=int)
                if len(succ)
                else np.array([], dtype=int)
            )
            hit = (
                np.searchsorted(csum, self.epr_pairs, side="left")
                if len(csum)
                else len(csum)
            )

            if len(csum) and hit < len(csum):
                attempts_run = int(hit + 1)
                pairs_generated = int(csum[attempts_run - 1])
                status = "completed"
            else:
                attempts_run = max_attempts
                pairs_generated = int(csum[-1]) if len(csum) else 0

            current_time = self.start + attempts_run * self.slot_duration
            completion = (
                current_time if current_time < self.end else self.end
            )
            self._update_resources_and_links(completion, attempts_run)
        else:
            completion = self.start

        burst = completion - self.start
        turnaround = max(0.0, completion - self.arrival)
        waiting = max(0.0, turnaround - burst)

        result = {
            "pga": self.name,
            "arrival_time": self.arrival,
            "start_time": self.start,
            "burst_time": burst,
            "completion_time": completion,
            "turnaround_time": turnaround,
            "waiting_time": waiting,
            "pairs_generated": pairs_generated,
            "status": status,
            "deadline": self.deadline if self.policy == "deadline" else None,
            "blocking_links": blocking_links,
        }
        self.log.append(result)
        return result


def 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: np.random.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: np.random.Generator | None = None,
    rng_arrivals: Dict[str, np.random.Generator] | None = None,
    instance_arrival_rate: float = 10.0,
    rates: Dict[Tuple[str, str], float] = None,
):
    log = []
    defer_counts = {}
    pga_release_times = {}
    pga_names = []
    seen_pgas = set()

    pga_route_links = {
        app: [
            tuple(sorted((u, v)))
            for u, v in zip(paths[0][:-1], paths[0][1:], strict=False)
        ]
        for app, paths in pga_network_paths.items()
    }
    resources = {link: 0.0 for link in all_links}
    link_busy = dict.fromkeys(all_links, 0.0)
    link_busy_record = dict.fromkeys(all_links, 0.0)
    link_waiting_routing = (
        {
            link: {"total_waiting_time": 0.0, "pga_waited": 0}
            for link in all_links
        }
        if full_dynamic and simple_paths is not None
        else None
    )
    link_waiting = {
        link: {"total_waiting_time": 0.0, "pga_waited": 0}
        for link in all_links
    }
    min_arrival = float("inf")
    max_completion = 0.0
    routing_decision_cpt = 0
    routing_decision_runtime = 0.0

    deadline_budgets = {
        app: app_specs[app].get("deadline_budget") for app in app_specs
    }
    base_release = {app: pga_rel_times.get(app, 0.0) for app in app_specs}
    max_instances = {
        app: max(0, int(app_specs[app].get("instances", 0)))
        for app in app_specs
    }
    release_indices = {app: 0 for app in app_specs}
    mean_instance_interarrival = 1.0 / instance_arrival_rate
    poisson_next_release = {
        app: float(base_release.get(app, 0.0)) for app in app_specs
    }

    events_queue = []
    ready_queue = []

    routing_metadata = {}
    pga_best = {}
    rerouting_candidates = {}
    if full_dynamic and simple_paths is not None:
        for app in app_specs:
            _t0 = time.perf_counter()
            routing_metadata[app] = compute_path_durations(
                pga_parameters[app],
                simple_paths=simple_paths,
                src=app_specs[app]["src"],
                dst=app_specs[app]["dst"],
                rates=rates,
            )
            routing_decision_runtime += time.perf_counter() - _t0
            min_fid = app_specs[app].get("min_fidelity", 0.0)
            feasible_durs = [
                dur
                for fid, _, _, dur in routing_metadata[app]
                if fid >= min_fid
            ]
            pga_best[app] = (
                min(feasible_durs) if feasible_durs else float("nan")
            )
    elif rerouting_mode:
        for app, app_paths in pga_network_paths.items():
            _t0 = time.perf_counter()
            rerouting_candidates[app] = compute_path_durations(
                pga_parameters[app],
                provisioned_paths=app_paths,
                rates=rates,
            )
            routing_decision_runtime += time.perf_counter() - _t0

    def enqueue_release(app: str) -> None:
        idx = release_indices[app]
        deadline_budget = deadline_budgets[app]

        if max_instances[app] > 0 and idx >= max_instances[app]:
            return

        release = poisson_next_release[app]
        _rng_arr = (
            rng_arrivals.get(app) if rng_arrivals is not None else None
        ) or rng
        poisson_next_release[app] = release + _rng_arr.exponential(
            mean_instance_interarrival
        )

        if release >= horizon_time:
            return

        deadline = release + deadline_budget
        heapq.heappush(
            events_queue,
            (release, deadline, release, app, idx, release, "release"),
        )
        release_indices[app] += 1

    for app in app_specs:
        enqueue_release(app)

    cur_t = 0.0

    while events_queue or ready_queue:
        if not ready_queue:
            cur_t = events_queue[0][0]

        if cur_t >= horizon_time:
            break

        while events_queue and events_queue[0][0] <= cur_t + EPS:
            (
                event_time,
                deadline,
                arrival_time,
                app,
                i,
                ready_time,
                event_type,
            ) = heapq.heappop(events_queue)
            if event_type == "release":
                enqueue_release(app)

            if event_time >= horizon_time:
                continue

            heapq.heappush(
                ready_queue,
                (deadline, ready_time, arrival_time, app, i, event_time),
            )

        if not ready_queue:
            continue

        while ready_queue:
            deadline, rdy_t, arrival_time, app, i, _ = heapq.heappop(
                ready_queue
            )
            at = float(arrival_time)
            if at < min_arrival:
                min_arrival = at

            pga_name = f"{app}{i}"
            if pga_name not in seen_pgas:
                seen_pgas.add(pga_name)
                pga_names.append(pga_name)
                pga_release_times[pga_name] = arrival_time

            routed_fid = np.nan
            if static_routing_mode:
                route_links = pga_route_links.get(app, [])
                selected_path = pga_network_paths[app][0]
                duration = durations.get(app, 0.0)
                last_available = max(
                    (resources.get(lk, 0.0) for lk in route_links),
                    default=0.0,
                )
            elif full_dynamic and simple_paths is not None:
                routing_decision_cpt += 1
                _t0 = time.perf_counter()
                routed, next_avail = dynamic_routing(
                    routing_metadata[app],
                    app_specs[app]["min_fidelity"],
                    deadline,
                    cur_t,
                    resources,
                    rng_routing if rng_routing is not None else rng,
                )
                routing_decision_runtime += time.perf_counter() - _t0
                if routed is None:
                    if next_avail is not None:
                        defer_counts[pga_name] = (
                            defer_counts.get(pga_name, 0) + 1
                        )
                        heapq.heappush(
                            events_queue,
                            (
                                next_avail,
                                deadline,
                                arrival_time,
                                app,
                                i,
                                rdy_t,
                                "resume"
                            ),
                        )
                    else:
                        log.append({
                            "pga": pga_name,
                            "arrival_time": arrival_time,
                            "ready_time": rdy_t,
                            "start_time": np.nan,
                            "burst_time": 0.0,
                            "completion_time": cur_t,
                            "turnaround_time": max(0.0, cur_t - arrival_time),
                            "waiting_time": max(0.0, cur_t - rdy_t),
                            "pairs_generated": 0,
                            "status": "drop",
                            "deadline": deadline,
                        })
                    continue
                (
                    selected_path,
                    route_links,
                    last_available,
                    duration,
                    routed_fid,
                ) = routed
            else:
                route_links = pga_route_links.get(app, [])
                selected_path = pga_network_paths[app][0]
                duration = durations.get(app, 0.0)

                last_available = 0.0
                for link in route_links:
                    last_available = max(
                        last_available, resources.get(link, 0.0)
                    )

                would_drop = last_available + duration > deadline + EPS
                if (
                    last_available > cur_t + EPS
                    and rerouting_mode
                    and would_drop
                ):
                    routing_decision_cpt += 1
                    _t0 = time.perf_counter()
                    alt_path = rerouting(
                        rerouting_candidates,
                        deadline,
                        cur_t,
                        app,
                        resources,
                    )
                    routing_decision_runtime += time.perf_counter() - _t0
                    if alt_path is not None:
                        (
                            selected_path,
                            route_links,
                            last_available,
                            duration,
                            routed_fid,
                        ) = alt_path

            _stamp = (
                {
                    "e2e_fidelity": routed_fid,
                    "pga_duration": duration,
                    "hops": len(selected_path) - 1,
                }
                if (full_dynamic and simple_paths is not None)
                or rerouting_mode
                else {}
            )
            if full_dynamic and simple_paths is not None:
                best = pga_best.get(app, float("nan"))
                _stamp["routing_efficiency"] = (
                    best / duration
                    if duration > 0 and not np.isnan(best)
                    else float("nan")
                )

            if last_available > cur_t + EPS:
                if last_available + duration <= deadline + EPS:
                    defer_counts[pga_name] = (
                        defer_counts.get(pga_name, 0) + 1
                    )
                    heapq.heappush(
                        events_queue,
                        (
                            last_available,
                            deadline,
                            arrival_time,
                            app,
                            i,
                            rdy_t,
                            "resume",
                        ),
                    )
                else:
                    log.append({
                        "pga": pga_name,
                        "arrival_time": arrival_time,
                        "ready_time": rdy_t,
                        "start_time": np.nan,
                        "burst_time": 0.0,
                        "completion_time": cur_t,
                        "turnaround_time": max(0.0, cur_t - arrival_time),
                        "waiting_time": max(0.0, cur_t - rdy_t),
                        "pairs_generated": 0,
                        "status": "drop",
                        "deadline": deadline,
                        **_stamp,
                    })
                continue

            start_time = cur_t
            completion = start_time + duration

            if (
                app_specs[app].get("policy") == "deadline"
                and deadline is not None
                and start_time + duration > deadline + EPS
            ):
                log.append({
                    "pga": pga_name,
                    "arrival_time": arrival_time,
                    "ready_time": rdy_t,
                    "start_time": np.nan,
                    "burst_time": 0.0,
                    "completion_time": cur_t,
                    "turnaround_time": max(0.0, cur_t - arrival_time),
                    "waiting_time": max(0.0, cur_t - rdy_t),
                    "pairs_generated": 0,
                    "status": "drop",
                    "deadline": deadline,
                    **_stamp,
                })
                continue

            recording = start_time >= warmup_time
            link_p_gens = [rates[lk] for lk in route_links]
            pga = PGA(
                name=pga_name,
                arrival=arrival_time,
                start=start_time,
                end=completion,
                route=selected_path,
                resources=resources,
                link_busy=link_busy_record if recording else link_busy,
                link_p_gens=link_p_gens,
                epr_pairs=int(pga_parameters[app]["epr_pairs"]),
                slot_duration=pga_parameters[app]["slot_duration"],
                rng=rng,
                log=log,
                policy=app_specs[app].get("policy"),
                p_swap=pga_parameters[app]["p_swap"],
                memory=pga_parameters[app]["memory"],
                deadline=deadline,
                route_links=route_links,
            )
            result = pga.run()
            result["ready_time"] = float(rdy_t)
            result["waiting_time"] = max(0.0, start_time - rdy_t)
            result.update(_stamp)

            if link_waiting_routing is not None:
                track_link_waiting(
                    result.get("waiting_time", 0.0),
                    link_waiting_routing,
                    blocking_links=result.get("blocking_links"),
                )

            if recording:
                track_link_waiting(
                    result.get("waiting_time", 0.0),
                    link_waiting,
                    blocking_links=result.get("blocking_links"),
                )
                if result["completion_time"] > horizon_time:
                    excess = result["completion_time"] - horizon_time
                    for lk in route_links:
                        if lk in link_busy_record:
                            link_busy_record[lk] = max(
                                0.0, link_busy_record[lk] - excess
                            )

            max_completion = max(max_completion, result["completion_time"])

            status = result.get("status", "")
            if status == "completed":
                continue

    df = pd.DataFrame(log)
    del log
    link_utilization = compute_link_utilization(
        link_busy_record, warmup_time, horizon_time,
    )

    return (
        df,
        pga_names,
        pga_release_times,
        link_utilization,
        link_waiting,
        routing_decision_cpt,
        routing_decision_runtime,
        defer_counts,
    )

1	"""
2	Simulation Of PGAs Scheduling
3	-----------------------------
4	This module provides classes and functions to simulate the scheduling of
5	Packet Generation Attempts (PGAs) in a quantum network. Each PGA tries to
6	generate entangled EPR pairs over a specified route within a defined time
7	window, considering resource availability and link busy times. The function,
8	`simulate_static`, simulates a static schedule of PGAs and returns performance
9	metrics, link utilizations, and other relevant data. While the function
10	`simulate_dynamic` implement a dynamic scheduling approach.
11	"""
12
13	import heapq	3✔
14	import re	3✔
15	import time	3✔
16	from typing import Any, Dict, List, Tuple	3✔
17
18	import numpy as np	3✔
19	import pandas as pd	3✔
20
21	from scheduling.routing import (	3✔
22	compute_path_durations,
23	dynamic_routing,
24	rerouting,
25	)
26	from utils.helper import compute_link_utilization, track_link_waiting	3✔
27
28	INIT_PGA_RE = re.compile(r"^([A-Za-z]+)(\d+)$")	3✔
29	EPS = 1e-12	3✔
30
31
32	class PGA:	3✔
33	def __init__(	3✔
34	self,
35	name: str,
36	arrival: float,
37	start: float,
38	end: float,
39	route: List[str],
40	resources: Dict[Tuple[str, str], float],
41	link_busy: Dict[Tuple[str, str], float],
42	link_p_gens: List[float] \| np.ndarray,
43	epr_pairs: int,
44	slot_duration: float,
45	rng: np.random.Generator,
46	log: List[Dict[str, Any]],
47	policy: str,
48	p_swap: float,
49	memory: int,
50	deadline: float \| None = None,
51	route_links: List[Tuple[str, str]] \| None = None,
52	) -> None:
53	"""Packet Generation Attempt (PGA) simulation. A PGA tries to
54	generate EPR pairs over a specified route within a defined time window,
55	considering resource availability and link busy times.
56
57	The possible outcomes for a PGA:
58	- If the PGA starts but cannot generate the required E2E EPR pairs
59	within its time window, it is marked as "failed".
60	- If the PGA successfully generates the required E2E EPR pairs within
61	its time window, it is marked as "completed".
62
63	The conditions for EPR generation:
64	- Each link attempts to generate EPR pairs in discrete time slots.
65	- The number of trials needed for a successful EPR pair generation on
66	each link follows a geometric distribution with success probability
67	`p_gen`.
68	- The first success across all links must occur within the memory
69	of the first generated pair to be considered valid.
70	- If there are swaps involved, each swap must also succeed based on
71	the swap probability `p_swap` for the end-to-end entanglement to be
72	successful.
73
74	Args:
75	name (str): PGA identifier.
76	arrival (float): Arrival time of the PGA in the simulation.
77	start (float): Start time of the PGA in the simulation.
78	end (float): End time of the PGA in the simulation.
79	route (List[str]): List of nodes in the PGA's route.
80	resources (Dict[Tuple[str, str], float]): Dictionary tracking
81	when undirected links become free.
82	link_busy (Dict[Tuple[str, str], float]): Dictionary to track busy
83	time of links.
84	link_p_gens (List[float] \| np.ndarray): Per-link probabilities of
85	generating an EPR pair in a single trial, in route_links order.
86	epr_pairs (int): Number of EPR pairs to generate for this PGA.
87	slot_duration (float): Duration of a time slot for EPR generation.
88	rng (np.random.Generator): Random number generator for
89	probabilistic events.
90	log (List[Dict[str, Any]]): Log to record PGA performance metrics.
91	policy (str): Scheduling policy for the PGA, either "best_effort"
92	or "deadline". If "deadline", the PGA will attempt to complete
93	within the maximum burst time defined in durations.
94	p_swap (float): Probability of swapping an EPR pair.
95	memory (int): Memory: number of independent link-generation trials
96	per slot.
97	deadline (float, optional): Deadline time for the PGA. Defaults to
98	None, which means no deadline.
99	"""
100	self.name = name	3✔
101	self.arrival = float(arrival)	3✔
102	self.start = float(start)	3✔
103	self.end = float(end)	3✔
104	self.route = route	3✔
105	self.resources = resources	3✔
106	self.link_busy = link_busy	3✔
107	self.epr_pairs = int(epr_pairs)	3✔
108	self.slot_duration = float(slot_duration)	3✔
109	self.rng = rng	3✔
110	self.log = log	3✔
111	self.policy = policy	3✔
112	self.deadline = None if deadline is None else float(deadline)	3✔
113	self.links = route_links	3✔
114	self.n_swap = max(0, len(self.route) - 2)	3✔
115	self.p_swap = float(p_swap)	3✔
116	self.memory = max(0, int(memory))	3✔
117	self.link_p_gens = np.asarray(link_p_gens, dtype=float)	3✔
118
119	def _simulate_e2e_attempts(self, max_attempts: int) -> np.ndarray:	3✔
120	"""Single end-to-end entanglement for a batch of attempts."""
121	if self.memory <= 0 or max_attempts <= 0:	3✔
122	return np.zeros(max_attempts, dtype=bool)	×
123
124	n_links = self.n_swap + 1	3✔
125	t_mem = self.memory	3✔
126	size = (max_attempts, n_links)	3✔
127
128	if np.any(self.link_p_gens <= 0.0):	3✔
129	return np.zeros(max_attempts, dtype=bool)	×
130	starts = self.rng.geometric(self.link_p_gens, size=size) - 1	3✔
131	ends = starts + (t_mem - 1)	3✔
132
133	candidate = starts.max(axis=1)	3✔
134	last_valid = ends.min(axis=1)	3✔
135
136	succ = (candidate < t_mem) & (last_valid >= candidate)	3✔
137
138	if self.n_swap > 0:	3✔
139	if self.p_swap < 1.0:	×
140	p_bsms = self.p_swap**self.n_swap	×
141	swap_ok = self.rng.random(max_attempts) < p_bsms	×
142	succ &= swap_ok	×
143
144	return succ	3✔
145
146	def _update_resources_and_links(	3✔
147	self,
148	completion: float,
149	attempts_run: int,
150	) -> None:
151	"""Mark nodes and links busy through ``completion``."""
152	for link in self.links:	3✔
153	self.resources[link] = max(	3✔
154	self.resources.get(link, 0.0), completion
155	)
156
157	if attempts_run > 0 and self.links:	3✔
158	busy = attempts_run * self.slot_duration	3✔
159	for link in self.links:	3✔
160	self.link_busy[link] = self.link_busy.get(link, 0.0) + busy	3✔
161
162	def run(self) -> Dict[str, Any]:	3✔
163	attempts_run = 0	3✔
164	pairs_generated = 0	3✔
165	wait_until = 0.0	3✔
166	blocking_links = []	3✔
167
168	for link in self.links:	3✔
169	lk_b_t = self.resources.get(link, 0.0)	3✔
170	if lk_b_t > wait_until:	3✔
171	wait_until = lk_b_t	3✔
172
173	for link in self.links:	3✔
174	lk_b_t = self.resources.get(link, 0.0)	3✔
175	if abs(lk_b_t - wait_until) < EPS:	3✔
176	blocking_links.append(link)	3✔
177
178	current_time = self.start	3✔
179	diff = self.end - self.start	3✔
180	t_budget = diff if diff > 0.0 else 0.0	3✔
181	status = "failed"	3✔
182
183	if t_budget > EPS and self.policy == "deadline":	3✔
184	max_attempts = int((t_budget + EPS) // self.slot_duration)	3✔
185	succ = self._simulate_e2e_attempts(max_attempts)	3✔
186	csum = (	3✔
187	np.cumsum(succ, dtype=int)
188	if len(succ)
189	else np.array([], dtype=int)
190	)
191	hit = (	3✔
192	np.searchsorted(csum, self.epr_pairs, side="left")
193	if len(csum)
194	else len(csum)
195	)
196
197	if len(csum) and hit < len(csum):	3✔
198	attempts_run = int(hit + 1)	3✔
199	pairs_generated = int(csum[attempts_run - 1])	3✔
200	status = "completed"	3✔
201	else:
202	attempts_run = max_attempts	3✔
203	pairs_generated = int(csum[-1]) if len(csum) else 0	3✔
204
205	current_time = self.start + attempts_run * self.slot_duration	3✔
206	completion = (	3✔
207	current_time if current_time < self.end else self.end
208	)
209	self._update_resources_and_links(completion, attempts_run)	3✔
210	else:
211	completion = self.start	×
212
213	burst = completion - self.start	3✔
214	turnaround = max(0.0, completion - self.arrival)	3✔
215	waiting = max(0.0, turnaround - burst)	3✔
216
217	result = {	3✔
218	"pga": self.name,
219	"arrival_time": self.arrival,
220	"start_time": self.start,
221	"burst_time": burst,
222	"completion_time": completion,
223	"turnaround_time": turnaround,
224	"waiting_time": waiting,
225	"pairs_generated": pairs_generated,
226	"status": status,
227	"deadline": self.deadline if self.policy == "deadline" else None,
228	"blocking_links": blocking_links,
229	}
230	self.log.append(result)	3✔
231	return result	3✔
232
233
234	def simulate_dynamic(	3✔
235	app_specs: Dict[str, Dict[str, Any]],
236	durations: Dict[str, float],
237	pga_parameters: Dict[str, Dict[str, float]],
238	pga_rel_times: Dict[str, float],
239	pga_network_paths: Dict[str, List[List[str]]],
240	rng: np.random.Generator,
241	full_dynamic: bool = True,
242	rerouting_mode: bool = False,
243	all_links: List[Tuple[str, str]] \| None = None,
244	simple_paths: Dict[str, List[List[str]]] \| None = None,
245	static_routing_mode: bool = False,
246	horizon_time: float \| None = None,
247	warmup_time: float = 0.0,
248	rng_routing: np.random.Generator \| None = None,
249	rng_arrivals: Dict[str, np.random.Generator] \| None = None,
250	instance_arrival_rate: float = 10.0,
251	rates: Dict[Tuple[str, str], float] = None,
252	):
253	log = []	3✔
254	defer_counts = {}	3✔
255	pga_release_times = {}	3✔
256	pga_names = []	3✔
257	seen_pgas = set()	3✔
258
259	pga_route_links = {	3✔
260	app: [
261	tuple(sorted((u, v)))
262	for u, v in zip(paths[0][:-1], paths[0][1:], strict=False)
263	]
264	for app, paths in pga_network_paths.items()
265	}
266	resources = {link: 0.0 for link in all_links}	3✔
267	link_busy = dict.fromkeys(all_links, 0.0)	3✔
268	link_busy_record = dict.fromkeys(all_links, 0.0)	3✔
269	link_waiting_routing = (	3✔
270	{
271	link: {"total_waiting_time": 0.0, "pga_waited": 0}
272	for link in all_links
273	}
274	if full_dynamic and simple_paths is not None
275	else None
276	)
277	link_waiting = {	3✔
278	link: {"total_waiting_time": 0.0, "pga_waited": 0}
279	for link in all_links
280	}
281	min_arrival = float("inf")	3✔
282	max_completion = 0.0	3✔
283	routing_decision_cpt = 0	3✔
284	routing_decision_runtime = 0.0	3✔
285
286	deadline_budgets = {	3✔
287	app: app_specs[app].get("deadline_budget") for app in app_specs
288	}
289	base_release = {app: pga_rel_times.get(app, 0.0) for app in app_specs}	3✔
290	max_instances = {	3✔
291	app: max(0, int(app_specs[app].get("instances", 0)))
292	for app in app_specs
293	}
294	release_indices = {app: 0 for app in app_specs}	3✔
295	mean_instance_interarrival = 1.0 / instance_arrival_rate	3✔
296	poisson_next_release = {	3✔
297	app: float(base_release.get(app, 0.0)) for app in app_specs
298	}
299
300	events_queue = []	3✔
301	ready_queue = []	3✔
302
303	routing_metadata = {}	3✔
304	pga_best = {}	3✔
305	rerouting_candidates = {}	3✔
306	if full_dynamic and simple_paths is not None:	3✔
307	for app in app_specs:	×
308	_t0 = time.perf_counter()	×
309	routing_metadata[app] = compute_path_durations(	×
310	pga_parameters[app],
311	simple_paths=simple_paths,
312	src=app_specs[app]["src"],
313	dst=app_specs[app]["dst"],
314	rates=rates,
315	)
316	routing_decision_runtime += time.perf_counter() - _t0	×
317	min_fid = app_specs[app].get("min_fidelity", 0.0)	×
318	feasible_durs = [	×
319	dur
320	for fid, _, _, dur in routing_metadata[app]
321	if fid >= min_fid
322	]
323	pga_best[app] = (	×
324	min(feasible_durs) if feasible_durs else float("nan")
325	)
326	elif rerouting_mode:	3✔
327	for app, app_paths in pga_network_paths.items():	×
328	_t0 = time.perf_counter()	×
329	rerouting_candidates[app] = compute_path_durations(	×
330	pga_parameters[app],
331	provisioned_paths=app_paths,
332	rates=rates,
333	)
334	routing_decision_runtime += time.perf_counter() - _t0	×
335
336	def enqueue_release(app: str) -> None:	3✔
337	idx = release_indices[app]	3✔
338	deadline_budget = deadline_budgets[app]	3✔
339
340	if max_instances[app] > 0 and idx >= max_instances[app]:	3✔
341	return	3✔
342
343	release = poisson_next_release[app]	3✔
344	_rng_arr = (	3✔
345	rng_arrivals.get(app) if rng_arrivals is not None else None
346	) or rng
347	poisson_next_release[app] = release + _rng_arr.exponential(	3✔
348	mean_instance_interarrival
349	)
350
351	if release >= horizon_time:	3✔
352	return	×
353
354	deadline = release + deadline_budget	3✔
355	heapq.heappush(	3✔
356	events_queue,
357	(release, deadline, release, app, idx, release, "release"),
358	)
359	release_indices[app] += 1	3✔
360
361	for app in app_specs:	3✔
362	enqueue_release(app)	3✔
363
364	cur_t = 0.0	3✔
365
366	while events_queue or ready_queue:	3✔
367	if not ready_queue:	3✔
368	cur_t = events_queue[0][0]	3✔
369
370	if cur_t >= horizon_time:	3✔
371	break	×
372
373	while events_queue and events_queue[0][0] <= cur_t + EPS:	3✔
374	(	3✔
375	event_time,
376	deadline,
377	arrival_time,
378	app,
379	i,
380	ready_time,
381	event_type,
382	) = heapq.heappop(events_queue)
383	if event_type == "release":	3✔
384	enqueue_release(app)	3✔
385
386	if event_time >= horizon_time:	3✔
387	continue	×
388
389	heapq.heappush(	3✔
390	ready_queue,
391	(deadline, ready_time, arrival_time, app, i, event_time),
392	)
393
394	if not ready_queue:	3✔
395	continue	×
396
397	while ready_queue:	3✔
398	deadline, rdy_t, arrival_time, app, i, _ = heapq.heappop(	3✔
399	ready_queue
400	)
401	at = float(arrival_time)	3✔
402	if at < min_arrival:	3✔
403	min_arrival = at	3✔
404
405	pga_name = f"{app}{i}"	3✔
406	if pga_name not in seen_pgas:	3✔
407	seen_pgas.add(pga_name)	3✔
408	pga_names.append(pga_name)	3✔
409	pga_release_times[pga_name] = arrival_time	3✔
410
411	routed_fid = np.nan	3✔
412	if static_routing_mode:	3✔
413	route_links = pga_route_links.get(app, [])	×
414	selected_path = pga_network_paths[app][0]	×
415	duration = durations.get(app, 0.0)	×
416	last_available = max(	×
417	(resources.get(lk, 0.0) for lk in route_links),
418	default=0.0,
419	)
420	elif full_dynamic and simple_paths is not None:	3✔
421	routing_decision_cpt += 1	×
422	_t0 = time.perf_counter()	×
423	routed, next_avail = dynamic_routing(	×
424	routing_metadata[app],
425	app_specs[app]["min_fidelity"],
426	deadline,
427	cur_t,
428	resources,
429	rng_routing if rng_routing is not None else rng,
430	)
431	routing_decision_runtime += time.perf_counter() - _t0	×
432	if routed is None:	×
433	if next_avail is not None:	×
434	defer_counts[pga_name] = (	×
435	defer_counts.get(pga_name, 0) + 1
436	)
437	heapq.heappush(	×
438	events_queue,
439	(
440	next_avail,
441	deadline,
442	arrival_time,
443	app,
444	i,
445	rdy_t,
446	"resume"
447	),
448	)
449	else:
450	log.append({	×
451	"pga": pga_name,
452	"arrival_time": arrival_time,
453	"ready_time": rdy_t,
454	"start_time": np.nan,
455	"burst_time": 0.0,
456	"completion_time": cur_t,
457	"turnaround_time": max(0.0, cur_t - arrival_time),
458	"waiting_time": max(0.0, cur_t - rdy_t),
459	"pairs_generated": 0,
460	"status": "drop",
461	"deadline": deadline,
462	})
463	continue	×
464	(	×
465	selected_path,
466	route_links,
467	last_available,
468	duration,
469	routed_fid,
470	) = routed
471	else:
472	route_links = pga_route_links.get(app, [])	3✔
473	selected_path = pga_network_paths[app][0]	3✔
474	duration = durations.get(app, 0.0)	3✔
475
476	last_available = 0.0	3✔
477	for link in route_links:	3✔
478	last_available = max(	3✔
479	last_available, resources.get(link, 0.0)
480	)
481
482	would_drop = last_available + duration > deadline + EPS	3✔
483	if (	3✔
484	last_available > cur_t + EPS
485	and rerouting_mode
486	and would_drop
487	):
488	routing_decision_cpt += 1	×
489	_t0 = time.perf_counter()	×
490	alt_path = rerouting(	×
491	rerouting_candidates,
492	deadline,
493	cur_t,
494	app,
495	resources,
496	)
497	routing_decision_runtime += time.perf_counter() - _t0	×
498	if alt_path is not None:	×
499	(	×
500	selected_path,
501	route_links,
502	last_available,
503	duration,
504	routed_fid,
505	) = alt_path
506
507	_stamp = (	3✔
508	{
509	"e2e_fidelity": routed_fid,
510	"pga_duration": duration,
511	"hops": len(selected_path) - 1,
512	}
513	if (full_dynamic and simple_paths is not None)
514	or rerouting_mode
515	else {}
516	)
517	if full_dynamic and simple_paths is not None:	3✔
518	best = pga_best.get(app, float("nan"))	×
519	_stamp["routing_efficiency"] = (	×
520	best / duration
521	if duration > 0 and not np.isnan(best)
522	else float("nan")
523	)
524
525	if last_available > cur_t + EPS:	3✔
526	if last_available + duration <= deadline + EPS:	3✔
527	defer_counts[pga_name] = (	3✔
528	defer_counts.get(pga_name, 0) + 1
529	)
530	heapq.heappush(	3✔
531	events_queue,
532	(
533	last_available,
534	deadline,
535	arrival_time,
536	app,
537	i,
538	rdy_t,
539	"resume",
540	),
541	)
542	else:
543	log.append({	3✔
544	"pga": pga_name,
545	"arrival_time": arrival_time,
546	"ready_time": rdy_t,
547	"start_time": np.nan,
548	"burst_time": 0.0,
549	"completion_time": cur_t,
550	"turnaround_time": max(0.0, cur_t - arrival_time),
551	"waiting_time": max(0.0, cur_t - rdy_t),
552	"pairs_generated": 0,
553	"status": "drop",
554	"deadline": deadline,
555	**_stamp,
556	})
557	continue	3✔
558
559	start_time = cur_t	3✔
560	completion = start_time + duration	3✔
561
562	if (	3✔
563	app_specs[app].get("policy") == "deadline"
564	and deadline is not None
565	and start_time + duration > deadline + EPS
566	):
567	log.append({	3✔
568	"pga": pga_name,
569	"arrival_time": arrival_time,
570	"ready_time": rdy_t,
571	"start_time": np.nan,
572	"burst_time": 0.0,
573	"completion_time": cur_t,
574	"turnaround_time": max(0.0, cur_t - arrival_time),
575	"waiting_time": max(0.0, cur_t - rdy_t),
576	"pairs_generated": 0,
577	"status": "drop",
578	"deadline": deadline,
579	**_stamp,
580	})
581	continue	3✔
582
583	recording = start_time >= warmup_time	3✔
584	link_p_gens = [rates[lk] for lk in route_links]	3✔
585	pga = PGA(	3✔
586	name=pga_name,
587	arrival=arrival_time,
588	start=start_time,
589	end=completion,
590	route=selected_path,
591	resources=resources,
592	link_busy=link_busy_record if recording else link_busy,
593	link_p_gens=link_p_gens,
594	epr_pairs=int(pga_parameters[app]["epr_pairs"]),
595	slot_duration=pga_parameters[app]["slot_duration"],
596	rng=rng,
597	log=log,
598	policy=app_specs[app].get("policy"),
599	p_swap=pga_parameters[app]["p_swap"],
600	memory=pga_parameters[app]["memory"],
601	deadline=deadline,
602	route_links=route_links,
603	)
604	result = pga.run()	3✔
605	result["ready_time"] = float(rdy_t)	3✔
606	result["waiting_time"] = max(0.0, start_time - rdy_t)	3✔
607	result.update(_stamp)	3✔
608
609	if link_waiting_routing is not None:	3✔
610	track_link_waiting(	×
611	result.get("waiting_time", 0.0),
612	link_waiting_routing,
613	blocking_links=result.get("blocking_links"),
614	)
615
616	if recording:	3✔
617	track_link_waiting(	3✔
618	result.get("waiting_time", 0.0),
619	link_waiting,
620	blocking_links=result.get("blocking_links"),
621	)
622	if result["completion_time"] > horizon_time:	3✔
623	excess = result["completion_time"] - horizon_time	×
624	for lk in route_links:	×
625	if lk in link_busy_record:	×
626	link_busy_record[lk] = max(	×
627	0.0, link_busy_record[lk] - excess
628	)
629
630	max_completion = max(max_completion, result["completion_time"])	3✔
631
632	status = result.get("status", "")	3✔
633	if status == "completed":	3✔
634	continue	3✔
635
636	df = pd.DataFrame(log)	3✔
637	del log	3✔
638	link_utilization = compute_link_utilization(	3✔
639	link_busy_record, warmup_time, horizon_time,
640	)
641
642	return (	3✔
643	df,
644	pga_names,
645	pga_release_times,
646	link_utilization,
647	link_waiting,
648	routing_decision_cpt,
649	routing_decision_runtime,
650	defer_counts,
651	)

Quang00 / qnscheduling / 26622922786

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous