与 multiprocessing.Pool 共享计数器

我想用multiprocessing.Value + multiprocessing.Lock在不同的进程之间共享一个计数器。例如：

import itertools as it
import multiprocessing

def func(x, val, lock):
    for i in range(x):
        i ** 2
    with lock:
        val.value += 1
        print('counter incremented to:', val.value)

if __name__ == '__main__':
    v = multiprocessing.Value('i', 0)
    lock = multiprocessing.Lock()

    with multiprocessing.Pool() as pool:
        pool.starmap(func, ((i, v, lock) for i in range(25)))
    print(counter.value())

这将引发以下异常：

RuntimeError：同步对象只能在之间共享 通过继承进行处理

我最困惑的是，相关的（尽管不完全相似）模式与multiprocessing.Process():

if __name__ == '__main__':
    v = multiprocessing.Value('i', 0)
    lock = multiprocessing.Lock()

    procs = [multiprocessing.Process(target=func, args=(i, v, lock))
             for i in range(25)]
    for p in procs: p.start()
    for p in procs: p.join()

现在，我认识到这是两件明显不同的事情：

• 第一个示例使用的工作进程数等于cpu_count()，并分割一个可迭代的range(25)它们之间
• 第二个示例创建 25 个工作进程和任务，每个进程和任务都有一个输入

也就是说：我如何与以下人员共享实例pool.starmap() (or pool.map()） 以这种方式？

我见过类似的问题here, here, and here，但这些方法似乎不适合.map()/.starmap()， 不管是否Value uses ctypes.c_int.

我意识到这种方法在技术上是有效的：

def func(x):
    for i in range(x):
        i ** 2
    with lock:
        v.value += 1
        print('counter incremented to:', v.value)

v = None
lock = None

def set_global_counter_and_lock():
    """Egh ... """
    global v, lock
    if not any((v, lock)):
        v = multiprocessing.Value('i', 0)
        lock = multiprocessing.Lock()

if __name__ == '__main__':
    # Each worker process will call `initializer()` when it starts.
    with multiprocessing.Pool(initializer=set_global_counter_and_lock) as pool:
        pool.map(func, range(25))

这真的是解决这个问题的最佳实践方法吗？

The RuntimeError使用时你会得到Pool是因为池方法的参数在通过（池内部）队列发送到工作进程之前已被腌制。 您尝试使用哪种池方法在这里无关紧要。当你只使用时不会发生这种情况Process因为不涉及队列。您只需使用以下命令即可重现错误pickle.dumps(multiprocessing.Value('i', 0)).

您的最后一个代码片段并不像您想象的那样工作。你不是sharing a Value，您正在为每个子进程重新创建独立的计数器。

如果您在 Unix 上并使用默认的启动方法“fork”，那么您只需not将共享对象作为参数传递到池方法中。 您的子进程将通过分叉继承全局变量。使用进程启动方法“spawn”（默认 Windows 和带有 Python 3.8+ 的 macOS）或“forkserver”，你必须使用initializer during Pool实例化，让子进程继承共享对象。

请注意，您不需要额外的multiprocessing.Lock在这里，因为multiprocessing.Value默认情况下，您可以使用内部的。

import os
from multiprocessing import Pool, Value #, set_start_method


def func(x):
    for i in range(x):
        assert i == i
        with cnt.get_lock():
            cnt.value += 1
            print(f'{os.getpid()} | counter incremented to: {cnt.value}\n')


def init_globals(counter):
    global cnt
    cnt = counter


if __name__ == '__main__':

    # set_start_method('spawn')

    cnt = Value('i', 0)
    iterable = [10000 for _ in range(10)]

    with Pool(initializer=init_globals, initargs=(cnt,)) as pool:
        pool.map(func, iterable)

    assert cnt.value == 100000

可能还值得注意的是，您不需要计数器shared在所有情况下。 如果您只需要跟踪某件事总共发生的频率，一种选择是在计算过程中保留单独的工作本地计数器，并在最后进行总结。 这可能会显着提高频繁更新计数器的性能，而在并行计算本身期间不需要同步。