基于MPI的并行遗传算法

用户1621951

发布于 2019-06-06 11:36:08

2.2K00

代码可运行

文章被收录于专栏：数据魔术师数据魔术师

运行总次数：0

代码可运行

基于MPI的并行遗传算法

求解港口船舶调度问题

在上一篇文章中我们大致了解到了MPI的基本概念以及其运行原理，并且学习了一些简单的MPI通信函数以及例子。在本篇中我们将会以实现遗传算法为例子，讲解一些更深入的MPI概念以及函数并投入使用。

初探并行编程技术之消息传递接口(Message Passing Interface, MPI)

遗传算法是一种通过模拟达尔文进化论的自然选择和遗传学机理的生物进化过程的计算模型，是一种通过模拟自然进化过程搜索最优解的方法。遗传算法为了模拟出自然进化的过程采取了基因编码的方式来表示一个个体，通过评估个体基因的适应度来得出其优劣程度。个体与个体之间通过遗传算子来产生新的个体，并通过个体适应度来筛选下一代，产生新的种群。循环往复，直至达到我们设定的终止条件为止。而后通过将最优个体的基因进行解码得到我们的解。

而一般编解码过程的复杂程度，适应度计算的耗时，遗传算子的操作以及种群大小都会影响到遗传算法求解速度。当我们的问题规模变大的时候，往往需要几个小时甚至几天遗传算法才能停止。

因此我们就需要用到并行计算方式去加速其求解过程，正好可以运用上MPI这一工具。

港口船舶调度问题

港口船舶调度问题可以描述为，一个泊位数为B的港口，在其工作时间T内，有N只船舶会先后停泊在港口进行作业，每只船舶v_i都有其泊位占用数l_i，到达时间为a_i，作业时长t_i，开始服务时间为s_i，占用港口为b_i~b_i+l_i。船舶调度需要满足以下约束：

（1）一个泊位在同一个时间只能由一只船占用。

（2）一个船舶只有在到达之后才能够由港口服务。

（3）船舶i的服务完成的时间si+ti如果超过T，那么该船舶则不能被服务。

假如我们有一个调度方案，服务成功的船舶集合为A，未能服务的船舶集合为B，集合A中每一条船开始服务的时间与占用的港口都已经确定，并且满足以上约束，那么我们可以用一个指标来衡量这个调度方案的优劣。通常的，如果每一条船等待的时间（亦即s_i-a_i）越短，不能服务的船只越少，并且全部船只服务完成的时间越早，那么可以说这个调度方案越优秀。那么我们可以用一个公式来表示其适应度:

其中P为可以服务的船只数量，Te表示全部船只服务完成的时间点。W_i表示该评估项的权重。该公式含义为，适应度F为所有成功服务的船的等待时间之和，未能服务的船只数量，服务完成时间的加权和。

然后我们需要进行编码，在这里我们采用整数编码，编码为船只的服务顺序，规定顺序在后的船只不允许比顺序在前的船只先服务。然后我们可以用一个1~n的数列来表示个体编码。

解码的过程根据整数编码，亦即前面所说的数列，来进行。在这里我用贪心算法来进行船只的服务，顺序遍历整数编码中的船只，对每一只船，总是将其安排到第一个可以服务的港口位置。可以证明，最优解就是按某种整数编码进行贪心得到的结果。

遗传算子在这里采用单亲繁殖的swap，insert，reverse操作。

上述模型基于MPI的实现

为了以MPI加速上述模型，我们首先需要分析模型的并行性。

纵观模型，由于遗传算法在每一代都会保持一个群体作为候选解集，而这些候选解集产生子代的过程相互独立，因此我们可以以此为切入点，将子代产生的过程的任务进行并行计算，然后再汇总作为新一代的子代。该过程可以用下面的流程图表示：

其中根节点主要进行的任务就是汇总，并向子节点分派任务，并向子节点收集其产生的子代个体。子节点主要进行的任务就是产生定量的子代。

至此我们的模型框架大致就出来了。而后需要完成的就是用MPI来实现我们的框架。

这个任务的关键点在与根节点向子节点任务的分派，以及子节点产生了定量子代个体之后个体的收集。

