避免 Numba 中的竞争条件

Question

这是一个玩具 njit 函数，它接受一个距离矩阵，循环遍历矩阵的每一行，并记录每列中的最小值以及最小值来自哪一行。但是，IIUC，使用 prange，这可能会导致竞争条件（尤其是对于较大的输入数组）：

from numba import njit, prange
import numpy as np

@njit
def some_transformation_func(D, row_i):
   """
   This function applies some transformation to the ith row (`row_i`) in the `D` matrix in place. 
   However, the transformation time is random (but all less than a second), which means 
   that the rows can take
   """

    # Apply some inplace transformation on the ith row of D


@njit(parallel=True)
def some_func(D):
    P = np.empty((D.shape[1]))
    I = np.empty((D.shape[1]), np.int64)
    P[:] = np.inf
    I[:] = -1

    for row in prange(D.shape[0]):
        some_transformation_func(D, row)
        for col in range(D.shape[1]):
            if P[col] > D[row, col]:
                P[col] = D[row, col]
                I[col] = row

    return P, I

if __name__ == "__main__":
    D = np.array([[4,1,6,9,9], 
                  [1,3,8,2,7], 
                  [2,8,0,0,1],
                  [3,7,4,6,5]
                 ])
    P, I = some_func(D)
    print(P)
    print(I)

    # [1. 1. 0. 0. 1.]
    # [1 0 2 2 2]

我如何确认是否存在竞争条件（特别是如果 D 非常大并且有更多的行和列）？而且，更重要的是，如果存在竞争条件，我该如何避免呢？

Answer 1

在这些情况下，不要将 prange 设置为数组的大小，最好的办法是手动将数据分成 n_threads 个块，然后分配进行相应处理，最后进行归约。所以，像这样：

from numba import njit, prange, config
import numpy as np

@njit
def wrapper_func(thread_idx, start_indices, stop_indices, D, P, I):
    for row in range(start_indices[thread_idx], stop_indices[thread_idx]):
        some_transformation_func(D, row)
        for col in range(D.shape[1]):
            if P[thread_idx, col] > D[row, col]:
                P[thread_idx, col] = D[row, col]
                I[thread_idx, col] = row


@njit
def some_transformation_func(D, row_i):
   """
   This function applies some transformation to the ith row (`row_i`) in the `D` matrix in place. 
   However, the transformation time is random (but all less than a second), which means 
   that the rows can take
   """

    # Apply some inplace transformation on the ith row of D


@njit(parallel=True)
def some_func(D):
    n_threads = config.NUMBA_NUM_THREADS  # Let's assume that there are 2 threads
    P = np.empty((n_threads, D.shape[1]))
    I = np.empty((n_threads, D.shape[1]), np.int64)
    P[:, :] = np.inf
    I[:, :] = -1

    start_indices = np.array([0, 2], np.int64)
    stop_indices = np.array([2, 4], np.int64)  # Note that these are exclusive

    for thread_idx in prange(n_threads):
        wrapper_func(thread_idx, start_indices, stop_indices, D, P, I)

    # Perform reduction from all threads and store results in P[0]
    for thread_idx in range(1, n_threads):
        for i in prange(l):
            if P[0, i] > P[thread_idx, i]:
                P[0, i] = P[thread_idx, i]
                I[0, i] = I[thread_idx, i]

    return P[0], I[0]

if __name__ == "__main__":
    D = np.array([[4,1,6,9,9], 
                  [1,3,8,2,7], 
                  [2,8,0,0,1],
                  [3,7,4,6,5]
                 ])
    P, I = some_func(D)
    print(P)
    print(I)

    # [1. 1. 0. 0. 1.]
    # [1 0 2 2 2]

请注意，这将花费您更多的内存（恰好 n_threads 更多的内存），但您可以从并行化中获益。此外，代码变得更清晰且更易于维护。需要做的是找出最好的方法来分块数据并确定 start_row 和 stop_row（独占）索引。

避免 Numba 中的竞争条件

Avoid Race Condition in Numba

python

race-condition

numba