Linux POSIX消息队列mq_receive偶发阻塞不返回问题排查

程序可正常编译为可执行文件，但使用完全相同的可执行文件与参数运行时，有时能正常执行并返回，有时却会卡在mq_receive处无法返回，后续代码（包括错误处理逻辑）均不执行。

阻塞代码片段 #

int64_t message[M];
for (;;) {
    // 检查是否没有更多消息需要接收
    mq_getattr(mq, &mqAttrs);
    if (mqAttrs.mq_curmsgs <= 0 && (*childProcessCounterPtr) <= 0) {
        break;
    }

    char charMessage[M * sizeof(int64_t)];
    ssize_t received = mq_receive(mq, charMessage, maxMsgSize, NULL);

    if (received == -1) { // 错误处理
        perror("mq_receive failed");
        exit(EXIT_FAILURE);
    } else { // 成功接收消息
        int messageSize = (int) received / sizeof(int64_t);
        memcpy(message, charMessage, received);

        for (int i = 0; i < messageSize; i++) {
            fprintf(outputFile, "%" PRId64 "\n", ((int64_t *) message)[i]);
        }
    }
}

相关代码片段 #

mq_open 初始化部分 #

mqd_t mq;
mq = mq_open(POSIX_MQ_NAME, O_CREAT | O_RDWR, 0666, NULL);
if (mq == -1) {
    perror("mq_open failed");
    exit(EXIT_FAILURE);
}

// 获取最大消息尺寸
struct mq_attr mqAttrs;
mq_getattr(mq, &mqAttrs);
maxMsgSize = mqAttrs.mq_msgsize;

mq_send 消息发送部分 #

int currPrimeCountInMsg = 0;
int64_t candidateNumber, message[M];

while (fscanf(file, "%" PRId64, &candidateNumber) == 1) {
    if (isPrime(candidateNumber)) {
        message[currPrimeCountInMsg] = candidateNumber; // 将质数加入消息
        currPrimeCountInMsg++;

        if (currPrimeCountInMsg == M) { // 消息已满时发送
            char charMessage[M * sizeof(int64_t)];
            memcpy(charMessage, message, M * sizeof(int64_t));

            if (mq_send(mq, charMessage, M * sizeof(int64_t), 0) == -1) { // 发送消息
                perror("mq_send (on child) failed");
               _exit(EXIT_FAILURE);
            }

            currPrimeCountInMsg = 0; // 重置消息计数器
        }
    }
}

收尾代码 #

程序末尾已执行消息队列的关闭与删除操作：

mq_close(mq);
mq_unlink(POSIX_MQ_NAME);

问题原因分析 #

1. 子进程未发送剩余消息：mq_send代码中，当fscanf循环结束后，若currPrimeCountInMsg > 0（存在未凑满M个的剩余质数），没有对应的发送逻辑，这部分消息会丢失。同时，时序巧合时会导致主进程阻塞：主进程检查到队列空但子进程仍在运行，调用mq_receive等待，此时子进程退出且无剩余消息发送，主进程就会一直阻塞。
2. 竞态条件导致判断失效：主进程中mq_getattr检查队列状态与调用mq_receive之间存在时间窗口，期间子进程可能已退出且无消息可发，导致主进程进入阻塞等待。
3. 默认阻塞接收模式：mq_receive默认以阻塞模式运行，当队列空且无发送方时，会无限期等待。

解决方法 #

1. 补充剩余消息发送逻辑：在mq_send的fscanf循环结束后，添加剩余消息的发送代码，确保所有质数都被发送：

// 处理循环结束后剩余的未发送消息
if (currPrimeCountInMsg > 0) {
    char charMessage[M * sizeof(int64_t)];
    memcpy(charMessage, message, currPrimeCountInMsg * sizeof(int64_t));
    if (mq_send(mq, charMessage, currPrimeCountInMsg * sizeof(int64_t), 0) == -1) {
        perror("mq_send remaining (on child) failed");
        _exit(EXIT_FAILURE);
    }
}

2. 使用非阻塞接收避免无限等待：修改mq_open的打开标志，添加O_NONBLOCK，并在mq_receive时处理无消息的情况：
   • 修改mq_open调用：

     mq = mq_open(POSIX_MQ_NAME, O_CREAT | O_RDWR | O_NONBLOCK, 0666, NULL);
     

   • 修改mq_receive的错误处理逻辑：

     ssize_t received = mq_receive(mq, charMessage, maxMsgSize, NULL);
     if (received == -1) {
         if (errno == EAGAIN) {
             // 无消息可接收，回到循环开头重新检查退出条件
             continue;
         } else {
             perror("mq_receive failed");
             exit(EXIT_FAILURE);
         }
     }
     

3. 优化退出判断逻辑：确保主进程仅在队列无消息且所有子进程已退出时才终止循环，结合非阻塞接收可以彻底避免竞态导致的阻塞。

内容的提问来源于stack exchange，提问作者DenizTuna