Python并行编程(十三)：进程池和mpi4py模块

爱被打了一巴掌 2021-12-13 03:59 367阅读 0赞

1、基本概念

多进程库提供了Pool类来实现简单的多进程任务。Pool类有以下方法：

\- apply()：直到得到结果之前一直阻塞。

\- apply\_async()：这是apply()方法的一个变体，返回的是一个result对象。这是一个异步的操作，在所有的子类执行之前不会锁住主进程。

\- map()：这是内置的map函数的并行版本，在得到结果之前一直阻塞，此方法将可迭代的数据的每一个元素作为进程池的一个任务来执行。

\- map\_async()：这是map的一个变体，返回一个result对象。如果指定了回调函数，回调函数应该是callable的，并且只接受一个参数。当result准备好时，会自动调用回调函数，除非调用失败。回调函数应该立即完成，否则，持有result的进程将被阻塞。

2、测试用例

创建四个进程池，然后使用map方法进行一个简单的计算。

import multiprocessing
    
    def function_square(data):
        result = data * data
        return result
    
    if __name__ == "__main__":
        inputs = list(range(100))
        pool = multiprocessing.Pool(processes=4)
        pool_outputs = pool.map(function_square, inputs)
        pool.close()
        pool.join()
        print("pool: ", pool_outputs)

pool.map方法将一些独立的任务提交给进程池。pool.map和内置map的执行结果相同，但pool.map是通过多个并行进程计算的。

3、mpi4py模块

Python提供了很多MPI模块写并行程序。其中mpi4py在MPI-1/2顶层构建，提供了面向对象的接口，紧跟C++绑定的MPI-2。MPI是C语言用户可以无需学习新的接口就可以使用这个库。

此模块包含的主要的应用：

\- 点对点通讯

\- 集体通讯

\- 拓扑

4、安装mpi4py

安装mpich：https://www.microsoft.com/en-us/download/confirmation.aspx?id=56727

下载并安装msmpisetup.exe

安装完成后安装目录如下：

![1095387-20181019100738476-1291286137.png][]

将bin目录添加到系统环境中：

![1095387-20181019100841605-424768067.png][]

用cmd输入并显示如下即为安装成功

![1095387-20181019101216845-1116488953.png][]

安装mpi4py

pip install mpi4py

MPI测试用例

from mpi4py import MPI
    
    def mpi_test(rank):
        print("I am rank %s" %rank)
    
    
    if __name__ == "__main__":
    
        comm = MPI.COMM_WORLD
        rank = comm.Get_rank()
        mpi_test(rank)
        print("Hello world from process", rank)

使用mpi运行文件

![1095387-20181019103255696-1670523618.png][]

在MPI中，并行程序中不同进程用一个非负整数来区别，如果我们有P个进程，那么rank会从0到P-1分配。

MPI拿到rank的函数如下：rank = comm.Get\_rank()

这个函数返回调用它的进程的rank，comm叫做交流者，用于区别不同的进程集合：comm = MPI.COMM\_WORLD

5、MPI点对点通讯

MPI提供的最实用的一个特性是点对点通讯。两个不同的进程之间可以通过点对点通讯交换数据：一个进程是接收者，一个进程是发送者。

Python的mpi4py通过下面两个函数提供了点对点通讯功能：

\- Comm.Send(data, process\_destination)：通过它在交流组中的排名来区分发送给不同进程的数据。

\- Comm.Recv(process\_source)：接收来自源进程的数据，也是通过在交流组中的排名来分分的。

Comm变量表示交流着，定义了可以互相通讯的进程组：

comm = MKPI.COMM\_WORLD

交换信息测试用例：

from mpi4py import MPI
    
    comm = MPI.COMM_WORLD
    rank = comm.rank
    print("My rank is :",rank)
    
    if rank == 0:
        data = 10000000
        destination_process = 4
        comm.send(data, dest=destination_process)
        print("sending data %s to process %d" %(data, destination_process))
    
    if rank == 1:
        destination_process = 8
        data = "hello,I am rank 1"
    
        comm.send(data, dest=destination_process)
        print("sending data %s to process %d" %(data, destination_process))
    
    if rank == 4:
        data = comm.recv(source=0)
        print("data received is = %s" %data)
    
    if rank == 8:
        data1 = comm.recv(source=1)
        print("data received is = %s" %data1)

运行结果：

![1095387-20181019115156873-2144384394.png][]

通过mpiexec -n 9运行9个互相通讯的进程，使用rank的值来区分每个进程。

整个过程分为两部分，发送者发送数据，接收者接收数据，二者必须都指定发送方/接收方，source=为指定发送者。如果有发送的数据没有被接收，程序会阻塞。

