关于Posix的消息队列

消息队列

可以认为是一个消息链表. 有足够写权限的线程可以往队列中放置消息, 有足够读权限的线程可以从队列中取走消息

在某个进程往一个队列写入消息前, 并不需要另外某个进程在该队列上等待消息的到达.

这跟管道和FIFO是相反的, 因为对于管道,FIFO来说, 除非读出者已经存在, 光有写入者是没有意义的

一个进程在往某消息队列写入消息后, 终止进程. 另一个进程某时刻读出该消息;

然而对于管带或FIFO而言, 当管道或FIFO的最后一次关闭发生时,仍在管道或FIFO中的数据将被抛弃

首先提一下, Posix消息队列和System V 消息队列之间有什么区别:

对Posix消息队列的读取总是返回最高优先级的最早消息; 对System V消息队列的读则可以返回任意指定优先级的消息

当往一个空队列放置一个消息时, Posix消息队列允许产生一个信号或者启动一个线程; System V则不提供类似机制

消息队列中每个消息都有如下的属性:

一个无符号整数优先级(Posix)或一个长整数类型(System V)

消息的数据部分长度

数据本身



基本操作函数:

mqd_t mq_open(const char *name, int oflag, ...
/* mode_t mode, struct mq_attr *attr */);

当我们的实际操作是创建一个新的队列时(即所要创建的队列不存在, 且oflag中已经指定O_CREAT), mode 和 attr参数是需要的

函数的返回值称为消息队列描述符, 这个值用作其他消息队列操作函数的第一个参数值

调用mq_close, 表示进程不再使用该描述符,但其消息队列并不从系统被删除

想要删除消息队列, 那么使用如下函数:

mqd_t mq_unlink(const char *name);

参数为调用mq_open的第一个参数值

每个消息队列有一个保存其当前打开着描述符数的引用计数器:

当一个消息队列引用计数大于0时, 其name就能删除, 但是该队列的析构要到最后一个mq_close发生时才进行

又因为Posix消息队列具备随内核的持续性, 即使当前没有打开着的消息队列, 该队列以及其上的各个消息也将一直存在, 直到调用mq_unlink并让它的引用计数达到0以删除该队列为止

现在先看看创建一个消息队列吧：

#include <stdio.h>
#include <stdlib.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <mqueue.h>

#define FILE_MODE 662

int main(int ac, char *av[])
{
int c, flags;
mqd_t mqd;
if(ac != 2)
{
fprintf(stderr, "To create message queue, file name is required\n");
return 0;
}

flags = O_CREAT | O_EXCL | O_RDWR;
//mqopen的第四个参数为NULL, 表示属性为默认
if((mqd = mq_open(av[1], flags, FILE_MODE, NULL)) < 0)
{
perror("error:");
exit(-1);
}

mq_close(mqd);
return 0;
}

我这里简化了书上的复杂操作. 但是, 在按照书上所说运行的时候却发生了特殊情况:

$./mqc /tmp/temp.1234
error:: Permission denied

在网上查找后, 也算是莫名其妙的解决了问题:
http://bbs.chinaunix.net/thread-1017072-1-1.html
通过 man 7 mq_overview 可以查到:

在使用root身份进行如下操作后:

$ sudo mkdir /dev/mqueue
$ sudo mount -t mqueue none /dev/mqueue

我们就可以顺利进行我们想要的操作了

$ ./mqc /myque.1234
$ ls /dev/mqueue/
myque.1234
$ sudo cat myque.1234
QSIZE:0          NOTIFY:0     SIGNO:0     NOTIFY_PID:0

有了创建函数后, 就需要有unlink函数：

#include <stdio.h>
#include <stdlib.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <mqueue.h>

int main(int ac, char *av[])
{
if(ac != 2)
{
fprintf(stderr, "To delete message queue, file name is required\n");
return 0;
}

if(mq_unlink(av[1]))
{
perror("error:");
exit(-1);
}
return 0;
}

删除函数就可以直接执行了

关于消息队列的属性

mqd_t mq_getattr(mqd_t mqdes, struct mq_attr *attr);
mqd_t mq_setattr(mqd_t mqdes, struct mq_attr *newattr, struct mq_attr *oldattr);

