python进程池剖析（三）

x625802392

　　　　之前文章对python中进程池的原理、数据流以及应用从代码角度做了简单的剖析，现在让我们回头看看标准库中对进程池的实现都有哪些值得我们学习的地方。我们知道，进程池内部由多个线程互相协作，向客户端提供可靠的服务，那么这些线程之间是怎样做到数据共享与同步的呢？在客户端使用apply/map函数向进程池分配任务时，使用self._taskqueue来存放任务元素，_taskqueue定义为Queue.Queue()，这是一个python标准库中的线程安全的同步队列，它保证通知时刻只有一个线程向队列添加或从队列获取元素。这样，主线程向进程池中分配任务（taskqueue.put），进程池中_handle_tasks线程读取_taskqueue队列中的元素，两个线程同时操作taskqueue，互不影响。进程池中有N个worker进程在等待任务下发，那么进程池中的_handle_tasks线程读取出任务后，又如何保证一个任务不被多个worker进程获取到呢？我们来看下_handle_tasks线程将任务读取出来之后如何交给worker进程的：



for taskseq, set_length in iter(taskqueue.get, None):
i = -1
for i, task in enumerate(taskseq):
if thread._state:
debug('task handler found thread._state != RUN')
break
try:
put(task)
except Exception as e:
job, ind = task[:2]
try:
cache[job]._set(ind, (False, e))
except KeyError:
pass
else:
if set_length:
debug('doing set_length()')
set_length(i+1)
continue
break
else:
debug('task handler got sentinel')
在从taskqueue中get到任务之后，对任务中的每个task，调用了put函数，这个put函数实际上是将task放入了管道，而主进程与worker进程的交互，正是通过管道来完成的。
再来看看worker进程的定义：
w = self.Process(target=worker,
args=(self._inqueue, self._outqueue,
self._initializer,
self._initargs, self._maxtasksperchild)
)
其中self._inqueue和self._outqueue为SimpleQueue()对象，实际是带锁的管道，上述_handle_task线程调用的put函数，即为SimpleQueue对象的方法。我们看到，这里worker进程定义均相同，所以进程池中的worker进程共享self._inqueue和self._outqueue对象，那么当一个task元素被put到共享的_inqueue管道中时，如何确保只有一个worker获取到呢，答案同样是加锁，在SimpleQueue()类的定义中，put以及get方法都带有锁，进行同步，唯一不同的是，这里的锁是用于进程间同步的。这样就保证了多个worker之间能够确保任务的同步。与分配任务类似，在worker进程运行完之后，会将结果put会_outqueue，_outqueue同样是SimpleQueue类对象，可以在多个进程之间进行互斥。
　　　　在worker进程运行结束之后，会将执行结果通过管道传回，进程池中有_handle_result线程来负责接收result，取出之后，通过调用_set方法将结果写回ApplyResult/MapResult对象，客户端可以通过get方法取出结果，这里通过使用条件变量进行同步，当_set函数执行之后，通过条件变量唤醒阻塞在get函数的主进程。
　　　　进程池终止工作通过调用Pool.terminate()来实现，这里的实现很巧妙，用了一个可调用对象，将终止Pool时的需要执行的回调函数先注册好，等到需要终止时，直接调用对象即可。



self._terminate = Finalize(
　　　　self, self._terminate_pool,
　　　　args=(self._taskqueue, self._inqueue, self._outqueue, self._pool,
self._worker_handler, self._task_handler,
self._result_handler, self._cache),
　　　　exitpriority=15
)
在Finalize类的实现了__call__方法，在运行self._terminate()时，就会调用构造self._terminate时传入的self._terminate_pool对象。
　　　　使用map/map_async函数向进程池中批量分配任务时，使用了生成器表达式：



self._taskqueue.put((((result._job, i, mapstar, (x,), {}) for i, x in enumerate(task_batches)), None))
生成器表达式很简单，只需把列表解析的的[]换成()即可，上述表达的列表解析表示为：
[(result._job, i, mapstar, (x,), {}) for i, x in enumerate(task_batches)]
这里使用生成器表达式的好处是，它相当于列表解析的扩展，是对内存有好的，因为它只是生成了一个生成器，当我们需要使用该生成器对应的逻辑目标数据时，它才会通过既定逻辑去生成该数据，所以不会大量占用内存。
　　　　在Pool中，_worker_handler线程负责监控、创建新的工作进程，在监控工作进程退出时，同时将退出的进程从进程池中删除掉。这类似于，一边遍历一边删除列表。我们来看下下面代码的实现：



>>> l = [1, 2, 3, 3, 4, 4, 4, 5]
>>> for i in l:
if i in [3, 4, 5]:
l.remove(i)

>>> l
[1, 2, 3, 4, 5]
　　我们看到l没有将所有的3和4都删除掉，这是因为remove改变了l的大小。再看下面的实现：



>>> l = [1, 2, 3, 3, 4, 4, 4, 5]
>>> for i in range(len(l)):
if l in [3, 4]:
del l

Traceback (most recent call last):
File "<pyshell#37>", line 2, in <module>
if l in [3, 4]:
IndexError: list index out of range
>>>
　　同样因为del l时，l的大小改变，继续访问下去导致访问越界。而标准库中的进程池给出了遍历删除的一个正确示例：



for i in reversed(range(len(self._pool))):
worker = self._pool
if worker.exitcode is not None:
worker.join()
cleaned = True
del self._pool
　　使用reversed，从后向前删除list中的元素，这样会保证所有符合删除条件的元素被删除掉：



>>> l = [1, 2, 3, 3, 4, 4, 4, 5]
>>> for i in reversed(range(len(l))):
if l in [3, 4, 5]:
del l

>>> l
[1, 2]
　　　　可以看出，一个篇幅并不算大的Pool模块，就有很多值得学习的地方。对于python亦或者其他语言，技能的提升，多阅读标准库中代码，是一个很不错的选择。对于我们经常使用，而不知其中实现奥秘的源码，多阅读源码，了解其技术实现，就像侯捷那本《STL源码剖析》中讲到的，源码之前，了无秘密。更重要的是，将这些漂亮而又高效的编码方式，运用在自己的工作中，让自己的代码也可以像标准库中的代码一样优雅，这可以说是每一个开发人员的追求。