Python中使用concurrent执行多进程任务

随着计算机技术的发展，诸如GPU和超算平台等越来越发达，这些技术的本质其实并没有带来算法上的革新，之所以能够提升计算的速度和规模，很大程度上是因为分布式和并行计算的优势。这里我们介绍一个简单的python自带的多进程的代码实现，使用的是concurrent这个工具，同时我们也会介绍如何更好的配置多进程的资源。

1.concurrent使用示例

concurrent是python自带的一个多进程实现仓库，不需要额外的安装。这里我们先介绍一个没有多进程的示例代码：

# sleep.pyimport timedef sleep(seconds):time.sleep(seconds)if __name__ == '__main__':times = [1] * 10time0 = time.time()for t in times:sleep(t)time1 = time.time()print ('The time cost is: {}s'.format(time1 - time0))

这个代码的功能其实就是休眠10s的时间，也比较容易，我们可以看一下这个代码的执行结果：

[dechin@dechin-manjaro concurrent]$ python3 sleep.py 
The time cost is: 10.014754295349121s

在我们统计时间的时候，发现会比10s的时间多出来一些，这部分时间不仅仅包含这个python程序执行的时间，跟时间的统计方式也有一定的关系，一般我们可以忽略这部分的gap时间。

我们假定上面这个程序中的sleep函数功能不是休眠1s的时间，而是执行一个耗时为1s的计算任务，而我们有很多的cpu，希望能够加速这个计算的过程，这个时候我们就需要用到多进程的技术，下面是修改为多进程之后的代码：

import concurrent.futures
import timedef sleep(seconds):time.sleep(seconds)if __name__ == '__main__':times = [1] * 10time0 = time.time()with concurrent.futures.ProcessPoolExecutor() as executor:executor.map(sleep, times)time1 = time.time()print ('The time cost is: {}s'.format(time1 - time0))

整个修改的方式也非常容易，就是把原代码中的for循环修改为concurrent的执行语句，让我们一起来看看执行的结果：

[dechin@dechin-manjaro concurrent]$ python3 concurrent_sleep.py 
The time cost is: 2.0304219722747803s

从结果上我们可以看到，运行的时间从10s降低到了2s。这说明我们的多进程任务带来了非常显著的优化效果，至于为什么优化之后是2s而不是3s或者1s，这个问题将在下一个章节中进行介绍。

2.多进程的最佳配置

使用多进程的方案，能有多大的加速效果，很大程度上是依赖于硬件的。理论上来说，如果有n个cpu核，我们就可以实现n倍的加速。但是大部分情况下会受限于算法或者任务本身，跟n倍的加速之间有一定的gap。首先让我们用ipython来看看本地电脑上有多少个cpu：

[dechin@dechin-manjaro concurrent]$ ipython
Python 3.8.5 (default, Sep  4 2020, 07:30:14) 
Type 'copyright', 'credits' or 'license' for more information
IPython 7.19.0 -- An enhanced Interactive Python. Type '?' for help.In [1]: import psutilIn [2]: psutil.cpu_count(logical=False)
Out[2]: 4In [3]: psutil.cpu_count(logical=True)
Out[3]: 8

这里我们使用的是psutil这个库，而不是常用的os或者multiprocessing，这是因为可以更好的区分逻辑核与物理核。我们本地的电脑上有4个物理核，每个物理核实际上对应于2个逻辑核，因此一共是有8个逻辑核。也就是说，理论上我们最大可以加速8倍的算法执行时间。让我们通过配置和修改一些参数来测试验证一下：

import concurrent.futures
import time
import sysdef sleep(seconds):time.sleep(seconds)if __name__ == '__main__':if sys.argv[1] == '-t':times = [1] * int(sys.argv[2]) # 获取命令行的时间输入参数time0 = time.time()with concurrent.futures.ProcessPoolExecutor() as executor:executor.map(sleep, times)time1 = time.time()print ('The time cost is: {}s'.format(time1 - time0))

在这个示例中，为了方便调整，我们把总的休眠时间定为命令行的输入参数，使用的是sys.argv这个函数来获取，注意获取到的参数是字符串格式的。这串代码的执行方法和执行结果如下所示：

