我希望能够对从文件中加载的某些输入数据进行并行计算。(由于文件可能非常大,因此我使用了生成器。)
在一定数量的项目上,我的代码运行正常,但超过这个阈值后,程序会挂起(一些工作进程无法结束)。
有什么建议吗?(我正在使用python2.7、8个CPU; 5000行还可以,7500行不行。)
首先,您需要一个输入文件。在bash中生成它:
for i in {0..10000}; do echo -e "$i"'\r' >> counter.txt; done
然后运行这个:
python2.7 main.py 100 counter.txt > run_log.txt
main.py:
#!/usr/bin/python2.7
import os, sys, signal, time
import Queue
import multiprocessing as mp
def eat_queue(job_queue, result_queue):
"""Eats input queue, feeds output queue
"""
proc_name = mp.current_process().name
while True:
try:
job = job_queue.get(block=False)
if job == None:
print(proc_name + " DONE")
return
result_queue.put(execute(job))
except Queue.Empty:
pass
def execute(x):
"""Does the computation on the input data
"""
return x*x
def save_result(result):
"""Saves results in a list
"""
result_list.append(result)
def load(ifilename):
"""Generator reading the input file and
yielding it row by row
"""
ifile = open(ifilename, "r")
for line in ifile:
line = line.strip()
num = int(line)
yield (num)
ifile.close()
print("file closed".upper())
def put_tasks(job_queue, ifilename):
"""Feeds the job queue
"""
for item in load(ifilename):
job_queue.put(item)
for _ in range(get_max_workers()):
job_queue.put(None)
def get_max_workers():
"""Returns optimal number of processes to run
"""
max_workers = mp.cpu_count() - 2
if max_workers < 1:
return 1
return max_workers
def run(workers_num, ifilename):
job_queue = mp.Queue()
result_queue = mp.Queue()
# decide how many processes are to be created
max_workers = get_max_workers()
print "processes available: %d" % max_workers
if workers_num < 1 or workers_num > max_workers:
workers_num = max_workers
workers_list = []
# a process for feeding job queue with the input file
task_gen = mp.Process(target=put_tasks, name="task_gen",
args=(job_queue, ifilename))
workers_list.append(task_gen)
for i in range(workers_num):
tmp = mp.Process(target=eat_queue, name="w%d" % (i+1),
args=(job_queue, result_queue))
workers_list.append(tmp)
for worker in workers_list:
worker.start()
for worker in workers_list:
worker.join()
print "worker %s finished!" % worker.name
if __name__ == '__main__':
result_list = []
args = sys.argv
workers_num = int(args[1])
ifilename = args[2]
run(workers_num, ifilename)
blocks=false
:我只是从各种代码片段中编译了代码的方案。但我不确定它实际上是做什么的 - 我很困惑(即使在这种情况下也是如此):如果它是false
,那么要读取输入的子进程不会等待下一个项目到来?(如果是这样,会发生什么?) - galapahblock=False
表示调用立即返回,无论队列中是否有任何内容。一般情况下,忽略 "block" 和 "timeout" 参数。对于绝大多数任务来说,这些东西都可以完美地使用,而使用它们可能会导致各种问题。需要很多经验才能知道何时需要非阻塞或超时调用 - 即使是那些有经验的人也不一定需要。;-) - Tim Peters