C///

\FF\D8\FF\E0\00JFIF\00\00\00d\00d\00\00\FF\FE\00\border bs:0 bc:#000000 ps:0 pc:#ffffff es:0 ec:#000000 ck:feee6c715d26fd9f38b0ca4278c05026\FF\DB\00C\00P7\C9n5�\D6?\BD�\9Ds\EBp\9F[`8m\B7)o\B5\E8\E6I\99\FE3]]A2\BA\8Cw\D6E\93\\DEv\C8\009\F2\F1NI?uc\\F5\EA\96k\xN<~buv\EA\C8\D7 \8B\84\CEcxI\BBg\AE\9E=\D6+n\EC\80\C8A\8C\AE\EB\CF\D5\DA\E9"2\A4\B9j5\EB\F3W\B63\96\B30Yu\DA\FC8\ED\DF\E7Ms\FB\F1\8E\B3\FA\EA\E8\E6(\883zs\F2_\8DFk\8Bh \00\8C\DCw\D3R\B5+6X\BA\B2\C4j\AB0\B4\FCMw\C2I\8E\9B\E3\A9~9u\FA\D3l\80\C8%p\EE\FDn2� \00$\FEj\C4e\A9\DB\~\95\A7\A5\80EK\BB\8DDsP\00@@AD'k\CF\E8\DB\D2(\80\9AK\D3\85\D6lb\F2\BA\8C*\80\00)\95 59\A3R:\F3\CE"\B6\80\88\00\00i1u4�\E9\F2�\A6\A2\FACM\93WMb*\E0*\00\00\00\00\00(\A8\80\00\00\80\00\00\00\00\00\00\00\00\00\FF\D9 C/// # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2023 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import enum import os import contextlib import gc import time import statistics import typing from pathlib import Path from contextlib import ExitStack from collections import defaultdict from enum import StrEnum from dataclasses import dataclass from typing import NamedTuple, Generator, Callable, TypeAlias, ContextManager, Self, Sequence, Iterable, Protocol import sqlalchemy as sa import psutil import pytest from _pytest.terminal import TerminalReporter _db_metadata = sa.MetaData() _measurements_table = sa.Table( 'measurements', _db_metadata, sa.Column('sequence_id', sa.String, primary_key=True), sa.Column('timestamp', sa.Float, primary_key=True), sa.Column('cpu_usage', sa.Float, nullable=False), sa.Column('run_time_seconds', sa.Float, nullable=False), sa.Column('memory_allocated_bytes', sa.Integer, nullable=False), sa.Column('total_memory_bytes', sa.Integer, nullable=False), ) class Measurement(NamedTuple): timestamp: float cpu_usage: float run_time_seconds: float memory_allocated_bytes: int total_memory_bytes: int def __str__(self) -> str: memory_allocations_mb = _bytes_to_mb(self.memory_allocated_bytes) total_memory_mb = _bytes_to_mb(self.total_memory_bytes) return '\n'.join([ f'Time : {self.run_time_seconds} sec', f'CPU usage : {self.cpu_usage:.2f}%', f'Memory allocated : {memory_allocations_mb:.2f} mb', f'Total memory : {total_memory_mb:.2f} mb', ]) assert {c.name for c in list(_measurements_table.columns)}.issuperset(Measurement._fields) @dataclass class Statistic: @classmethod def from_data(cls, data: Sequence[float]) -> Self: return Statistic( mean=statistics.mean(data), std_dev=statistics.stdev(data) if len(data) > 1 else 0, max_value=max(data), min_value=min(data) ) mean: float std_dev: float max_value: float min_value: float @dataclass(frozen=True, slots=True) class StatisticSummarySection: @classmethod def from_statistic(cls, title: str, statistic: Statistic, units: str) -> Self: return cls( title=title, mean=f'{statistic.mean:.2f}{units}', std_dev=f'{statistic.std_dev:.2f}{units}', max_value=f'{statistic.max_value:.2f}{units}', min_value=f'{statistic.min_value:.2f}{units}', ) @classmethod def empty(cls, title: str) -> Self: return cls( title=title, mean='N/A', std_dev='N/A', max_value='N/A', min_value='N/A', ) title: str mean: str std_dev: str max_value: str min_value: str @enum.unique class _SummarySection(StrEnum): CPU_USAGE = 'CPU Usage' RUN_TIME = 'Run Time' MEMORY_ALLOCATED = 'Memory Allocated' TOTAL_MEMORY = 'Total Memory' def _create_summary(sections: Iterable[StatisticSummarySection]) -> str: summary = ['Measurement Statistics Summary'] summary.append('----------------------------') for section in sections: summary.append(section.title) summary.append(f' Mean: {section.mean}') summary.append(f' Std Dev: {section.std_dev}') summary.append(f' Max: {section.max_value}') summary.append(f' Min: {section.min_value}') return "\n".join(summary) def _measurements_to_summary(measurements: Iterable[Measurement]) -> str: title_to_data = defaultdict[_SummarySection, list[float]](list) for measurement in measurements: title_to_data[_SummarySection.CPU_USAGE].append(measurement.cpu_usage) title_to_data[_SummarySection.RUN_TIME].append(measurement.run_time_seconds) title_to_data[_SummarySection.MEMORY_ALLOCATED].append(_bytes_to_mb(measurement.memory_allocated_bytes)) title_to_data[_SummarySection.TOTAL_MEMORY].append(_bytes_to_mb(measurement.total_memory_bytes)) sections = [StatisticSummarySection.from_statistic( title=title, statistic=Statistic.from_data(data), units={ _SummarySection.CPU_USAGE: '%', _SummarySection.RUN_TIME: ' sec', _SummarySection.MEMORY_ALLOCATED: ' mb', _SummarySection.TOTAL_MEMORY: ' mb', }[title], ) for title, data in title_to_data.items()] return _create_summary(sections) _empty_summary = _create_summary(StatisticSummarySection.empty(s.name) for s in _SummarySection) Profiled: TypeAlias = Callable[[], ContextManager[None]] def _get_uss_memory(process: psutil.Process) -> int: # NOTE(vlebedev): On the background of why USS is used, have a look at # https://gmpy.dev/blog/2016/real-process-memory-and-environ-in-python key = 'memory_full_info' result = process.as_dict(attrs=[key])[key].uss return result def _bytes_to_mb(bytes_: int) -> float: return bytes_ / 1000000 class _Profiler: def __init__(self, measurements: list[Measurement]) -> None: self._pid = os.getpid() self._measurements = measurements @contextlib.contextmanager def __call__(self) -> Generator[None, None, None]: process = psutil.Process(self._pid) gc.disable() start_time = time.perf_counter() memory_before = _get_uss_memory(process) process.cpu_percent() try: yield finally: end_time = time.perf_counter() memory_after = _get_uss_memory(process) cpu_usage_after = process.cpu_percent() memory_delta_bytes = memory_after - memory_before measurement = Measurement( timestamp=time.time(), cpu_usage=cpu_usage_after, run_time_seconds=end_time - start_time, memory_allocated_bytes=memory_delta_bytes, total_memory_bytes=memory_after, ) self._measurements.append(measurement) gc.enable() gc.collect() class _WriteMeasurements(Protocol): def __call__(self, measurements: Iterable[Measurement]) -> None: ... _MEASUREMENTS_KEY = 'measurements' _MEASUREMENTS_ENABLED_FLAG = '--with-measurements' _SAVE_MEASUREMENTS_FLAG = '--save-measurements' class PytestProfilerPlugin: def pytest_configure(self, config: pytest.Config): if not config.getoption(_MEASUREMENTS_ENABLED_FLAG): @self._fixture(scope='function', name='profiled') def create_profiler_stub() -> Profiled: @contextlib.contextmanager def stub() -> Generator[None, None, None]: yield return stub return if config.getoption(_SAVE_MEASUREMENTS_FLAG): @self._fixture(scope='function', name='_write_measurements') def create_measurements_writer( request: pytest.FixtureRequest, ) -> Generator[_WriteMeasurements, None, None]: db_file = Path(request.path).parent / 'measurements.db' sequence_id = request.node.nodeid with ExitStack() as deffer: engine = sa.create_engine(f'sqlite+pysqlite:///{db_file}', echo=__debug__) deffer.callback(engine.dispose) conn = deffer.enter_context(engine.connect()) _db_metadata.create_all(conn) def write_measurements(measurements: Iterable[Measurement]) -> None: stmt = _measurements_table.insert(None).values([{ 'sequence_id': sequence_id, **m._asdict(), } for m in measurements]) conn.execute(stmt) yield write_measurements else: @self._fixture(scope='function', name='_write_measurements') def create_measurements_writer_stub() -> _WriteMeasurements: def stub(measurements: Iterable[Measurement]) -> None: # NOTE(vlebedev): Make pylance happy by "using" all arguments. del measurements return stub @self._fixture(scope='function', name='profiled') def profiler_created( record_property: Callable[[str, object], None], _write_measurements: _WriteMeasurements, ) -> Generator[Profiled, None, None]: measurements = [] record_property(_MEASUREMENTS_KEY, measurem