Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/simulator/common.py
| # Copyright (C) 2021 The Software Heritage developers | # Copyright (C) 2021 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| from dataclasses import dataclass, field | from dataclasses import dataclass, field | ||||
| from datetime import datetime, timedelta | from datetime import datetime, timedelta | ||||
| import textwrap | import textwrap | ||||
| from typing import Dict, List, Tuple | from typing import Dict, List, Optional, Tuple | ||||
| import uuid | import uuid | ||||
| import plotille | import plotille | ||||
| from simpy import Environment as _Environment | from simpy import Environment as _Environment | ||||
| from simpy import Store | from simpy import Store | ||||
| from swh.model.model import OriginVisitStatus | from swh.model.model import OriginVisitStatus | ||||
| from swh.scheduler.interface import SchedulerInterface | from swh.scheduler.interface import SchedulerInterface | ||||
| Show All 9 Lines | class SimulationReport: | ||||
| """Total count of finished visits""" | """Total count of finished visits""" | ||||
| visit_runtimes: Dict[Tuple[str, bool], List[float]] = field(default_factory=dict) | visit_runtimes: Dict[Tuple[str, bool], List[float]] = field(default_factory=dict) | ||||
| """Collected visit runtimes for each (status, eventful) tuple""" | """Collected visit runtimes for each (status, eventful) tuple""" | ||||
| metrics: List[Tuple[datetime, List[SchedulerMetrics]]] = field(default_factory=list) | metrics: List[Tuple[datetime, List[SchedulerMetrics]]] = field(default_factory=list) | ||||
| """Collected scheduler metrics for every timestamp""" | """Collected scheduler metrics for every timestamp""" | ||||
| def record_visit(self, duration: float, eventful: bool, status: str) -> None: | latest_snapshots: Dict[Tuple[str, str], bytes] = field(default_factory=dict) | ||||
| """Collected latest snapshots for origins""" | |||||
| def record_visit( | |||||
| self, | |||||
| origin: Tuple[str, str], | |||||
| duration: float, | |||||
| status: str, | |||||
| snapshot=Optional[bytes], | |||||
| ) -> None: | |||||
| eventful = False | |||||
| if status == "full": | |||||
| eventful = snapshot != self.latest_snapshots.get(origin) | |||||
| self.latest_snapshots[origin] = snapshot | |||||
| self.total_visits += 1 | self.total_visits += 1 | ||||
| self.visit_runtimes.setdefault((status, eventful), []).append(duration) | self.visit_runtimes.setdefault((status, eventful), []).append(duration) | ||||
| def record_metrics(self, timestamp: datetime, metrics: List[SchedulerMetrics]): | def record_metrics(self, timestamp: datetime, metrics: List[SchedulerMetrics]): | ||||
| self.metrics.append((timestamp, metrics)) | self.metrics.append((timestamp, metrics)) | ||||
| @property | @property | ||||
| def useless_visits(self): | def useless_visits(self): | ||||
| ▲ Show 20 Lines • Show All 89 Lines • Show Last 20 Lines | |||||