mqpxtest.c

VxWorks入门例程

/* mqPXTest.c - examples using POSIX message queues */

/* The mqPXCom() routine spawns two tasks that
 * communicate using the message queue.            */

/* The mqPXNotify() routine illustrates the use
 * of mq_notify() to request notification via signal
 * of new messages in a queue. To simplify the example,
 * a single task both sends and receives a message.   */

/* The mqPXAttr() routine illustrates how to use
 * mq_setattr() and mq_getattr() to set and get the
 * attributes of the message queue.                  */

/* includes */  
#include "vxWorks.h"  
#include "mqueue.h"  
#include "fcntl.h"  
#include "errno.h"  
#include "taskLib.h"
#include "stdio.h"
#include "signal.h"

/* defines */
#define HI_PRIO     31
#define LOW_PRIO     0  
#define MSG_SIZE    16 
#define MQ_NAME "myMessQ"
#define MSG     "hello world"

/* forward declarations */
int receiveTask(void);
int sendTask(void);
static void NotificationHandle(int,siginfo_t*,void*);
static void mqRead(mqd_t);
 
/***************************************************************************/

int mqPXCom(void)
{
    if(taskSpawn("tRcvTask",95,0,4000,(FUNCPTR)receiveTask,0,0,0,0,0,0,0,0,0,0)==-1)
    {
        printf("taskSpawn of tRcvTask failed\n");
        return(-1);
    }
    if(taskSpawn("tSndTask",100,0,4000,(FUNCPTR)sendTask,0,0,0,0,0,0,0,0,0,0)==-1)
    {
        printf("taskSpawn of tSendTask failed\n");
        return(-1);
    }
  
    return(0);
}
  
int receiveTask(void)
{
    mqd_t    mqPXId;
    char     msg[MSG_SIZE];
    int      prio;

    if((mqPXId=mq_open(MQ_NAME,O_RDWR|O_CREAT,0,NULL))==(mqd_t)-1)
    {
        printf("receiveTask: mq_open failed\n");
        return(-1);
    }

    if(mq_receive(mqPXId,msg,MSG_SIZE,&prio)==-1)
    {
        printf("receiveTask: mq_receive failed\n");
        return(-1);
    }
    printf("receiveTask: Msg of priority %d received:\n\t%s\n",prio,msg);
    
    return(0);
}

int sendTask(void)
{
    mqd_t    mqPXId;
  
    if((mqPXId=mq_open(MQ_NAME,O_RDWR,0,NULL))==(mqd_t)-1)
    {
        printf("sendTask: mq_open failed\n");
        return(-1);
    }

    if(mq_send(mqPXId,MSG,sizeof(MSG),HI_PRIO)==-1)
    {
        printf("sendTask: mq_send failed\n");
        return(-1);
    }
    printf("sendTask: mq_send succeeded\n");
      
    if(mq_close(mqPXId)==-1)
    {
        printf("mq_close failed\n");
        return(-1);
    }

    if(mq_unlink(MQ_NAME)==-1)
    {
        printf("mq_unlink failed\n");
        return(-1);
    }

    return(0);
}

/***************************************************************************/

