多处理：如何在多个进程之间共享一个字典？
Posted
技术标签:
【中文标题】多处理：如何在多个进程之间共享一个字典？【英文标题】：multiprocessing: How do I share a dict among multiple processes? 【发布时间】：2011-10-13 12:53:53 【问题描述】：一个程序，它创建多个在可加入队列Q 上工作的进程，并最终可能操纵全局字典D 来存储结果。 （因此每个子进程都可以使用D 来存储其结果，并查看其他子进程产生的结果）
如果我在子进程中打印字典 D，我会看到对其进行的修改（即在 D 上）。但是主进程加入Q后，如果我打印D，就是空dict！
我知道这是一个同步/锁定问题。谁能告诉我这里发生了什么，以及如何同步对 D 的访问？
【问题讨论】：
这至少在使用 osx 10.14.4 的 python 3.7.2 上不能按预期工作 字典不同步，其内容被其他进程重写。但是，multiprocessing.Manager().list() 按预期工作。
【参考方案1】：
一般答案涉及使用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
【讨论】：
感谢发件人。确实， D = multiprocessing.Manager().dict() 解决了我的问题。我正在使用 D = dict()。 @LorenzoBelli，如果您要询问对经理的访问是否同步，我相信答案是肯定的。multiprocessing.Manager() 返回一个 instance of SyncManager，它的名字暗示了很多！
@senderle 我想与子进程共享父进程的 numpy 随机状态。我试过使用Manager，但还是没有运气。你能看看我的问题here，看看你能不能提供一个解决方案？如果我每次生成随机数时都执行np.random.seed(None)，我仍然可以获得不同的随机数，但这不允许我使用父进程的随机状态，这不是我想要的。非常感谢任何帮助。
@RadioControlled 很高兴编写更新，但简而言之，虽然我认为您不能直接实现这一点，但您可以轻松创建具有相同键和值的新托管字典，并使用它而不是原来的。这对您的情况是否足够？
@senderle，这就是我最终所做的。所以答案是你必须这样做。【参考方案2】：
多处理不像线程。每个子进程都会得到一份主进程内存的副本。通常状态是通过通信（管道/套接字）、信号或共享内存共享的。
多处理为您的用例提供了一些抽象 - 通过使用代理或共享内存将共享状态视为本地：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【讨论】：
非常感谢。你让我找到了/a 解决方案：multiprocessing.Manager().dict()。 有人能详细说明“每个子进程都将获得主进程内存的副本”这句话的含义吗？ @Itsme2003 默认情况下，衍生进程无法访问父进程的内存（这是与线程的关键区别之一）。因此，当进程需要父进程的对象时，它必须创建它的副本（而不是获取对实际对象的引用）。上面的答案详细说明了如何在进程之间共享对象。 因为这经常被误认为：只要不修改对象，至少在通常的Linux设置中，对象实际上只会在内存中存储一次。更改后将立即复制。如果您需要节省内存并且不修改对象，这可能非常重要。【参考方案3】：除了这里的@senderle，有些人可能还想知道如何使用multiprocessing.Pool 的功能。
好消息是.Pool() 实例有一个.Pool() 方法，它模仿了所有熟悉的***multiprocessing API。
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'
这是一个稍微不同的示例，每个进程只是将其进程 ID 记录到全局 DictProxy 对象 d。
【讨论】：
【参考方案4】：我想分享我自己的工作，它比 Manager 的 dict 更快，并且比使用大量内存且不适用于 Mac OS 的 pyshmht 库更简单、更稳定。虽然我的 dict 只适用于纯字符串并且目前是不可变的。 我使用线性探测实现并将键和值对存储在表之后的单独内存块中。
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')
【讨论】：
Github？文档？我们如何使用这个工具？ 我现在知道from pyshmht import HashTable 找不到哈希表模块【参考方案5】：
也许你可以试试 pyshmht，共享基于内存的 Python 哈希表扩展。
通知
未经过全面测试，仅供参考。
目前缺少用于多处理的锁定/sem 机制。
【讨论】：
以上是关于多处理：如何在多个进程之间共享一个字典？的主要内容，如果未能解决你的问题，请参考以下文章