使用 MPI for Python 並行化遺傳演算法

（點擊

上方藍字

，快速關注我們）

來源：伯樂在線專欄作者 - iPytLab

如有好文章投稿，請點擊 → 這裡了解詳情

前言

本文中作者使用MPI的Python介面mpi4py來將自己的遺傳演算法框架GAFT進行多進程並行加速。並對加速效果進行了簡單測試。

項目鏈接:

• 
  

  GitHub: https://github.com/PytLab/gaft



• 
  

  PyPI: https://pypi.python.org/pypi/gaft

正文

我們在用遺傳演算法優化目標函數的時候，函數通常都是高維函數，其導數一般比較難求取。這樣我們的適應度函數計算通常都是比較費時的計算。

例如在使用遺傳演算法尋找最優結構時候通常需要調用量化軟體進行第一性原理計算結構的total energy，這是非常費時的過程; 例如我們優化力場參數的時候，以力場計算出的能量同基準能量之前的誤差作為適應度，也需要調用相應的力場程序獲取總能量來求取，同樣這個過程也是相對耗時的。

這就會導致一個問題，當我們的

種群比較大

的時候，我們需要利用適應度信息來產生下一代種群，這時候每一代繁殖的過程將會很耗時。但有幸的是，種群的選擇交叉變異過程對於種群中的個體都是

相互獨立

的過程，我們可以將這一部分進行並行處理來加速遺傳演算法的迭代。

使用mpi4py

由於實驗室的集群都是MPI環境，我還是選擇使用MPI介面來將代碼並行化，這裡我還是用了MPI介面的Python版本mpi4py來將代碼並行化。關於mpi4py的使用，我之前寫過一篇博客專門做了介紹，可以參見《Python多進程並行編程實踐-mpi4py的使用》

將mpi4py的介面進一步封裝

為了能讓mpi的介面在GAFT中更方便的調用，我決定將mpi4py針對遺傳演算法中需要用的地方進行進一步封裝，為此我單獨寫了個MPIUtil類, 詳細代碼參見gaft/mpiutil.py。

封裝通信子常用的介面

例如進程同步, 獲取rank，進程數，判斷是否為主進程等。

class

MPIUtil

(

object

)

:

    

def

__init__

(

self

)

:

        

logger_name

=

"gaft.{}"

.

format

(

self

.

__class__

.

__name__

)

        

self

.

_logger

=

logging

.

getLogger

(

logger_name

)

 

    

# Wrapper for common MPI interfaces.

    

def

barrier

(

self

)

:

        

if

MPI_INSTALLED

:

            

mpi_comm

=

MPI

.

COMM_WORLD

            

mpi_comm

.

barrier

()

 

    

@

property

    

def

rank

(

self

)

:

        

if

MPI_INSTALLED

:

            

mpi_comm

=

MPI

.

COMM_WORLD

            

return

mpi_comm

.

Get_rank

()

        

else

:

            

return

0

 

    

@

property

    

def

size

(

self

)

:

        

if

MPI_INSTALLED

:

            

mpi_comm

=

MPI

.

COMM_WORLD

            

return

mpi_comm

.

Get_size

()

        

else

:

            

return

1

 

    

@

property

    

def

is_master

(

self

)

:

        

return

self

.

rank

==

0

組內集合通信介面

由於本次並行化的任務是在種群繁衍時候進行的，因此我需要將上一代種群進行劃分，劃分成多個子部分，然後在每個進程中對劃分好的子部分進行選擇交叉變異等遺傳操作。在最後將每個字部分得到的子種群進行收集合併。為此寫了幾個劃分和收集的介面:

def

split_seq

(

self

,

sequence

)

:

    

"""

    Split the sequence according to rank and processor number.

    """

    

starts

=

[

i

for

i

in

range

(

0

,

len

(

sequence

),

len

(

sequence

)

//

self

.

size

)]

    

ends

=

starts

[

1

:

]

+

[

len

(

sequence

)]

    

start

,

end

=

list

(

zip

(

starts

,

ends

))[

self

.

rank

]

 

    

return

sequence

[

start

:

end

]

 

def

split_size

(

self

,

size

)

:

    

"""

    Split a size number(int) to sub-size number.

    """

    

if

size

<

self

.

size

:

        

warn_msg

=

(

"Splitting size({}) is smaller than process "

+

                    

"number({}), more processor would be "

+

                    

"superflous"

).

format

(

size

,

self

.

size

)

        

