代码之家 › 专栏 › 技术社区 › Brad Solomon

使用multiprocessing.starmap()在进程之间共享计数器

multiprocessing python-3.x python

Brad Solomon · 技术社区 · 7 年前

我想用 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() ,并拆分了一个iterable range(25) 他们之间
第二个示例创建了25个工作进程和任务,每个进程和任务都有一个输入

pool.starmap() (或 pool.map()

我见过类似的问题 here , here ,及 here ,但这些方法似乎并不适用于 .map() / .starmap() ,关于是否 Value 使用 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))

这真的是最好的实践方式吗?

1 回复 | 直到 6 年前

Darkonaut 6 年前

这个 RuntimeError 当你使用 Pool 您尝试使用的池方法与此无关。当你仅仅使用 Process 因为没有队列。您只需使用 pickle.dumps(multiprocessing.Value('i', 0)) .

您的上一个代码片段无法按您认为的方式工作。你不是 A. Value

如果您在Unix上使用默认的启动方法“fork”,那么只需使用不将共享对象作为参数传递到池方法中。 macOS with Python 3.8+ initializer 在期间 水塘 实例化,让子进程继承共享对象。

注意,你不需要额外的钱 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

共享如果您只需要记录某件事情发生的总频率,那么可以选择在计算过程中保留单独的工作者本地计数器,并在最后进行汇总。