Microbench

Microbench是一个用于基准测试Python函数的小型Python包，可选地捕获额外的运行时/环境信息。它在集群/分布式环境中非常有用，其中相同的函数在不同的环境中运行，并且设计为可扩展以添加新功能。除了基准测试外，这还可以通过记录关键Python包的版本或甚至全局环境中加载的所有包来帮助重现性。其他捕获的元数据可以包括CPU和RAM使用情况、环境变量和硬件规格。

要求

Microbench默认没有Python标准库之外的依赖，尽管建议使用pandas来检查结果。但是，某些混入（扩展）有特定的要求

• 需要安装line_profiler包来进行逐行代码基准测试。
• MBInstalledPackages需要setuptools，这不是标准库的一部分，但通常可用。
• 核心CPU、总RAM和遥测扩展需要psutil。
• NVIDIA GPU插件需要nvidia-smi实用工具，该工具通常随NVIDIA显卡驱动程序一起提供。它需要在您的PATH中。

安装

要使用pip安装

pip install microbench

用法

Microbench是为基准测试Python函数而设计的。以下示例将假设您已经定义了一个名为myfunction的Python函数，您希望对其进行基准测试

def myfunction(arg1, arg2, ...):
    ...

最小示例

首先，创建一个基准测试套件，指定配置和要捕获的信息。

以下是一个最小、完整的示例

from microbench import MicroBench

basic_bench = MicroBench()

要将基准测试附加到您的函数上，只需使用装饰器basic_bench，如下所示：

@basic_bench
def myfunction(arg1, arg2, ...):
    ...

就这么多！当调用myfunction()时，元数据将被捕获到io.StringIO()缓冲区中，可以按照以下方式读取（使用pandas库）：

import pandas as pd
results = pd.read_json(basic_bench.outfile, lines=True)

上述示例捕获了字段start_time（开始时间）、finish_time（结束时间）、run_durations（每个函数调用，默认为秒），function_name（函数名称）、timestamp_tz（时区名称，参见此README中的时区部分），以及duration_counter（用于计算持续时间的函数名称，参见此README中的持续时间计时部分）。

微基准测试可以捕获来自环境、资源使用和硬件的许多其他类型的元数据，以下将介绍。

扩展示例

以下是一个使用混入（以MB前缀的类名）扩展功能更完整的示例。请注意，可以向构造函数（在此情况下some_info=123）提供关键字参数来指定要捕获的附加信息。我们还将指定iterations=3，这意味着将被调用的函数将执行3次（返回的结果始终是最后一次运行的结果）并且会捕获每次运行的计时。我们指定了自定义持续时间计数器，time.monotonic而不是默认的time.perf_counter（有关说明，请参见此README后面的持续时间计时部分）。此示例还指定了outfile选项，该选项将元数据追加到磁盘上的文件。

from microbench import *
import numpy, pandas, time

class MyBench(MicroBench, MBFunctionCall, MBPythonVersion, MBHostInfo):
    outfile = '/home/user/my-benchmarks'
    capture_versions = (numpy, pandas)  # Or use MBGlobalPackages/MBInstalledPackages
    env_vars = ('SLURM_ARRAY_TASK_ID', )

benchmark = MyBench(some_info=123, iterations=3, duration_counter=time.monotonic)

上述示例中的env_vars选项指定了一个要捕获为env_<变量名称>的环境变量列表。在此示例中，slurm数组任务ID将被存储为env_SLURM_ARRAY_TASK_ID。如果未设置环境变量，则值将为null。

要捕获软件包版本，您可以直接指定它们（如上所示），或者捕获全局环境中每个软件包的版本。在以下示例中，我们将自动捕获microbench、numpy和pandas的版本。

from microbench import *
import numpy, pandas

class Bench2(MicroBench, MBGlobalPackages):
    outfile = '/home/user/bench2'

bench2 = Bench2()

如果您想更进一步，并捕获每个可导入软件包的版本，有一个混入可以实现这一点

from microbench import *

class Bench3(MicroBench, MBInstalledPackages):
    pass

bench3 = Bench3()

检查结果

每个结果都是一个JSON对象。当使用outfile选项时，每个@benchmark调用的JSON对象都会存储在文件的单独一行。上述最小示例的输出（针对单次运行）将类似于以下内容

{"start_time": "2018-08-06T10:28:24.806493+00:00", "finish_time": "2018-08-06T10:28:24.867456+00:00", "run_durations": [0.60857599999999999], "function_name": "my_function", "timestamp_tz": "UTC", "duration_counter": "perf_counter"}

