python并发与并行（十二） ———— 考虑用concurrent.futures实现真正的并行计算

有些Python程序写到一定阶段，性能就再也上不去了。即便优化了代码，程序的执行速度可能还是达不到要求。考虑到现在的计算机所装配的CPU核心数量越来越多，所以我们很自然地就想到用并行方式来解决这个问题。那么接下来就必须思考，如何将代码所要执行的计算任务划分成多个独立的部分并在各自的核心上面平行地运行。

Python的全局解释器锁（global interpreter lock，GIL）导致我们没办法用线程来实现真正的并行，所以先把这种方案排除掉。另一种常见的方案，是把那些对性能要求比较高的（performance-critical）代码用C语言重写成扩展模块。C语言比Python更接近底层硬件，因此运行速度要比Python快，这样的话，有些任务可能根本就不需要做并行，而是单单用C语言重写一遍就好。另外，C扩展还可以启动原生线程（native thread），这种线程不受Python解释器制约，也不必考虑GIL的问题，它们能够平行地运行，从而发挥出多核CPU的优势。Python里面针对C扩展而设计的那些API，有详细的文档可以参考，所以这是个很好的备选方案。大家在开发扩展模块的时候，还可以借助SWIG（https://github.com/swig/swig） 与CLIF（https://github.com/google/clif） 等工具。

然而，用C语言重写Python代码，代价是比较高的。因为有些代码在Python之中很简洁，但是改写成C代码之后，就变得特别难懂、特别复杂了。在移植过程中，我们还必须做大量的测试，以确保移植过去的那些代码跟原来的Python代码效果相同，并且不会引入bug。有的时候，这些工作确实很有意义，所以Python行业里面出现了大量的C扩展模块，用来迅速执行各种任务，例如文本解析、图像合成、矩阵运算等。另外还有Cython（https://cython.org/） 与Numba（https://numba.pydata.org/） 这样的开源工具帮我们顺利地向C语言移植。

问题是，在大多数情况下，我们不能只把整个程序里的一小段代码移植到C语言，因为程序的性能之所以缓慢，通常是由多个因素造成的，而不是说只要消除了其中某一个主要因素，整个程序的性能就会大幅提升。要想把C语言在底层硬件与线程方面的优势发挥出来，必须把程序里的许多代码都迁移过去，这会让测试量激增，而且容易引入新的bug。所以，还是得想想有没有什么好办法，能够在Python语言自身的范围内，解决这种复杂的并行计算问题。

Python内置的multiprocessing模块提供了多进程机制，这种机制很容易通过内置的concurrent.futures模块来使用，这可能就是我们要找的理想方案。这种方案可以启动许多条子进程（child process），这些进程是独立于主解释器的，它们有各自的解释器与相应的全局解释器锁，因此这些子进程可以平行地运行在CPU的各个核心上面。每条子进程都能够充分利用它所在的这个核心来执行运算。这些子进程都有指向主进程的链接，用来接收所要执行的计算任务并返回结果。

例如，现在要用Python来执行某种计算量很大的工作，而且想把CPU里的各个核心充分利用起来。笔者写了下面这个计算最大公约数的函数，来模拟刚才讲的那种工作。在实际程序中，我们要写的可能是一种运算量更大的算法（例如用纳维-斯托克斯方程（Navier-Stokes equation）研究流体动力学）。

def gcd(pair):
    a, b = pair
    low = min(a, b)
    for i in range(low, 0, -1):
        if a % i == 0 and b % i == 0:
            return i
    assert False, 'Not reachable'

如果把有待求解最大公约数的那些元组按照先后顺序交给这个函数去执行，那么程序花费的总时间就会随着元组的数量呈正比例上升，因为我们根本就没有做平行计算。

import time

def gcd(pair):
    a, b = pair
    low = min(a, b)
    for i in range(low, 0, -1):
        if a % i == 0 and b % i == 0:
            return i
    assert False, 'Not reachable'
    
NUMBERS = [
    (1963309, 2265973), (2030677, 3814172),
    (1551645, 2229620), (2039045, 2020802),
    (1823712, 1924928), (2293129, 1020491),
    (1281238, 2273782), (3823812, 4237281),
    (3812741, 4729139), (1292391, 2123811),
]

def main():
    start = time.time()
    results = list(map(gcd, NUMBERS))
    end = time.time()
    delta = end - start
    print(f'Took {delta:.3f} seconds')

if __name__ == '__main__':
    main()

output:

Took 0.494 seconds

直接把这种代码分给多条Python线程去执行，是不会让程序提速的，因为它们全都受制于同一个Python全局解释器锁（GIL），无法真正平行地运行在各自的CPU核心上面。现在就来演示这一点。笔者使用concurrent.futures模块里面的ThreadPoolExecutor类，并允许它最多可以启用两条工作线程。

from concurrent.futures import ThreadPoolExecutor
import time

def gcd(pair):
    a, b = pair
    low = min(a, b)
    for i in range(low, 0, -1):
        if a % i == 0 and b % i == 0:
            return i
    assert False, 'Not reachable'


NUMBERS = [
    (1963309, 2265973), (2030677, 3814172),
    (1551645, 2229620), (2039045, 2020802),
    (1823712, 1924928), (2293129, 1020491),
    (1281238, 2273782), (3823812, 4237281),
    (3812741, 4729139), (1292391, 2123811),
]

