我正在尝试解决一个涉及大量子问题的大数值问题,并且我正在使用 Python 的多处理模块(特别是 Pool.map)将不同的独立子问题拆分到不同的核心上。每个子问题都涉及计算大量子子问题,如果尚未由任何进程计算它们,我试图通过将它们存储到文件中来有效地内存这些结果,否则跳过计算并从文件中读取结果。
我遇到了文件的并发问题:不同的进程有时会检查是否已经计算了子子问题(通过查找将存储结果的文件),看到它没有,运行计算,然后尝试将结果同时写入同一个文件。如何避免写这样的冲突?
最佳答案
@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()
关于Python多处理安全地写入文件,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/13446445/