self

.

_logger

.

warning

(

warn_msg

)

        

splited_sizes

=

[

1

]

*

size

+

[

0

]

*

(

self

.

size

-

size

)

    

elif

size

%

self

.

size

!=

0

:

        

residual

=

size

%

self

.

size

        

splited_sizes

=

[

size

//

self

.

size

]

*

self

.

size

        

for

i

in

range

(

residual

)

:

            

splited_sizes

[

i

]

+=

1

    

else

:

        

splited_sizes

=

[

size

//

self

.

size

]

*

self

.

size

 

    

return

splited_sizes

[

self

.

rank

]

 

def

merge_seq

(

self

,

seq

)

:

    

"""

    Gather data in sub-process to root process.

    """

    

if

self

.

size

==

1

:

        

return

seq

 

    

mpi_comm

=

MPI

.

COMM_WORLD

    

merged_seq

=

mpi_comm

.

allgather

(

seq

)

    

return

list

(

chain

(

*

merged_seq

))

用於限制程序在主進程執行的裝飾器

有些函數例如日誌輸出，數據收集的函數，我只希望在主進程執行，為了方便，寫了個裝飾器來限制函數在主進程中執行:

def

master_only

(

func

)

:

    

"""

    Decorator to limit a function to be called

    only in master process in MPI env.

    """

    

@

wraps

(

func

)

    

def

_call_in_master_proc

(

*

args

,

**

kwargs

)

:

        

if

mpi

.

is_master

:

            

return

func

(

*

args

,

**

kwargs

)

 

    

return

_call_in_master_proc

在遺傳演算法主循環中添加並行

主要在種群繁衍中對種群針對進程數進行劃分然後並行進行遺傳操作併合並子種群完成並行，代碼改動很少。詳見:https://github.com/PytLab/gaft/blob/master/gaft/engine.py#L67

# Enter evolution iteration.

for

g

in

range

(

ng

)

:

    

# Scatter jobs to all processes.

    

local_indvs

=

[]

    

local_size

=

mpi

.

split_size

(

self

.

population

.

size

//

2

)

 

    

# Fill the new population.

    

for

_

in

range

(

local_size

)

:

        

# Select father and mother.

        

parents

=

self

.

selection

.

select

(

self

.

population

,

fitness

=

self

.

fitness

)

        

# Crossover.

        

children

=

self

.

crossover

.

cross

(

*

parents

)

        

# Mutation.

        

children

=

[

self

.

mutation

.

mutate

(

child

)

for

child

in

children

]

        

# Collect children.

        

local_indvs

.

extend

(

children

)

 

    

# Gather individuals from all processes.

    

indvs

=

mpi

.

merge_seq

(

local_indvs

)

    

# The next generation.

    

self

.

population

.

individuals

=

indvs

測試加速效果

測試一維搜索

下面我針對項目中的一維優化的例子進行並行加速測試來看看加速的效果。例子代碼在/examples/ex01/

由於自己本子核心數量有限，我把gaft安裝在實驗室集群上使用MPI利用多核心進行並行計算一維優化，種群大小為50，代數為100，針對不同核心數可以得到不同的優化時間和加速比。可視化如下圖:

核心數與優化時間的關係:

核心數與加速比:

測試力場優化

這裡我對自己要研究的對象進行加速測試，這部分代碼並未開源，針對每個個體的適應度計算都需要調用其他的計算程序，因此此過程相比直接有函數表達式的目標函數計算要耗時很多。

同樣，我針對不同核心數看看使用MPI在集群上加速的效果:

核心數與優化時間的關係:

核心數與加速比:

可見針對上述兩個案例，MPI對遺傳演算法的加速還是比較理想的，程序可以扔到集群上飛起啦~~~

總結

本文主要總結了使用mpi4py對遺傳演算法進行並行化的方法和過程，並對加速效果進行了測試，可見MPI對於遺傳演算法框架GAFT的加速效果還是較為理想的。帶有MPI並行的遺傳演算法框架目前也已更新並上傳至GitHub(https://github.com/PytLab/gaft) 歡迎圍觀[]~(￣▽￣)~*

看完本文有收穫？請轉發分享給更多人

關注「大數據與機器學習文摘」，成為Top 1%

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！

本站內容充實豐富，博大精深，小編精選每日熱門資訊，隨時更新，點擊「搶先收到最新資訊」瀏覽吧！

請您繼續閱讀更多來自 Python開發者 的精彩文章: