如何从多个进程中增加共享计数器？

Question

我在使用 multiprocessing 模块时遇到问题。我正在使用Pool 的工人及其map 方法来同时分析大量文件。每次处理一个文件时，我都希望更新一个计数器，以便我可以跟踪还有多少文件需要处理。这是示例代码：

import os
import multiprocessing

counter = 0


def analyze(file):
    # Analyze the file.
    global counter
    counter += 1
    print counter


if __name__ == '__main__':
    files = os.listdir('/some/directory')
    pool = multiprocessing.Pool(4)
    pool.map(analyze, files)

我找不到解决办法。

Answer 1

问题是 counter 变量在您的进程之间没有共享：每个单独的进程都在创建自己的本地实例并增加它。

请参阅文档中的this section，了解可用于在进程之间共享状态的一些技术。在您的情况下，您可能希望在您的工作人员之间共享一个 Value 实例

这是您示例的工作版本（带有一些虚拟输入数据）。请注意，它使用了我在实践中会尽量避免的全局值：

from multiprocessing import Pool, Value
from time import sleep

counter = None

def init(args):
    ''' store the counter for later use '''
    global counter
    counter = args

def analyze_data(args):
    ''' increment the global counter, do something with the input '''
    global counter
    # += operation is not atomic, so we need to get a lock:
    with counter.get_lock():
        counter.value += 1
    print counter.value
    return args * 10

if __name__ == '__main__':
    #inputs = os.listdir(some_directory)

    #
    # initialize a cross-process counter and the input lists
    #
    counter = Value('i', 0)
    inputs = [1, 2, 3, 4]

    #
    # create the pool of workers, ensuring each one receives the counter 
    # as it starts. 
    #
    p = Pool(initializer = init, initargs = (counter, ))
    i = p.map_async(analyze_data, inputs, chunksize = 1)
    i.wait()
    print i.get()

Answer 2

没有竞争条件错误的计数器类：

class Counter(object):
    def __init__(self):
        self.val = multiprocessing.Value('i', 0)

    def increment(self, n=1):
        with self.val.get_lock():
            self.val.value += n

    @property
    def value(self):
        return self.val.value

Answer 3

一个极其简单的例子，从 jkp 的回答中改变：

from multiprocessing import Pool, Value
from time import sleep

counter = Value('i', 0)
def f(x):
    global counter
    with counter.get_lock():
        counter.value += 1
    print("counter.value:", counter.value)
    sleep(1)
    return x

with Pool(4) as p:
    r = p.map(f, range(1000*1000))

Answer 4

更快的 Counter 类，无需两次使用 Value 的内置锁

class Counter(object):
    def __init__(self, initval=0):
        self.val = multiprocessing.RawValue('i', initval)
        self.lock = multiprocessing.Lock()

    def increment(self):
        with self.lock:
            self.val.value += 1

    @property
    def value(self):
        return self.val.value

https://eli.thegreenplace.net/2012/01/04/shared-counter-with-pythons-multiprocessing https://docs.python.org/2/library/multiprocessing.html#multiprocessing.sharedctypes.Value https://docs.python.org/2/library/multiprocessing.html#multiprocessing.sharedctypes.RawValue

Answer 5

这是基于与其他答案中提出的不同方法的问题的解决方案。它使用带有multiprocessing.Queue 对象的消息传递（而不是带有multiprocessing.Value 对象的共享内存）和过程安全（原子）内置递增和递减运算符@987654323 @ 和 -=（而不是引入自定义的 increment 和 decrement 方法），因为你要求它。

首先，我们定义一个类Subject 用于实例化一个对象，该对象将是父进程的本地对象，其属性将被递增或递减：

import multiprocessing


class Subject:

    def __init__(self):
        self.x = 0
        self.y = 0

接下来，我们定义一个类Proxy 用于实例化一个对象，该对象将成为远程代理，子进程将通过该代理请求父进程检索或更新Subject 对象的属性。进程间通信将使用两个multiprocessing.Queue 属性，一个用于交换请求，一个用于交换响应。请求的格式为(sender, action, *args)，其中sender 是发件人名称，action 是操作名称（'get'、'set'、'increment' 或'decrement' 属性值），而@ 987654339@ 是参数元组。响应格式为value（对'get' 请求）：

class Proxy(Subject):

    def __init__(self, request_queue, response_queue):
        self.__request_queue = request_queue
        self.__response_queue = response_queue

    def _getter(self, target):
        sender = multiprocessing.current_process().name
        self.__request_queue.put((sender, 'get', target))
        return Decorator(self.__response_queue.get())

    def _setter(self, target, value):
        sender = multiprocessing.current_process().name
        action = getattr(value, 'action', 'set')
        self.__request_queue.put((sender, action, target, value))

    @property
    def x(self):
        return self._getter('x')

    @property
    def y(self):
        return self._getter('y')

    @x.setter
    def x(self, value):
        self._setter('x', value)

    @y.setter
    def y(self, value):
        self._setter('y', value)