这个分派与收集的过程，就是典型的一对多与多对一通信的例子，恰好可以用上我们上一篇学习的MPI_Bcast与MPI_Gather函数。但是我们之前学习的通信函数都是传递MPI自带的数据类型，在这里我们要进行传递的是遗传算法的个体，而这个个体包括其整数编码以及其适应度，因此我们还需要进行额外的操作。

为了在MPI通信中传递自定义的数据，MPI并行库提供了多种多样的方式供我们选用。

我们可以定义一个新的类型，其由已知MPI类型构成。在这里我们介绍一个最通用的类型生成器，MPI_Type_struct，这个函数允许我们定义自己的结构体并在MPI通信中进行传输。其函数原型为：

例如我们想定义这样一个结构，其构成为：

Struct MyStruct {
    Int array[7];
    Double fitness;
};

根据C++内存对齐原则(这里的int为4字节，double为8字节)，可知结构体中num数组的偏移为0，fitness浮点数的偏移为32，因此可以如下定义：

MPI_Datatype  newtype;
Int count = 2;
Int B[] = {7, 1};
Int D[] = {0, 32};
MPI_Datatype T[] = {MPI_INT, MPI_DOUBLE};
MPI_Type_struct(count, B, D, T, &newtype);

其中B数组存放的是每个块中的元素个数，D数组存放的是每一个块的偏移，T数组中存放的是块中元素的类型(该类型必须是MPI已知类型，包括自定义类型)，结果的类型句柄存放在newtype中。

得到了新结构的类型句柄还不能直接用，我们要需要向MPI提交这个类型，用法就是调用MPI_Type_commit函数，其原型为：

即如下使用：

MPI_Type_commit(&newtype);

// …use newtype
MPI_Send(buf, 1, newtype, dest, tag, comm);

至此，我们就可以像对待MPI原生类型一样使用我们定义的新类型了。

但是这样的自定义结构有一个缺点，就是只能传输结构体内的数据。意思就是，假如我有一个结构体如下定义：

Struct MyStruct {
    Int *array;
    Double fitness;
};

其中array是一个int型指针，程序运行时我会动态给他分配一段内存，而进行MPI通信的时候，我也希望传输的是这个数组指针指向的内存，但是自定义结构无法完成该任务，只能传输这个指针，而单单这个指针的值毫无意义。

为了完成这个工作，MPI给我们提供了一个强大的工具，MPI_Pack，能够将数据打包到一段内存中，其函数原型为：

参数inbuf, incount, datatype, outcount, comm与之前所述的函数参数没什么区别。参数outbuf表示的是我们打包的数据存放的缓冲区的起始地址，参数position表示的是每一次打包操作完成后内存偏移。如下图所示：

再回到之前的例子中，我们有一个这样的结构体

Struct MyStruct {
    Int *array;
    Double fitness;
};

假设我们有一这样一个结构的变量var，并且已经给var中array分配了n个int类型内存大小，那么为了传输这段结构，我们可以用以下代码进行打包

Char outbuf [outcount];
Int position = 0;
// 打包array指向的数组
MPI_Pack(var.array, n,MPI_INT, outbuf, outcount, &position, MPI_COMM_WORLD);

现在outbuf数组里面存有我们打包的数据，其数据大小为position。在进行数据传输的时候我们将outbuf看成一个由position个MPI_PACKED类型组成缓冲区进行发送，接受进程通过常规MPI通信函数接受到这一段内存之后，可以通过MPI_Unpack函数进行数据解包。其函数原型为：

MPI_Unpack函数的功能与MPI_Pack相反，用以下代码可以将接收到的数据解包：

Char inbuf [insize];
MyStruct var;
Var.array = new int[n];    // n为array数组大小 
… // 接收pack的打包数据到inbuf内
Int position = 0;
// 解包数据到array指向的数组
MPI_Unpack(inbuf, insize, &position, var.array, n, MPI_INT, MPI_COMM_WORLD);
// 解包数据到fitness
MPI_Unpack(inbuf, insize, &position, var.fitness, 1, MPI_DOUBLE, MPI_COMM_WORLD);

需要注意，解包的顺序与数据类型、大小必须与打包的完全一致，否则解包的时候将会出错。

至此，我们已经成功传输了自定义的数据。

算法模型

遗传个体与种群定义如下：

