python3使用concurrent执行多进程任务

技术背景

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

concurrent使用示例

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

# sleep.py

import time

def sleep(seconds):
    time.sleep(seconds)

if __name__ == '__main__':
    times = [1] * 10
    time0 = 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,希望能够加速这个计算的过程,这个时候我们就需要用到多进程的技术,下面是修改为多进程之后的代码:

# concurrent_sleep.py

import concurrent.futures
import time

def sleep(seconds):
    time.sleep(seconds)

if __name__ == '__main__':
    times = [1] * 10
    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))

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

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

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

多进程的最佳配置

使用多进程的方案,能有多大的加速效果,很大程度上是依赖于硬件的。理论上来说,如果有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 psutil

In [2]: psutil.cpu_count(logical=False)
Out[2]: 4

In [3]: psutil.cpu_count(logical=True)
Out[3]: 8

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

# concurrent_sleep.py

import concurrent.futures
import time
import sys

def 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 16
The time cost is: 2.0304934978485107s

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

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

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

# concurrent_sleep.py

import concurrent.futures
import time
import sys

def 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 17
The time cost is: 5.032677173614502s

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

获取返回值

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

# concurrent_sleep.py

import concurrent.futures
import time
import sys

def sleep(seconds):
    time.sleep(seconds)
    return 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:
        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: 16s
The time cost is: 4.034112930297852s

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

总结概要

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

版权声明

本文首发链接为:https://www.cnblogs.com/dechinphy/p/concurrent.html
作者ID:DechinPhy
更多原著文章请参考:https://www.cnblogs.com/dechinphy/

相关文章

本文从多个角度分析了vi编辑器保存退出命令。我们介绍了保存...
Python中的回车和换行是计算机中文本处理中的两个重要概念,...
SQL Server启动不了错误1067是一种比较常见的故障,主要原因...
信息模块是一种可重复使用的、可编程的、可扩展的、可维护的...
本文从电脑配置、PyCharm版本、Java版本、配置文件以及程序冲...
本文主要从多个角度分析了安装SQL Server 2012时可能出现的错...