statistics.py

from __future__ import annotations

from typing import Any, List, Optional

from pydantic import BaseModel, computed_field


class RequestRecord(BaseModel):
    id: int = -1
    num_input_tokens: int = -1
    tokens: List[int] = []
    error_tokens: int = 0
    start_timestamp: int = -1
    first_token_timestamp: int = -1
    end_timestamp: int = -1
    decode_iteration: int = 0

    def register_event(self,
                       is_error: bool,
                       is_final: bool,
                       timestamp: int,
                       decoding_iter: int,
                       tokens: List[int],
                       first_token_timestamp: int = None) -> None:
        if is_final:
            self.end_timestamp = timestamp
        elif self.first_token_timestamp == -1:
            self.first_token_timestamp = timestamp

        if first_token_timestamp is not None and is_final:
            self.first_token_timestamp = first_token_timestamp

        if is_error:
            self.error_tokens += 1

        self.tokens += tokens
        self.decode_iteration = decoding_iter

    @computed_field
    def num_total_output_tokens(self) -> int:
        """
        Returns the total number of output tokens generated by the request.
        """
        return len(self.tokens)

    @computed_field
    def num_generated_tokens(self) -> int:
        """
        Returns the number of generated (OSL - 1) tokens by the request.
        """
        return self.num_total_output_tokens - 1

    @computed_field
    def generation_time(self) -> int:
        """
        Returns the generation time of the request (E2E Latency - TTFT).
        """
        return self.end_to_end_latency - self.time_to_first_token

    @computed_field
    def time_to_first_token(self) -> int:
        return (self.first_token_timestamp -
                self.start_timestamp if self.first_token_timestamp > 0 else 0.0)

    @computed_field
    def intertoken_latency(self) -> float:
        """
        Returns the time-per-output-token latency of the request [(OSL - 1) / (E2E Latency - TTFT)].
        """
        return ((self.end_timestamp - self.first_token_timestamp) /
                self.num_generated_tokens
                if self.num_generated_tokens > 0 else 0.0)

    @computed_field
    def end_to_end_latency(self) -> int:
        """
        Returns the end-to-end latency of the request (end time - start time).
        """
        return self.end_timestamp - self.start_timestamp

    @computed_field
    def output_token_throughput(self) -> float:
        """
        Returns the total token throughput of the request (Total output tokens / E2E Latency).
        """
        return float(self.num_total_output_tokens) / self.end_to_end_latency

    @computed_field
    def generation_token_throughput(self) -> float:
        return (self.num_generated_tokens / self.generation_time)


class PercentileStats(BaseModel):
    p50: float
    p90: float
    p95: float
    p99: float
    minimum: float
    maximum: float
    average: float

    @classmethod
    def from_iterable(cls, values: List[Any]) -> PercentileStats:
        length = len(values)
        sorted_values = sorted(values)
        return cls(
            p50=sorted_values[int(length * 0.50)],
            p90=sorted_values[int(length * 0.90)],
            p95=sorted_values[int(length * 0.95)],
            p99=sorted_values[int(length * 0.99)],
            average=float(sum(values)) / length,
            minimum=min(values),
            maximum=max(values),
        )


class BenchmarkStatistics(BaseModel):
    # Time-related Properties
    total_latency_ns: float

    # Token-related Properties
    total_output_tokens: int
    total_input_tokens: int

    # General Information
    num_requests: int
    issue_rate_ns: float

    # Speculative Information
    acceptance_length: float

    # Energy Monitoring
    total_energy: Optional[float] = None

    # Percentile-related Statistics
    request_latency_percentiles: Optional[PercentileStats] = None
    output_throughput_percentiles: Optional[PercentileStats] = None
    token_percentiles: Optional[PercentileStats] = None
    tpot_percentiles: Optional[PercentileStats] = None
    ttft_percentiles: Optional[PercentileStats] = None
    generation_tp_percentiles: Optional[PercentileStats] = None
    generation_latency_percentiles: Optional[PercentileStats] = None
    # Percentile-related Speculative Statistics
    num_draft_tokens_percentiles: Optional[PercentileStats] = None
    num_accepted_draft_tokens_percentiles: Optional[PercentileStats] = None
    draft_acceptance_rate_percentiles: Optional[PercentileStats] = None
    acceptance_length_percentiles: Optional[PercentileStats] = None

    @computed_field
    def sum_per_request_latencies_ns(self) -> float:
        return self.request_latency_percentiles.average * self.num_requests

    @computed_field
    def avg_concurrent_requests(self) -> int:
        return self.sum_per_request_latencies_ns / self.total_latency_ns

    @computed_field
    def generation_tokens(self) -> int:
        return int(self.total_output_tokens - self.num_requests)

    @computed_field
    def total_generation_time_ns(self) -> float:
        return self.generation_latency_percentiles.average * self.num_requests

    @computed_field
    def per_user_time_per_output_token_ns(self) -> float:
        return self.tpot_percentiles.average

    @computed_field
    def per_user_time_to_first_token_ns(self) -> float:
        return self.ttft_percentiles.average

    @computed_field
    def per_user_generation_token_throughput_ns(self) -> float:
        return self.generation_tp_percentiles.average

    @computed_field
    def request_throughput_ns(self) -> float:
        return float(self.num_requests) / self.total_latency_ns

    @computed_field
    def average_input_length(self) -> float:
        return float(self.total_input_tokens) / self.num_requests

    @computed_field
    def average_output_length(self) -> float:
        return float(self.total_output_tokens) / self.num_requests

    @computed_field
    def output_throughput_tok_ns(self) -> float:
        return float(self.total_output_tokens) / self.total_latency_ns

    @computed_field
    def total_token_throughput_tok_ns(self) -> float:
        return float(self.total_input_tokens +
                     self.total_output_tokens) / self.total_latency_ns

    @computed_field
    def output_throughput_tok_ns_per_user(self) -> float:
        return self.output_throughput_percentiles.average

    @computed_field
    def output_tps_per_w(self) -> Optional[float]:
        if not self.total_energy:
            return None
        return float(self.total_output_tokens / self.total_energy)

    @computed_field
    def total_gpu_power(self) -> Optional[float]:
        if not self.total_energy or not self.total_latency_ns:
            return None
        return float((self.total_energy * 1e9) / self.total_latency_ns)