[dechin@dechin-manjaro concurrent]$ python3 concurrent_sleep.py -t 16The time cost is: 2.0304934978485107s

在上面的执行结果中，我们发现原本需要16s的休眠任务，在多进程场景下被加速到了2s，刚好符合我们对逻辑核的加速倍数的预期。但是前面提到，能否加速逻辑核的倍数，跟任务本身的算法也有关系。比如在本用例中，如果算法分配的结果就是需要17个子算法来进行实现，那么在每个逻辑核上面执行了2次的休眠任务之后，又有一个逻辑核需要再执行1次休眠任务，而此时其他的逻辑核都需要等待这个执行任务的逻辑核的任务结束。具体的执行结果如下所示：

[dechin@dechin-manjaro concurrent]$ python3 concurrent_sleep.py -t 17The time cost is: 3.0313029289245605s

这个结果也验证了我们此前的预想，因为16s的任务执行需要2s的时间，而执行完16s的任务之后，又需要等待剩下的一个1s时长的任务执行完毕，总耗时3s。这里我们没有配置max_worker的情况下，会按照系统中最高的逻辑核数来进行多进程的任务分配，但是在实际场景中我们需要考虑多种因素的限制，如内存和进程数的均衡配置（在大内存任务中，如果进程全开，有可能导致内存不足的问题）。只有在满足所有系统约束条件的情况下，才能发挥硬件最好的性能。在下面的代码中我们将给出如何配置执行任务的核数的方案：

import concurrent.futures
import time
import sysdef sleep(seconds):time.sleep(seconds)if __name__ == '__main__':if sys.argv[1] == '-t':times = [1] * int(sys.argv[2])time0 = time.time()with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:executor.map(sleep, times)time1 = time.time()print ('The time cost is: {}s'.format(time1 - time0))

配置方法也很容易，就是在ProcessPoolExecutor函数的入参中引入max_worker，这里我们先将最大使用的核数设置为4，再来重新看一下上述用例的执行结果：

[dechin@dechin-manjaro concurrent]$ python3 concurrent_sleep.py -t 16
The time cost is: 4.032958030700684s
[dechin@dechin-manjaro concurrent]$ python3 concurrent_sleep.py -t 17The time cost is: 5.032677173614502s

对于16s的任务，因为开了4个核的并行，因此在4s的时间内完成了任务。而17s的任务，同样是需要多等待1s的时间，总耗时为5s。

3.获取返回值

如果任务可以互相独立的去执行，互相之间并不需要通信，那自然是最理想的情况。但是更多的情况下，我们是要收集各个进程的返回值的，通过这个返回值，在各个进程之间进行通信。而在concurrent的情况下，map函数的返回值直接就是所有进程的返回值所组成的列表，这更加方便了我们的任务执行。

import concurrent.futures
import time
import sysdef sleep(seconds):time.sleep(seconds)return seconds
'''
学习中遇到问题没人解答？小编创建了一个Python学习交流群：711312441
''' 
if __name__ == '__main__':if sys.argv[1] == '-t':times = [1] * int(sys.argv[2])time0 = time.time()with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:results = executor.map(sleep, times)print ('The total sleep cpu time is: {}s'.format(sum(results)))time1 = time.time()print ('The time cost is: {}s'.format(time1 - time0))

在这个用例中，我们将map函数的结果存储到results这一参数中，最后对results进行求和的操作，这个简单的示例中，返回的结果实际上就是总的输入的休眠时间。执行结果如下所示：

[dechin@dechin-manjaro concurrent]$ python3 concurrent_sleep.py -t 16
The total sleep cpu time is: 16sThe time cost is: 4.034112930297852s

可以看到，所有的返回值被成功的获取到。

总结

多进程技术是独立于算法任务本身的一种优化技术，通过python中的concurrent库，我们可以非常容易的实现多进程的任务，用来优化已有的算法。这里我们也给出了一些多进程配置信息的参考方案，在GPU和超算相关的领域上，都能够发挥较大的用途。