"""Calculate application efficiency and performance portability."""
from __future__ import annotations
from typing import Iterable
import numpy as np
import pandas as pd
from loguru import logger
from ppbcc.constants import (
APPLICATION,
APPLICATION_EFFICIENCY,
DESCRIPTION,
HARDWARE,
PARADIGM,
PERFORMANCE_PORTABILITY,
PRECISION,
PROBLEM_SIZE,
)
from ppbcc.performance_portability.selection import (
AVERAGE_OVER_EFFICIENCY,
AVERAGE_OVER_PP,
AVERAGE_SIZE,
BEST_SIZE,
WORST_SIZE,
_application_label,
_convert_runtime_to_ns,
)
def _harmonic_mean_or_zero(
values: pd.Series, platform_count: int, non_zero_pp: bool
) -> float:
"""Calculate PP, optionally omitting unsupported platforms.
Args:
values: One application efficiency per platform.
platform_count: Size of the platform set PP is defined over.
non_zero_pp: Whether platforms with zero efficiency are omitted.
Returns:
The harmonic mean, or zero if a required platform is unsupported.
"""
numeric = values.to_numpy(dtype=float)
if non_zero_pp:
numeric = numeric[numeric > 0.0]
if len(numeric) == 0:
return 0.0
return float(len(numeric) / np.reciprocal(numeric).sum())
if len(numeric) != platform_count or np.any(numeric <= 0.0):
return 0.0
return float(len(numeric) / np.reciprocal(numeric).sum())
def _validate_average_over(average_over: str) -> None:
"""Reject an unknown ``--average-over`` mode."""
if average_over not in {AVERAGE_OVER_PP, AVERAGE_OVER_EFFICIENCY}:
raise ValueError(f"Unsupported average-over mode: {average_over!r}")
[docs]
def calculate_metrics(
df: pd.DataFrame,
description_is_workload: bool,
non_zero_pp: bool = False,
hardware_universe: Iterable[str] | None = None,
application_universe: Iterable[str] | None = None,
) -> tuple[pd.DataFrame, pd.DataFrame]:
"""Calculate application efficiency and performance portability.
Duplicate measurements are reduced to their median runtime. Application
efficiencies are computed per hardware/workload and then arithmetically
averaged over workloads. PP is the harmonic mean across all selected
hardware. By default, an absent result contributes zero and makes PP zero.
In non-zero PP mode, platforms with zero efficiency are omitted from the
harmonic mean; the application-efficiency output still retains those zeros.
Args:
df: Selected benchmark rows.
description_is_workload: Whether Description belongs to the workload
key rather than the application label.
non_zero_pp: Whether to exclude unsupported platforms from PP.
hardware_universe: Optional complete platform set. This is used for
per-size scaling so a platform with no row at one size contributes
zero rather than disappearing from that size's PP calculation.
application_universe: Optional complete application set. This is used
for per-size calculations so an implementation with no row at one
size receives zero rather than disappearing from that size.
Returns:
A pair ``(efficiency, portability)``. The first DataFrame has one row
per application/hardware; the second has one row per application.
Raises:
ValueError: If there are no positive numeric wall-clock runtimes.
"""
data = df.copy()
data[APPLICATION] = [
_application_label(paradigm, description, description_is_workload)
for paradigm, description in zip(data[PARADIGM], data[DESCRIPTION])
]
data["Runtime (ns)"] = _convert_runtime_to_ns(data)
invalid = data["Runtime (ns)"].isna() | data["Runtime (ns)"].le(0)
if invalid.any():
logger.warning(f"Dropping {int(invalid.sum())} rows with invalid runtimes")
data = data.loc[~invalid].copy()
if data.empty:
raise ValueError("No positive numeric wall-clock runtimes remain.")
workload_columns = [PROBLEM_SIZE]
if PRECISION in data.columns and data[PRECISION].notna().any():
workload_columns.append(PRECISION)
if description_is_workload:
workload_columns.append(DESCRIPTION)
measurement_keys = [HARDWARE, APPLICATION, *workload_columns]
runtimes = (
data.groupby(measurement_keys, dropna=False, as_index=False)["Runtime (ns)"]
.median()
.sort_values(measurement_keys)
)
best_keys = [HARDWARE, *workload_columns]
runtimes["Best Runtime (ns)"] = runtimes.groupby(best_keys, dropna=False)[
"Runtime (ns)"
].transform("min")
runtimes[APPLICATION_EFFICIENCY] = (
runtimes["Best Runtime (ns)"] / runtimes["Runtime (ns)"]
)
applications = sorted(
set(application_universe)
if application_universe is not None
else set(runtimes[APPLICATION].unique())
)
hardware = sorted(
set(hardware_universe)
if hardware_universe is not None
else set(runtimes[HARDWARE].unique())
)
workloads = runtimes[workload_columns].drop_duplicates()
full_index = pd.MultiIndex.from_frame(
pd.DataFrame(
[
(application, platform, *workload)
for application in applications
for platform in hardware
for workload in workloads.itertuples(index=False, name=None)
],
columns=[APPLICATION, HARDWARE, *workload_columns],
)
)
indexed = runtimes.set_index([APPLICATION, HARDWARE, *workload_columns])
complete = indexed.reindex(full_index)
complete.index.names = [APPLICATION, HARDWARE, *workload_columns]
complete[APPLICATION_EFFICIENCY] = complete[APPLICATION_EFFICIENCY].fillna(0.0)
complete = complete.reset_index()
efficiency = (
complete.groupby([APPLICATION, HARDWARE], as_index=False)[
APPLICATION_EFFICIENCY
]
.mean()
.sort_values([APPLICATION, HARDWARE])
)
portability = (
efficiency.groupby(APPLICATION, as_index=False)[APPLICATION_EFFICIENCY]
.agg(_harmonic_mean_or_zero, len(hardware), non_zero_pp)
.rename(columns={APPLICATION_EFFICIENCY: PERFORMANCE_PORTABILITY})
.sort_values(PERFORMANCE_PORTABILITY, ascending=False)
)
logger.debug(f"Performance portability:\n{portability.to_string(index=False)}")
return efficiency, portability
[docs]
def calculate_metrics_by_size(
df: pd.DataFrame,
description_is_workload: bool,
non_zero_pp: bool = False,
) -> tuple[pd.DataFrame, pd.DataFrame]:
"""Calculate both metrics independently for each problem size.
Args:
df: Selected rows for one benchmark problem.
description_is_workload: Whether Description belongs to the workload key.
non_zero_pp: Whether unsupported platforms are excluded from PP.
Returns:
Efficiency and portability DataFrames that retain problem size.
"""
numeric_sizes = pd.to_numeric(df[PROBLEM_SIZE], errors="coerce")
sizes = sorted(numeric_sizes.dropna().unique())
if not sizes:
raise ValueError("No numeric problem sizes remain for the scaling plot.")
platforms = sorted(df[HARDWARE].unique())
applications = sorted(
{
_application_label(paradigm, description, description_is_workload)
for paradigm, description in zip(df[PARADIGM], df[DESCRIPTION])
}
)
efficiency_frames: list[pd.DataFrame] = []
portability_frames: list[pd.DataFrame] = []
for size in sizes:
size_rows = df.loc[
np.isclose(numeric_sizes.to_numpy(dtype=float), size, equal_nan=False)
]
efficiency, portability = calculate_metrics(
size_rows,
description_is_workload,
non_zero_pp=non_zero_pp,
hardware_universe=platforms,
application_universe=applications,
)
efficiency.insert(1, PROBLEM_SIZE, float(size))
portability.insert(1, PROBLEM_SIZE, float(size))
efficiency_frames.append(efficiency)
portability_frames.append(portability)
return (
pd.concat(efficiency_frames, ignore_index=True),
pd.concat(portability_frames, ignore_index=True),
)
[docs]
def calculate_scaling_metrics(
df: pd.DataFrame,
description_is_workload: bool,
non_zero_pp: bool = False,
) -> pd.DataFrame:
"""Calculate performance portability independently for each problem size."""
_, portability = calculate_metrics_by_size(
df,
description_is_workload,
non_zero_pp=non_zero_pp,
)
return portability
[docs]
def calculate_average_size_metrics(
df: pd.DataFrame,
description_is_workload: bool,
non_zero_pp: bool = False,
average_over: str = AVERAGE_OVER_PP,
) -> tuple[pd.DataFrame, pd.DataFrame]:
"""Average application efficiency and PP over sizes.
Application efficiency is calculated independently at every size and then
arithmetically averaged, so every problem size has equal weight. PP is
reduced in one of two ways:
* ``pp``: PP is calculated at every size and the scores are averaged
arithmetically.
* ``efficiency``: the size sweep is treated as one benchmark. PP is the
harmonic mean over platforms of the size-averaged efficiencies, i.e.
exactly the PP of the plotted application-efficiency panel.
The two differ whenever an application's platform ranking changes with the
size: the arithmetic mean of harmonic means is not a harmonic mean.
Args:
df: Selected rows for one benchmark problem.
description_is_workload: Whether Description belongs to the workload key.
non_zero_pp: Whether unsupported platforms are excluded from PP.
average_over: ``pp`` or ``efficiency``, see above.
Returns:
Size-averaged application efficiency and performance portability.
Raises:
ValueError: If ``average_over`` is unsupported.
"""
_validate_average_over(average_over)
per_size_efficiency, per_size_portability = calculate_metrics_by_size(
df,
description_is_workload,
non_zero_pp=non_zero_pp,
)
efficiency = (
per_size_efficiency.groupby([APPLICATION, HARDWARE], as_index=False)[
APPLICATION_EFFICIENCY
]
.mean()
.sort_values([APPLICATION, HARDWARE])
)
if average_over == AVERAGE_OVER_EFFICIENCY:
portability = (
efficiency.groupby(APPLICATION, as_index=False)[APPLICATION_EFFICIENCY]
.agg(_harmonic_mean_or_zero, df[HARDWARE].nunique(), non_zero_pp)
.rename(columns={APPLICATION_EFFICIENCY: PERFORMANCE_PORTABILITY})
)
else:
portability = per_size_portability.groupby(APPLICATION, as_index=False)[
PERFORMANCE_PORTABILITY
].mean()
return efficiency, portability.sort_values(
PERFORMANCE_PORTABILITY, ascending=False
)
[docs]
def calculate_extreme_size_metrics(
df: pd.DataFrame,
description_is_workload: bool,
mode: str,
non_zero_pp: bool = False,
) -> tuple[pd.DataFrame, pd.DataFrame]:
"""Select each application's best or worst metrics over problem sizes.
Application efficiency is reduced independently for every application and
hardware pair. Performance portability is reduced independently for every
application, so both Cascade panels represent the requested size-axis
extreme of the metric they display.
Args:
df: Selected rows for one benchmark problem.
description_is_workload: Whether Description belongs to the workload key.
mode: ``best`` for maxima or ``worst`` for minima.
non_zero_pp: Whether unsupported platforms are excluded from PP.
Returns:
Application-efficiency and performance-portability extrema.
Raises:
ValueError: If mode is not ``best`` or ``worst``.
"""
if mode not in {BEST_SIZE, WORST_SIZE}:
raise ValueError(f"Unsupported size-extreme mode: {mode!r}")
per_size_efficiency, per_size_portability = calculate_metrics_by_size(
df,
description_is_workload,
non_zero_pp=non_zero_pp,
)
reduction = "max" if mode == BEST_SIZE else "min"
efficiency = (
per_size_efficiency.groupby([APPLICATION, HARDWARE], as_index=False)[
APPLICATION_EFFICIENCY
]
.agg(reduction)
.sort_values([APPLICATION, HARDWARE])
)
portability = (
per_size_portability.groupby(APPLICATION, as_index=False)[
PERFORMANCE_PORTABILITY
]
.agg(reduction)
.sort_values(PERFORMANCE_PORTABILITY, ascending=False)
)
return efficiency, portability
[docs]
def calculate_export_metrics(
df: pd.DataFrame,
description_is_workload: bool,
non_zero_pp: bool = False,
average_over: str = AVERAGE_OVER_PP,
) -> tuple[pd.DataFrame, pd.DataFrame]:
"""Calculate export metrics without combining sizes or precisions.
Returns one efficiency row per application, size, precision, and hardware,
and one portability row per application, size, and precision. Additional
rows with Problem Size ``average`` contain per-precision arithmetic means
over the independently calculated size efficiencies. Their PP is the mean
of the per-size PP scores for ``average_over="pp"``, or the harmonic mean
of the averaged efficiencies for ``average_over="efficiency"`` (see
:func:`calculate_average_size_metrics`).
"""
_validate_average_over(average_over)
dimensions = [PROBLEM_SIZE, PRECISION]
platforms = sorted(df[HARDWARE].unique())
applications = sorted(
{
_application_label(paradigm, description, description_is_workload)
for paradigm, description in zip(df[PARADIGM], df[DESCRIPTION])
}
)
efficiency_frames: list[pd.DataFrame] = []
portability_frames: list[pd.DataFrame] = []
for values, rows in df.groupby(dimensions, dropna=False, sort=True):
if not isinstance(values, tuple):
values = (values,)
efficiency, portability = calculate_metrics(
rows,
description_is_workload,
non_zero_pp=non_zero_pp,
hardware_universe=platforms,
application_universe=applications,
)
for position, (column, value) in enumerate(zip(dimensions, values), start=1):
efficiency.insert(position, column, value)
portability.insert(position, column, value)
efficiency_frames.append(efficiency)
portability_frames.append(portability)
if not efficiency_frames:
raise ValueError("No problem-size/precision groups remain for CSV export.")
efficiency = pd.concat(efficiency_frames, ignore_index=True)
portability = pd.concat(portability_frames, ignore_index=True)
average_efficiency = (
efficiency.groupby(
[APPLICATION, PRECISION, HARDWARE],
as_index=False,
dropna=False,
)[APPLICATION_EFFICIENCY]
.mean()
.sort_values([APPLICATION, PRECISION, HARDWARE])
)
average_efficiency.insert(1, PROBLEM_SIZE, AVERAGE_SIZE)
if average_over == AVERAGE_OVER_EFFICIENCY:
average_portability = (
average_efficiency.groupby(
[APPLICATION, PRECISION],
as_index=False,
dropna=False,
)[APPLICATION_EFFICIENCY]
.agg(_harmonic_mean_or_zero, len(platforms), non_zero_pp)
.rename(columns={APPLICATION_EFFICIENCY: PERFORMANCE_PORTABILITY})
.sort_values([APPLICATION, PRECISION])
)
else:
average_portability = (
portability.groupby(
[APPLICATION, PRECISION],
as_index=False,
dropna=False,
)[PERFORMANCE_PORTABILITY]
.mean()
.sort_values([APPLICATION, PRECISION])
)
average_portability.insert(1, PROBLEM_SIZE, AVERAGE_SIZE)
return (
pd.concat([efficiency, average_efficiency], ignore_index=True),
pd.concat([portability, average_portability], ignore_index=True),
)
[docs]
def calculate_runtimes(
df: pd.DataFrame,
description_is_workload: bool,
time_column: str,
remove_description: bool = False,
) -> pd.DataFrame:
"""Reduce benchmark rows to one runtime per application and platform.
Duplicate measurements are reduced to their median. Rows whose runtime is
missing or not positive are dropped with a warning, because a time bar on
a logarithmic axis cannot show them.
Args:
df: Selected benchmark rows of one problem size and precision.
description_is_workload: Whether Description belongs to the workload
key rather than the application label.
time_column: Runtime column to reduce, e.g. ``Kernel Time``.
remove_description: Whether implementation variants of one paradigm
are combined; the fastest variant then stands for the paradigm.
Returns:
One row per application, hardware and precision with the runtime in
nanoseconds in ``Runtime (ns)``.
Raises:
ValueError: If no positive runtime remains.
"""
data = df.copy()
data[APPLICATION] = [
_application_label(
paradigm, "" if remove_description else description, description_is_workload
)
for paradigm, description in zip(data[PARADIGM], data[DESCRIPTION])
]
data["Runtime (ns)"] = _convert_runtime_to_ns(data, time_column)
invalid = data["Runtime (ns)"].isna() | data["Runtime (ns)"].le(0)
if invalid.any():
dropped = sorted(
set(zip(data.loc[invalid, APPLICATION], data.loc[invalid, HARDWARE]))
)
logger.warning(
f"Dropping {int(invalid.sum())} rows without a positive "
f"{time_column!r}: {dropped}"
)
data = data.loc[~invalid].copy()
if data.empty:
raise ValueError(f"No positive {time_column!r} values remain.")
keys = [APPLICATION, HARDWARE, PRECISION]
medians = data.groupby(
[*keys, PARADIGM, DESCRIPTION], dropna=False, as_index=False
)["Runtime (ns)"].median()
return (
medians.groupby(keys, dropna=False, as_index=False)["Runtime (ns)"]
.min()
.sort_values(keys)
.reset_index(drop=True)
)