线程中阻塞 MPI_Recv 的 CPU 使用率

Posted 2023-03-27

【中文标题】线程中阻塞 MPI_Recv 的 CPU 使用率

【英文标题】：CPU usage of a blocking MPI_Recv in a thread

【发布时间】：2018-08-18 06:57:48

【问题描述】：

我必须编写一个 MPI 库，其中每个进程都在执行一些独立的任务，但应该对可能从其他进程不可预测地发送的一些消息做出反应。 这些消息的发送和接收都是库的一部分，我不能假设库函数会被频繁调用以跟踪立即发送的进度或检查接收队列。如果接收进程正在做一些计算，发送进程可能会被阻塞一段不可预知的时间。

我目前感兴趣的解决方案是让每个 MPI 进程生成一个 pthread 线程，该线程固定在其自己的 CPU 上，使用循环中的阻塞接收来接收这些消息。正如我所担心的那样，我的实验表明该线程占用了一半的 CPU 时间（我希望阻塞接收能够以某种方式与内核一起工作以避免这种情况）。

我通过在一个进程的一个线程中使用假计算函数、在另一个线程中使用阻塞接收以及另一个发送消息以供第一个进程接收的进程来衡量这种行为，但仅当计算完成时，这在计算之后和发送消息之前由屏障强制执行。每个进程只有一个线程参与屏障，因此它可以工作。这可以确保接收线程在其他线程进行计算时真正卡住等待消息。然后我测量计算时间。设置如下所示：

             +                 +
             | P0              | P1
          +--+--+              |
          |     |              |
compute() |     | Recv(1)      |
          |     |              |
          +--------------------+ Barrier
          |     |              |
          |     |              | Send(0)
          |     |              |
          +     +              +

我尝试将阻塞接收更改为 MPI_Iprobe 循环，该循环会将 CPU 让给另一个线程，这样如果没有消息要接收，则不会占用太多 CPU 时间，因为我使用了 sleep(0) 函数作为pthread_yield 或sched_yield 需要特权才能将调度策略更改为实时策略我不确定我是否需要。 然后nanosleep函数来控制间隔。

一个简单的版本如下所示：

int flag;
while (1)

    MPI_Iprobe(1, 0, comm, &flag, MPI_STATUS_IGNORE);
    if (flag == 1) break;
    sleep(0);


MPI_Recv(NULL, 0, MPI_INT, 1, 0, comm, MPI_STATUS_IGNORE);

这似乎可以解决我的问题。在我的实验中，计算线程所花费的时间几乎与没有其他线程一样，而如果我只使用阻塞接收 MPI_Recv 或者我没有使用 sleep(0)，则这次是两倍。

这里是我用来衡量这个的代码：

#define COMPUTE_LOOP_ITER 200000000

void compute()

    int p[2];
    for (int i = 0; i < COMPUTE_LOOP_ITER; ++i)
    
        p[i%2] = i;
    


void * thread_recv_message(void * arg)

    MPI_Comm comm = *(MPI_Comm*) arg;

    int flag;
    while (1)
    
        MPI_Iprobe(1, 0, comm, &flag, MPI_STATUS_IGNORE);

        if (flag == 1) break;
        sleep(0);
    

    MPI_Recv(NULL, 0, MPI_INT, 1, 0, comm, MPI_STATUS_IGNORE);

    return NULL;


// Returns the compute() time on p0, 0 on others
double test(MPI_Comm comm)

    int s, p;
    double res = 0;
    MPI_Comm_rank(comm, &s);
    MPI_Comm_size(comm, &p);

    if (p != 2)
    
        fprintf(stderr, "Requires 2 processes and no more in comm\n");
        fflush(stderr);

        MPI_Abort(comm, 1);
    

    // Pin each process to its own core
    int cpuid = sched_getcpu();
    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    CPU_SET(cpuid, &cpuset);
    pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);

    if (s == 0)
    
        pthread_t thr;
        pthread_attr_t attr;

        // Make sure the new thread is pinned on the same core
        pthread_attr_init(&attr);
        pthread_attr_setaffinity_np(&attr, sizeof(cpu_set_t), &cpuset);

        pthread_create(&thr, &attr, thread_recv_message, &comm);

        double t1,t2;
        t1 = MPI_Wtime();

        compute();

        t2 = MPI_Wtime();

        MPI_Barrier(comm);

        res = t2 - t1;
        pthread_join(thr, NULL);
    
    else // s == 1
    
        MPI_Barrier(comm);
        MPI_Send(NULL, 0, MPI_INT, 0, 0, comm);
    

    MPI_Barrier(comm);

    return res;

由于我几乎没有使用 MPI 的经验，也没有使用线程的经验，所以这个解决方案对我来说似乎很脆弱，我不知道我是否可以依赖它。

我在使用 Linux 内核版本 4.4.0 的 Ubuntu 16.04 上使用 mpich 3.2

这个问题主要是就这个问题和我目前的解决方案征求意见或讨论。如果需要，我可以解释更多我的测试方法或提供更多代码。

【问题讨论】：

如果这种探测循环比阻塞的接收调用快，那就大错特错了。探测循环做了很多工作。

我测量的不是接收消息的时间，而是计算在尝试接收消息时在同一核心上的并行线程中花费的时间。在进行计算时实际上没有收到任何消息。

那么有些事情是非常、非常、非常错误的。探测循环做了很多工作，不断地主动检查从未到达的消息。

这就是为什么另一个线程中的计算需要多两倍的时间。如果我在阻塞接收之前移除探测循环，它大致相同。我比较的不是在接收之前有没有探测循环的效率。但是当探测循环意识到队列中没有消息而不是立即尝试时，尝试将 CPU 让给计算线程（这里通过睡眠）。

【参考方案1】：

由于在compute 和thread_recv_message 之间的示例中没有数据依赖关系，因此很难判断接收到的数据究竟做了什么。我也不确定句子片段“应该对可能从其他进程不可预测地发送的一些消息做出反应”中使用的“不可预测”的具体含义。

如果您确定 rank x 将在某个时候向 rank y 发送数据，那么 MPI_Irecv 和 MPI_Test 将实现这种通信样式而不阻塞线程调用或完成接收请求。您可以将这些 Irecv 和 Test 调用与您的计算交错，或者可能每 2、64 或 128 次计算循环迭代调用它们一次，或者任何合适的方法。

如果接收等级事先不知道它将从哪个等级接收或数据的大小，那么您可能希望使用MPI_Probe 或MPI_Iprobe 并使用返回的MPI_Status 结构。对Iprobe 的调用可以与您的计算交错，类似于我对Irecv 的描述。

也可以使用 MPI_Alltoall 或 MPI_Allgather 之类的集合来提供与 Probes 的许多排名相似的功能，例如通过交换包含字节计数的数组，该数组将在排名对之间发送和接收随后的点对点呼叫。如果您可以保证所有等级最终都会到达集体调用，那么这种方法可能会很好地利用只有 MPI 内部才可用的实现细节。您还可以使用等效的非阻塞集合（Ialltoall、Iallgather 等）将此步骤与您的计算重叠。