每个消息队列有四个属性, mq_getattr返回所有的这些属性, mq_setattr设置其中的某个属性

struct mq_attr {
long mq_flags;       /* Flags: 0 or O_NONBLOCK */
long mq_maxmsg;      /* Max. # of messages on queue */
long mq_msgsize;     /* Max. message size (bytes) */
long mq_curmsgs;     /* # of messages currently in queue */
}

再回到mq_open函数定义处, 指向mq_attr的指针可以作为该函数的第四个参数传递, 从而在创建队列初就设置好每个消息的最大长度和允许存在的最大消息数量,另外两个成员被忽略

使用mq_setattr给所指定队列设置属性, 但是只使用由attr指向的mq_attr结构的mq_flags成员, 以设置或清除非阻塞标志,
其他三个成员则被忽略(其中两个只能在创建队列时指定, 还有一个及时获取). 当然, mq_setattr的最后一个参数用于接收之前的属性和当前状态

了解了消息队列的属性后, 我们可以重新编写创建消息队列的函数了：

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <mqueue.h>
#include <unistd.h>

#define FILE_MODE S_IRUSR | S_IWUSR | S_IROTH

int main(int ac, char *av[])
{
int c, flags;
mqd_t mqd;
struct mq_attr attr;
memset(&attr, '\0', sizeof(attr));

if(ac < 2)
{
fprintf(stderr, "To create message queue,enough info is required\n");
return 0;
}

opterr = 0;
//如果要设置消息队列的属性, 那么最大长度和最大大小要么都设置,要么都不设置
int concur = 0;
flags = O_CREAT | O_RDWR;

while((c=getopt(ac, av, "em:z:")) != -1)
{
switch(c)
{
case 'e':                //是否要O_EXCL标志
flags |= O_EXCL;
break;
case 'm':                //设置最大消息数目
concur++;
attr.mq_maxmsg = atoi(optarg);
break;
case 'z':                //设置最大消息大小
concur++;
attr.mq_msgsize = atoi(optarg);
break;
}
}

//如果只设置了m和z其中一个, 那么是不能被接收的
if(optind != ac-1 || concur == 1)
{
fprintf(stderr, "Usage : [-e] [-m maxmsg -z msgsize] <name>\n");
exit(-1);
}

if((mqd = mq_open(av[optind], flags,FILE_MODE , (attr.mq_maxmsg!=0)?&attr:NULL)) < 0)
{
perror("error:");
exit(-1);
}

mq_close(mqd);
return 0;
}

要注意的是：

（1）attr.mq_maxmsg 不能超过文件 /proc/sys/fs/mqueue/msg_max 中的数值，我的机器上面是10。

（2）attr.mq_msgsize不能超过 /proc/sys/fs/mqueue/msgsize_max 的数值。我这里是8192

（3）消息队列名称前面必须加上斜杆。

往消息队列中放置消息和取出消息

mqd_t mq_send(mqd_t mqdes, const char *msg_ptr, size_t msg_len, unsigned msg_prio);
ssize_t mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len, unsigned *msg_prio);

mq_receive函数总是返回指定队列中最高优先级的最早消息, 而且该优先级能随该消息的内容和长度一起返回

且mq_receive的len参数的值不能小于能加到所指定队列中的消息的最大大小

//发送
#include <stdio.h>
#include <stdlib.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <mqueue.h>

int main(int ac, char **av)
{
mqd_t mqd;
void *ptr;
size_t len;
unsigned prio;

if(ac != 4){
fprintf(stderr, "Usage: mqs <queue name> <#bytes> <priority>\n");
exit(-1);
}

len = atoi(av[2]);
prio = atoi(av[3]);

if((mqd = mq_open(av[1], O_WRONLY)) < 0){
perror("mq_open error : ");
exit(-1);
}

ptr = calloc(len, sizeof(char));
if(NULL == ptr){
perror("calloc error");
exit(-1);
}

if(mq_send(mqd, ptr, len, prio) < 0){
perror("mq_send error");
exit(-1);
}

return 0;
}

//接收
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <mqueue.h>

