使用MPI並行化遺傳算法框架GAFT

時間 2019-11-17

標籤使用 mpi 並行遺傳算法框架 gaft 简体版

原文原文鏈接

前言

本文中做者使用MPI的Python接口mpi4py來將本身的遺傳算法框架GAFT進行多進程並行加速。並對加速效果進行了簡單測試。python

項目連接:git

GitHub: github.com/PytLab/gaft
PyPI: pypi.python.org/pypi/gaft

正文

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

例如在使用遺傳算法尋找最優結構時候一般須要調用量化軟件進行第一性原理計算結構的total energy，這是很是費時的過程; 例如咱們優化力場參數的時候，以力場計算出的能量同基準能量以前的偏差做爲適應度，也須要調用相應的力場程序獲取總能量來求取，一樣這個過程也是相對耗時的。算法

這就會致使一個問題，當咱們的種羣比較大的時候，咱們須要利用適應度信息來產生下一代種羣，這時候每一代繁殖的過程將會很耗時。但有幸的是，種羣的選擇交叉變異過程對於種羣中的個體都是相互獨立的過程，咱們能夠將這一部分進行並行處理來加速遺傳算法的迭代。編程

使用mpi4py

因爲實驗室的集羣都是MPI環境，我仍是選擇使用MPI接口來將代碼並行化，這裏我仍是用了MPI接口的Python版本mpi4py來將代碼並行化。關於mpi4py的使用，我以前寫過一篇博客專門作了介紹，能夠參見《Python多進程並行編程實踐-mpi4py的使用》app

將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複製代碼

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

主要在種羣繁衍中對種羣針對進程數進行劃分而後並行進行遺傳操做併合並子種羣完成並行，代碼改動不多。詳見:github.com/PytLab/gaft…

# 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複製代碼