itsource

다중 처리:여러 프로세스에서 딕트를 공유하려면 어떻게 해야 합니까?

mycopycode 2023. 6. 11. 10:39
반응형

다중 처리:여러 프로세스에서 딕트를 공유하려면 어떻게 해야 합니까?

으로, 조인가큐능동에여러프생램로성그는하,Q 결국에는 .D결과를 저장합니다.(각 자식 프로세스가 사용할 수 있도록 함)D결과를 저장하고 다른 하위 프로세스가 어떤 결과를 생성하는지 확인합니다.)

사전 D를 자식 프로세스로 인쇄하면 사전에 수행된 수정 사항(예: D)이 표시됩니다.하지만 메인 프로세스가 Q에 합류한 후 D를 인쇄하면 빈 딕트입니다!

동기화/잠금 문제인 것으로 알고 있습니다.누가 여기서 무슨 일이 일어나고 있는지, 어떻게 D에 대한 액세스를 동기화할 수 있는지 알려줄 수 있습니까?

일반적인 대답은 객체를 사용하는 것을 포함합니다.문서에서 수정됨:

from multiprocessing import Process, Manager

def f(d):
    d[1] += '1'
    d['2'] += 2

if __name__ == '__main__':
    manager = Manager()

    d = manager.dict()
    d[1] = '1'
    d['2'] = 2

    p1 = Process(target=f, args=(d,))
    p2 = Process(target=f, args=(d,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

    print d

출력:

$ python mul.py 
{1: '111', '2': 6}

멀티프로세싱은 스레드화와 다릅니다.각 하위 프로세스는 기본 프로세스의 메모리 복사본을 가져옵니다.일반적으로 상태는 통신(파이프/소켓), 신호 또는 공유 메모리를 통해 공유됩니다.

다중 처리는 프록시 또는 공유 메모리를 사용하여 로컬로 처리되는 공유 상태인 사용 사례에 대해 몇 가지 추상화를 사용할 수 있게 합니다. http://docs.python.org/library/multiprocessing.html#sharing-state-between-processes

관련 섹션:

가 여기에 것 은 @senderle의 하는지 궁금해 할 도 있습니다.multiprocessing.Pool.

은 좋은점그곳이 입니다..Pool()에서 에한방까지.manager 최상 를 친 한 두 API 하 인 스 턴 는 스 모 방multiprocessing.

from itertools import repeat
import multiprocessing as mp
import os
import pprint

def f(d: dict) -> None:
    pid = os.getpid()
    d[pid] = "Hi, I was written by process %d" % pid

if __name__ == '__main__':
    with mp.Manager() as manager:
        d = manager.dict()
        with manager.Pool() as pool:
            pool.map(f, repeat(d, 10))
        # `d` is a DictProxy object that can be converted to dict
        pprint.pprint(dict(d))

출력:

$ python3 mul.py 
{22562: 'Hi, I was written by process 22562',
 22563: 'Hi, I was written by process 22563',
 22564: 'Hi, I was written by process 22564',
 22565: 'Hi, I was written by process 22565',
 22566: 'Hi, I was written by process 22566',
 22567: 'Hi, I was written by process 22567',
 22568: 'Hi, I was written by process 22568',
 22569: 'Hi, I was written by process 22569',
 22570: 'Hi, I was written by process 22570',
 22571: 'Hi, I was written by process 22571'}

는 각 " " " " " " " " " " " " " " " " " " " " " " " " " " 기록하는 입니다.DictProxyd.

매니저의 명령보다 빠르고 메모리를 많이 사용하고 Mac OS에서는 작동하지 않는 pyshht 라이브러리보다 더 간단하고 안정적인 내 작업을 공유하고 싶습니다.비록 내 딕트는 평범한 문자열에서만 작동하고 현재 불변입니다.선형 프로빙 구현을 사용하고 키와 값 쌍을 표 뒤의 별도 메모리 블록에 저장합니다.

from mmap import mmap
import struct
from timeit import default_timer
from multiprocessing import Manager
from pyshmht import HashTable


class shared_immutable_dict:
    def __init__(self, a):
        self.hs = 1 << (len(a) * 3).bit_length()
        kvp = self.hs * 4
        ht = [0xffffffff] * self.hs
        kvl = []
        for k, v in a.iteritems():
            h = self.hash(k)
            while ht[h] != 0xffffffff:
                h = (h + 1) & (self.hs - 1)
            ht[h] = kvp
            kvp += self.kvlen(k) + self.kvlen(v)
            kvl.append(k)
            kvl.append(v)

        self.m = mmap(-1, kvp)
        for p in ht:
            self.m.write(uint_format.pack(p))
        for x in kvl:
            if len(x) <= 0x7f:
                self.m.write_byte(chr(len(x)))
            else:
                self.m.write(uint_format.pack(0x80000000 + len(x)))
            self.m.write(x)

    def hash(self, k):
        h = hash(k)
        h = (h + (h >> 3) + (h >> 13) + (h >> 23)) * 1749375391 & (self.hs - 1)
        return h

    def get(self, k, d=None):
        h = self.hash(k)
        while True:
            x = uint_format.unpack(self.m[h * 4:h * 4 + 4])[0]
            if x == 0xffffffff:
                return d
            self.m.seek(x)
            if k == self.read_kv():
                return self.read_kv()
            h = (h + 1) & (self.hs - 1)

    def read_kv(self):
        sz = ord(self.m.read_byte())
        if sz & 0x80:
            sz = uint_format.unpack(chr(sz) + self.m.read(3))[0] - 0x80000000
        return self.m.read(sz)

    def kvlen(self, k):
        return len(k) + (1 if len(k) <= 0x7f else 4)

    def __contains__(self, k):
        return self.get(k, None) is not None

    def close(self):
        self.m.close()

uint_format = struct.Struct('>I')


def uget(a, k, d=None):
    return to_unicode(a.get(to_str(k), d))


def uin(a, k):
    return to_str(k) in a


def to_unicode(s):
    return s.decode('utf-8') if isinstance(s, str) else s


def to_str(s):
    return s.encode('utf-8') if isinstance(s, unicode) else s


def mmap_test():
    n = 1000000
    d = shared_immutable_dict({str(i * 2): '1' for i in xrange(n)})
    start_time = default_timer()
    for i in xrange(n):
        if bool(d.get(str(i))) != (i % 2 == 0):
            raise Exception(i)
    print 'mmap speed: %d gets per sec' % (n / (default_timer() - start_time))


def manager_test():
    n = 100000
    d = Manager().dict({str(i * 2): '1' for i in xrange(n)})
    start_time = default_timer()
    for i in xrange(n):
        if bool(d.get(str(i))) != (i % 2 == 0):
            raise Exception(i)
    print 'manager speed: %d gets per sec' % (n / (default_timer() - start_time))


def shm_test():
    n = 1000000
    d = HashTable('tmp', n)
    d.update({str(i * 2): '1' for i in xrange(n)})
    start_time = default_timer()
    for i in xrange(n):
        if bool(d.get(str(i))) != (i % 2 == 0):
            raise Exception(i)
    print 'shm speed: %d gets per sec' % (n / (default_timer() - start_time))


if __name__ == '__main__':
    mmap_test()
    manager_test()
    shm_test()

노트북 성능 결과는 다음과 같습니다.

mmap speed: 247288 gets per sec
manager speed: 33792 gets per sec
shm speed: 691332 gets per sec

단순 사용 예:

ht = shared_immutable_dict({'a': '1', 'b': '2'})
print ht.get('a')

파이썬용 메모리 기반 해시 테이블 확장을 공유하는 pyshht를 시도해 볼 수도 있습니다.

공지

  1. 당신이 참고할 수 있도록, 그것은 완전히 테스트되지 않았습니다.

  2. 현재 다중 처리를 위한 잠금/셈 메커니즘이 부족합니다.

나의 경우, 일관된 출력을 얻지 못하고 있습니다. 예를 들어, 다음과 같습니다.__total_count__항상 20살은 아닙니다.

from itertools import repeat
import multiprocessing as mp
import os
import pprint
from functools import partial
import numpy as np
import time

def counter(value, d: dict) -> None:
    if value not in d:
        d["__unique_count__"] += 1
        d[value] = 1
    else:
        d[value] += 1

    d["__total_count__"] += 1


if __name__ == '__main__':
    mp.freeze_support()
    with mp.Manager() as manager:
        d = manager.dict()
        d["__unique_count__"] = 0 
        d["__total_count__"] = 0
        numbers = np.random.randint(0,5,size=100)
        print(len(numbers))
        with manager.Pool() as pool:
            pool.map(partial(counter, d=d), numbers)
        # `d` is a DictProxy object that can be converted to dict
        final_d = dict(d)
        pprint.pprint(final_d)
        print(final_d["__unique_count__"], final_d["__total_count__"])

출력1

100
{0: 26,
 1: 16,
 2: 26,
 3: 14,
 4: 18,
 '__total_count__': 92,
 '__unique_count__': 5}
5 92

출력2

100
{0: 10,
 1: 21,
 2: 28,
 3: 22,
 4: 19,
 '__total_count__': 95,
 '__unique_count__': 5}
5 95

언급URL : https://stackoverflow.com/questions/6832554/multiprocessing-how-do-i-share-a-dict-among-multiple-processes

반응형

'itsource' 카테고리의 다른 글

경고, FIREBASE_CONFIG 환경 변수가 없습니다.firebase-admin 초기화가 실패합니다.  (0) 2023.06.11
문자열에서 선행 및 후행 0을 제거하는 방법은 무엇입니까?파이썬  (0) 2023.06.11
ggplot2를 사용하여 축에 정수 값만 표시하는 방법  (0) 2023.06.11
이 Microsoft의 모호성을 어떻게 해결할 수 있습니까?사무실. 엑셀 전화?  (0) 2023.06.11
B가 잘못되면 A를 롤백합니다. 스프링 부트, jdbct 템플릿  (0) 2023.06.11

'itsource'의 다른글

  • 현재글다중 처리:여러 프로세스에서 딕트를 공유하려면 어떻게 해야 합니까?

관련글

댓글

티스토리툴바