comm.send()和comm.recv()函数都是阻塞的函数，他们会一直阻塞调用者，直到数据使用完成，同时在MPI中，有两种方式发送和接收数据：

\- buffer模式

\- 同步模式

在buffer模式中，只要需要发送的数据被拷贝到buffer中，执行权就会交回到主程序，此时数据并非已经发送/接收完成。在同步模式中，只有函数真正的结束发送/接收任务之后才会返回。

6、避免死锁

mpi4py没有提供特定的功能来解决这种情况，但是提供了一些程序员必须遵守的规则来避免死锁的问题。

出现死锁的情况：

from mpi4py import MPI
    
    comm = MPI.COMM_WORLD
    rank = comm.rank
    print("my rank is :",rank)
    
    if rank == 1:
        data_send = "a"
        destination_process = 5
        source_process = 5
        data_received = comm.recv(source=source_process)
        comm.send(data_send, dest=destination_process)
    
        print("sending data %s to process %d" %(data_send, destination_process))
        print("data received is = %s" %data_received)
    
    if rank == 5:
        data_send = "b"
        destination_process = 1
        source_process = 1
        data_received = comm.recv(source=source_process)
        comm.send(data_send, dest=destination_process)
        print("sending data %s to process %d" % (data_send, destination_process))
        print("data received is = %s" % data_received)

运行结果：

![1095387-20181019142616250-2089618437.png][]

进程1和进程5产生阻塞，程序阻塞。

此时两个进程都在等待对方，发生阻塞，因为recv和send都是阻塞的，两个函数都先使用的recv，所以调用者都在等待他们完成。所以讲上述代码改为如下即可解决阻塞：

from mpi4py import MPI
    
    comm = MPI.COMM_WORLD
    rank = comm.rank
    print("my rank is :",rank)
    
    if rank == 1:
        data_send = "a"
        destination_process = 5
        source_process = 5
        comm.send(data_send, dest=destination_process)
        data_received = comm.recv(source=source_process)
    
        print("sending data %s to process %d" %(data_send, destination_process))
        print("data received is = %s" %data_received)
    
    if rank == 5:
        data_send = "b"
        destination_process = 1
        source_process = 1
        data_received = comm.recv(source=source_process)
        comm.send(data_send, dest=destination_process)
        print("sending data %s to process %d" % (data_send, destination_process))
        print("data received is = %s" % data_received)

将其中一个函数的recv和send顺序调换。

运行结果：

![1095387-20181019142943065-39428718.png][]

也可通过Sendrecv函数解决，代码如下：

from mpi4py import MPI
    
    comm = MPI.COMM_WORLD
    rank = comm.rank
    print("my rank is :",rank)
    
    if rank == 1:
        data_send = "a"
        destination_process = 5
        source_process = 5
        # comm.send(data_send, dest=destination_process)
        # data_received = comm.recv(source=source_process)
        data_received = comm.sendrecv(data_send, dest=destination_process, source=source_process)
    
        print("sending data %s to process %d" %(data_send, destination_process))
        print("data received is = %s" %data_received)
    
    if rank == 5:
        data_send = "b"
        destination_process = 1
        source_process = 1
        # data_received = comm.recv(source=source_process)
        # comm.send(data_send, dest=destination_process)
        data_received = comm.sendrecv(data_send, dest=destination_process, source=source_process)
        print("sending data %s to process %d" % (data_send, destination_process))
        print("data received is = %s" % data_received)

运行结果：

![1095387-20181019143401681-1836138641.png][]

7、集体通讯：Broadcast

在并行代码的开发中，会经常需要在多个进程间共享某个变量运行时的值，或操作多个进程提供的变量。MPI库提供了在多个进程之间交换信息的方法，将所有进程变成通讯者的这种方法叫做集体交流。因此，一个集体交流通常是2个以上的进程，也可以称为广播——一个进程将消息发送给其他进程。mpi4py模块通过以下方式提供广播的功能：

buf = comm.bcast(data_to_share, rank_of_root_process)

这个函数将root消息中包含的信息发送给属于comm通讯组其他的进程，每个进程必须通过相同的root和comm来调用它。

![1095387-20181019152911198-1562690982.png][]

测试代码：

from mpi4py import MPI
    
    comm = MPI.COMM_WORLD
    rank = comm.Get_rank()
    
    if rank == 0:
        variable_to_share = 100
    else:
        variable_to_share = None
    
    variable_to_share = comm.bcast(variable_to_share, root=0)
    print("process = %d  variable shared = %d" %(rank, variable_to_share))

