第 46 章 System V 消息队列 (System V Message Queues)

// 摘自《The Linux Programming Interface》 第 46 章
#include <sys/msg.h>
int msgget(key_t key, int msgflg);    /* 返回 msqid 或 -1 */

// 摘自《The Linux Programming Interface》 第 46 章
struct mbuf {
    long mtype;                /* 必须 > 0 */
    char mtext[1024];
};

/* 发 */
struct mbuf msg = {1, "hello"};
msgsnd(msqid, &msg, strlen("hello") + 1, 0);    /* 阻塞 */

/* 收 */
msgrcv(msqid, &msg, sizeof(msg.mtext), 0, 0);   /* 取任意第一条，阻塞 */
printf("type=%ld body=%s\n", msg.mtype, msg.mtext);

/* 删除——msg/sem 立即；shm 标记 */
msgctl(msqid, IPC_RMID, NULL);

/* 取内核关联数据 */
struct msqid_ds ds;
msgctl(msqid, IPC_STAT, &ds);

/* 改 owner/permissions/msg_qbytes（4 个字段） */
ds.msg_perm.uid = newuid;
ds.msg_qbytes = new_max;
msgctl(msqid, IPC_SET, &ds);

/* Linux 扩展 */
int maxind = msgctl(0, MSG_INFO, (struct msqid_ds *)&msginfo);   /* 拿 maxind */
int msqid = msgctl(ind, MSG_STAT, &ds);                          /* 按 index 拿 id */

// 摘自《The Linux Programming Interface》 第 46 章
struct msqid_ds {
    struct ipc_perm msg_perm;       /* 权限 */
    time_t msg_stime;               /* 最后 msgsnd 时间 */
    time_t msg_rtime;               /* 最后 msgrcv 时间 */
    time_t msg_ctime;               /* 最后变化时间 */
    unsigned long __msg_cbytes;     /* 当前字节数 */
    msgqnum_t msg_qnum;             /* 当前消息数 */
    msglen_t  msg_qbytes;           /* 上限字节数 */
    pid_t msg_lspid;                /* 最后 msgsnd PID */
    pid_t msg_lrpid;                /* 最后 msgrcv PID */
};

$ ipcs -l
$ cat /proc/sys/kernel/msgmni   # 最大队列数
748
$ cat /proc/sys/kernel/msgmax   # 单条最大字节
8192
$ cat /proc/sys/kernel/msgmnb   # 单队列最大字节
16384

// 摘自《The Linux Programming Interface》 第 46 章 Listing 46-6
#define _GNU_SOURCE
int maxind = msgctl(0, MSG_INFO, (struct msqid_ds *)&msginfo);
printf("maxind: %d\n", maxind);
for (int ind = 0; ind <= maxind; ind++) {
    struct msqid_ds ds;
    int msqid = msgctl(ind, MSG_STAT, &ds);
    if (msqid == -1) {
        if (errno != EINVAL && errno != EACCES) errMsg("...");
        continue;
    }
    printf("%4d %8d 0x%08lx %7ld\n",
           ind, msqid, (unsigned long)ds.msg_perm.__key, (long)ds.msg_qnum);
}

// 摘自《The Linux Programming Interface》 第 46 章 Listing 46-7/8
/* 客户端：创建自己的消息队列（用 IPC_PRIVATE） */
clientId = msgget(IPC_PRIVATE, S_IRUSR | S_IWUSR | S_IWGRP);
atexit(removeQueue);     /* 退出时删 */

/* 构造请求：包含自己的 clientId + pathname */
req.mtype = 1;     /* 给 server 看的 */
req.clientId = clientId;
strncpy(req.pathname, argv[1], sizeof(req.pathname));
msgsnd(serverId, &req, REQ_MSG_SIZE, 0);

/* 等 server 回复（用 mtype=clientId 选自己的） */
while ((msgLen = msgrcv(clientId, &resp, RESP_MSG_SIZE, 0, 0)) > 0) {
    if (resp.mtype == RESP_MT_END) break;
    if (resp.mtype == RESP_MT_DATA) write(STDOUT, resp.data, msgLen);
    if (resp.mtype == RESP_MT_FAILURE) { fprintf(stderr, "%s", resp.data); break; }
}

// 摘自《The Linux Programming Interface》 第 46 章 Listing 46-8
/* 监听 serverId 的所有消息；fork 子进程处理每个请求 */
for (;;) {
    msgLen = msgrcv(serverId, &req, REQ_MSG_SIZE, 0, 0);
    if (msgLen == -1 && errno == EINTR) continue;   /* SIGCHLD 中断 */
    pid = fork();
    if (pid == 0) {
        serveRequest(&req);    /* 读文件并发 RESP_MT_DATA 消息到 req.clientId */
        _exit(EXIT_SUCCESS);
    }
}
/* SIGCHLD handler 调用 waitpid 回收 zombie */