如何使 numba @jit 使用所有 cpu 内核（并行化 numba @jit）

Posted 2023-02-16

【中文标题】如何使 numba @jit 使用所有 cpu 内核（并行化 numba @jit）

【英文标题】：How to make numba @jit use all cpu cores (parallelize numba @jit)

【发布时间】：2018-01-18 12:01:37

【问题描述】：

我正在使用 numbas @jit 装饰器在 python 中添加两个 numpy 数组。如果我使用@jit 与python 相比，性能是如此之高。

但是，即使我传入@numba.jit(nopython = True, parallel = True, nogil = True)，它也没有利用所有 CPU 内核。

有什么方法可以使用 numba @jit 的所有 CPU 内核。

这是我的代码：

import time                                                
import numpy as np                                         
import numba                                               

SIZE = 2147483648 * 6                                      

a = np.full(SIZE, 1, dtype = np.int32)                     

b = np.full(SIZE, 1, dtype = np.int32)                     

c = np.ndarray(SIZE, dtype = np.int32)                     

@numba.jit(nopython = True, parallel = True, nogil = True) 
def add(a, b, c):                                          
    for i in range(SIZE):                                  
        c[i] = a[i] + b[i]                                 

start = time.time()                                        
add(a, b, c)                                               
end = time.time()                                          

print(end - start)                                        

【问题讨论】：

您的示例不使用prange - 我找到了您的问题，因为我想知道为什么prange 不能并行运行，但我在@jit 中缺少parallel = True.. .

【参考方案1】：

您可以将parallel=True 传递给任何 numba jitted 函数，但这并不意味着它总是使用所有内核。您必须了解 numba 使用一些启发式方法来使代码并行执行，有时这些启发式方法根本无法在代码中找到要并行化的任何内容。当前有一个pull request，因此如果无法使其“并行”，它会发出警告。所以它更像是一个“请尽可能使其并行执行”参数而不是“强制并行执行”。

但是，如果您确实知道可以并行化代码，则始终可以手动使用线程或进程。只是适应example of using multi-threading from the numba docs：

#!/usr/bin/env python
from __future__ import print_function, division, absolute_import

import math
import threading
from timeit import repeat

import numpy as np
from numba import jit

nthreads = 4
size = 10**7  # CHANGED

# CHANGED
def func_np(a, b):
    """
    Control function using Numpy.
    """
    return a + b

# CHANGED
@jit('void(double[:], double[:], double[:])', nopython=True, nogil=True)
def inner_func_nb(result, a, b):
    """
    Function under test.
    """
    for i in range(len(result)):
        result[i] = a[i] + b[i]

def timefunc(correct, s, func, *args, **kwargs):
    """
    Benchmark *func* and print out its runtime.
    """
    print(s.ljust(20), end=" ")
    # Make sure the function is compiled before we start the benchmark
    res = func(*args, **kwargs)
    if correct is not None:
        assert np.allclose(res, correct), (res, correct)
    # time it
    print(':>5.0f ms'.format(min(repeat(lambda: func(*args, **kwargs),
                                          number=5, repeat=2)) * 1000))
    return res

def make_singlethread(inner_func):
    """
    Run the given function inside a single thread.
    """
    def func(*args):
        length = len(args[0])
        result = np.empty(length, dtype=np.float64)
        inner_func(result, *args)
        return result
    return func

def make_multithread(inner_func, numthreads):
    """
    Run the given function inside *numthreads* threads, splitting its
    arguments into equal-sized chunks.
    """
    def func_mt(*args):
        length = len(args[0])
        result = np.empty(length, dtype=np.float64)
        args = (result,) + args
        chunklen = (length + numthreads - 1) // numthreads
        # Create argument tuples for each input chunk
        chunks = [[arg[i * chunklen:(i + 1) * chunklen] for arg in args]
                  for i in range(numthreads)]
        # Spawn one thread per chunk
        threads = [threading.Thread(target=inner_func, args=chunk)
                   for chunk in chunks]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()
        return result
    return func_mt


func_nb = make_singlethread(inner_func_nb)
func_nb_mt = make_multithread(inner_func_nb, nthreads)

a = np.random.rand(size)
b = np.random.rand(size)

correct = timefunc(None, "numpy (1 thread)", func_np, a, b)
timefunc(correct, "numba (1 thread)", func_nb, a, b)
timefunc(correct, "numba (%d threads)" % nthreads, func_nb_mt, a, b)

我突出显示了我更改的部分，其他所有内容都是从示例中逐字复制的。这利用了我机器上的所有内核（4 核机器因此 4 个线程），但没有显示出显着的加速：

numpy (1 thread)       539 ms
numba (1 thread)       536 ms
numba (4 threads)      442 ms

在这种情况下，多线程缺乏（很多）加速是因为加法是一种带宽受限的操作。这意味着从数组中加载元素并将结果放入结果数组中要比实际加法花费更多的时间。

在这些情况下，您甚至会因为并行执行而看到速度变慢！

只有当函数更复杂并且与加载和存储数组元素相比实际操作需要大量时间时，您才会看到并行执行的巨大改进。 numba 文档中的示例是这样的：

def func_np(a, b):
    """
    Control function using Numpy.
    """
    return np.exp(2.1 * a + 3.2 * b)

@jit('void(double[:], double[:], double[:])', nopython=True, nogil=True)
def inner_func_nb(result, a, b):
    """
    Function under test.
    """
    for i in range(len(result)):
        result[i] = math.exp(2.1 * a[i] + 3.2 * b[i])

这实际上（几乎）与线程数成比例，因为两次乘法、一次加法和一次调用math.exp 比加载和存储结果要慢得多：

func_nb = make_singlethread(inner_func_nb)
func_nb_mt2 = make_multithread(inner_func_nb, 2)
func_nb_mt3 = make_multithread(inner_func_nb, 3)
func_nb_mt4 = make_multithread(inner_func_nb, 4)

a = np.random.rand(size)
b = np.random.rand(size)

correct = timefunc(None, "numpy (1 thread)", func_np, a, b)
timefunc(correct, "numba (1 thread)", func_nb, a, b)
timefunc(correct, "numba (2 threads)", func_nb_mt2, a, b)
timefunc(correct, "numba (3 threads)", func_nb_mt3, a, b)
timefunc(correct, "numba (4 threads)", func_nb_mt4, a, b)

结果：

numpy (1 thread)      3422 ms
numba (1 thread)      2959 ms
numba (2 threads)     1555 ms
numba (3 threads)     1080 ms
numba (4 threads)      797 ms

【参考方案2】：

为了完整起见，在 2018 年（numba v 0.39）你可以这样做

from numba import prange

并在您的原始函数定义中将range 替换为prange，就是这样。

这立即使 CPU 利用率达到 100%，在我的情况下，运行时间从 2.9 秒加快到 1.7 秒（对于 SIZE = 2147483648 * 1，在 16 核 32 线程的机器上）。

更复杂的内核通常可以通过传入fastmath=True 来加快速度。