Python 多处理池:在执行任务期间动态设置进程数

Question

我们在开发机器上提交 Python 2.7 中的大型 CPU 密集型作业(由许多独立的并行进程组成)，一次持续数天。当这些作业与大量进程一起运行时，机器的响应速度会减慢很多。理想情况下，我想限制白天开发代码时可用的 CPU 数量，并在夜间尽可能高效地运行尽可能多的进程。

Python 多处理库允许您在启动池时指定进程数。有没有办法在每次启动新任务时动态更改此数字？

例如，允许20 个进程在 19-07 小时运行，允许 10 个进程在 07-19 小时运行。

一种方法是检查使用大量 CPU 的活动进程数。这就是我希望它的工作方式：

from multiprocessing import Pool
import time 
pool = Pool(processes=20)
def big_task(x):
while check_n_process(processes=10) is False:
time.sleep(60*60)
x += 1
return x 

x = 1
multiple_results = [pool.apply_async(big_task, (x)) for i in range(1000)]
print([res.get() for res in multiple_results])

但是我需要编写"check_n_process"函数。

还有其他想法可以解决这个问题吗？

(代码需要在 Python 2.7 中运行 - bash 实现是不可行的)。

Answer 1

Pythonmultiprocessing.Pool不提供更改正在运行Pool的工作器数量的方法。一个简单的解决方案是依靠第三方工具。

billiard提供的池用于提供此类功能。

像Celery或Luigi这样的任务队列框架肯定允许灵活的工作负载，但要复杂得多。

如果使用外部依赖项不可行，可以尝试以下方法。根据这个答案，您可以设置基于信号量的限制机制。

from threading import Semaphore, Lock
from multiprocessing import Pool
def TaskManager(object):
def __init__(self, pool_size):
self.pool = Pool(processes=pool_size)
self.workers = Semaphore(pool_size)
# ensures the semaphore is not replaced while used
self.workers_mutex = Lock()  
def change_pool_size(self, new_size):
"""Set the Pool to a new size."""
with self.workers_mutex:  
self.workers = Semaphore(new_size)
def new_task(self, task):
"""Start a new task, blocks if queue is full."""
with self.workers_mutex:
self.workers.acquire()
self.pool.apply_async(big_task, args=[task], callback=self.task_done))
def task_done(self):
"""Called once task is done, releases the queue is blocked."""
with self.workers_mutex:
self.workers.release()

如果超过 X 个工作人员忙，池将阻止进一步尝试安排您的big_tasks。通过控制此机制，您可以限制并发运行的进程数量。当然，这意味着您放弃了Pool排队机制。

task_manager = TaskManager(20)
while True:
if seven_in_the_morning():
task_manager.change_pool_size(10)
if seven_in_the_evening():
task_manager.change_pool_size(20)
task = get_new_task()
task_manager.new_task()  # blocks here if all workers are busy

Answer 2

这是非常不完整的(也是一个老问题)，但您可以通过跟踪正在运行的进程并仅在有利时才调用 apply_async() 来管理负载; 如果每个作业的运行时间少于永久，您可以通过在工作时间或 os.getloadavg() 过高时调度较少的作业来降低负载。 我这样做是为了在运行多个"scp"时管理网络负载，以逃避我们内部网络上的流量整形(不要告诉任何人！