运行结果：

![1095387-20181019155136898-1744479578.png][]

rank等于0的root进程初始化了一个变量，variable\_to\_share，值为100，然后声明了一个广播variable\_to\_share = comm.bcast(variable\_to\_share, root=0)

这个变量将通过通讯组发送给其他进程。

集体通讯允许组中的多个进程同时进行数据交流。在mpi4py模块中，只提供了阻塞版本的集体通讯(阻塞调用者，直到缓存中的数据全部安全发送。)

广泛应用的集体通讯应该是：

\- 组中的进程提供通讯的屏障

\- 通讯方式包括：

\- 将一个进程的数据广播到组中其他进程中

\- 从其他进程收集数据发给一个进程

\- 从一个进程散播数据到其他进程中

\- 减少操作

8、集体通讯：Scatter

scatter函数和广播很像，但是不同的是comm.bcast将相同的数据发送给所有在监听的进程，comm.scatter可以将数据放在数据中，发送给不同的进程。

![1095387-20181019163936480-2146818012.png][]

comm.scatter函数接收一个array，根据进程的rank将其中的元素发给不同的进程，第一个元素发送给进程0，第二个元素发给进程1，以此类推。

测试用例：

from mpi4py import MPI
    
    comm = MPI.COMM_WORLD
    rank = comm.Get_rank()
    
    # array_to_share = ["a","b","c","d","e","f","g","h","i","j"]
    if rank == 0:
        array_to_share = [0,1,2,3,4,5,6,7,8,9]
    else:
        array_to_share = None
    
    recvbuf = comm.scatter(array_to_share, root=0)
    print("Process = %d  recvbuf = %s" %(rank, recvbuf))

执行结果：

![1095387-20181019164732535-399075072.png][]

注意：列表中的元素个数，需要个进程保持一致。否则会出现如下错误。

![1095387-20181019165002404-1395321471.png][]

9、集体通讯：gather

gather函数基本上是反向的scatter，即收集所有进程发送到root进程数据。方法如下：

recvbuf = comm.gather(sendbuf, rank_of_root_process)

sendbuf是要发送的数据，rank\_of\_root\_process代表要接收数据的进程。

![1095387-20181019171441437-602087030.png][]

测试用例：

from mpi4py import MPI
    
    
    comm = MPI.COMM_WORLD
    size = comm.Get_size()
    # print(size)
    rank = comm.Get_rank()
    data = "process %s" %rank
    # print("start %s"%data)
    data = comm.gather(data, root=0)
    # print(data)
    if rank == 0:
        print("rank = %s receiving data to other process" %rank)
        for i in range(1, size):
            #data[i] = (i+1) ** 2
            value = data[i]
            print("process %s receiving %s from process %s" %(rank, value, i))
        # print(data)

执行结果：

![1095387-20181019173941439-2146636800.png][]

10、使用Alltoall通讯

Alltoall集体通讯结合了scatter和gather的功能。在mpi4py中，有以下三类的Alltoall集体通讯。

\- comm.Alltoall(sendbuf, recvbuf);

\- comm.Alltoallv(sendbuf, recvbuf);

\- comm.Alltoallw(sendbuf, recvbuf);

Alltoall测试用例：

from mpi4py import MPI
    import numpy
    
    comm = MPI.COMM_WORLD
    size = comm.Get_size()
    rank = comm.Get_rank()
    
    a_size = 1
    
    # print("numpy arange: %s" %numpy.arange(size, dtype=int))
    senddata = (rank+1)*numpy.arange(size, dtype=int)
    recvdata = numpy.empty(size * a_size, dtype=int)
    print("senddata is %s , recvdata is %s" %(senddata, recvdata))
    # print("Recvdata is %s: , \n numpy.empty is %s" %(recvdata, numpy.empty(size * a_size, dtype=int)))
    
    comm.Alltoall(senddata, recvdata)
    print("process %s sending %s, receiving %s" %(rank, senddata, recvdata))

运行结果：

![1095387-20181022110509571-909374627.png][]

comm.alltoall方法将task j的sendbuf的第j个对象拷贝到task i中，recvbuf的第j个对象，一一对应。发送过程如图：

![1095387-20181022110645667-278950697.png][]

可以将左右两个方格看做xy轴，结果一一对应，如左图的(0,0)对应的值为0，其对应的有图的值为右图的(0,0)也为0。左图的3,4对应的值为16，右图(4,3)也为16。

P0包含的数据\[0 1 2 3 4\]，它将值0赋值给自己，1传给进程P1，2传给进程P2，3传给进程P3，以此类推。