typedef unsigned short No;      // 船舶编号
typedef double Fitness;         // 适应度，用浮点数表示
typedef std::vector<No> Order;  // 整数编码，表示船舶服务顺序
typedef std::pair<Order, Fitness> Individual;  // 遗传个体
typedef std::vector<Individual> GROUP;  // 遗传种群

算法流程如下，先进行种群初始化，然后进行一定次数的迭代，迭代完成后搜寻种群中

Individual GA::start_evl()
{
    // 在根节点初始化种群，并将其广播到每一个子节点
    init();

    // 为了简单起见，这里进行固定次数迭代
    for (int i = 0; i < iter_times; ++i)
    {
        iter();
    }

    // 迭代完成后在根节点的种群中选取最优个体
    Individual best;
    get_best(&best);

    return best;
}

在init方法中，根节点产生一个随机解，然后将该随机解广播到每一个子节点，随后每一个节点都将该随机解填充到自己的种群，也就是group_1中。

void GA::init()
{
    Individual init_inv;

    // 在根节点产生一个随机个体
    for (int i = 1; i <= nShips; ++i) init_inv.first.push_back(i);
    if (rnk == 0) disturb(init_inv.first);

    // 广播随机个体
    MPI_Bcast(&init_inv.first[0], nShips, MPI_UNSIGNED_SHORT, 0, MPI_COMM_WORLD);
    // 根据解计算适应度
    init_inv.second = berth.decode(init_inv.first);

    // 将初始解填充到初始种群
    for (int i = 0; i < m_group_size; ++i)
    {
        group_1.push_back(init_inv);
    }
}

在iter方法中，每一个节点产生定量的子代到另外一个种群缓冲区group_2中，产生完毕后将group_2所有个体进行打包，然后与其他节点的group_2用MPI_Allgather进行呼唤，并且将收集到的所有个体解包到group_1中。那么一次迭代完成后，每一个节点的种群group_1便是更新后的种群。

void GA::iter()
{
    // group_2为节点产生的子代的缓冲区
    GROUP group_2;
    double sum_fitness = 0;
    // 计算总适应度和
    for (int i = 0; i < group_size; ++i)
    {
        sum_fitness += group_1[i].second;
    }
    // 产生定量子代
    for (int i = 0; i < per_size; ++i)
    {
        double p1;          // 轮盘赌个体选择概率
        double p2;          // 染色体选择概率操作
        double acc_p;       // 累加概率
        int chose_index;    // 轮盘赌选择个体下标
        // 轮盘赌选择个体
        p1 = random_probability();
        for (chose_index = 0, acc_p = 0; acc_p <= p1 && chose_index < group_size; ++chose_index)
        {
            acc_p += group_1[chose_index].second / sum_fitness;
        }
        chose_index -= 1;
        Individual& chose_indiv = group_1[chose_index];
        // 进行无性繁殖
        p2 = random_probability();
        //获取遗传算子
        Gene_func p_func = get_gen_func_by_p(p2);   
        Individual individual;
        individual.first = (this->*p_func)(chose_indiv.first);
        // 计算个体适应度
        individual.second = t_berth.decode(individual.first);
        group_2[i] = move(individual);
    }
    int position = 0;
    // 将自身产生的个体打包到发送缓冲区perbuf中
    for (int i = 0; i < per_size; ++i)
    {
        MPI_Pack(&group_2.at(i).first[0], nShips, MPI_UNSIGNED_SHORT, perbuf, packsize, &position, MPI_COMM_WORLD);
        MPI_Pack(&group_2.at(i).second, 1, MPI_DOUBLE, perbuf, packsize, &position, MPI_COMM_WORLD);
    }
    // 所有节点产生的个体互相交换
    MPI_Allgather(perbuf, position, MPI_PACKED, gbuf, position, MPI_PACKED, MPI_COMM_WORLD);
    int insize = position * sz;
    position = 0;
    for (int i = 0; i < per_size*sz; ++i)
    {
        MPI_Unpack(gbuf, insize, &position, &group_1[i].first[0], nShips, MPI_UNSIGNED_SHORT, MPI_COMM_WORLD);
        MPI_Unpack(gbuf, insize, &position, &group_1[i].second, 1, MPI_DOUBLE, MPI_COMM_WORLD);
    }
}

最后在根节点的种群中选择最优个体输出即可。

运行结果

分别分配1个任务，24个任务，96个任务进行并行计算，运行结果如下