97

我正在尝试解决一个涉及大量子问题的大数值问题，并且我正在使用 Python 的多处理模块（特别是 Pool.map）将不同的独立子问题拆分到不同的核心上。每个子问题都涉及计算大量子子问题，如果尚未由任何进程计算，我试图通过将它们存储到文件中来有效地记忆这些结果，否则跳过计算并从文件中读取结果。

我遇到了文件的并发问题：不同的进程有时会检查是否已经计算了子子问题（通过查找将存储结果的文件），看看它没有，运行计算，然后尝试将结果同时写入同一个文件。如何避免写这样的冲突？

4

2 回答 2

158

@GP89 ​​提到了一个很好的解决方案。使用队列将写入任务发送到对文件具有唯一写入权限的专用进程。所有其他工作人员都具有只读访问权限。这将消除碰撞。这是一个使用 apply_async 的示例，但它也适用于 map：

import multiprocessing as mp
import time

fn = 'c:/temp/temp.txt'

def worker(arg, q):
    '''stupidly simulates long running process'''
    start = time.clock()
    s = 'this is a test'
    txt = s
    for i in range(200000):
        txt += s 
    done = time.clock() - start
    with open(fn, 'rb') as f:
        size = len(f.read())
    res = 'Process' + str(arg), str(size), done
    q.put(res)
    return res

def listener(q):
    '''listens for messages on the q, writes to file. '''

    with open(fn, 'w') as f:
        while 1:
            m = q.get()
            if m == 'kill':
                f.write('killed')
                break
            f.write(str(m) + '\n')
            f.flush()

def main():
    #must use Manager queue here, or will not work
    manager = mp.Manager()
    q = manager.Queue()    
    pool = mp.Pool(mp.cpu_count() + 2)

    #put listener to work first
    watcher = pool.apply_async(listener, (q,))

    #fire off workers
    jobs = []
    for i in range(80):
        job = pool.apply_async(worker, (i, q))
        jobs.append(job)

    # collect results from the workers through the pool result queue
    for job in jobs: 
        job.get()

    #now we are done, kill the listener
    q.put('kill')
    pool.close()
    pool.join()

if __name__ == "__main__":
   main()
于 2012-11-23T13:38:36.637 回答
2

在我看来，您需要使用Manager将结果临时保存到列表中，然后将列表中的结果写入文件。此外，用于starmap传递您要处理的对象和托管列表。第一步是构建要传递给的参数starmap，其中包括托管列表。

from multiprocessing import Manager
from multiprocessing import Pool  
import pandas as pd

def worker(row, param):
    # do something here and then append it to row
    x = param**2
    row.append(x)

if __name__ == '__main__':
    pool_parameter = [] # list of objects to process
    with Manager() as mgr:
        row = mgr.list([])

        # build list of parameters to send to starmap
        for param in pool_parameter:
            params.append([row,param])

        with Pool() as p:
            p.starmap(worker, params)

从这一点开始，您需要决定如何处理列表。如果您有大量 RAM 和庞大的数据集，请随意使用 pandas 进行连接。然后，您可以非常轻松地将文件保存为 csv 或 pickle。

        df = pd.concat(row, ignore_index=True)

        df.to_pickle('data.pickle')
        df.to_csv('data.csv')
于 2016-12-21T17:02:35.040 回答