| """ |
| Business Source License 1.1 |
| |
| Copyright (C) 2025 Raman Marozau, raman@worktif.com |
| Use of this software is governed by the Business Source License included in the LICENSE file and at www.mariadb.com/bsl11. |
| |
| Change Date: Never |
| On the date above, in accordance with the Business Source License, use of this software will be governed by the open source license specified in the LICENSE file. |
| Additional Use Grant: Free for personal and non-commercial research use only. |
| |
| SPDX-License-Identifier: BUSL-1.1 |
| """ |
|
|
| import datetime |
| import os |
|
|
| import json |
| import warnings |
| from pathlib import Path |
| from typing import List |
|
|
| import numpy as np |
| import pandas as pd |
| from matplotlib import pyplot as plt |
| from pandas.core.interchange.dataframe_protocol import DataFrame |
| from scipy.signal import savgol_filter |
|
|
|
|
| class BenchmarksAnalysis: |
| """ |
| Class for managing the analysis of benchmarks. |
| |
| This class provides functionality to handle benchmark data including loading, |
| processing into a DataFrame, and generating various visual plots for performance |
| analysis. It ensures proper organization of benchmark files and provides specific |
| plot types to assess performance metrics and acceleration factors across a range |
| of input sizes. |
| |
| :ivar benchmarks_plot_dir: Directory for storing generated benchmark plots. |
| :type benchmarks_plot_dir: pathlib.Path |
| :ivar test_number: The integer representation of the test identifier. |
| :type test_number: int |
| """ |
|
|
| def __init__(self, test_id: str) -> None: |
| self.benchmarks_plot_dir = self.benchmarks_setup() |
| self.test_number = int(test_id) |
|
|
| @staticmethod |
| def benchmarks_setup() -> Path: |
| """ |
| Creates and initializes a directory for storing benchmark plots. |
| |
| This static method generates a uniquely named directory based on the current |
| date and time in the "purem_benchmarks" directory relative to the script's |
| location. If the directory does not already exist, it will be created. The |
| directory name includes a timestamp in the format "YYYY-MM-DD_HH_MM_SS__FFF". |
| |
| :return: A `Path` object pointing to the created directory. |
| :rtype: Path |
| """ |
| current_time = datetime.datetime.now() |
| benchmarks_id: str = f"""purem_benchmarks_{current_time.strftime("%Y-%m-%d_%H_%M_%S__%f")[:-4]}""" |
| current_dir: str = os.path.dirname(__file__) |
| benchmarks_plot_dir: Path = Path(current_dir) / "purem_benchmarks" / benchmarks_id |
| os.makedirs(benchmarks_plot_dir, exist_ok=True) |
|
|
| return benchmarks_plot_dir |
|
|
| @staticmethod |
| def extract_size_from_filepath(filepath: str) -> int: |
| """ |
| Extract the `size` value from the given file path. |
| Assumes the size is encoded as the last part before the file extension. |
| """ |
| split_filepath = filepath.split("_") |
| return int(split_filepath[-1].split(".")[0]) |
|
|
| def run(self) -> None: |
| """ |
| Executes the process of generating benchmarks and acceleration plots, as well as creating |
| and saving Markdown-formatted data frames. This function orchestrates the pipeline by |
| calling several helper methods to handle plotting and data frame construction. |
| |
| :raises AnyRelevantError: Description of why and when it might be raised. |
| |
| :return: None |
| """ |
| benchmarks_df = self.benchmarks_data_frame() |
| self.benchmarks_plot(benchmarks_df) |
| self.acceleration_plot(benchmarks_df) |
|
|
| markdown_df = self.markdown_data_frame() |
| self.markdown_save(markdown_df) |
|
|
| def load_benchmarks_files(self) -> List[str]: |
| """ |
| Loads benchmark files matching a specific test number from a designated directory. |
| The function identifies files in the benchmarks directory that match a given test number, |
| extracts a unique timestamp from the matching files, and retrieves all files that share |
| the same timestamp. This ensures that all related benchmark files for a specific test are |
| collected efficiently. |
| |
| :param test_number: An integer representing the test number used to filter and locate |
| benchmark files. The test number will be zero-padded to ensure a four-digit format. |
| :return: A list of file paths for all benchmark files associated with the specified test |
| number and same timestamp. |
| :rtype: List[str] |
| :raises FileNotFoundError: If no benchmark files matching the specified test number |
| are found in the benchmarks directory. |
| """ |
| benchmarks_dir = ".benchmarks/Darwin-CPython-3.11-64bit" |
| search_pattern = str(self.test_number).zfill(4) |
|
|
| matched_files = [f for f in os.listdir(benchmarks_dir) if f.startswith(search_pattern) and f.endswith(".json")] |
|
|
| if not matched_files: |
| raise FileNotFoundError(f"No benchmark files found starting with {search_pattern} in {benchmarks_dir}") |
|
|
| file_stamp = matched_files[0].split("_")[2] |
| return [ |
| os.path.join(benchmarks_dir, f) |
| for f in os.listdir(benchmarks_dir) |
| if f.split("_")[2] == file_stamp and f.endswith(".json") |
| ] |
|
|
| def benchmarks_data_frame(self) -> pd.DataFrame: |
| """ |
| Generates a Pandas DataFrame containing benchmark data extracted |
| from all related files. The method processes multiple files, loads |
| their content, extracts benchmark statistics, and organizes them |
| into a structured tabular format. |
| |
| :param self: The instance of the class containing this method. |
| |
| :returns: A pandas DataFrame where each row represents a single |
| benchmark entry. The columns include 'size', 'func', 'min', |
| 'max', 'mean', 'stddev', and 'ops', corresponding to the |
| benchmark statistics and other metadata. |
| :rtype: pd.DataFrame |
| """ |
| files = self.load_benchmarks_files() |
|
|
| records = [] |
|
|
| for filepath in files: |
| with open(filepath, "r") as f: |
| data = json.load(f) |
|
|
| filepath_split = filepath.split("_") |
|
|
| for entry in data["benchmarks"]: |
| size = self.extract_size_from_filepath(filepath) |
| func = entry["params"]["func_name"] |
| stats = entry["stats"] |
| records.append({ |
| "size": size, |
| "func": func, |
| "min": stats["min"], |
| "max": stats["max"], |
| "mean": stats["mean"], |
| "stddev": stats["stddev"], |
| "ops": stats["ops"] |
| }) |
|
|
| return pd.DataFrame(records) |
|
|
| def benchmarks_plot(self, df: DataFrame) -> None: |
| """ |
| Generates and saves benchmarking plots for various performance metrics. |
| |
| This function iterates over a defined list of metrics such as "min", "max", "mean", |
| "stddev", and "ops", and generates two types of plots for each metric: |
| one for the full range of input sizes and another for large input sizes only. |
| Each plot visualizes the relationship between input sizes and metric values for all |
| functions provided in the DataFrame. The x-axis is set to log scale for both |
| plots. The y-axis is also set to log scale for all metrics except "stddev". |
| The generated plots are saved as PNG files in a directory defined by |
| `self.benchmarks_plot_dir`. |
| |
| :param df: A DataFrame containing benchmarking data. The DataFrame must have the |
| following columns: |
| - "func": Names of the functions being benchmarked. |
| - "size": Input sizes corresponding to the benchmarks. |
| - Metric columns (e.g., "min", "max", "mean", "stddev", "ops"): Values of the |
| respective performance metrics. |
| """ |
| metrics = ["min", "max", "mean", "stddev", "ops"] |
|
|
| for metric in metrics: |
| plt.figure(figsize=(10, 6)) |
| for func_name in df["func"].unique(): |
| data_by_func = df[df["func"] == func_name].sort_values("size") |
| plt.plot(data_by_func["size"], data_by_func[metric], label=func_name, marker='o') |
| plt.title(f"{metric.upper()} vs Size (Full Range)") |
| plt.xlabel("Input Size") |
| plt.ylabel(metric) |
| plt.legend() |
| plt.grid(True) |
| plt.xscale("log") |
| if metric != "stddev": |
| plt.yscale("log") |
| plt.tight_layout() |
| plt.savefig(f"{self.benchmarks_plot_dir}/benchmark_{metric}_full.png") |
| plt.close() |
|
|
| plt.figure(figsize=(10, 6)) |
| for func_name in df["func"].unique(): |
| data_by_func = df[(df["func"] == func_name) & (df["size"] > 1e5)].sort_values("size") |
| if not data_by_func.empty: |
| plt.plot(data_by_func["size"], data_by_func[metric], label=func_name, marker='o') |
| plt.title(f"{metric.upper()} vs Size (Large Inputs Only)") |
| plt.xlabel("Input Size (>1e5)") |
| plt.ylabel(metric) |
| plt.legend() |
| plt.grid(True) |
| plt.xscale("log") |
| if metric != "stddev": |
| plt.yscale("log") |
| plt.tight_layout() |
| plt.savefig(f"{self.benchmarks_plot_dir}/benchmark_{metric}_large.png") |
| plt.close() |
|
|
| def acceleration_plot(self, df: pd.DataFrame) -> None: |
| """ |
| Plots the acceleration of the "Purem" implementation relative to other functions/libraries |
| based on their performance for different input sizes. The plot shows acceleration as a |
| function of the input size as a logarithmic graph, detailing the relative performance |
| gain or loss. |
| |
| This function is intended to visualize and compare performance metrics for computational |
| functions with respect to a "Purem" baseline. It ensures that shared input sizes |
| are used for comparison and applies smoothing (Savitzky-Golay filter) to the acceleration |
| curve for better interpretability. |
| |
| :param df: A pandas DataFrame containing benchmarking results with the following |
| columns: |
| - "func": The name of the function or library being benchmarked. |
| - "size": The size of the input data for the benchmark. |
| - "ops": The respective operations per second or performance metric for each |
| function. |
| |
| :return: None |
| """ |
| func_names = df["func"].unique() |
|
|
| assert any("Purem" in name for name in func_names), "Purem is missed!" |
|
|
| df_purem = df[df["func"].str.contains("Purem")].sort_values("size") |
|
|
| plt.figure(figsize=(14, 8)) |
|
|
| for func_name in func_names: |
| if "Purem" in func_name: |
| continue |
|
|
| df_other = df[df["func"] == func_name].sort_values("size") |
| common_sizes = np.intersect1d(df_purem["size"], df_other["size"]) |
|
|
| if len(common_sizes) == 0: |
| continue |
|
|
| purem_ops = df_purem[df_purem["size"].isin(common_sizes)]["ops"].values |
| other_ops = df_other[df_other["size"].isin(common_sizes)]["ops"].values |
|
|
| acceleration = purem_ops / other_ops |
|
|
| try: |
| if len(acceleration) >= 5: |
| acceleration_smooth = savgol_filter(acceleration, window_length=5, polyorder=2) |
| else: |
| acceleration_smooth = acceleration |
| except Exception as e: |
| warnings.warn(f"Smoothing skipped: {e}") |
| acceleration_smooth = acceleration |
|
|
| plt.plot(common_sizes, acceleration_smooth, marker='o', label=f"Purem vs {func_name.partition(':')[2].strip()}") |
|
|
| plt.xscale('log') |
| plt.yscale('log') |
| plt.axhline(y=1, color='black', linestyle='--', linewidth=1, label='Equal Performance (x1)') |
| plt.xlabel('Input Size', fontsize=14) |
| plt.ylabel('Acceleration (Purem / Other)', fontsize=14) |
| plt.title('Purem Acceleration Relative to Other Libraries (Production Level)', fontsize=16) |
| plt.grid(True, which="both", linestyle='--', linewidth=0.6) |
| plt.legend(fontsize=12) |
| plt.tight_layout() |
| plt.savefig(f"{self.benchmarks_plot_dir}/benchmark_acceleration_large.png") |
| plt.close() |
|
|
| def markdown_data_frame(self) -> pd.DataFrame: |
| """ |
| Converts benchmark data from files into a structured pandas DataFrame. |
| |
| This method processes a collection of JSON files that contain benchmark |
| results. It reads the files, extracts relevant statistical information |
| about the benchmarking process, and formats this data into a structured |
| pandas DataFrame. Each record in the resultant DataFrame represents a |
| specific benchmark test and includes its relevant metadata and statistical |
| metrics. |
| |
| :return: |
| A pandas DataFrame where each row corresponds to a benchmark with |
| columns that include details such as 'Elements', 'Function', 'OPS', |
| 'Min Time (s)', 'Max Time (s)', 'Mean Time (s)', and 'Std Dev'. |
| :rtype: |
| pd.DataFrame |
| """ |
| files = self.load_benchmarks_files() |
|
|
| records_markdown = [] |
|
|
| for filepath in files: |
| with open(filepath, "r") as f: |
| data = json.load(f) |
|
|
| filepath_split = filepath.split("_") |
|
|
| for b in data['benchmarks']: |
| size = self.extract_size_from_filepath(filepath) |
| func_name = b['params']['func_name'] |
| stats = b['stats'] |
| records_markdown.append({ |
| "Elements": size, |
| "Function": func_name, |
| "OPS": stats["ops"], |
| "Min Time (s)": stats["min"], |
| "Max Time (s)": stats["max"], |
| "Mean Time (s)": stats["mean"], |
| "Std Dev": stats["stddev"] |
| }) |
|
|
| return pd.DataFrame(records_markdown) |
|
|
| def markdown_save(self, df: pd.DataFrame) -> None: |
| """ |
| Saves a markdown file summarizing benchmark results in a human-readable format. The function processes |
| a pandas DataFrame containing benchmark data, formats specific columns for scientific notation or fixed-point |
| notation, and groups the results by the "Elements" column. Each group is written as a separate section |
| in the markdown file, where the file is saved to the benchmarks directory. |
| |
| :param df: A pandas DataFrame containing benchmark results. Expected columns include "Elements", |
| "OPS", "Min Time (s)", "Max Time (s)", "Mean Time (s)", and "Std Dev". |
| :return: None |
| """ |
| df = df.sort_values(by=["Elements", "OPS"], ascending=[True, False]) |
|
|
| df["Min Time (s)"] = df["Min Time (s)"].map(lambda x: f"{x:.2e}") |
| df["Max Time (s)"] = df["Max Time (s)"].map(lambda x: f"{x:.2e}") |
| df["Mean Time (s)"] = df["Mean Time (s)"].map(lambda x: f"{x:.2e}") |
| df["Std Dev"] = df["Std Dev"].map(lambda x: f"{x:.2e}") |
| df["OPS"] = df["OPS"].map(lambda x: f"{x:.2f}") |
|
|
| grouped = df.groupby('Elements') |
|
|
| with open(f"{self.benchmarks_plot_dir}/benchmarks_table.md", "w") as f: |
| f.write("# Benchmark Results\n\n") |
| for size, group in grouped: |
| f.write(f"### Elements: {size}\n\n") |
| f.write(group.drop(columns=["Elements"]).to_markdown(index=False)) |
| f.write("\n\n") |
|
|
