了解 Python 中的多处理/管理器队列

Question

我很难理解使用 Python 进行多处理的管理器队列。

我的老师给了我这个代码：

def check_prime(n): test if n is a prime or not and returns a boolean.

def chunks(lst, n):
    for i in range(0, len(lst), n):
        yield lst[i:i + n]

def find_prime_worker(input, output):
    for chunk in iter(input.get,None):
        print('(Pid: {})'.format(getpid()))
        primes_found = list(filter(check_prime,chunk))
        output.put(primes_found)

from multiprocessing import Pool,Manager
from itertools import chain

def calculate_primes(ncore,N,chunksize):
    with Manager() as manager:
        input = manager.Queue()
        output = manager.Queue()

        with Pool(ncore) as p:
            it = p.starmap_async(find_prime_worker,[(input,output)]*ncore)
            for r in chunks(range(1,N),chunksize):
                input.put(r)
            for i in range(ncore): input.put(None)
            it.wait()
            output.put(None)

        res = list(chain(*list(iter(output.get,None))))
    return res

calculate_primes(8,100,5)

1. 如果在 find_prime_worker 函数中，我添加

   print('(Pid: {})'.format(getpid()))

calculate_primes() 函数返回一个空结果 []。 如果我删除此行，则功能可以正常工作。我想知道在不破坏正确功能的情况下正在执行什么进程ID...请问该怎么做？

1. 我不理解 calculate_primes() 函数：我们定义了 2 个管理器队列输入和输出。

我不明白为什么我们用 starmap_async 传递给 find_prime_worker 函数，参数大小为 [(input,output)]*ncore。

非常感谢。

Answer 1

如果要打印进程id，需要：

from os import getpid

如果没有该语句，您的工作函数 find_prime_worker 将引发异常，但由于您只是通过调用 it.wait() 等待调用工作函数的结果准备就绪，而不是使用 it.get() 实际检索结果，您不会将异常反映回主进程。将您的代码更改为return_values = it.get()，您将看到异常。添加import 语句，异常将消失，return_values 将是来自find_prime_worker 的所有返回值的列表，这将是[None] * 8，因为该函数隐式返回None，并且它正在被调用ncore 或 8 次。顺便说一句，it 这个名字的选择很糟糕，因为它暗示（至少对我而言）调用starmap_async 方法返回的是一个迭代器，而实际上它是返回的是一个multiprocessing.pool.AsycResult 实例。在我看来，result 将是这个返回值的更好名称。

这让我想到了你的第二个问题。该程序试图找到从 1 到 99 的所有素数，并通过创建一个带有 ncore 进程的进程池来做到这一点，其中 ncore 设置为 8。[(input,output)]*ncore 创建一个包含 8 个元组的列表，其中每个元组是 @987654339 @。因此，it = p.starmap_async(find_prime_worker,[(input,output)]*ncore) 语句将创建 8 个任务，其中每个任务都使用参数 input 和 output 调用 find_prime_worker。因此，这些任务是相同的，它们通过从输入队列input 获取消息来读取它们的输入，并将它们的结果写入输出队列output。输入队列上的消息只是数字 1 到 99，被分成大小为 5 的范围：range(1, 6)、range(6, 11)、range(11, 16) ... range(96, 100)。

---

仅在您感兴趣的情况下提供额外信息

该代码的不寻常之处在于，多处理池在内部使用队列来传递参数，并消除了为此目的显式创建队列的必要性。然而，代码使用池的内部队列来传递托管队列参数，这些参数保存的值可以通过池的参数传递机制更容易透明地处理：

def check_prime(n):
    # Not a real implemntation and not all the values between 1 and 100:
    if n in (3, 5, 7, 11, 13, 17, 19, 23, 97):
        return True
    return False

def chunks(lst, n):
    for i in range(0, len(lst), n):
        yield lst[i:i + n]

def find_prime_worker(chunk):
    from os import getpid

    print('(Pid: {})'.format(getpid()))
    return list(filter(check_prime, chunk))


def calculate_primes(ncore, N, chunksize):
    from multiprocessing import Pool

    with Pool(ncore) as p:
        results = p.map(find_prime_worker, chunks(range(1, N), chunksize))
    res = []
    for result in results:
        res.extend(result)
    return res

