Python - 使用 joblib 进行循环并行化

Question

我需要一些帮助来准确理解我做了什么/为什么我的代码没有按我预期的那样运行。

我已经开始使用 joblib 来尝试通过并行运行(大)循环来加速我的代码。

我是这样使用它的:

from joblib import Parallel, delayed
def frame(indeces, image_pad, m):

    XY_Patches = np.float32(image_pad[indeces[0]:indeces[0]+m, indeces[1]:indeces[1]+m,  indeces[2]])
    XZ_Patches = np.float32(image_pad[indeces[0]:indeces[0]+m, indeces[1],                  indeces[2]:indeces[2]+m])
    YZ_Patches = np.float32(image_pad[indeces[0],                 indeces[1]:indeces[1]+m,  indeces[2]:indeces[2]+m])

    return XY_Patches, XZ_Patches, YZ_Patches


def Patch_triplanar_para(image_path, patch_size):

    Image, Label, indeces =  Sampling(image_path)

    n = (patch_size -1)/2
    m = patch_size

    image_pad = np.pad(Image, pad_width=n, mode='constant', constant_values = 0)

    A = Parallel(n_jobs= 1)(delayed(frame)(i, image_pad, m) for i in indeces)
    A = np.array(A)
    Label = np.float32(Label.reshape(len(Label), 1))
    R, T, Y =  np.hsplit(A, 3)

    return R, T, Y, Label

我一直在试验“n_jobs”，期望增加它会加快我的功能。但是，当我增加 n_jobs 时，速度会显着降低。在没有“并行”的情况下运行此代码时，事情会变慢，直到我将作业数从 1 增加。

为什么会这样？我知道我运行的作业越多，脚本越快？我用错了吗？

谢谢!

Answer 1

可能你的问题是因为image_pad是一个大数组。在您的代码中，您正在使用 joblib 的默认 multiprocessing 后端。这个后端创建了一个工作池，每个工作池都是一个 Python 进程。然后将函数的输入数据复制 n_jobs 次并广播给池中的每个工作人员，这会导致严重的开销。引自 joblib的文档:

By default the workers of the pool are real Python processes forked using the multiprocessing module of the Python standard library when n_jobs != 1. The arguments passed as input to the Parallel call are serialized and reallocated in the memory of each worker process.

This can be problematic for large arguments as they will be reallocated n_jobs times by the workers.

As this problem can often occur in scientific computing with numpy based datastructures, joblib.Parallel provides a special handling for large arrays to automatically dump them on the filesystem and pass a reference to the worker to open them as memory map on that file using the numpy.memmap subclass of numpy.ndarray. This makes it possible to share a segment of data between all the worker processes.

Note: The following only applies with the default "multiprocessing" backend. If your code can release the GIL, then using backend="threading" is even more efficient.

所以如果是这种情况，你应该切换到线程后端，如果你能够在调用 frame 时释放全局解释器锁，或者切换到 的共享内存方法作业库.

docs假设 joblib 提供了一个可能有用的自动 memmap 转换。