def main():    
    start = time.time()
    pool = ThreadPoolExecutor(max_workers=2)
    results = list(pool.map(gcd, NUMBERS))
    end = time.time()
    delta = end - start
    print(f'Took {delta:.3f} seconds')

if __name__ == '__main__':
    main()
    

output:

Took 0.518 seconds

由于要启动线程池并和它通信，这种写法比单线程版本还慢。 但是请注意，只需要变动一行代码就能让程序出现奇效，也就是把ThreadPoolExecutor改成concurrent.futures模块里的ProcessPoolExecutor。这样一改，程序立刻就快了起来。

from concurrent.futures import ProcessPoolExecutor
import time

def gcd(pair):
    a, b = pair
    low = min(a, b)
    for i in range(low, 0, -1):
        if a % i == 0 and b % i == 0:
            return i
    assert False, 'Not reachable'


NUMBERS = [
    (1963309, 2265973), (2030677, 3814172),
    (1551645, 2229620), (2039045, 2020802),
    (1823712, 1924928), (2293129, 1020491),
    (1281238, 2273782), (3823812, 4237281),
    (3812741, 4729139), (1292391, 2123811),
]

def main():
    start = time.time()
    pool = ProcessPoolExecutor(max_workers=2)  # The one change
    results = list(pool.map(gcd, NUMBERS))
    end = time.time()
    delta = end - start
    print(f'Took {delta:.3f} seconds')

    
if __name__ == '__main__':
    main()

output:

Took 0.295 seconds

程序变得比原来快多了。这是为什么呢？因为ProcessPool-Executor类会执行下面这一系列的步骤（当然，这实际上是由multiprocessing模块里的底层机制所推动的）。

1）从包含输入数据的NUMBERS列表里把每个元素取出来，以便交给map。

2）用pickle模块对每个元素做序列化处理，把它转成二进制形式。

3）将序列化之后的数据，从主解释器所在的进程经由本地socket复制到子解释器所在的进程。

4）在子进程里面，用pickle模块对数据做反序列化处理，把它还原成Python对象。

5）引入包含gcd函数。

6）把刚才还原出来的那个对象交给gcd函数去处理，此时，其他子进程也可以把它们各自的那份数据交给它们各自的gcd函数执行。

7）对执行结果做序列化处理，把它转化成二进制形式。

8）将二进制数据通过socket复制到上级进程。

9）在上级进程里面对二进制数据做反序列化处理，把它还原成Python对象。

10）把每条子进程所给出的结果都还原好，最后合并到一个list里面返回。

从开发者这边来看，这个过程似乎很简单，但实际上，multiprocessing模块与ProcessPoolExecutor类要做大量的工作才能实现出这样的并行效果。同样的效果，假如改用其他语言来做，那基本上只需要用一把锁或一项原子操作就能很好地协调多个线程，从而实现并行。但这在Python里面不行，所以我们才考虑通过ProcessPoolExecutor来实现。然而这样做的开销很大，因为它必须在上级进程与子进程之间做全套的序列化与反序列化处理。

这个方案对那种孤立的而且数据利用度较高的任务来说，比较合适。所谓孤立（isolated），这里指每一部分任务都不需要跟程序里的其他部分共用状态信息。所谓数据利用度较高（high-leverage），这里指任务所使用的原始材料以及最终所给出的结果数据量都很小，因此上级进程与子进程之间只需要互传很少的信息就行，然而在把原始材料加工成最终产品的过程中，却需要做大量运算。刚才那个求最大公约数的任务就属于这样的例子，当然还有很多涉及其他数学算法的任务，也是如此。

如果你面对的计算任务不具备刚才那两项特征，那么使用ProcessPoolExecutor所引发的开销可能就会盖过因为并行而带来的好处。在这种情况下，我们可以考虑直接使用multiprocessing所提供的一些其他高级功能，例如共享内存（shared memory）、跨进程的锁（cross-process lock）、队列（queue）以及代理（proxy）等。但是，这些功能都相当复杂，即便两个Python线程之间所要共享的进程只有一条，也是要花很大工夫才能在内存空间里面将这些工具安排到位。假如需要共享的进程有很多条，而且还涉及socket，那么这种代码理解起来会更加困难。

总之，不要刚一上来，就立刻使用跟multiprocessing这个内置模块有关的机制，而是可以先试着用ThreadPoolExecutor来运行这种孤立且数据利用度较高的任务。把这套方案实现出来之后，再考虑向ProcessPoolExecutor方案迁移。如果ProcessPoolExecutor方案也无法满足要求，而且其他办法也全都试遍了，那么最后可以考虑直接使用multiprocessing模块里的高级功能来编写代码。

总结：

▪ 把需要耗费大量CPU资源的计算任务改用C扩展模块来写，或许能够有效提高程序的运行速度，同时又让程序里的其他代码依然能够利用Python语言自身的特性。但是，这样做的开销比较大，而且容易引入bug。 ▪ Python自带的multiprocessing模块提供了许多强大的工具，让我们只需要耗费很少的精力，就可以把某些类型的任务平行地放在多个CPU核心上面处理。 ▪ 要想发挥出multiprocessing模块的优势，最好是通过concurrent.futures模块及其ProcessPoolExecutor类来编写代码，因为这样做比较简单。 ▪ 只有在其他方案全都无效的情况下，才可以考虑直接使用multiprocessing里面的高级功能（那些功能用起来相当复杂）。