# Required for Windows:
if __name__ == '__main__':
    print(calculate_primes(8, 100, 5))

打印：

(Pid: 9440)
(Pid: 9440)
(Pid: 9440)
(Pid: 9440)
(Pid: 9440)
(Pid: 9112)
(Pid: 9440)
(Pid: 9112)
(Pid: 9440)
(Pid: 9440)
(Pid: 9112)
(Pid: 14644)
(Pid: 14644)
(Pid: 9112)
(Pid: 9440)
(Pid: 14644)
(Pid: 9112)
(Pid: 9112)
(Pid: 14644)
(Pid: 9440)
[3, 5, 7, 11, 13, 17, 19, 23, 97]

但让我们坚持使用代码的原始方法。如果您有兴趣，可以使用性能更高的multiprocessing.Queue 重写代码，而不是当前使用的托管队列。但是，它不能作为参数传递给多处理池工作函数。相反，您必须使用 multiprocessing.pool.Pool 构造函数的 initializer 和 initargs 参数来使用这些队列引用为池中的每个进程初始化全局变量。并且由于工作函数现在将引用队列作为全局变量，它不再需要任何参数，并且starmap_async 或任何map 函数似乎不再合适，而不向函数find_prime_worker 引入一个虚拟参数，即没用过。所以apply_async方法似乎是更合乎逻辑的选择：

def init_processes(inq, outq):
    global input, output
    input = inq
    output = outq

def check_prime(n):
    # Not a real implementation and not all the values between 1 and 100:
    if n in (3, 5, 7, 11, 13, 17, 19, 23, 97):
        return True
    return False

def chunks(lst, n):
    for i in range(0, len(lst), n):
        yield lst[i:i + n]

def find_prime_worker():
    from os import getpid

    for chunk in iter(input.get, None):
        print('(Pid: {})'.format(getpid()))
        primes_found = list(filter(check_prime, chunk))
        output.put(primes_found)

def calculate_primes(ncore, N, chunksize):
    from multiprocessing import Pool, Queue
    from itertools import chain

    input = Queue()
    output = Queue()

    with Pool(ncore, initializer=init_processes, initargs=(input, output)) as p:
        results = [p.apply_async(find_prime_worker) for _ in range(ncore)]
        for r in chunks(range(1, N), chunksize):
            input.put(r)
        for i in range(ncore):
            input.put(None)
        # Wait for all tasks to complete.
        # The actual return values are not of interest;
        # they are just None -- this is just for demo purposes but
        # calling result.get() will raise an exception if find_prime_worker
        # raised an exception:
        return_values = [result.get() for result in results]
        print(return_values)

        output.put(None)
        res = list(chain(*list(iter(output.get,None))))
    return res

# Required for Windows:
if __name__ == '__main__':
    print(calculate_primes(8, 100, 5))

打印：