开始和结束时间以ISO-8601格式的时间戳给出，默认为UTC时区（请参见此README中的时区部分）。

运行持续时间以秒给出，默认使用time.perf_counter函数捕获，但可以覆盖（请参见此README后面的持续时间计时部分）。

最简单的方法是详细检查结果，就是将它们加载到pandas数据框中。

# Read results directly from active benchmark suite
benchmark.get_results()

# Or, equivalently when using a file, read it using pandas directly
import pandas
results = pandas.read_json('/home/user/my-benchmarks', lines=True)

Pandas具有强大的数据处理能力。例如，可以通过Python版本计算平均运行时间。

# Calculate overall runtime
results['runtime'] = results['finish_time'] - results['start_time']

# Average overall runtime by Python version
results.groupby('python_version')['runtime'].mean()

还有更多高级操作可用。建议阅读pandas教程。

行分析器支持

Microbench还支持line_profiler，该工具显示Python代码每一行的执行时间。注意，这将减慢你的代码速度，所以仅在需要时使用它，但它在发现函数中的瓶颈时很有用。需要安装line_profiler包（例如：pip install line_profiler）。

from microbench import MicroBench, MBLineProfiler
import pandas

# Create our benchmark suite using the MBLineProfiler mixin
class LineProfilerBench(MicroBench, MBLineProfiler):
    pass

lpbench = LineProfilerBench()

# Decorate our function with the benchmark suite
@lpbench
def my_function():
    """ Inefficient function for line profiler """
    acc = 0
    for i in range(1000000):
        acc += i

    return acc

# Call the function as normal
my_function()

# Read the results into a Pandas DataFrame
results = lpbench.get_results()

# Get the line profiler report as an object
lp = MBLineProfiler.decode_line_profile(results['line_profiler'][0])

# Print the line profiler report
MBLineProfiler.print_line_profile(results['line_profiler'][0])

上一个示例的最后一条语句将打印行分析器报告，显示每一行代码的执行时间。示例

Timer unit: 1e-06 s

Total time: 0.476723 s
File: /home/user/my_test.py
Function: my_function at line 12

Line #      Hits         Time  Per Hit   % Time  Line Contents
==============================================================
    12                                               @lpbench
    13                                               def my_function():
    14                                                   """ Inefficient function for line profiler """
    15         1          2.0      2.0      0.0          acc = 0
    16   1000001     217874.0      0.2     45.7          for i in range(1000000):
    17   1000000     258846.0      0.3     54.3              acc += i
    18
    19         1          1.0      1.0      0.0          return acc

NVIDIA GPU支持

可以使用MBNvidiaSmi插件捕获有关NVIDIA GPU的属性。这需要当前PATH中可用nvidia-smi实用程序。

默认情况下，会捕获gpu_name（型号号）和memory.total属性。可以使用类或对象级别的变量nvidia_attributes指定额外属性。要查看可用的属性，请运行nvidia-smi --help-query-gpu。

默认情况下，将轮询所有安装的GPU。要限制特定GPU，请指定nvidia_gpus属性为GPU ID的元组，可以是基于零的GPU索引（重启之间可能更改，不推荐），GPU UUID或PCI总线ID。您可以通过运行nvidia-smi -L来找到GPU UUID。

下面是一个指定可选的nvidia_attributes和nvidia_gpus字段的示例

from microbench import MicroBench, MBNvidiaSmi

class GpuBench(MicroBench, MBNvidiaSmi):
    outfile = '/home/user/gpu-benchmarks'
    nvidia_attributes = ('gpu_name', 'memory.total', 'pcie.link.width.max')
    nvidia_gpus = (0, )  # Usually better to specify GPU UUIDs here instead

gpu_bench = GpuBench()

遥测支持

我们使用“遥测”一词来指函数由并行运行的线程定期执行期间捕获的元数据。例如，这可以用来查看内存使用随时间的变化。

遥测支持需要psutil库。

Microbench会自动启动和清理监控线程。最终用户只需要定义一个telemetry静态方法，该方法接受一个psutil.Process对象，并返回作为字典的遥测数据。

默认遥测收集间隔是每60秒，如果需要，可以使用telemetry_interval类变量进行自定义。

下面是一个捕获每90秒内存使用的最小示例

from microbench import MicroBench

class TelemBench(MicroBench):
    telemetry_interval = 90

    @staticmethod
    def telemetry(process):
        return process.memory_full_info()._asdict()

telem_bench = TelemBench()

扩展microbench

Microbench包含一些混合功能，如上述扩展示例中所述。

