第二个线程无法通过 C/C++/linux 中线程 1 发送的消息队列接收消息

Question

我正在模拟一个 udp 服务器，它通过 udp 客户端接收多个请求消息。我需要在这个 udp 服务器示例代码中生成 2 个线程。线程 1 将在所有 udp 请求到达时接收它们，并通过消息队列将其发送到线程 2 进行处理。

线程 1(SndMq) 能够通过 udp 接收请求消息，也能够使用结构将其放入消息队列中。但问题是 线程 2(RcvMq) 甚至无法读取消息队列中的单个消息。请提出任何解决方案。

错误部分（线程2函数）：

void* RcvMq(void*) {

        time_t t2;
        time(&t2);

        msgst *msg = new msgst;

        int mqid = msgget((key_t)12345, 0666 | IPC_CREAT);
        if (mqid == -1) {
                printf("thread 2: msgget failed with error: %d\n", errno);
                exit(EXIT_FAILURE);
        }
        else
            printf("\nthread 2: got MQ id: %d\n", mqid);

        long receive_m_type = 123;

        printf("thread 2: going to enter while(1) loop\n");
        while(1) {
                printf("\nthread 2: inside while(1) loop, going to msgrcv()\n");
                if ( msgrcv(mqid, (void *)&msg, BUFSIZE, receive_m_type, 0) < 0 ) {
                        //printf("\nthread 2: msgrcv failed with error: [%s]\n", strerror(errno));
                        printf("\nthread 2: msgrcv failed with error\n");
                        exit(EXIT_FAILURE);
                }
                else {
                        printf("thread 2: going to ctime()\n");
                        printf("\nthread 2: DeQueued at: %s\n", ctime(&t2));
                }
                printf("\nthread 2: going to print struct's buffer\n");
                printf("thread 2: Message received at MQ rcv is: %s \n", msg->buffer);
        }
}

udp 服务器模拟器代码

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <time.h>
#include <netdb.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <sys/types.h>
#include <sys/msg.h>
#include<pthread.h>

#define PORT 5000
#define BUFSIZE 2048

pthread_t sndtid;
pthread_t rcvtid;

typedef struct msgst  {
        long int mtype;
        char buffer[BUFSIZE];
    }msgst;

void* SndMq(void*) {

        time_t t1;
        time(&t1);
        struct sockaddr_in myaddr;  /* our address */
        struct sockaddr_in remaddr; /* remote address */
        socklen_t addrlen = sizeof(remaddr);    /* length of addresses */
        int recvlen;                /* # bytes received */
        int fd;                     /* our socket */
        int rc, on=1;
        char buf[BUFSIZE];  /* receive buffer */
        size_t buflen;
        /* create a UDP socket: socket, setsockopt, bind */
        if ((fd = socket(AF_INET, SOCK_DGRAM, 0)) < 0) {
                perror("cannot create socket\n");
                return 0;
        }
        if((rc = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (char *)&on, sizeof(on))) < 0) {
                perror("thread 1: UDP Server setsockopt() - ERROR");
                close(fd);
                exit(-1);
        }
        /* bind the socket to any valid IP address and a specific port */
        memset((char *)&myaddr, 0, sizeof(myaddr));
        myaddr.sin_family = AF_INET;
        myaddr.sin_addr.s_addr = htonl(INADDR_ANY);
        //myaddr.sin_port = htons(SERVICE_PORT);
        myaddr.sin_port = htons(PORT);
        if (    bind(fd, (struct sockaddr *)&myaddr, sizeof(myaddr) ) < 0) {
                perror("bind failed");
                return 0;
        }

        printf("\nthread 1: Server waiting on port %d\n", PORT);

        //MsgQ implementation:
        int mqid = msgget((key_t)12345, 0666 | IPC_CREAT);
        if(mqid == -1) {
                printf("thread 1: msgget() failed, errno: %d\n",errno );
        }
        else
            printf("thread 1: created MQ id: %d\n", mqid);

        /* now loop, receiving data and printing what we received */
        for (;;) {

                //rcv from udp client
                recvlen = recvfrom(fd, buf, BUFSIZE, 0, (struct sockaddr *)&remaddr, &addrlen);
                printf("\nthread 1: received %d bytes\n", recvlen);
                if (recvlen > 0) {
                        buf[recvlen] = 0;
                        printf("thread 1: received message from UDP client: %s\n", buf);
                }

                //send received msg via MsgQ
                msgst *msgptr = new msgst;
                msgptr->mtype = 123;
                strcpy(msgptr->buffer, buf);
                //msgptr->buffer=buf;
                //memcpy(msgptr->buffer, buf, BUFSIZE);
                buflen = strlen(msgptr->buffer);

                if (msgsnd(mqid, (void *)&msgptr, buflen, IPC_NOWAIT) == -1) {
                        printf("thread 1: msgsnd failed\n");
                        exit(EXIT_FAILURE);
                }
                printf("thread 1: added Req to MsgQ at: %s\n", ctime(&t1));
        }/* never exits */
    }