(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
(Pid: 10516)
[None, None, None, None, None, None, None, None]
[3, 5, 7, 11, 13, 17, 19, 23, 97]

这里有几点需要注意。首先，虽然我们使用的池大小为 8，但我们可以看到单个进程 10516 处理了所有块。这可能是因为check_prime 是如此微不足道，以至于calculate_primes 任务能够从输入队列中读取一个块，为块中的 5 个数字中的每一个调用一次 check_prime 并返回读取所有其他块并在池中的其他进程甚至有机会运行之前重复调用。

其次，在创建多处理池以及通过队列发送参数和获取结果时存在开销，否则您将没有这些开销。 find_prime_worker 必须足够占用 CPU，以便并行处理补偿额外的开销。我怀疑在这种情况下它确实如此，至少对于 find_prime_worker 的这种实现不会。

第三，这是非常微妙的，使用multiprocessing.Queue 需要一点小心。在上面的代码中，我们知道find_prime_worker 不会返回除None 以外的任何内容，因此return_values = [result.get() for result in results] 语句主要用于演示目的，并确保find_prime_worker 引发的任何异常都不会被忽视。但是，一般来说，如果您只想等待所有提交的任务完成而不关心返回值，而不是保存从 apply_async 调用返回的所有 AsyncResult 实例，然后在这些实例上调用 get例如，可以只调用p.close()，然后调用p.join()，在执行这些调用后，您可以确定所有提交的任务都已完成（可能有例外）。这是因为在池实例上调用join 将等待所有池进程退出，这只会在完成它们的工作或某些异常情况时发生。但我没有在这段代码中使用该方法，因为在读取写入该队列的所有消息之前，您绝不能加入已写入多处理队列的进程。

但是我们可以改变逻辑，这样我们就知道我们已经从队列中读取了所有可能的消息，方法是让每个工作任务在处理完所有块后写入None 记录：

def init_processes(inq, outq):
    global input, output
    input = inq
    output = outq

def check_prime(n):
    # Not a real implemntation and not all the values between 1 and 100:
    if n in (3, 5, 7, 11, 13, 17, 19, 23, 97):
        return True
    return False

def chunks(lst, n):
    for i in range(0, len(lst), n):
        yield lst[i:i + n]

def find_prime_worker():
    from os import getpid

    for chunk in iter(input.get, None):
        print('(Pid: {})'.format(getpid()))
        primes_found = list(filter(check_prime, chunk))
        if primes_found:
            output.put(primes_found)
    output.put(None)

def calculate_primes(ncore, N, chunksize):
    from multiprocessing import Pool, Queue

    input = Queue()
    output = Queue()

    with Pool(ncore, initializer=init_processes, initargs=(input, output)) as p:
        results = [p.apply_async(find_prime_worker) for _ in range(ncore)]
        for r in chunks(range(1, N), chunksize):
            input.put(r)
        for i in range(ncore):
            input.put(None)
        # We assume find_prime_worker does not raise and exception and
        # there will be ncore None records on the output queue:
        none_seen = 0
        res = []
        while none_seen < ncore:
            result = output.get()
            if result is None:
                none_seen += 1
            else:
                res.extend(result)
        # Now we can, to be "tidy":
        p.close()
        p.join()
    return res

# Required for Windows:
if __name__ == '__main__':
    print(calculate_primes(8, 100, 5))

打印：

(Pid: 15796)
(Pid: 14644)
(Pid: 19100)
(Pid: 15796)
(Pid: 15796)
(Pid: 14644)
(Pid: 15796)
(Pid: 19100)
(Pid: 14644)
(Pid: 14644)
(Pid: 19100)
(Pid: 15796)
(Pid: 14644)
(Pid: 8924)
(Pid: 15796)
(Pid: 19100)
(Pid: 14644)
(Pid: 8924)
(Pid: 15796)
(Pid: 19100)
[3, 5, 17, 19, 7, 23, 11, 13, 97]

请注意，我向find_prime_worker 添加了附加检查，因为将空列表写入输出队列是没有意义的：

        if primes_found:
            output.put(primes_found)

最后，使用输入和输出队列的正常用例是不使用多处理池，而是创建自己的multirpocessing.Process 实例，实际上是使用这些Process实例和队列来实现您自己的池：

def check_prime(n):
    # Not a real implementation and not all the values between 1 and 100:
    if n in (3, 5, 7, 11, 13, 17, 19, 23, 97):
        return True
    return False

def chunks(lst, n):
    for i in range(0, len(lst), n):
        yield lst[i:i + n]

def find_prime_worker(input, output):
    from os import getpid

    for chunk in iter(input.get, None):
        print('(Pid: {})'.format(getpid()))
        primes_found = list(filter(check_prime, chunk))
        if primes_found:
            output.put(primes_found)
    output.put(None)

def calculate_primes(ncore, N, chunksize):
    from multiprocessing import Process, Queue

    input = Queue()
    output = Queue()

    processes = [Process(target=find_prime_worker, args=(input, output))
                 for _ in range(ncore)
                 ]
    for process in processes:
        process.start()
    for r in chunks(range(1, N), chunksize):
        input.put(r)
    for i in range(ncore):
        input.put(None)

    none_seen = 0
    res = []
    while none_seen < ncore:
        result = output.get()
        if result is None:
            none_seen += 1
        else:
            res.extend(result)

    for process in processes:
        process.join()

    return res

# Required for Windows:
if __name__ == '__main__':
    print(calculate_primes(8, 100, 5))

打印：

(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
(Pid: 18112)
[3, 5, 7, 11, 13, 17, 19, 23, 97]