您还可以向基准测试套件添加函数，在运行时捕获额外信息。这些函数必须以capture_前缀开头，以便在函数开始之前自动运行，或者以capturepost_前缀开头，以便在函数完成时自动运行。它们接受单个参数，bm_data，这是一个要扩展额外数据的字典。应小心避免覆盖现有的键名。

以下是一个捕获机器类型（例如：i386，x86_64等）的示例

from microbench import MicroBench
import platform

class Bench(MicroBench):
    outfile = '/home/user/my-benchmarks'

    def capture_machine_platform(self, bm_data):
        bm_data['platform'] = platform.machine()

benchmark = Bench()

扩展JSONEncoder

Microbench将数据编码为JSON，但有时Microbench会遇到默认情况下无法编码为JSON的数据类型（例如自定义对象或类）。例如，在使用MBFunctionCall和MBReturnValue时，如果任何参数或返回值（分别）无法编码为JSON，将显示警告，并将值替换为占位符以允许元数据捕获继续，并显示警告。

如果您希望实际捕获这些值，您需要指定一种将对象转换为JSON的方法。这可以通过扩展microbench.JSONEncoder并实现对象类型的测试以及字符串、列表或字典的转换来完成。

例如，要使用str(graph)作为表示来捕获来自igraph包的Graph对象，我们可以执行以下操作（注意，我们可以使用我们想要的任何表示，例如，如果我们想以更详细或更简略的方式捕获对象）

import microbench as mb
from igraph import Graph

# Extend the JSONEncoder to encode Graph objects
class CustomJSONEncoder(mb.JSONEncoder):
    def default(self, o):
        # Encode igraph.Graph objects as strings
        if isinstance(o, Graph):
            return str(o)

        # Add further isinstance(o, ...) cases here
        # if needed

        # Make sure to call super() to handle
        # default cases
        return super().default(o)

# Define your benchmark class as normal
class Bench(mb.MicroBench, mb.MBReturnValue):
    pass

# Create a benchmark suite with the custom JSON
# encoder from above
bench = Bench(json_encoder=CustomJSONEncoder)

# Attach the benchmark suite to our function
@bench
def return_a_graph():
    return Graph(2, ((0, 1), (0, 2)))

# This should now work without warnings or errors
return_a_graph()

Redis支持

默认情况下，microbench将输出追加到文件中，但输出可以指向其他地方，例如redis -一个内存中的、网络化的数据源。当共享文件系统不可用的情况下，此选项非常有用。

Redis支持需要redis-py。

要使用此功能，请从MicroBench继承而不是从MicroBenchRedis继承，并指定redis连接和键名，如下面的示例所示

from microbench import MicroBenchRedis

class RedisBench(MicroBenchRedis):
    # redis_connection contains arguments for redis.StrictClient()
    redis_connection = {'host': 'localhost', 'port': 6379}
    redis_key = 'microbench:mykey'

benchmark = RedisBench()

要检索结果，可以直接使用redis包

import redis
import pandas

# Establish the connection to redis
rconn = redis.StrictRedis(host=..., port=...)

# Read the redis data from 'myrediskey' into a list of byte arrays
redis_data = redis.lrange('myrediskey', 0, -1)

# Convert the list into a single string
json_data = '\n'.join(r.decode('utf8') for r in redis_data)

# Read the string into a pandas dataframe
results = pandas.read_json(json_data, lines=True)

运行时影响考虑因素

运行时影响取决于捕获的信息和平台。一般来说，捕获环境变量、Python包版本和函数的计时信息的影响可以忽略不计。捕获遥测和调用外部程序（如用于GPU信息的nvidia-smi）影响较大，尽管后者是每次调用的一个操作，并且通常不到一秒。遥测捕获间隔应相对频繁（例如，每分钟或两分钟，而不是每秒）以避免显著的影响。

持续时间计时

默认情况下，run_durations使用time.perf_counter函数以秒为单位给出，这对于大多数用例应该足够。您可以使用返回浮点数或整数的任何函数来获取时间。一些示例包括，如果您想以纳秒为单位获取时间，可以使用time.perf_counter_ns，或者如果您需要一个不受时钟调整影响的单调时钟（对于非常长时间运行的代码），可以使用time.monotonic。在创建基准测试套件对象时使用duration_counter=...参数，如本README的扩展示例部分所示。

时区

Microbench默认以UTC时区捕获start_time和finish_time。这可以通过在创建基准类时传递tz=...参数来覆盖，其中值是一个时区对象（例如，使用pytz库创建）。

反馈

请注意，这是一个新创建的、实验性的包。请通过GitHub问题告诉我您的反馈或功能请求。