void* RcvMq(void*) {

        time_t t2;
        time(&t2);

        msgst *msg = new msgst;

        int mqid = msgget((key_t)12345, 0666 | IPC_CREAT);
        if (mqid == -1) {
                printf("thread 2: msgget failed with error: %d\n", errno);
                exit(EXIT_FAILURE);
        }
        else
            printf("\nthread 2: got MQ id: %d\n", mqid);

        long receive_m_type = 123;

        printf("thread 2: going to enter while(1) loop\n");
        while(1) {
                printf("\nthread 2: inside while(1) loop, going to msgrcv()\n");
                if ( msgrcv(mqid, (void *)&msg, BUFSIZE, receive_m_type, 0) < 0 ) {
                        //printf("\nthread 2: msgrcv failed with error: [%s]\n", strerror(errno));
                        printf("\nthread 2: msgrcv failed with error\n");
                        exit(EXIT_FAILURE);
                }
                else {
                        printf("thread 2: going to ctime()\n");
                        printf("\nthread 2: DeQueued at: %s\n", ctime(&t2));
                }
                printf("\nthread 2: going to print struct's buffer\n");
                printf("thread 2: Message received at MQ rcv is: %s \n", msg->buffer);
        }
}
int main() {

        int rtn=0;

        rtn = pthread_create(&(sndtid), NULL, &SndMq, NULL);
        if (rtn != 0)
                printf("\ncan't create thread SndMq, error:[%s]", strerror(rtn));
        else
                printf("\n\ncreated SndMq thread\n");

        rtn = pthread_create(&(rcvtid), NULL, &RcvMq, NULL);
        if (rtn != 0)
                printf("\ncan't create thread RcvMq :[%s]", strerror(rtn));
        else
                printf("created RcvMq thread\n");

        printf("main(): All threads spawned successfully, now waiting for threads to end...\n");
        pthread_join(sndtid, NULL);
        pthread_join(rcvtid, NULL);
        printf("\nThreads ended, main() exiting now...\n");

        return 0;
}

输出：

created SndMq thread
created RcvMq thread
main(): All threads spawned successfully, now waiting for threads to end...

thread 2: got MQ id: 6422528
thread 2: going to enter while(1) loop

thread 2: inside while(1) loop, going to msgrcv()

thread 1: Server waiting on port 5000
thread 1: created MQ id: 6422528

thread 1: received 1500 bytes
thread 1: received message from UDP client: <TransactionId>1</TransactionId>
thread 1: added Req to MsgQ at: Sat Apr  5 18:16:45 2014


thread 1: received 1500 bytes
thread 1: received message from UDP client: <TransactionId>2</TransactionId>
thread 1: added Req to MsgQ at: Sat Apr  5 18:16:45 2014


thread 1: received 1500 bytes
thread 1: received message from UDP client: <TransactionId>3</TransactionId>
thread 1: added Req to MsgQ at: Sat Apr  5 18:16:45 2014


thread 1: received 1500 bytes
thread 1: received message from UDP client: <TransactionId>4</TransactionId>
thread 1: added Req to MsgQ at: Sat Apr  5 18:16:45 2014


thread 1: received 1500 bytes
thread 1: received message from UDP client: <TransactionId>5</TransactionId>
thread 1: added Req to MsgQ at: Sat Apr  5 18:16:45 2014


thread 1: received 1500 bytes
thread 1: received message from UDP client: <TransactionId>6</TransactionId>
thread 1: added Req to MsgQ at: Sat Apr  5 18:16:45 2014


thread 1: received 1500 bytes
thread 1: received message from UDP client: <TransactionId>7</TransactionId>
thread 1: added Req to MsgQ at: Sat Apr  5 18:16:45 2014


thread 1: received 1500 bytes
thread 1: received message from UDP client: <TransactionId>8</TransactionId>
thread 1: added Req to MsgQ at: Sat Apr  5 18:16:45 2014


thread 1: received 1500 bytes
thread 1: received message from UDP client: <TransactionId>9</TransactionId>
thread 1: added Req to MsgQ at: Sat Apr  5 18:16:45 2014


thread 1: received 1500 bytes
thread 1: received message from UDP client: <TransactionId>10</TransactionId>
thread 1: added Req to MsgQ at: Sat Apr  5 18:16:45 2014

Answer 1

(1) 您的主要问题是 C 错误。您传递的是 ptr 的地址（指向 ptr 的 ptr）而不是指针。

msgst *msgptr = new msgst;
//...

if (msgsnd(mqid, (void *)&msgptr, buflen, IPC_NOWAIT) == -1) 

应该是

if (msgsnd(mqid, (void *) msgptr, buflen, IPC_NOWAIT) == -1)

因此，您在 &msgptr 处写入任何随机垃圾，并且由于您正在写入 msgtype 123，然后尝试稍后读取 msgtype 123（但写入垃圾），您将永远（或很少！）找到该消息类型。

你在消息接收端犯了同样的错误：

msgst *msg = new msgst;
//.......

if ( msgrcv(mqid, (void *)&msg, BUFSIZE, receive_m_type, 0) < 0 ) 

(2) 你可能还没有找到它，但你有内存泄漏，因为你new 缓冲区但从来没有delete 它。无论如何，您实际上并不需要动态内存分配。将缓冲区分配在堆栈上会更容易，因为无论如何它都会被msgsnd 复制，然后你就完成了。这也可以消除您在 (1) 中遇到的问题。

(3) 您将在msgsnd 上遇到 IPC_NOWAIT 问题。在消费者处理之前，很容易在生产者端填充队列。至少您应该检查 EAGAIN 并在写入失败时采取任何适当的操作。

Answer 2

可能与所询问的问题无关。

但是在SndMq() 中，如果recvfrom() 接收到BUFSIZE 字节，那么代码将超出buf 的范围：

    buf[recvlen] = 0;

更新：

要解决此问题，请更改此行

recvlen = recvfrom(fd, buf, BUFSIZE, 0, (struct sockaddr *)&remaddr, &addrlen);

成为

recvlen = recvfrom(fd, buf, BUFSIZE - 1, 0, (struct sockaddr *)&remaddr, &addrlen);