我有以下脚本,我想通过在 CPU 核心之间拆分任务来加快处理速度。
myscript.py
from queue import Queue
from worker_script import Worker
task_queue = Queue(maxsize=1)
while some_other_functions:
task_queue.put(Some_Stuff) ## Step happens randomly anytime.
## Hence number of input tasks cannot be known
Worker(task_queue).start()
task_queue.join()
## further script that needs the above task_queue to be successfully completed first
worker_script.py
class Worker:
def __init__(self, task_queue):
self.task_queue = task_queue
def run(self):
while True:
task_object = self.task_queue.get()
do_some_big_calculation(task_object) #### The Step I want to use Multiprocess
self.task_queue.task_done()
如果我要将多重处理应用于
do_some_big_calculation(task_object)
,我将如何将task_done()
方法应用于task_queue
?
我正在申请 task_done()
,以便 task_queue.join()
拦截者知道分配的所有任务都已完成。
我在
worker_script.py
中的实现是:
import multiprocessing
class Worker:
def __init__(self, task_queue):
self.task_queue = task_queue
def run(self):
num_processes = multiprocessing.cpu_count() - 2
jobs = []
with multiprocessing.Pool(processes=num_processes) as pool:
while True:
new_job = pool.apply_async(self.sub_worker)
jobs.append(new_job)
def sub_worker(self):
while True:
task_object = self.task_queue.get()
do_some_big_calculation(task_object) #### The Step I want to use Multiprocess
self.task_queue.task_done()
我依靠
.get()
来阻止多重处理,直到另一个任务可用。
然而,我修改后的实现陷入了我无法弄清楚的地方。
对我的实施或全新解决方案有任何帮助,我们将不胜感激! 谢谢!
你说你“被困在某个地方”,但你还没有发布一个最小的、可重现的例子,所以很难确定你被困在哪里以及为什么被困(无论被困意味着什么)。
我确实注意到你的
run
方法有一个无限循环提交“作业”,所以程序永远不会完成。在某些时候,run
必须决定终止,然后它可以向task_queue.join()
发出调用以等待所有任务完成。但是,如果您想在 task_done
上调用方法 self.task_queue
,那么 self.task_queue
必须是 multiprocessing.JoinableQueue
的实例,但事实并非如此。您还可以在 start
实例上调用不存在的 Worker
方法。
您根本不需要使用
multiprocesssing.JoinableQueue
实例,因为 multiprocessing.pool Pool
实例有自己的内部队列用于提交作业和接收结果。您的代码应该看起来更像:
worker_script.py
import multiprocessing
class Worker:
def __init__(self):
# Why multiprocessing.cpu_count() - 2?
num_processes = multiprocessing.cpu_count() - 2
self._pool = Pool(num_processes)
def submit_task(self, task_object):
# Node need to save returned AsyncResult instance unless you need a
# return value from sub_worker:
self.pool.apply_async(self.sub_worker, args=(task_object,))
def terminate(self):
"""Wait for all submitted tasks to complete."""
self._pool.close()
self._pool.join()
def sub_worker(self, task_object):
do_some_big_calculation(task_object)
myscript.py
def main():
worker = Worker()
while some_other_functions:
worker.submit_task(task_object)
# Wait for all submitted tasks to complete:
worker.terminate()
if __name__ == '__main__'
main()