int main(int ac, char *av[])
{
mqd_t mqd;
int c, n;
char *buf;
unsigned prio;
int flags = O_RDONLY;
struct mq_attr attr;

while((c = getopt(ac, av, "n")) != -1)
{
switch(c)                    //可加入非阻塞标志
{
case 'n':
flags |= O_NONBLOCK;
}
}

if(optind != ac -1)
{
fprintf(stderr, "Usage: mqr
<mq name>\n");
exit(-1);
}

if((mqd = mq_open(av[1], flags)) < 0){
perror("mq_open error : ");
exit(-1);
}

mq_getattr(mqd, &attr);
buf = calloc(attr.mq_msgsize, sizeof(char));
if(NULL == buf){
perror("calloc error");
exit(-1);
}

if((n=mq_receive(mqd, buf, attr.mq_msgsize, &prio)) < 0){
perror("mq_receive error");
exit(-1);
}
printf("read %d bytes , priority : %u\n", n, prio);

return 0;
}

执行结果:(mqs 为发送, mqr为接收)

$ ./mqs /haha 100 9999
$ ./mqs /haha 100 99999            //优先级不能大于特定值
mq_send error: Invalid argument
$ ./mqs /haha 150 99
$ ./mqs /haha 200 99
$./mqr /haha
read 100 bytes , priority : 9999    //优先接收优先级高的
$ ./mqr /haha
read 150 bytes , priority : 99
$ ./mqr /haha
read 200 bytes , priority : 99
$ ./mqr /haha
//阻塞在等待中
$ ./mqr -n /haha                //非阻塞, 返回错误
mq_receive error: Resource temporarily unavailable

可以观察到, 当消息队列为空时,若是阻塞, mq_reveive函数也会阻塞住; 若是非阻塞, 那么何时来消息我们可能就要轮询, 显然两者都不合适

关于mq_notify函数

Posix消息队列允许异步事件通知, 以告知何时有一个消息放置到了某个空消息队列中, 以下两种方式可选:

1 产生一个信号

2 创建一个线程来执行一个指定函数

mqd_t mq_notify(mqd_t mqdes, const struct sigevent *notification);

函数为指定队列创建或删除异步事件通知

typedef union sigval
{
int sival_int;
void *sival_ptr;
} sigval_t;

typedef struct sigevent
{
sigval_t sigev_value;        //passed to signal handler or thraed
int sigev_signo;            //signal numnber if signal
int sigev_notify;            //SIGEV_{NONE, SIGNAL, THREAD}

union
{
int _pad[__SIGEV_PAD_SIZE];

/* When SIGEV_SIGNAL and SIGEV_THREAD_ID set, LWP ID of the
thread to receive the signal.  */
__pid_t _tid;

struct
{
void (*_function) (sigval_t);    /* Function to start.  */
void *_attribute;            /* Really pthread_attr_t.  */
} _sigev_thread;
} _sigev_un;
} sigevent_t;

如果mq_notify函数的notification非空, 那么当前进程希望在有一个消息到达所指定的先前为空的队列时得到通知. 即"该进程被注册为接收该队列的通知"

如果notification为空, 而且当前进程目前被注册为接收所指定队列的通知, 那么已存在的注册将被撤销

任意时刻只能有一个进程可以被注册为接收某个给定队列的通知

当有一个消息到达先前为空的消息队列,而且已有一个进程被注册为接收该队列的通知, 只有在没有任何线程阻塞在该队列的mq_receive调用中的前提下, 通知才会发出. 即在mq_receive中的阻塞比任何通知的注册都要优先

当通知被发送给注册进程时, 其注册就被撤销, 如果想的话, 需要重新注册(所以一般情况下, 都是在信号处理函数中的一开始就再次调用mq_notify进行重新注册)

现在来看看:

#include <stdio.h>
#include <stdlib.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <mqueue.h>
#include <signal.h>

mqd_t mqd;
void *buff;
struct mq_attr attr;
struct sigevent sigev;

static void sig_usr1(int);

int main(int ac, char *av[])
{
if(ac != 2){
fprintf(stderr,"Usage: mqn1 <name>\n");
exit(-1);
}

if((mqd = mq_open(av[1], O_RDONLY)) < 0){
perror("mq_open error : ");
exit(-1);
}
mq_getattr(mqd, &attr);
buff = calloc(attr.mq_msgsize, sizeof(char));
if(NULL == buff){
perror("calloc error");
exit(-1);
}

if(signal(SIGUSR1, sig_usr1) == SIG_ERR){
perror("signal errpr");
exit(-1);
}
//注册称为接收消息的进程
sigev.sigev_signo = SIGUSR1;
sigev.sigev_notify = SIGEV_SIGNAL;
mq_notify(mqd, &sigev);

for(;;)
pause();
return 0;
}

//设置信号处理函数, 在接收到信号后调用接收函数接收消息
static void sig_usr1(int s)
{
ssize_t n;
mq_notify(mqd, &sigev);            //remember to register first
n = mq_receive(mqd, buff, attr.mq_msgsize, NULL);
printf("SIGUSR1 received, read %d bytes\n", n);
}

运行过后, 发现并没有什么问题.

但是, 此程序其实存在着极大的安全隐患, 因为它是在信号处理函数中调用mq_notify, mq_receive和printf的, 这些函数在此处调用并不安全(不是异步安全函数)

既然如此, 那么我们就令信号处理函数如下所示:

static void sig_usr1(int s)
{
mq_flag = 1;
}

//在主函数中:
sigprocmask(SIG_BLOCK, &newmask, &oldmask);
for(;;)
{
while(mq_flag == 0)
sigsuspend(&zeromask);
mq_flag = 0;
mq_notify(mqd, &sigev);
n = mq_receive(mqd, buff, attr.mq_msgsize, NULL);
printf("receive %d bytes\n", n);
}

这样, 显然就解决了之前的问题, 但是貌似又遇到了新的问题, 就是

在信号处理函数返回之后, 成功取出消息之前, 如果消息队列中又来了好几个消息. 那么岂不是以后再也不会有信号来通知此函数去读取了, 因为只有在消息队列为空的情况下才会发生信号, 然而此时消息队列中最早的消息虽然已经提醒进程去取, 但消息还未被取出, 后来的消息就不属于消息队列为空时候的消息了, 那么也就不会发出信号

于是, 我们在这边干等, 消息队列中的消息也在等着我们去读取, 但谁也不知到谁的存在, 之后再来几个消息也不会有任何反应了

解决的办法是, 使用非阻塞模式循环读取, 直到返回EAGAIN这种错误

在主函数中:

mqd = mq_open(..., O_RDONLY | O_NONBLOCK);
...
sigprocmask(SIG_BLOCK, &newmask, &oldmask);
for(;;)
{
while(mq_flag == 0)
sigsuspend(&zeromask);
mq_flag = 0;
mq_notify(mqd, &sigev);
while((n = mq_receive(mqd, buff, attr.mq_msgsize, NULL)) >= 0)
printf("receive %d bytes\n", n);
if(errno != EAGAIN)
err_sys("...");
}

除了使信号处理函数改变等待量, 我们还有更简单的方法:

使用sigwait() 函数

sigpromask(SIG_BOOCK, &newmask, NULL);        //先阻塞该信号
for(;;)
{
sigwait(&newmask, &signo);
if(signo == SIGUSR1){
mq_notify(mqd, &sigev);
while((n = mq_receive(mqd, buff, attr.mq_msgsize, NULL)) >= 0)
printf("receive %d bytes\n", n);
if(errno != EAGAIN)
err_sys("...");
}
}

之前在多线程与信号中认识过sigwait, 了解其原子操作操作, 也知道在多线程中, 必须使用pthread_sigmask, 而不是sigprocmask

除了使用sigwait, 我们还能使用select多路复用来解决这个问题, 虽然消息队列的文件描述符不是普通的能被select接收的文件描述符, 但我们选择其他方法. 前提是write函数是异步安全函数, 然后在信号处理函数与select之间架起一个管道, 可以在接收到信号后, 从信号处理函数中向select那端发送数据告知已经可以从消息队列取数据了

除了使用信号, sigevent结构体中还有一个选项, 就是使用线程来处理

当有消息传递到空消息队列中时, 自动调用一个线程去处理, 当然, 依旧是非阻塞处理

static void notify_thread(union sigval a)
{
ssize_t n;
void *buf = (char *)malloc(attr.mq_msgsize);
mq_notify(mqd, ...);

while((n=mq_receive(...)) > 0)
printf(...);

if(errno != EAGAIN)
exit(-1);

free(buf);
pthread_exit(NULL);
}