multiprocessing库之pool浅析

通信队列

前面说道为了解决进程间通信通道容量限制带来的问题，multiprocessing.pool库创建了一个内部缓存来存放任务；此外进程池需要与外部交换任务本身及其结果（也就是worker函数的inqueue/outqueue参数），这就是Pool类内的三个队列：

• _taskqueue：queue.Queue实例，它既是线程间通信队列，也充当Pool类内部的缓存（内部是collections.deque对象），用于存放来不及送到进程池执行的任务对象；
• _inqueue：multiprocessing.SimpleQueue实例，用于传送任务对象到进程池；
• _outqueue：multiprocessing.SimpleQueue实例，用于从进程池传送result对象到外部；

multiprocessing.SimpleQueue是用于进程间通信的队列，因此在操作系统层面观察到它的存在。简单来说，它（在非windows操作系统上）就是os.pipe返回的对象。用pipe在进程间传送数据，这就是multiprocessing库中队列的真相。

pipe的缓冲区在Linux操作系统上默认为64K，可见向其写入内容的进程有很大可能会频繁阻塞，这一点决定了Pool的内部设计。

线程分工

在讲线程分工之前必须介绍Pool类两个重要的内部数据结构：

• _pool：列表，存放所有活动的工作进程；
• _cache：列表，存放所有未得到执行结果的result对象；

不难发现在Pool类的__init__方法中创建了如下三个线程：

• _worker_handler：对应静态方法_handle_workers，主要负责维护进程池；
• _task_handler：对应静态方法_handle_tasks，主要负责传送任务给工作进程；
• _result_handler：对应静态方法_handle_results，主要负责接收任务执行结果；

_worker_handler线程做的事情其实很简单：每隔0.1秒检查一次_pool中每个工作进程的exitcode，如果exitcode不为None就代表该工作进程已经终止。这样的话_worker_handler线程就要执行清理工作并将工作进程补足。

_task_handler线程的作用主要是扮演主线程的分身：

• 从_taskqueue读取（主线程写入的）消息，转发到_inqueue（这个动作很可能阻塞）；
• 如果是批量加入的任务，将其拆解为单个任务再转发；

_result_handler线程的角色比较简单，就是不断的从队列中取得worker返回的结果，并执行result对象的_set方法来触发回调函数并设定任务执行结果。

但是它还有一项特殊的任务，就是只要进程池不是强制关闭或操作系统出现错误，就必须保证收回所有任务的执行结果。这部分代码简化后如下：

while cache and thread._state != TERMINATE:
    try:
        task = get()
    except (OSError, EOFError):
        return

    if task is None:
        continue
    job, i, obj = task
    try:
        cache[job]._set(i, obj)
    except KeyError:
        pass

最后看看主线程的一个重要任务：添加任务。我们选取Pool类一系列用于添加任务的接口中最简单的一个：

def apply_async(self, func, args=(), kwds={}, callback=None,
        error_callback=None):
    if self._state != RUN:
        raise ValueError("Pool not running")
    result = ApplyResult(self._cache, callback, error_callback)
    self._taskqueue.put(([(result._job, None, func, args, kwds)], None))
    return result

可以看到Pool将任务填入_taskqueue队列之后，将包装好的ApplyResult对象返回给了调用者。构造ApplyResult对象时将Pool的内部数据结构_cache包装了进去，这样_result_handler线程要从_cache中清除已完成任务就很简单了。

数据流向

说到这里数据流向就比较清楚了：

task的流向：

main_thread -> _taskqueue -> _task_handler => _inqueue => workers

result的流向：

workers => _outqueue => _result_handler

在这里，"->"表示进程内的数据流，"=>"表示进程间的数据流。

但是除此以外，还有进程池关闭时的控制流。

为了优雅的关闭进程池，必须保证所有线程/进程都是正常完成所有处理之后自行退出。所以Pool类的close方法仅仅设置了主线程以及_worker_handler线程的状态为CLOSE就结束了，剩下的工作需要由各线程依次通过队列发送一个特殊的消息来通知所有线程/进程（这里选择的特殊消息是None）。

这个发送None消息的动作由_worker_handler发起通知到_task_handler，但其他线程/进程的通知消息则全部由_task_handler发送：

_worker_handler -> _taskqueue -> _task_handler

_task_handler -> _outqueue -> _result_handler
_task_handler（循环N次） => _inqueue  => workers