如何从传递给多处理的函数返回反字典？

Question

我有一份CSV文件的清单。我想对其中的每一个做一组操作，然后生成一个计数器dict，并且我想从所有CSV文件中验证包含单个计数器dict的主列表。我希望并行处理每个csv文件，然后从每个文件返回计数器dict。我在这里找到了一个类似的解决方案：How can I recover the return value of a function passed to multiprocessing.Process?

我用了大卫·卡伦提出的解决方案。这个解决方案对于字符串非常有效，但是当我试图返回一个计数器dict或一个普通dict时。所有的CSV文件都会被处理，直到send_end.send(结果)，在执行时它会永远挂在那里，然后抛出一个内存错误。我正在Linux服务器上运行这个程序，它有足够的内存来创建计数器dicts列表。

我使用了以下代码：

import multiprocessing

#get current working directory
cwd = os.getcwd()

#take a list of all files in cwd
files = os.listdir(cwd)

#defining the function that needs to be done on all csv files
def worker(f,send_end):
    infile= open(f) 
    #read liens in csv file
    lines = infile.readlines()
    #split the lines by "," and store it in a list of lists
    master_lst = [line.strip().split(“,”) for line in lines]
    #extract the second field in each sublist 
    counter_lst = [ element[1] for element in master_lst]
    print “Total elements in the list” + str(len(counter_lst))
    #create a dictionary of count elements
    a = Counter(counter_lst)
    # return the counter dict
    send_end.send(a)

def main():
    jobs = []
    pipe_list = []
    for f in files:
        if f.endswith('.csv'):
           recv_end, send_end = multiprocessing.Pipe(duplex=False)
           p = multiprocessing.Process(target=worker, args=(f, send_end))
           jobs.append(p)
           pipe_list.append(recv_end)
           p.start()

    for proc in jobs:
       proc.join()
    result_list = [x.recv() for x in pipe_list]
    print len(result_list)

if __name__ == '__main__':
     main()

我得到的错误如下：

Process Process-42:
Traceback (most recent call last):
  File "/usr/lib64/python2.7/multiprocessing/process.py", line 258, in 
  _bootstrap
  self.run()
  File "/usr/lib64/python2.7/multiprocessing/process.py", line 114, in run
  self._target(*self._args, **self._kwargs)
  File "/home/amm/python/collapse_multiprocessing_return.py", line 32, in 
  worker
  a = Counter(counter_lst)
  File "/usr/lib64/python2.7/collections.py", line 444, in __init__
  self.update(iterable, **kwds)
  File "/usr/lib64/python2.7/collections.py", line 526, in update
  self[elem] = self_get(elem, 0) + 1
 MemoryError
 Process Process-17:
 Traceback (most recent call last):
 Process Process-6:
 Traceback (most recent call last):
 File "/usr/lib64/python2.7/multiprocessing/process.py", line 258, in 
 _bootstrap
 File "/usr/lib64/python2.7/multiprocessing/process.py", line 258, in 
 _bootstrap
 Process Process-8:
 Traceback (most recent call last):
 File "/usr/lib64/python2.7/multiprocessing/process.py", line 258, in 
 _bootstrap
 self.run()
 self.run()
 self.run()
 File "/usr/lib64/python2.7/multiprocessing/process.py", line 114, in run
 File "/usr/lib64/python2.7/multiprocessing/process.py", line 114, in run
 self._target(*self._args, **self._kwargs)
 File "/usr/lib64/python2.7/multiprocessing/process.py", line 114, in run
 File "/home/amm/python/collapse_multiprocessing_return.py", line 32, in 
 worker
 self._target(*self._args, **self._kwargs)
 self._target(*self._args, **self._kwargs)
 File "/home/amm/python/collapse_multiprocessing_return.py", line 32, in 
 worker
 File "/home/amm/python/collapse_multiprocessing_return.py", line 32, in 
 worker
 a = Counter(counter_lst_lst)
 a = Counter(counter_lst_lst)
 a = Counter(counter_lst_lst)
 File "/usr/lib64/python2.7/collections.py", line 444, in __init__
 File "/usr/lib64/python2.7/collections.py", line 444, in __init__
 File "/usr/lib64/python2.7/collections.py", line 444, in __init__
 self.update(iterable, **kwds)
 File "/usr/lib64/python2.7/collections.py", line 526, in update
 self[elem] = self_get(elem, 0) + 1
 MemoryError
 self.update(iterable, **kwds)
 self.update(iterable, **kwds)
 File "/usr/lib64/python2.7/collections.py", line 526, in update
 File "/usr/lib64/python2.7/collections.py", line 526, in update
 self[elem] = self_get(elem, 0) + 1
 self[elem] = self_get(elem, 0) + 1
 MemoryError
 MemoryError
 Process Process-10:
 Traceback (most recent call last):
 File "/usr/lib64/python2.7/multiprocessing/process.py", line 258, in 
 _bootstrap
 self.run()
 File "/usr/lib64/python2.7/multiprocessing/process.py", line 114, in run
 self._target(*self._args, **self._kwargs)
 File "/home/amm/python/collapse_multiprocessing_return.py", line 32, in 
 worker
 a = Counter(counter_lst)
 File "/usr/lib64/python2.7/collections.py", line 444, in __init__
 self.update(iterable, **kwds)
 File "/usr/lib64/python2.7/collections.py", line 526, in update
 self[elem] = self_get(elem, 0) + 1
 MemoryError
 ^Z
 [18]+  Stopped                 collapse_multiprocessing_return.py

现在，在send_end.send(a)中代替"a“，如果我替换f，则使用文件名。它在目录中打印csv文件的数量(在本例中，len(result_list)就是这样做的)。但是当计数器返回"a“时，它就会永远卡住，抛出上面的错误。

我想让代码通过计数器切块接收结束，没有任何错误/问题。这附近有工作吗？有人能提出一个可行的解决方案吗？

p.s:我是多处理模块的新手，如果这个问题听起来很幼稚的话，很抱歉。此外，我尝试了multiprocessing.Manager()，但是得到了一个类似的错误

Answer 1

您的回溯提到了Process Process-42:，因此至少有42个进程正在创建。您正在为每个CSV文件创建一个进程，这是无用的，可能会导致内存错误。

使用multiprocessing.Pool.map可以更简单地解决问题。worker函数也可以大大缩短：

def worker(f):
    with open(f) as infile:
        return Counter(line.strip().split(",")[1]
                       for line in infile)

def main():
    pool = multiprocessing.Pool()
    result_list = pool.map(worker, [f for f in files if f.endswith('.csv')])

不向池传递参数意味着它将创建尽可能多的进程，就像您有CPU内核一样。更多的使用可能提高性能，也可能不会提高性能。