python - 如何在 python/numpy 中自动化 BLAS 的环境变量相关基准测试？

Question

我需要一些帮助来弄清楚如何在 python 中自动化基准测试。

我正在通过 python 中的 numpy 测试线程对 BLAS 库调用的影响。在 linux 环境中，OpenBLAS 中的线程是通过环境变量 OMP_NUM_THREADS 控制的。我想做一个测试，我将 OMP_NUM_THREADS 从 1 增加到最大值，在每个线程计数处为例程计时，然后最终操纵所有线程计数的总计时。

问题如下。环境变量可以在 python 中设置，但它们只影响子进程或子 shell。所以我可以使用以下驱动程序代码正确运行我的基准测试:

#!/usr/bin/env python                                                                                                     # driver script for thread test
import os

thread_set =[1,2,4,8,16]
for thread in thread_set:

    os.environ['OMP_NUM_THREADS']='{:d}'.format(thread)
    os.system("echo $OMP_NUM_THREADS")
    os.system("numpy_test")

和 numpy_test 脚本:

#!/usr/bin/env python
#timing test for numpy dot product (using OpenBLAS)                                                      
#based on http://stackoverflow.com/questions/11443302/compiling-numpy-with-openblas-integration
import sys
import timeit

setup = "import numpy; x = numpy.random.random((1000,1000))"
count = 5

t = timeit.Timer("numpy.dot(x, x.T)", setup=setup)
dot_time = t.timeit(count)/count
print("dot: {:7.3g} sec".format(dot_time))

但分析这是一个非常手动的过程。

特别是，我无法将 numpy_test 中的值 dot_time 返回到我的外包装例程，因此我无法以任何方式分析我的测试结果自动化时尚。例如，我想绘制 dot_time 与线程数的关系图，或者评估 dot_time/线程数是否恒定。

如果我尝试通过定义一个 python 测试函数(避免上面的 os.system() 方法)完全在 python 实例中进行类似的测试，然后在 thread in thread_set 循环，然后测试函数的所有实例都继承相同的 OMP_NUM_THREADS 值(父 python shell 的值)。所以这个测试失败了:

#!/usr/bin/env python
#attempt at testing threads that doesn't work
#(always uses inherited value of OMP_NUM_THREADS)
import os

import sys
import timeit

def test_numpy():
    setup = "import numpy; x = numpy.random.random((1000,1000))"
    count = 5

    t = timeit.Timer("numpy.dot(x, x.T)", setup=setup)
    dot_time = t.timeit(count)/count
    print("dot: {:7.3g} sec".format(dot_time))
    return dot_time

thread_set =[1,2,4,8,16]
for thread in thread_set:
    os.environ['OMP_NUM_THREADS']='{:d}'.format(thread)
    os.system("echo $OMP_NUM_THREADS")
    time_to_run = test_numpy()
    print(time_to_run)

失败的原因在于 thread 的每个实例都需要相同的时间，因为 test_numpy() 总是继承父级中 OMP_NUM_THREADS 的值环境，而不是通过 os.environ() 设置的值。但是，如果这样的事情可行，那么进行我需要做的分析将是微不足道的。

在实际测试中，我将运行超过 1000 个排列，因此自动化是关键。鉴于此，如果您能回答以下任何问题，我将不胜感激:

1. 如何从这样的子进程返回一个值 (dot_time)？有没有比读/写文件更优雅的解决方案？

2. 是否有更好的方法来构建这种(依赖于环境变量的)测试？

提前谢谢你。

Answer 1

你可以这样做:

import subprocess

os.environ['OMP_NUM_THREADS'] = '{:d}'.format(thread)
proc = subprocess.Popen(["numpy_test"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout, stderr = proc.communicate()

然后您将在标准输出中获得 numpy_test 脚本的输出。一般来说，我认为 subprocess.call 和 subprocess.Popen 优于 os.system。