创建一个在MPI进程中保持同步的计数器

Question

我在使用基本通信和组MPI2方法方面有相当多的经验,并且使用MPI进行了相当多的令人尴尬的并行仿真工作.到目前为止,我已经构建了我的代码,以便有一个调度节点和一堆工作节点.调度节点具有将与模拟器一起运行的参数文件列表.它使用参数文件为每个工作节点播种.工作节点运行其模拟,然后请求调度节点提供的另一个参数文件.运行所有参数文件后,调度节点会在关闭自身之前关闭每个工作节点.

参数文件通常命名为"Par_N.txt",其中N是标识整数(例如,-N = 1-1000).所以我在想,如果我可以创建一个计数器,并且可以在所有节点上同步这个计数器,我就可以省去调度节点,并使系统更简单一些.理论上这听起来很简单,在实践中我怀疑它有点困难,因为我需要确保计数器在被改变时被锁定等等.并且认为可能存在MPI的内置方式处理这个(事情.有什么想法吗？我在想这个吗？

Answer 1

实现共享计数器并非易事,但是一旦你执行它并将它放在某个库中,你可以用它做很多事情.

在使用MPI-2的书中,如果你要实现这些东西,你必须亲自动手,其中一个例子(代码可在线获得)是一个共享计数器."不可扩展"的应该可以运行到几十个进程 - 计数器是一个0..size-1整数的数组,每个等级一个,然后`get next work item#'操作包括锁定窗口,阅读每个人对计数器的贡献(在这种情况下,他们已经采取了多少项),更新自己的(++),关闭窗口,并计算总数.这一切都是通过被动单侧操作完成的.(更好的缩放只使用树而不是1-d数组).

所以使用你会说排名0主持柜台,并且每个人都继续做工作单位并更新计数器以获得下一个计数器,直到没有更多的工作; 然后你在障碍物或其他东西等待并最终确定.

一旦你有这样的东西 - 使用共享值来获得下一个工作单位 - 工作,那么你可以推广到更复杂的方法.因此,正如suzterpatt建议的那样,每个人在开始时"分享"工作单位的工作都很好,但是如果一些人比其他人完成得更快,该怎么办？现在通常的答案是偷工作; 每个人都将他们的工作单位列在一个队列中,然后当一个人没有工作时,它会从另一个人的另一端偷走工作单位,直到没有剩下的工作为止.这实际上是master-worker的完全分布式版本,其中不再有单个主分区工作.一旦你有一个共享计数器工作,你可以从那些互斥量,并从中可以实现出列.但是,如果简单的共享计数器运行良好,您可能不需要去那里.

更新: 好的,所以这是一个hacky尝试共享计数器 - 我在MPI-2书中的简单版本的版本:似乎工作,但我不会说比这更强大的东西(没有玩过这个东西很长一段时间).有一个简单的计数器实现(对应于MPI-2书中的非缩放版本),有两个简单的测试,一个大致对应于你的工作案例; 每个项目更新计数器以获得工作项目,然后执行"工作"(睡眠随机时间量).在每次测试结束时,打印出计数器数据结构,这是每个等级完成的增量数.

#include <mpi.h>
#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>

struct mpi_counter_t {
    MPI_Win win;
    int  hostrank ;
    int  myval;
    int *data;
    int rank, size;
};

struct mpi_counter_t *create_counter(int hostrank) {
    struct mpi_counter_t *count;

    count = (struct mpi_counter_t *)malloc(sizeof(struct mpi_counter_t));
    count->hostrank = hostrank;
    MPI_Comm_rank(MPI_COMM_WORLD, &(count->rank));
    MPI_Comm_size(MPI_COMM_WORLD, &(count->size));

    if (count->rank == hostrank) {
        MPI_Alloc_mem(count->size * sizeof(int), MPI_INFO_NULL, &(count->data));
        for (int i=0; i<count->size; i++) count->data[i] = 0;
        MPI_Win_create(count->data, count->size * sizeof(int), sizeof(int),
                       MPI_INFO_NULL, MPI_COMM_WORLD, &(count->win));
    } else {
        count->data = NULL;
        MPI_Win_create(count->data, 0, 1,
                       MPI_INFO_NULL, MPI_COMM_WORLD, &(count->win));
    }
    count -> myval = 0;

    return count;
}

int increment_counter(struct mpi_counter_t *count, int increment) {
    int *vals = (int *)malloc( count->size * sizeof(int) );
    int val;

    MPI_Win_lock(MPI_LOCK_EXCLUSIVE, count->hostrank, 0, count->win);

    for (int i=0; i<count->size; i++) {

        if (i == count->rank) {
            MPI_Accumulate(&increment, 1, MPI_INT, 0, i, 1, MPI_INT, MPI_SUM,
                           count->win);
        } else {
            MPI_Get(&vals[i], 1, MPI_INT, 0, i, 1, MPI_INT, count->win);
        }
    }

    MPI_Win_unlock(0, count->win);
    count->myval += increment;

    vals[count->rank] = count->myval;
    val = 0;
    for (int i=0; i<count->size; i++)
        val += vals[i];

    free(vals);
    return val;
}

void delete_counter(struct mpi_counter_t **count) {
    if ((*count)->rank == (*count)->hostrank) {
        MPI_Free_mem((*count)->data);
    }
    MPI_Win_free(&((*count)->win));
    free((*count));
    *count = NULL;

    return;
}

void print_counter(struct mpi_counter_t *count) {
    if (count->rank == count->hostrank) {
        for (int i=0; i<count->size; i++) {
            printf("%2d ", count->data[i]);
        }
        puts("");
    }
}

int test1() {
    struct mpi_counter_t *c;
    int rank;
    int result;

    c = create_counter(0);

    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    result = increment_counter(c, 1);
    printf("%d got counter %d\n", rank, result);

    MPI_Barrier(MPI_COMM_WORLD);
    print_counter(c);
    delete_counter(&c);
}


int test2() {
    const int WORKITEMS=50;

    struct mpi_counter_t *c;
    int rank;
    int result = 0;

    c = create_counter(0);

    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    srandom(rank);

    while (result < WORKITEMS) {
        result = increment_counter(c, 1);
        if (result <= WORKITEMS) {
             printf("%d working on item %d...\n", rank, result);
             sleep(random() % 10);
         } else {
             printf("%d done\n", rank);
         }
    }

    MPI_Barrier(MPI_COMM_WORLD);
    print_counter(c);
    delete_counter(&c);
}

int main(int argc, char **argv) {

    MPI_Init(&argc, &argv);

    test1();
    test2();

    MPI_Finalize();
}