Python多處理：如何在多個進程中共享一個字典？

Question

一個程序，用於創建幾個在可連接隊列上工作的進程，Q，最終可能會操作全局字典D來存儲結果。 （因此每個子進程可以使用D來存儲其結果並查看其他子進程正在生成的結果）

如果我在子進程中打印字典D，則會看到已完成的修改（即在D）。但是在主進程加入Q後，如果我打印D，這是一個空字典！

我知道這是一個同步/鎖定問題。有人可以告訴我這裏發生了什麼，以及我如何能夠同步訪問D？

Answer 1

多處理不像線程。每個子進程都會獲得主進程內存的副本。通常狀態通過通信（管道/套接字），信號或共享內存共享。

多重提出了一些可供您使用情況的抽象 - 共享的狀態會被視爲本地使用代理或共享內存：http://docs.python.org/library/multiprocessing.html#sharing-state-between-processes

相關章節：

• http://docs.python.org/library/multiprocessing.html#shared-ctypes-objects
• http://docs.python.org/library/multiprocessing.html#module-multiprocessing.managers

Answer 2

一般的答案是使用Manager對象。從文檔改編：

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} 

Answer 3

也許你可以試試pyshmht，共享的基於內存的哈希表擴展爲Python。

注意

1. 這不完全測試，僅供參考。

2. 它目前缺乏多處理的鎖定/ sem機制。

Answer 4

我想與大家分享我自己的工作比經理的字典更快，更簡單，比使用的內存噸和適用於Mac OS不起作用pyshmht庫更穩定。雖然我的字典只適用於普通字符串，並且目前是不可變的。 我使用線性探測實現並將鍵和值對存儲在表之後的單獨內存塊中。

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