int mqPXNotify(void)
{
    struct mq_attr attr;
    struct sigevent sigNotify;
    struct sigaction mySigAction;
    mqd_t mqPXId;
    
    mySigAction.sa_sigaction=NotificationHandle;
    mySigAction.sa_flags=SA_SIGINFO;
    sigemptyset(&mySigAction.sa_mask);

    if(sigaction(SIGUSR1,&mySigAction,NULL)==-1)
    {
        printf("sigaction failed\n");
        return(-1);
    }

    attr.mq_flags=O_NONBLOCK;
    attr.mq_maxmsg=2;
    attr.mq_msgsize=MSG_SIZE;

    if((mqPXId=mq_open(MQ_NAME,O_CREAT|O_RDWR,0,&attr))==(mqd_t)-1)
    {
        printf("mq_open failed\n");
        return(-1);
    }

    sigNotify.sigev_signo=SIGUSR1;
    sigNotify.sigev_notify=SIGEV_SIGNAL;
    sigNotify.sigev_value.sival_int=(int)mqPXId;

    if(mq_notify(mqPXId,&sigNotify)==-1)
    {
        printf("mq_notify failed\n");
        return(-1);
    }

    mqRead(mqPXId);

    if(mq_send(mqPXId,MSG,sizeof(MSG),LOW_PRIO)==-1)
    {
        printf("mq_send failed\n");
        return(-1);
    }

    if(mq_close(mqPXId)==-1)
    {
        printf("mq_close failed\n");
        return(-1);
    }

    if(mq_unlink(MQ_NAME)==-1)
    {
        printf("mq_unlink failed\n");
        return(-1);
    }
    
    return(0);
}

static void NotificationHandle(int sig,siginfo_t *pInfo,void *pSigContext)
{
    struct sigevent sigNotify;
    mqd_t mqPXId;
    mqPXId=(mqd_t)pInfo->si_value.sival_int;

    sigNotify.sigev_signo=pInfo->si_signo;
    sigNotify.sigev_value=pInfo->si_value;
    sigNotify.sigev_notify=SIGEV_SIGNAL;

    if(mq_notify(mqPXId,&sigNotify)==-1)
    {
        printf("mq_notify failed\n");

        return;
    }

    mqRead(mqPXId);
}

static void mqRead(mqd_t mqPXId)
{
    char msg[MSG_SIZE];
    int prio;

    while(mq_receive(mqPXId,msg,MSG_SIZE,&prio)!=-1)
    {
        printf("mqRead: received message:\n\t%s\n",msg);
    }
    if(errno!=EAGAIN)
    {
        printf("mq_receive: errno=%d\n",errno);
    }
}

/***************************************************************************/

int mqPXAttr(void)
{
    mqd_t mqPXId;
    struct mq_attr attr;
    struct mq_attr oldAttr;
    char buffer[MSG_SIZE];
    int prio;

    attr.mq_flags=0;
    attr.mq_maxmsg=1;
    attr.mq_msgsize=16;

    if((mqPXId=mq_open(MQ_NAME,O_CREAT|O_RDWR,0,&attr))==(mqd_t)-1)
        return(-1);
    printf("mq_open without non-blocking succeeded\n");

    attr.mq_flags=O_NONBLOCK;
    if(mq_setattr(mqPXId,&attr,&oldAttr)==-1)
        return(-1);
    else
    {
        if(oldAttr.mq_flags&O_NONBLOCK)
            return(-1);
        printf("mq_setattr turning on non-blocking succeeded\n");
    }

    if(mq_receive(mqPXId,buffer,MSG_SIZE,&prio)==-1)
    {
        if(errno!=EAGAIN)
            return(-1);
        printf("mq_receive with non-blocking didn't block on empty queue\n");
    }
    else
        return(-1);

    if(mq_getattr(mqPXId,&oldAttr)==-1)
        return(-1);
    else
    {
        if(!(oldAttr.mq_flags&O_NONBLOCK)||(oldAttr.mq_curmsgs!=0))
            return(-1);
        printf("queue attributes are:
	non-blocking is: %s
	message size is: %d
	max message in queue: %d
	no. of current msgs in queue: %d\n",
        oldAttr.mq_flags&O_NONBLOCK?"ON":"OFF",
        oldAttr.mq_msgsize,oldAttr.mq_maxmsg,oldAttr.mq_curmsgs);
    }

    if(mq_close(mqPXId)==-1)
    {
        printf("mq_close failed\n");
        return(-1);
    }

    if(mq_unlink(MQ_NAME)==-1)
    {
        printf("mq_unlink failed\n");
        return(-1);
    }

    return(0);
}