然后，我们定义类Decorator 来装饰Proxy 对象的getter 返回的int 对象，以便通知其setter 是否增加或减少运算符+= 和-=通过添加action 属性来使用，在这种情况下，setter 请求'increment' 或'decrement' 操作而不是'set' 操作。递增和递减运算符+= 和-= 调用相应的扩充赋值特殊方法__iadd__ 和__isub__（如果它们已定义），并退回到始终定义的赋值特殊方法__add__ 和__sub__对于int 对象（例如proxy.x += value 等价于proxy.x = proxy.x.__iadd__(value) 等价于proxy.x = type(proxy).x.__get__(proxy).__iadd__(value) 等价于type(proxy).x.__set__(proxy, type(proxy).x.__get__(proxy).__iadd__(value))）：

class Decorator(int):

    def __iadd__(self, other):
        value = Decorator(other)
        value.action = 'increment'
        return value

    def __isub__(self, other):
        value = Decorator(other)
        value.action = 'decrement'
        return value

然后，我们定义函数worker将在子进程中运行并请求递增和递减操作：

def worker(proxy):
    proxy.x += 1
    proxy.y -= 1

最后，我们定义一个单独的请求队列向父进程发送请求，以及多个响应队列向子进程发送响应：

if __name__ == '__main__':
    subject = Subject()
    request_queue = multiprocessing.Queue()
    response_queues = {}
    processes = []
    for index in range(4):
        sender = 'child {}'.format(index)
        response_queues[sender] = multiprocessing.Queue()
        proxy = Proxy(request_queue, response_queues[sender])
        process = multiprocessing.Process(
            target=worker, args=(proxy,), name=sender)
        processes.append(process)
    running = len(processes)
    for process in processes:
        process.start()
    while subject.x != 4 or subject.y != -4:
        sender, action, *args = request_queue.get()
        print(sender, 'requested', action, *args)
        if action == 'get':
            response_queues[sender].put(getattr(subject, args[0]))
        elif action == 'set':
            setattr(subject, args[0], args[1])
        elif action == 'increment':
            setattr(subject, args[0], getattr(subject, args[0]) + args[1])
        elif action == 'decrement':
            setattr(subject, args[0], getattr(subject, args[0]) - args[1])
    for process in processes:
        process.join()

当+= 和-= 是进程安全时，程序保证终止。如果您通过注释Decorator 的相应__iadd__ 或__isub__ 来删除进程安全，则程序只会偶然终止（例如，proxy.x += value 相当于proxy.x = proxy.x.__iadd__(value)，但如果@987654375 则回退到proxy.x = proxy.x.__add__(value) @ 没有定义，相当于proxy.x = proxy.x + value 相当于proxy.x = type(proxy).x.__get__(proxy) + value 相当于type(proxy).x.__set__(proxy, type(proxy).x.__get__(proxy) + value)，所以没有添加action 属性，setter 请求'set' 操作而不是@987654381 @操作）。

流程安全会话示例（原子+= 和-=）：

child 0 requested get x
child 0 requested increment x 1
child 0 requested get y
child 0 requested decrement y 1
child 3 requested get x
child 3 requested increment x 1
child 3 requested get y
child 2 requested get x
child 3 requested decrement y 1
child 1 requested get x
child 2 requested increment x 1
child 2 requested get y
child 2 requested decrement y 1
child 1 requested increment x 1
child 1 requested get y
child 1 requested decrement y 1

示例进程不安全会话（非原子+= 和-=）：

child 2 requested get x
child 1 requested get x
child 0 requested get x
child 2 requested set x 1
child 2 requested get y
child 1 requested set x 1
child 1 requested get y
child 2 requested set y -1
child 1 requested set y -1
child 0 requested set x 1
child 0 requested get y
child 0 requested set y -2
child 3 requested get x
child 3 requested set x 2
child 3 requested get y
child 3 requested set y -3  # the program stalls here

Answer 6

我正在使用 PyQT5 中的进程栏，所以我同时使用线程和池

import threading
import multiprocessing as mp
from queue import Queue

def multi(x):
    return x*x

def pooler(q):
    with mp.Pool() as pool:
    count = 0
    for i in pool.imap_unordered(ggg, range(100)):
        print(count, i)
        count += 1
        q.put(count)

def main():
    q = Queue()
    t = threading.Thread(target=thr, args=(q,))
    t.start()
    print('start')
    process = 0
    while process < 100:
        process = q.get()
        print('p',process)
if __name__ == '__main__':
    main()

我把这个放在 Qthread worker 中，它可以在可接受的延迟下工作