swh.scheduler.simulator.common module

class swh.scheduler.simulator.common.SimulationReport(total_visits: int = 0, visit_runtimes: Dict[Tuple[str, bool], List[float]] = <factory>, scheduler_metrics: List[Tuple[datetime.datetime, List[swh.scheduler.model.SchedulerMetrics]]] = <factory>, visit_metrics: List[Tuple[datetime.datetime, int]] = <factory>, latest_snapshots: Dict[Tuple[str, str], bytes] = <factory>)[source]

Bases: object

DURATION_THRESHOLD = 3600

Max duration for histograms

total_visits: int = 0

Total count of finished visits

visit_runtimes: Dict[Tuple[str, bool], List[float]]

Collected visit runtimes for each (status, eventful) tuple

scheduler_metrics: List[Tuple[datetime.datetime, List[swh.scheduler.model.SchedulerMetrics]]]

Collected scheduler metrics

This is a list of couples (timestamp, [SchedulerMetrics,]): the list of scheduler metrics collected at given timestamp.

visit_metrics: List[Tuple[datetime.datetime, int]]

Collected visit metrics over time

latest_snapshots: Dict[Tuple[str, str], bytes]

Collected latest snapshots for origins

record_visit(origin: Tuple[str, str], duration: float, status: str, snapshot=typing.Union[bytes, NoneType]) None[source]
record_metrics(timestamp: datetime.datetime, scheduler_metrics: List[swh.scheduler.model.SchedulerMetrics])[source]
property uneventful_visits

Number of uneventful, full visits

runtime_histogram(status: str, eventful: bool) str[source]
metrics_plot() str[source]
metrics_csv(fobj: TextIO) None[source]

Export scheduling metrics in a csv file

format(with_plots=True)[source]
class swh.scheduler.simulator.common.Task(visit_type: str, origin: str, backend_id: uuid.UUID = <factory>)[source]

Bases: object

visit_type: str
origin: str
backend_id: uuid.UUID
class swh.scheduler.simulator.common.TaskEvent(task: swh.scheduler.simulator.common.Task, status: swh.model.model.OriginVisitStatus, eventful: bool = False)[source]

Bases: object

task: swh.scheduler.simulator.common.Task
status: swh.model.model.OriginVisitStatus
eventful: bool = False
class swh.scheduler.simulator.common.Queue(env, capacity=inf)[source]

Bases: simpy.resources.store.Store

Model a queue of objects to be passed between processes.

slots_remaining()[source]
class swh.scheduler.simulator.common.Environment(start_time: datetime.datetime)[source]

Bases: simpy.core.Environment

report: swh.scheduler.simulator.common.SimulationReport
scheduler: swh.scheduler.interface.SchedulerInterface
property time

Get the current simulated wall clock time