相同的P1的数据为\[0 2 4 6 8\] ， 它将0传给P0，2传给P1，4传给P2，以此类推。

All-to-all定制通讯也叫全部交换，这种操作经常用于各种并发算法中，比如快速傅里叶变换，矩阵变换，样本排序以及一些数据库的 Join 操作。

11、简化操作

同comm.gather一样，comm.reduce接收一个数组，每一个元素是一个进程的输入，然后返回一个数组，每一个元素是进程的输出，返回给root进程。输出的元素包含了简化的结果。

简化定义如下：comm.Reduce(sendbuf, recvbuf, rank\_of\_root\_process, op = type\_of\_reduction\_operation)

这里需要注意的是，参数op和comm.gather不同，它代表你想应用在数据上的操作，mpi4py模块代表定义了一系列的简化操作，包括：

\- MPI.MAX：返回最大的元素

\- MPI.MIN：返回最小的元素

\- MPI.SUM：对所有的元素相加

\- MPI.PROD：对所有元素相乘

\- MPI.LAND：对所有元素进行逻辑操作

\- MPI.MAXLOC：返回最大值，以及拥有它的进程

\- MPI.MINLOC：返回最小值，以及拥有它的进程

测试用例：

import numpy as np
    from mpi4py import MPI
    
    comm = MPI.COMM_WORLD
    size = comm.size
    rank = comm.rank
    array_size = 3
    recvdata = np.zeros(array_size, dtype=np.int)
    senddata = (rank+1)*np.arange(size, dtype=np.int)
    print("+++++++++++++%s+++++++++++++%s++++++++++++" %(recvdata, senddata))
    print("Process %s sending %s" %(rank, senddata))
    comm.Reduce(senddata, recvdata, root=0, op=MPI.SUM)
    print("on task %s, after Reduce: data = %s" %(rank, recvdata))

执行结果：

![1095387-20181022142700991-1309201219.png][]

MPI.SUM为求和操作，过程如下：

![1095387-20181022143216420-1965465083.png][]

简化操作将每个task的第i个元素相加，然后放回到P0进程(root进程)的第i个元素中。

转载于:https://www.cnblogs.com/dukuan/p/9811057.html

[1095387-20181019100738476-1291286137.png]: /images/20211210/7f7670a9a39246b68e5969c8372f6901.png
[1095387-20181019100841605-424768067.png]: /images/20211210/2e0b1e75f01d4018840fd1d713682f1d.png
[1095387-20181019101216845-1116488953.png]: /images/20211210/7139a8867434478690db8d4d3bc3ca8b.png
[1095387-20181019103255696-1670523618.png]: /images/20211210/6117320fb20a44fa93653bfb65358f1b.png
[1095387-20181019115156873-2144384394.png]: /images/20211210/aab44db4d8f444d5a37f1855079d3982.png
[1095387-20181019142616250-2089618437.png]: /images/20211210/1d2127631a9a4cab8773c1f80c0d9af3.png
[1095387-20181019142943065-39428718.png]: /images/20211210/1b95b7f8c51e453cb33c4f1bf6341c71.png
[1095387-20181019143401681-1836138641.png]: /images/20211210/d82971c982fb4ecfa71591d38b98bcaa.png
[1095387-20181019152911198-1562690982.png]: /images/20211210/481d0c1de9a24fdaacb143ae76ac17e1.png
[1095387-20181019155136898-1744479578.png]: /images/20211210/60c19f6d347e44f69799f05a89ad990a.png
[1095387-20181019163936480-2146818012.png]: /images/20211210/04dbd0864b3b4c3aacb125cb1219c211.png
[1095387-20181019164732535-399075072.png]: /images/20211210/5d5f63aeb3964370b4420a4aba9846a4.png
[1095387-20181019165002404-1395321471.png]: /images/20211210/dac2a6e705ac46a4b1066601ba94bbc2.png
[1095387-20181019171441437-602087030.png]: /images/20211210/0e53c2e5c80242b0bc0a7607fbd2f001.png
[1095387-20181019173941439-2146636800.png]: /images/20211210/d1ac0204162043718aedf3cf534f7c85.png
[1095387-20181022110509571-909374627.png]: /images/20211210/0d7fdf8f2de241dd917c57e4cf2ae8f0.png
[1095387-20181022110645667-278950697.png]: /images/20211210/fc83043c69a449d890708bbffc302318.png
[1095387-20181022142700991-1309201219.png]: /images/20211210/6d27f5e6ece8405aa729319ea903aa77.png
[1095387-20181022143216420-1965465083.png]: /images/20211210/cf9616f6120c4f65ab26a7012f044bc0.png