一、概述
CPU從三十多年前的8086，到十年前的奔騰，再到當下的多核i7。一開始，以單核cpu的主頻為目標，架構(gòu)的改良和集成電路工藝的進步使得cpu的性能高速上升，單核cpu的主頻從老爺車的MHz階段一度接近4GHz高地。然而，也因為工藝和功耗等的限制，單核cpu遇到了人生的天花板，急需轉(zhuǎn)換思維，以滿足無止境的性能需求。多核cpu在此登上歷史舞臺。給你的老爺車多加兩個引擎，讓你有法拉利的感覺。現(xiàn)時代，連手機都到處叫囂自己有4核8核處理器的時代，PC就更不用說了。
扯遠了，anyway，對于俺們程序員來說，如何利用如此強大的引擎完成我們的任務(wù)才是我們要考慮的。隨著大規(guī)模數(shù)據(jù)處理、大規(guī)模問題和復雜系統(tǒng)求解需求的增加，以前的單核編程已經(jīng)有心無力了。如果程序一跑就得幾個小時，甚至一天，想想都無法原諒自己。那如何讓自己更快的過度到高大上的多核并行編程中去呢?哈哈，廣大人民的力量!
目前工作中我所接觸到的并行處理框架主要有MPI、OpenMP和MapReduce(Hadoop)三個(CUDA屬于GPU并行編程，這里不提及)。MPI和Hadoop都可以在集群中運行，而OpenMP因為共享存儲結(jié)構(gòu)的關(guān)系，不能在集群上運行，只能單機。另外，MPI可以讓數(shù)據(jù)保留在內(nèi)存中，可以為節(jié)點間的通信和數(shù)據(jù)交互保存上下文，所以能執(zhí)行迭代算法，而Hadoop卻不具有這個特性。因此，需要迭代的機器學習算法大多使用MPI來實現(xiàn)。當然了，部分機器學習算法也是可以通過設(shè)計使用Hadoop來完成的。(淺見，如果錯誤，希望各位不吝指出，謝謝)。
本文主要介紹Python環(huán)境下MPI編程的實踐基礎(chǔ)。
二、MPI與mpi4py
MPI是MessagePassingInterface的簡稱，也就是消息傳遞。消息傳遞指的是并行執(zhí)行的各個進程具有自己獨立的堆棧和代碼段，作為互不相關(guān)的多個程序獨立執(zhí)行，進程之間的信息交互完全通過顯示地調(diào)用通信函數(shù)來完成。
Mpi4py是構(gòu)建在mpi之上的python庫，使得python的數(shù)據(jù)結(jié)構(gòu)可以在進程(或者多個cpu)之間進行傳遞。
2.1、MPI的工作方式
很簡單，就是你啟動了一組MPI進程，每個進程都是執(zhí)行同樣的代碼!然后每個進程都有一個ID，也就是rank來標記我是誰。什么意思呢?假設(shè)一個CPU是你請的一個工人，共有10個工人。你有100塊磚頭要搬，然后很公平，讓每個工人搬10塊。這時候，你把任務(wù)寫到一個任務(wù)卡里面，讓10個工人都執(zhí)行這個任務(wù)卡中的任務(wù)，也就是搬磚!這個任務(wù)卡中的“搬磚”就是你寫的代碼。然后10個CPU執(zhí)行同一段代碼。需要注意的是，代碼里面的所有變量都是每個進程獨有的，雖然名字相同。
例如，一個腳本test.py，里面包含以下代碼：
frommpi4pyimportMPI
print("helloworld'')
print("myrankis:%d"%MPI.rank)
然后我們在命令行通過以下方式運行：
#mpirun–np5pythontest.py
-np5指定啟動5個mpi進程來執(zhí)行后面的程序。相當于對腳本拷貝了5份，每個進程運行一份，互不干擾。在運行的時候代碼里面唯一的不同，就是各自的rank也就是ID不一樣。所以這個代碼就會打印5個helloworld和5個不同的rank值，從0到4.
2.2、點對點通信
點對點通信(Point-to-PointCommunication)的能力是信息傳遞系統(tǒng)最基本的要求。意思就是讓兩個進程直接可以傳輸數(shù)據(jù)，也就是一個發(fā)送數(shù)據(jù)，另一個接收數(shù)據(jù)。接口就兩個，send和recv，來個例子：
importmpi4py.MPIasMPI
comm=MPI.COMM_WORLD
comm_rank=comm.Get_rank()
comm_size=comm.Get_size()
#pointtopointcommunication
data_send=[comm_rank]*5
comm.send(data_send,dest=(comm_rank+1)%comm_size)
data_recv=comm.recv(source=(comm_rank-1)%comm_size)
print("myrankis%d,andIreceived:"%comm_rank)
printdata_recv
啟動5個進程運行以上代碼，結(jié)果如下：
myrankis0,andIreceived:
[4,4,4,4,4]
myrankis1,andIreceived:
[0,0,0,0,0]
myrankis2,andIreceived:
[1,1,1,1,1]
myrankis3,andIreceived:
[2,2,2,2,2]
myrankis4,andIreceived:
[3,3,3,3,3]
可以看到，每個進程都創(chuàng)建了一個數(shù)組，然后把它傳遞給下一個進程，最后的那個進程傳遞給第一個進程。comm_size就是mpi的進程個數(shù)，也就是-np指定的那個數(shù)。MPI.COMM_WORLD表示進程所在的通信組。
但這里面有個需要注意的問題，如果我們要發(fā)送的數(shù)據(jù)比較小的話，mpi會緩存我們的數(shù)據(jù)，也就是說執(zhí)行到send這個代碼的時候，會緩存被send的數(shù)據(jù)，然后繼續(xù)執(zhí)行后面的指令，而不會等待對方進程執(zhí)行recv指令接收完這個數(shù)據(jù)。但是，如果要發(fā)送的數(shù)據(jù)很大，那么進程就是掛起等待，直到接收進程執(zhí)行了recv指令接收了這個數(shù)據(jù)，進程才繼續(xù)往下執(zhí)行。所以上述的代碼發(fā)送[rank]*5沒啥問題，如果發(fā)送[rank]*500程序就會半死不活的樣子了。因為所有的進程都會卡在發(fā)送這條指令，等待下一個進程發(fā)起接收的這個指令，但是進程是執(zhí)行完發(fā)送的指令才能執(zhí)行接收的指令，這就和死鎖差不多了。所以一般，我們將其修改成以下的方式：
importmpi4py.MPIasMPI
comm=MPI.COMM_WORLD
comm_rank=comm.Get_rank()
comm_size=comm.Get_size()
data_send=[comm_rank]*5
ifcomm_rank==0:
comm.send(data_send,dest=(comm_rank+1)%comm_size)
ifcomm_rank>0:
data_recv=comm.recv(source=(comm_rank-1)%comm_size)
comm.send(data_send,dest=(comm_rank+1)%comm_size)
ifcomm_rank==0:
data_recv=comm.recv(source=(comm_rank-1)%comm_size)
print("myrankis%d,andIreceived:"%comm_rank)
printdata_recv
第一個進程一開始就發(fā)送數(shù)據(jù)，其他進程一開始都是在等待接收數(shù)據(jù)，這時候進程1接收了進程0的數(shù)據(jù)，然后發(fā)送進程1的數(shù)據(jù)，進程2接收了，再發(fā)送進程2的數(shù)據(jù)……知道最后進程0接收最后一個進程的數(shù)據(jù)，從而避免了上述問題。
一個比較常用的方法是封一個組長，也就是一個主進程，一般是進程0作為主進程leader。主進程將數(shù)據(jù)發(fā)送給其他的進程，其他的進程處理數(shù)據(jù)，然后返回結(jié)果給進程0。換句話說，就是進程0來控制整個數(shù)據(jù)處理流程。
2.3、群體通信
點對點通信是A發(fā)送給B，一個人將自己的秘密告訴另一個人，群體通信(CollectiveCommunications)像是拿個大喇叭，一次性告訴所有的人。前者是一對一，后者是一對多。但是，群體通信是以更有效的方式工作的。它的原則就一個：盡量把所有的進程在所有的時刻都使用上!我們在下面的bcast小節(jié)講述。
群體通信還是發(fā)送和接收兩類，一個是一次性把數(shù)據(jù)發(fā)給所有人，另一個是一次性從所有人那里回收結(jié)果。
1)廣播bcast
將一份數(shù)據(jù)發(fā)送給所有的進程。例如我有200份數(shù)據(jù)，有10個進程，那么每個進程都會得到這200份數(shù)據(jù)。
importmpi4py.MPIasMPI
comm=MPI.COMM_WORLD
comm_rank=comm.Get_rank()
comm_size=comm.Get_size()
ifcomm_rank==0:
data=range(comm_size)
data=comm.bcast(dataifcomm_rank==0elseNone,root=0)
print'rank%d,got:'%(comm_rank)
printdata
結(jié)果如下：
rank0,got:
[0,1,2,3,4]
rank1,got:
[0,1,2,3,4]
rank2,got:
[0,1,2,3,4]
rank3,got:
[0,1,2,3,4]
rank4,got:
[0,1,2,3,4]
Root進程自己建了一個列表，然后廣播給所有的進程。這樣所有的進程都擁有了這個列表。然后愛干嘛就干嘛了。
對廣播最直觀的觀點是某個特定進程將數(shù)據(jù)一一發(fā)送給每個進程。假設(shè)有n個進程，那么假設(shè)我們的數(shù)據(jù)在0進程，那么0進程就需要將數(shù)據(jù)發(fā)送給剩下的n-1個進程，這是非常低效的，復雜度是O(n)。那有沒有高效的方式?一個最常用也是非常高效的手段是規(guī)約樹廣播：收到廣播數(shù)據(jù)的所有進程都參與到數(shù)據(jù)廣播的過程中。首先只有一個進程有數(shù)據(jù)，然后它把它發(fā)送給第一個進程，此時有兩個進程有數(shù)據(jù);然后這兩個進程都參與到下一次的廣播中，這時就會有4個進程有數(shù)據(jù)，……，以此類推，每次都會有2的次方個進程有數(shù)據(jù)。通過這種規(guī)約樹的廣播方法，廣播的復雜度降為O(logn)。這就是上面說的群體通信的高效原則：充分利用所有的進程來實現(xiàn)數(shù)據(jù)的發(fā)送和接收。
2)散播scatter
將一份數(shù)據(jù)平分給所有的進程。例如我有200份數(shù)據(jù)，有10個進程，那么每個進程會分別得到20份數(shù)據(jù)。
importmpi4py.MPIasMPI
comm=MPI.COMM_WORLD
comm_rank=comm.Get_rank()
comm_size=comm.Get_size()
ifcomm_rank==0:
data=range(comm_size)
printdata
else:
data=None
local_data=comm.scatter(data,root=0)
print'rank%d,got:'%comm_rank
printlocal_data
結(jié)果如下：
[0,1,2,3,4]
rank0,got:
0
rank1,got:
1
rank2,got:
2
rank3,got:
3
rank4,got:
4
這里root進程創(chuàng)建了一個list，然后將它散播給所有的進程，相當于對這個list做了劃分，每個進程獲得等分的數(shù)據(jù)，這里就是list的每一個數(shù)。(主要根據(jù)list的索引來劃分，list索引為第i份的數(shù)據(jù)就發(fā)送給第i個進程)。如果是矩陣，那么就等分的劃分行，每個進程獲得相同的行數(shù)進行處理。
需要注意的是，MPI的工作方式是每個進程都會執(zhí)行所有的代碼，所以每個進程都會執(zhí)行scatter這個指令，但只有root執(zhí)行它的時候，它才兼?zhèn)浒l(fā)送者和接收者的身份(root也會得到屬于自己的數(shù)據(jù))，對于其他進程來說，他們都只是接收者而已。
3)收集gather
那有發(fā)送，就有一起回收的函數(shù)。Gather是將所有進程的數(shù)據(jù)收集回來，合并成一個列表。下面聯(lián)合scatter和gather組成一個完成的分發(fā)和收回過程：
importmpi4py.MPIasMPI
comm=MPI.COMM_WORLD
comm_rank=comm.Get_rank()
comm_size=comm.Get_size()
ifcomm_rank==0:
data=range(comm_size)
printdata
else:
data=None
local_data=comm.scatter(data,root=0)
local_data=local_data*2
print'rank%d,gotanddo:'%comm_rank
printlocal_data
combine_data=comm.gather(local_data,root=0)
ifcomm_rank==0:
printcombine_data
結(jié)果如下：
[0,1,2,3,4]
rank0,gotanddo:
0
rank1,gotanddo:
2
rank2,gotanddo:
4
rank4,gotanddo:
8
rank3,gotanddo:
6
[0,2,4,6,8]
Root進程將數(shù)據(jù)通過scatter等分發(fā)給所有的進程，等待所有的進程都處理完后(這里只是簡單的乘以2)，root進程再通過gather回收他們的結(jié)果，和分發(fā)的原則一樣，組成一個list。Gather還有一個變體就是allgather，可以理解為它在gather的基礎(chǔ)上將gather的結(jié)果再bcast了一次。啥意思?意思是root進程將所有進程的結(jié)果都回收統(tǒng)計完后，再把整個統(tǒng)計結(jié)果告訴大家。這樣，不僅root可以訪問combine_data，所有的進程都可以訪問combine_data了。
4)規(guī)約reduce
規(guī)約是指不但將所有的數(shù)據(jù)收集回來，收集回來的過程中還進行了簡單的計算，例如求和，求最大值等等。為什么要有這個呢?我們不是可以直接用gather全部收集回來了，再對列表求個sum或者max就可以了嗎?這樣不是累死組長嗎?為什么不充分使用每個工人呢?規(guī)約實際上是使用規(guī)約樹來實現(xiàn)的。例如求max，完成可以讓工人兩兩pk后，再返回兩兩pk的最大值，然后再對第二層的最大值兩兩pk，直到返回一個最終的max給組長。組長就非常聰明的將工作分配下工人高效的完成了。這是O(n)的復雜度，下降到O(logn)(底數(shù)為2)的復雜度。
importmpi4py.MPIasMPI
comm=MPI.COMM_WORLD
comm_rank=comm.Get_rank()
comm_size=comm.Get_size()
ifcomm_rank==0:
data=range(comm_size)
printdata
else:
data=None
local_data=comm.scatter(data,root=0)
local_data=local_data*2
print'rank%d,gotanddo:'%comm_rank
printlocal_data
all_sum=comm.reduce(local_data,root=0,op=MPI.SUM)
ifcomm_rank==0:
print'sumis:%d'%all_sum
結(jié)果如下：
[0,1,2,3,4]
rank0,gotanddo:
0
rank1,gotanddo:
2
rank2,gotanddo:
4
rank3,gotanddo:
6
rank4,gotanddo:
8
sumis:20
可以看到，最后可以得到一個sum值。
以上內(nèi)容為大家介紹了Python多核編程mpi4py實踐，希望對大家有所幫助，如果想要了解更多Python相關(guān)知識，請關(guān)注IT培訓機構(gòu):千鋒教育。http://www.e7g2kmi.cn/
京公網(wǎng)安備 11010802030320號