我正在使用 Pythonmultiprocessing.Manager
共享对一个进程将生成而其他进程将查看的数据集的访问权限。但是,我遇到了返回的 dict 代理manager.dict()
不支持的问题iteritems()
。
我可以迭代items()
,但这意味着构建一个包含字典中所有项目的新元组,这是一个很大的数字。有没有办法在不构造中间列表/元组的情况下做到这一点,从而只使用恒定数量的额外内存?
注意:如果解决方案要求生成过程暂停以进行迭代,也可以。
我正在使用 Pythonmultiprocessing.Manager
共享对一个进程将生成而其他进程将查看的数据集的访问权限。但是,我遇到了返回的 dict 代理manager.dict()
不支持的问题iteritems()
。
我可以迭代items()
,但这意味着构建一个包含字典中所有项目的新元组,这是一个很大的数字。有没有办法在不构造中间列表/元组的情况下做到这一点,从而只使用恒定数量的额外内存?
注意:如果解决方案要求生成过程暂停以进行迭代,也可以。
您可以迭代keys()
以减少内存占用。您必须防止密钥被删除。
否则,这里有一个示例,它有两种不同的方法,可以让您遍历 dict 中的项目。此示例中的iteritems()
方法仅适用于创建管理器对象的进程和管理器对象创建的子进程。那是因为需要 manager 对象来创建新的代理,而其他进程无权访问它。该iteritems2()
方法适用于其他进程,因为它不依赖于在这些进程中创建新代理。
import multiprocessing as mp
import multiprocessing.managers
class mydict(dict):
def __init__(self, *args, **kwargs):
dict.__init__(self, *args, **kwargs)
self.iters = {}
def iteritems(self):
print "iteritems", mp.current_process()
return dict.iteritems(self)
def _iteritems_start(self):
print "_iteritems_start", mp.current_process()
i = dict.iteritems(self)
self.iters[id(i)] = i
return id(i)
def _iteritems_next(self, iter_id):
try:
return self.iters[iter_id].next()
except StopIteration:
del self.iters[iter_id]
return None
class mydict_proxy(mp.managers.DictProxy):
def iteritems(self):
print "iteritems proxy", mp.current_process()
return self._callmethod("iteritems")
def iteritems2(self):
print "iteritems2 proxy", mp.current_process()
iter_id = self._callmethod("_iteritems_start")
def generator():
while True:
a = self._callmethod("_iteritems_next",
(iter_id,))
if a == None:
return
yield a
return generator()
_method_to_typeid_ = { "iteritems": "Iterator" }
_exposed_ = mp.managers.DictProxy._exposed_
_exposed_ += ("iteritems", "_iteritems_start", "_iteritems_next")
class mymanager(mp.managers.BaseManager):
pass
mymanager.register("mydict", mydict, mydict_proxy)
mymanager.register("Iterator", proxytype = mp.managers.IteratorProxy,
create_method = False)
def other(d):
for k, v in d.iteritems2():
d[k] = v.lower()
for k, v in d.iteritems():
d[k] = ord(v)
def main():
manager = mymanager()
manager.start()
d = manager.mydict(list(enumerate("ABCDEFGHIJKLMNOP")))
for (k, v) in d.iteritems():
print k, v
proc = mp.Process(target = other, args = (d,))
proc.start()
proc.join()
for (k, v) in d.iteritems():
print k, v
if __name__ == "__main__":
main()
请注意,虽然此代码可能更节省内存,但它可能会慢很多。
您可以使用SyncManager类来注册您自己的类型。然后,您可以在该类型上实现方法,例如仅从字典中获取有限数量的项目。
这是一个帮助您入门的示例:
import multiprocessing
from multiprocessing import managers
class TakerDict(dict):
"""Like a dict, but allows taking a limited number of items."""
def take(self, items=1):
"""Take the first `items` items."""
return [item for _, item in zip(range(items), self.items())]
# NOTE: add other dict methods to the tuple if you need them.
TakerProxy = managers.MakeProxyType('TakerProxy', ('take',))
managers.SyncManager.register('taker', TakerDict, TakerProxy)
if __name__ == '__main__':
manager = multiprocessing.Manager()
taker = manager.taker()
# in other processes, use e.g. taker.take(5)
因此,为了限制内存使用,您将不得不重复调用管理器进程来获取下一批元素。
但是,要做到这一点,您的 dict 必须支持索引(因此您可以从特定偏移量恢复)。由于您无法访问 dict 中元素的基本顺序,因此最好使用列表来代替(例如manager.list()
)。然后在您的子进程中,请求len()
列表的 ,并按切片索引以获得适当大小的批次——您不需要为此注册任何代理类型。
iteritems()
是为了一个列表字典。您可以使用 for 循环。或者您可以说sorted()
which 将返回排序列表中的键,然后遍历该列表并执行dict[key]
. 希望有帮助。如果有更好的方法。请与我分享。我很想知道。