Python 다중 처리 및 공유 카운터

Python 다중 처리 및 공유 카운터

---

다중 처리 모듈에 문제가 있습니다. 많은 파일에서 데이터를로드하기 위해 map 메서드와 함께 작업자 풀을 사용하고 있으며 각 파일에 대해 사용자 지정 함수로 데이터를 분석합니다. 파일이 처리 될 때마다 카운터를 업데이트하여 처리해야 할 파일 수를 추적하고 싶습니다. 다음은 샘플 코드입니다.

def analyze_data( args ):
    # do something 
    counter += 1
    print counter


if __name__ == '__main__':

    list_of_files = os.listdir(some_directory)

    global counter
    counter = 0

    p = Pool()
    p.map(analyze_data, list_of_files)

이에 대한 해결책을 찾을 수 없습니다.

---

문제는 counter변수가 프로세스간에 공유되지 않는다는 것입니다. 각 개별 프로세스는 자체 로컬 인스턴스를 생성하고이를 증가시킵니다.

프로세스간에 상태를 공유하기 위해 사용할 수있는 몇 가지 기술은 설명서 의이 섹션 을 참조하십시오 . 귀하의 경우 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()

---

경쟁 조건 버그가없는 카운터 클래스 :

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

---

Value의 내장 잠금을 두 번 사용하지 않고 더 빠른 Counter 클래스

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

---

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))

참조 URL : https://stackoverflow.com/questions/2080660/python-multiprocessing-and-a-shared-counter