阅读 153

python3使用concurrent执行多进程任务

python3使用concurrent执行多进程任务

编程之家收集整理的这篇文章主要介绍了python3使用concurrent执行多进程任务,编程之家小编觉得挺不错的,现在分享给大家,也给大家做个参考。



技术背景

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

concurrent使用示例

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

# sleep.pyimport timedef 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.pyimport concurrent.futuresimport timedef 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]$ ipythonPython 3.8.5 (default,Sep  4 2020,07:30:14) Type 'copyright','credits' or 'license' for more informationIPython 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倍的算法执行时间。让我们通过配置和修改一些参数来测试验证一下:

# concurrent_sleep.pyimport concurrent.futuresimport timeimport 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的情况下,会按照系统中最高的逻辑核数来进行多进程的任务分配,但是在实际场景中我们需要考虑多种因素的限制,如内存和进程数的均衡配置(在大内存任务中,如果进程全开,有可能导致内存不足的问题)。只有在满足所有系统约束条件的情况下,才能发挥硬件最好的性能。在下面的代码中我们将给出如何配置执行任务的核数的方案:

# concurrent_sleep.pyimport concurrent.futuresimport timeimport 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 16The 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。

获取返回值

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

# concurrent_sleep.pyimport concurrent.futuresimport timeimport sysdef sleep(seconds):    time.sleep(seconds)    return secondsif __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 16The total sleep cpu time is: 16sThe time cost is: 4.034112930297852s

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

总结概要

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

版权声明

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

总结

以上是编程之家为你收集整理的python3使用concurrent执行多进程任务全部内容,希望文章能够帮你解决python3使用concurrent执行多进程任务所遇到的程序开发问题。

如果觉得编程之家网站内容还不错,欢迎将编程之家网站推荐给程序员好友。

本图文内容来源于网友网络收集整理提供,作为学习参考使用,版权属于原作者。


文章分类
后端
版权声明:本站是系统测试站点,无实际运营。本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容, 请发送邮件至 XXXXXXo@163.com 举报,一经查实,本站将立刻删除。
相关推荐