Python：如何检查 multiprocessing.Pool 中待处理任务的数量？

Question

我有一小部分工作人员 (4) 和一个非常大的任务列表 (5000~)。我正在使用一个池并使用 map_async() 发送任务。因为我正在运行的任务相当长，所以我强制将块大小设置为 1，以便一个长进程无法容纳一些较短的进程。

我想做的是定期检查还有多少任务需要提交。我知道最多有 4 个会处于活动状态，我关心的是还剩下多少个需要处理。

我用谷歌搜索了一下，找不到任何人这样做。

一些简单的代码可以提供帮助：

import multiprocessing
import time

def mytask(num):
    print('Started task, sleeping %s' % num)
    time.sleep(num)

pool = multiprocessing.Pool(4)
jobs = pool.map_async(mytask, [1,2,3,4,5,3,2,3,4,5,2,3,2,3,4,5,6,4], chunksize=1)
pool.close()

while True:
    if not jobs.ready():
        print("We're not done yet, %s tasks to go!" % <somethingtogettasks>)
        jobs.wait(2)
    else:
        break

Answer 1

看起来 jobs._number_left 就是您想要的。 _ 表示它是一个内部值，可能会根据开发人员的突发奇想而改变，但这似乎是获取该信息的唯一方法。

Answer 2

假设您使用的是 apply_async，您可以通过查看 Pool._cache 属性来检查待处理作业的数量。这是存储 ApplyResult 的位置，直到它们可用并且等于待处理的 ApplyResult 数量。

import multiprocessing as mp
import random
import time


def job():
    time.sleep(random.randint(1,10))
    print("job finished")

if __name__ == '__main__':
    pool = mp.Pool(5)
    for _ in range(10):
        pool.apply_async(job)

    while pool._cache:
        print("number of jobs pending: ", len(pool._cache))
        time.sleep(2)

    pool.close()
    pool.join()

Answer 3

据我所知，这不是一个完美的方法，但如果您使用 Pool.imap_unordered() 函数而不是 map_async，则可以拦截正在处理的元素。

import multiprocessing
import time

process_count = 4

def mytask(num):
    print('Started task, sleeping %s' % num)
    time.sleep(num)
    # Actually, you should return the job you've created here.
    return num

pool = multiprocess.Pool(process_count)
jobs  = []
items = [1,2,3,4,5,3,2,3,4,5,2,3,2,3,4,5,6,4]
job_count = 0
for job in pool.imap_unordered(mytask, items):
    jobs.append(job)
    job_count += 1

    incomplete = len(items) - job_count
    unsubmitted = max(0, incomplete - process_count)

    print "Jobs incomplete: %s. Unsubmitted: %s" % incomplete, unsubmitted

pool.close()

我正在减去 process_count，因为您几乎可以假设所有进程都将在以下两个例外之一的情况下进行处理：1）如果您使用迭代器，则可能没有更多的项目可供使用和处理和 2) 您剩余的物品可能少于 4 件。我没有为第一个异常编写代码。但如果您需要的话，这样做应该很容易。不管怎样，你的例子使用了一个列表，所以你不应该遇到这个问题。

编辑：我还意识到您正在使用 While 循环，这使得您看起来像是在尝试定期更新某些内容，例如每半秒或其他内容。我作为示例给出的代码不会这样做。我不确定这是否有问题。

Answer 4

我有类似的要求：跟踪进度，根据结果执行临时工作，随时干净地停止所有处理。我的处理方法是使用 apply_async 一次发送一个任务。我所做的工作的高度简化版本：

maxProcesses = 4
q = multiprocessing.Queue()
pool = multiprocessing.Pool()
runlist = range(100000)
sendcounter = 0
donecounter = 0
while donecounter < len(runlist):
    if stopNowBooleanFunc():  # if for whatever reason I want to stop processing early
        if donecounter == sendcounter:  # wait til already sent tasks finish running
            break
    else:  # don't send new tasks if it's time to stop
        while sendcounter < len(runlist) and sendcounter - donecounter < maxProcesses:
            pool.apply_async(mytask, (runlist[sendcounter], q))
            sendcounter += 1

    while not q.empty():  # process completed results as they arrive
        aresult = q.get()
        processResults(aresult)
        donecounter += 1

请注意，我使用队列而不是返回结果。