如何让多处理池既不启动新进程又不终止当前 运行 进程？
How do you get multiprocessing Pool to not spin up new processes but also not terminate currently running processes?
我在 Python 2.7 中使用 Python multiprocessing.Pool class。我有大量工作只能在一天中的特定时间段内完成 运行。每项工作都需要一些时间。我想将作业限制为一次最多 运行 n 个并行。
Pool 功能可以很好地限制并行作业的数量，但在我尝试结束作业时似乎出现了问题。当我的 window 结束时，我希望当前 运行ning 作业完成处理。我不希望开始新的工作。我一直在尝试使用 Pool.close() 来做到这一点，这确实让我的 运行ning 进程按预期完成，但从实验来看，队列中但尚未开始处理的作业仍将即使在 Pool 关闭后也提交处理。
另一个选项 Pool.terminate() 甚至会主动关闭 运行ning 作业，这与预期的行为背道而驰。
Function
Allows running jobs to finish
Prevents new jobs from starting
.terminate()
No
Yes
.close()
Yes
No
Desired behaviour
Yes
Yes
首先，你不应该使用 Python2.7，它已经被弃用一段时间了。
您应该使用 concurrent.futures 标准库中的 ProcessPoolExecutor 并在激活 cancel_futures 标志的情况下调用 .shutdown() 方法，让执行程序完成启动的作业但取消任何待处理的工作。
from concurrent.futures import ProcessPoolExecutor
parallel_jobs = 4 # The pool size
executor = ProcessPoolExecutor(parallel_jobs)
future_1 = executor.submit(work_1, argument_1)
...
future_n = executor.submit(work_n, argument_n)
...
# At some point when the time window ends and you need to stop jobs:
executor.shutdown(cancel_futures=True)
# Some futures such as future_n may have been cancelled here, you can check that:
if future_n.cancelled():
print("job n has been cancelled")
# Or you can try to get the result while checking for CancelledError:
try:
result_n = future_n.result()
except CancelledError:
print("job n hasn't finished in the given time window")
这里有一个取消的例子：
from concurrent.futures import ThreadPoolExecutor, as_completed, wait
from time import sleep
# The job to execute concurrently
def foo(i: int) -> str:
sleep(0.2)
print(i)
return f"{i}"
e = ThreadPoolExecutor(4)
# Jobs are scheduled concurrently, this call does not block
futures = [e.submit(foo, i) for i in range(100)]
# Shutdown during execution and cancel pending jobs
e.shutdown(cancel_futures=True)
# Gather completed results
results = [f.result() for f in futures if not f.cancelled()]
print(results)
如果执行此代码，您会看到 100 个计划的作业并未全部完成，只有一些是因为执行程序已在其间关闭。
我在 Python 2.7 中使用 Python multiprocessing.Pool class。我有大量工作只能在一天中的特定时间段内完成 运行。每项工作都需要一些时间。我想将作业限制为一次最多 运行 n 个并行。
Pool 功能可以很好地限制并行作业的数量，但在我尝试结束作业时似乎出现了问题。当我的 window 结束时，我希望当前 运行ning 作业完成处理。我不希望开始新的工作。我一直在尝试使用 Pool.close() 来做到这一点，这确实让我的 运行ning 进程按预期完成，但从实验来看，队列中但尚未开始处理的作业仍将即使在 Pool 关闭后也提交处理。
另一个选项 Pool.terminate() 甚至会主动关闭 运行ning 作业，这与预期的行为背道而驰。
| Function | Allows running jobs to finish | Prevents new jobs from starting |
|---|---|---|
| .terminate() | No | Yes |
| .close() | Yes | No |
| Desired behaviour | Yes | Yes |
首先，你不应该使用 Python2.7，它已经被弃用一段时间了。
您应该使用 concurrent.futures 标准库中的 ProcessPoolExecutor 并在激活 cancel_futures 标志的情况下调用 .shutdown() 方法，让执行程序完成启动的作业但取消任何待处理的工作。
from concurrent.futures import ProcessPoolExecutor
parallel_jobs = 4 # The pool size
executor = ProcessPoolExecutor(parallel_jobs)
future_1 = executor.submit(work_1, argument_1)
...
future_n = executor.submit(work_n, argument_n)
...
# At some point when the time window ends and you need to stop jobs:
executor.shutdown(cancel_futures=True)
# Some futures such as future_n may have been cancelled here, you can check that:
if future_n.cancelled():
print("job n has been cancelled")
# Or you can try to get the result while checking for CancelledError:
try:
result_n = future_n.result()
except CancelledError:
print("job n hasn't finished in the given time window")
这里有一个取消的例子：
from concurrent.futures import ThreadPoolExecutor, as_completed, wait
from time import sleep
# The job to execute concurrently
def foo(i: int) -> str:
sleep(0.2)
print(i)
return f"{i}"
e = ThreadPoolExecutor(4)
# Jobs are scheduled concurrently, this call does not block
futures = [e.submit(foo, i) for i in range(100)]
# Shutdown during execution and cancel pending jobs
e.shutdown(cancel_futures=True)
# Gather completed results
results = [f.result() for f in futures if not f.cancelled()]
print(results)
如果执行此代码，您会看到 100 个计划的作业并未全部完成，只有一些是因为执行程序已在其间关闭。