线程抓住其他线程来自队列（Linux）的IPC消息

Question

当我在Linux上有两个不同的IPC消息队列时，有时会从错误队列中获取消息。

下面的玩具程序显示问题，可以在不同的处理器上重复。

任何帮助非常感谢！

伯特

/* 
    To compile; 
    gcc MM.c -o mm -fno-stack-protector -pthread 

    We want Mickey to send a message to Minnie exclusively. 
    We want Donald to send a message to pluto exclusively. 

    Problem: Pluto intercepts Minnie's messages. 

    Listing gives: 

    $ ./mm 
    Mickey thread successfully started. 
    Minnie thread successfully started. 
    Pluto thread successfully started. 
    Donald thread successfully started. 
    Donald sent a message to Pluto. 
    Mickey sent a message to Minnie. 
    Pluto received: Sit, Pluto! 

    Minnie received: Hello, Minnie! 

    Mickey sent a message to Minnie. (100 times) 

    Pluto received: Hello, Minnie! 
*/ 
#include <sys/types.h> 
#include <sys/msg.h> 
#include <sys/ipc.h> 
#include <string.h> 
#include <stdio.h> 

pthread_t t1,t2,t3,t4; 

// Mickey send 
key_t ipcMickey; 
int mqMickeyid; 
char helloMickeymsg[] = {"Hello, Minnie!"}; 
struct { long type; char text[100]; } myMickeymsg; 

// Minnie get 
int mqMinnieid; 
struct { long type; char text[100]; } myMinniemsg; 

// Donald send 
key_t ipcDonald; 
int mqDonaldid; 
char helloDonaldmsg[] = {"Sit, Pluto!"}; 
struct { long type; char text[100]; } myDonaldmsg; 

// Pluto get 
int mqPlutoid; 
struct { long type; char text[100]; } myPlutomsg; 

static void * DONALDthreadFunc(void *arg) 
{ 
    printf("Donald thread successfully started.\n"); 
    char *s = (char *) arg; 

    while(1) 
    { 
     //send 
     memset(myDonaldmsg.text, 0, 100); 
     strncpy(myDonaldmsg.text, helloDonaldmsg, strlen(helloDonaldmsg)); 
     myDonaldmsg.type = 1; 
     msgsnd(mqDonaldid, &myDonaldmsg, sizeof(myDonaldmsg), 0); 
     printf("Donald sent a message to Pluto.\r\n"); 
     sleep(4); 
    } 

    /* just a formality */ 
    return (void *) strlen(s); 
} 

static void * PLUTOthreadFunc(void *arg) 
{ 
    printf("Pluto thread successfully started.\n"); 
    char *s = (char *) arg; 

    while(1) 
    { 
     //receive 
     mqPlutoid = msgget(ipcDonald, 0); 
     msgrcv(mqPlutoid, &myPlutomsg, sizeof(myPlutomsg), 0, 0); 
     printf("Pluto received: %s\r\n\r\n", myPlutomsg.text); 
     sleep(1); 
    } 

    /* just a formality */ 
    return (void *) strlen(s); 
} 

static void * MICKEYthreadFunc(void *arg) 
{ 
    printf("Mickey thread successfully started.\n"); 
    char *s = (char *) arg; 

    while(1) 
    { 
     //send 
     memset(myMickeymsg.text, 0, 100); 
     strncpy(myMickeymsg.text, helloMickeymsg, strlen(helloMickeymsg)); 
     myMickeymsg.type = 1; 
     msgsnd(mqMickeyid, &myMickeymsg, sizeof(myMickeymsg), 0); 
     printf("Mickey sent a message to Minnie.\r\n"); 
     usleep(10000); 
    } 

    /* just a formality */ 
    return (void *) strlen(s); 
} 

static void * MINNIEthreadFunc(void *arg) 
{ 
    printf("Minnie thread successfully started.\n"); 
    char *s = (char *) arg; 

    while(1) 
    { 
     //receive 
     mqMinnieid = msgget(ipcMickey, 0); 
     msgrcv(mqMinnieid, &myMinniemsg, sizeof(myMinniemsg), 0, 0); 
     printf("Minnie received: %s\r\n\r\n", myMinniemsg.text); 
     sleep(3); 
    } 

    return (void *) strlen(s); 
} 

int main (void) 
{ 
     ipcMickey = ftok("/tmp/mqmickey", 63); 
     mqMickeyid = msgget(ipcMickey, IPC_CREAT | 0666); 

    ipcDonald = ftok("/tmp/mqdonald", 69); 
     mqDonaldid = msgget(ipcDonald, IPC_CREAT | 0666); 

    pthread_create(&t1, NULL, MICKEYthreadFunc, "Mickey sends\n"); 
    pthread_create(&t2, NULL, MINNIEthreadFunc, "Minnie replies\n"); 
    pthread_create(&t3, NULL, DONALDthreadFunc, "Donald sends\n"); 
    pthread_create(&t4, NULL, PLUTOthreadFunc, "Pluto replies\n"); 

    while(1) 
    { 
     sleep(5); 
    } 
} 

Answer 1

问题是你错过了错误处理。

包括

ipcMickey = ftok("/tmp/mqmickey", 63); 
if (-1==ipcMickey) 
{ 
    perror("ipcMickey"); 
    exit(255); 
} 
ipcDonald = ftok("/tmp/mqdonald", 69); 
if (-1==ipcDonald) 
{ 
    perror("ipcDonald"); 
    exit(255); 
} 

，你很快就会发现

./mm 
ipcDonald: No such file or directory 

的ftok返回-1在这种情况下。如果这两个文件丢失，无论IPC键将是-1，这意味着所有流量将共享相同的端口:)

所以

touch /tmp/mqmickey /tmp/mqdonald 

修复它。有人统计（与usleep(random()%10000)替换睡觉）：

gcc MM.c -o mm -O3 -fno-stack-protector -pthread 
time ./mm | { trap "" INT; sort | uniq -c | tee stats; } 
    16047 
     1 Donald sent a 
    8054 Donald sent a message to Pluto. 
     1 Donald thread successfully started. 
    8040 Mickey sent a message to Minnie. 
     1 Mickey thread successfully started. 
    8065 Minnie received: Hello, Minnie! 
     1 Minnie thread successfully started. 
    7982 Pluto received: Sit, Pluto! 
     1 Pluto thread successfully started. 


real 0m40.814s 
user 0m0.168s 
sys 0m0.092s 

Answer 2

这里被稍微修改程序的版本：

/* 
    To compile; 
    gcc MM.c -o mm -fno-stack-protector -pthread 

    We want Mickey to send a message to Minnie exclusively. 
    We want Donald to send a message to pluto exclusively. 

    Problem: Pluto intercepts Minnie's messages. 

    Listing gives: 

    $ ./mm 
    Mickey thread successfully started. 
    Minnie thread successfully started. 
    Pluto thread successfully started. 
    Donald thread successfully started. 
    Donald sent a message to Pluto. 
    Mickey sent a message to Minnie. 
    Pluto received: Sit, Pluto! 

    Minnie received: Hello, Minnie! 

    Mickey sent a message to Minnie. (100 times) 

    Pluto received: Hello, Minnie! 
*/ 
#include <sys/types.h> 
#include <sys/msg.h> 
#include <sys/ipc.h> 
#include <string.h> 
#include <stdio.h> 

pthread_t t1,t2,t3,t4; 

// Mickey send 
key_t ipcMickey; 
int mqMickeyid; 
char helloMickeymsg[] = {"Hello, Minnie!"}; 
struct { long type; char text[100]; } myMickeymsg; 

// Minnie get 
int mqMinnieid; 
struct { long type; char text[100]; } myMinniemsg; 

// Donald send 
key_t ipcDonald; 
int mqDonaldid; 
char helloDonaldmsg[] = {"Sit, Pluto!"}; 
struct { long type; char text[100]; } myDonaldmsg; 

// Pluto get 
int mqPlutoid; 
struct { long type; char text[100]; } myPlutomsg; 

static void * DONALDthreadFunc(void *arg) 
{ 
    printf("Donald thread successfully started.\n"); 
    char *s = (char *) arg; 

    while(1) 
    { 
     //send 
     memset(myDonaldmsg.text, 0, 100); 
     strncpy(myDonaldmsg.text, helloDonaldmsg, strlen(helloDonaldmsg)); 
     myDonaldmsg.type = 1; 
     msgsnd(mqDonaldid, &myDonaldmsg, sizeof(myDonaldmsg), 0); 
     printf("Donald sent a message to Pluto.\r\n"); 
     sleep(4); 
    } 

    /* just a formality */ 
    return (void *) strlen(s); 
} 

static void * PLUTOthreadFunc(void *arg) 
{ 
    printf("Pluto thread successfully started.\n"); 
    char *s = (char *) arg; 

    while(1) 
    { 
     //receive 
     mqPlutoid = msgget(ipcDonald, 0); 
     msgrcv(mqPlutoid, &myPlutomsg, sizeof(myPlutomsg), 0, 0); 
     printf("Pluto received: %s\r\nPluto uses MQ with id = %d\r\n", myPlutomsg.text,mqPlutoid); 
     sleep(1); 
    } 

    /* just a formality */ 
    return (void *) strlen(s); 
} 

static void * MICKEYthreadFunc(void *arg) 
{ 
    printf("Mickey thread successfully started.\n"); 
    char *s = (char *) arg; 

    while(1) 
    { 
     //send 
     memset(myMickeymsg.text, 0, 100); 
     strncpy(myMickeymsg.text, helloMickeymsg, strlen(helloMickeymsg)); 
     myMickeymsg.type = 1; 
     msgsnd(mqMickeyid, &myMickeymsg, sizeof(myMickeymsg), 0); 
     printf("Mickey sent a message to Minnie.\r\n"); 
     usleep(10000); 
    } 

    /* just a formality */ 
    return (void *) strlen(s); 
} 

static void * MINNIEthreadFunc(void *arg) 
{ 
    printf("Minnie thread successfully started.\n"); 
    char *s = (char *) arg; 

    while(1) 
    { 
     //receive 
     mqMinnieid = msgget(ipcMickey, 0); 
     msgrcv(mqMinnieid, &myMinniemsg, sizeof(myMinniemsg), 0, 0); 
     printf(" Minnie received: %s\r\nMinnie uses MQ with id = %d\r\n", myMinniemsg.text,mqMinnieid); 
     sleep(3); 
    } 

    return (void *) strlen(s); 
} 

int main (void) 
{ 
    ipcMickey = ftok("./mqmickey", 63); 
    mqMickeyid = msgget(ipcMickey, IPC_CREAT | 0666); 
    printf("mqMickeyid=%d\n",mqMickeyid); 

    ipcDonald = ftok("./mqdonald", 69); 
    mqDonaldid = msgget(ipcDonald, IPC_CREAT | 0666); 
    printf("mqDonaldid=%d\n",mqDonaldid); 

    pthread_create(&t1, NULL, MICKEYthreadFunc, "Mickey sends\n"); 
    pthread_create(&t2, NULL, MINNIEthreadFunc, "Minnie replies\n"); 
    pthread_create(&t3, NULL, DONALDthreadFunc, "Donald sends\n"); 
    pthread_create(&t4, NULL, PLUTOthreadFunc, "Pluto replies\n"); 

    while(1) 
    { 
     sleep(5); 
    } 
} 

我发现，由于某些原因，msgget失败，都MQS都具有相同的ID，那么它是一个线程首先踢的比赛。

随着你的程序的这款改装版，这样做：

touch mqmickey 
touch mqdonald 

先创建文件。

你应该看到消息去正确的收件人。

Answer 3

非常感谢你sehe和VJo，

下面是节目的修改版本，考虑到所有的你的意见。

希望这对其他人有用。

伯特

/* 
    We want Mickey to send a message to Minnie exclusively. 
    We want Donald to send a message to pluto exclusively. 

    To compile: 
    gcc mm.c -o mm -fno-stack-protector -pthread 

    Requires: 
    syslog enabled and started. 

    To test: 
    time ./mm | { trap "" INT; sort | uniq -c | tee stats; } 

    Results: 
    1503373 Donald sent a message to Pluto. 
      1 Donald thread successfully started. 
    1423964 Mickey sent a message to Minnie. 
      1 Mickey thread successfully started. 
    1423958 Minnie received: Hello, Minnie! 
      1 Minnie thread successfully started. 
    1503333 Pluto received: Sit, Pluto! 
      1 Pluto thread successfully started. 

    real 0m17.133s 
    user 0m16.053s 
    sys 0m5.248s 

*/ 
#include <sys/types.h> 
#include <sys/msg.h> 
#include <sys/ipc.h> 
#include <string.h> 
#include <stdio.h> 
#include <stdlib.h> 
#include <syslog.h> 


#define MQMICKEY "/tmp/MQMickey" 
#define MQDONALD "/tmp/MQDonald" 

pthread_t t1,t2,t3,t4; 
FILE *fMickeypointer; 
FILE *fDonaldpointer; 

// Mickey send 
key_t ipcMickey; 
int mqMickeyid; 
char helloMickeymsg[] = {"Hello, Minnie!"}; 
struct { long type; char text[100]; } myMickeymsg; 

// Minnie get 
int mqMinnieid; 
struct { long type; char text[100]; } myMinniemsg; 

// Donald send 
key_t ipcDonald; 
int mqDonaldid; 
char helloDonaldmsg[] = {"Sit, Pluto!"}; 
struct { long type; char text[100]; } myDonaldmsg; 

// Pluto get 
int mqPlutoid; 
struct { long type; char text[100]; } myPlutomsg; 

static void * DONALDthreadFunc(void *arg) 
{ 
    printf("Donald thread successfully started.\n"); 
    char *s = (char *) arg; 

    while(1) 
    { 
     //send 
     memset(myDonaldmsg.text, 0, 100); 
     strncpy(myDonaldmsg.text, helloDonaldmsg, strlen(helloDonaldmsg)); 
     myDonaldmsg.type = 1; 
     msgsnd(mqDonaldid, &myDonaldmsg, sizeof(myDonaldmsg), 0); 
     printf("Donald sent a message to Pluto.\r\n"); 
    } 

    /* just a formality */ 
    return (void *) strlen(s); 
} 

static void * PLUTOthreadFunc(void *arg) 
{ 
    printf("Pluto thread successfully started.\n"); 
    char *s = (char *) arg; 

    while(1) 
    { 
     //receive 
     mqPlutoid = msgget(ipcDonald, 0); 
     msgrcv(mqPlutoid, &myPlutomsg, sizeof(myPlutomsg), 0, 0); 
     printf("Pluto received: %s\r\n", myPlutomsg.text); 
    } 

    /* just a formality */ 
    return (void *) strlen(s); 
} 

static void * MICKEYthreadFunc(void *arg) 
{ 
    printf("Mickey thread successfully started.\n"); 
    char *s = (char *) arg; 

    while(1) 
    { 
     //send 
     memset(myMickeymsg.text, 0, 100); 
     strncpy(myMickeymsg.text, helloMickeymsg, strlen(helloMickeymsg)); 
     myMickeymsg.type = 1; 
     msgsnd(mqMickeyid, &myMickeymsg, sizeof(myMickeymsg), 0); 
     printf("Mickey sent a message to Minnie.\r\n"); 
    } 

    /* just a formality */ 
    return (void *) strlen(s); 
} 

static void * MINNIEthreadFunc(void *arg) 
{ 
    printf("Minnie thread successfully started.\n"); 
    char *s = (char *) arg; 

    while(1) 
    { 
     //receive 
     mqMinnieid = msgget(ipcMickey, 0); 
     msgrcv(mqMinnieid, &myMinniemsg, sizeof(myMinniemsg), 0, 0); 
     printf(" Minnie received: %s\r\n", myMinniemsg.text); 
    } 

    return (void *) strlen(s); 
} 

int main (void) 
{ 
    /* 
     MUST create the target files first. 

     The return of ftok is -1 if file is missing. If BOTH files are missing, both ipc keys 
     will be -1, meaning all traffic shares the same port. Result: one queue will acquire 
     messages not intended for it! 
    */ 
    fMickeypointer = fopen(MQMICKEY, "a"); 
    if(fMickeypointer == NULL) 
    { 
     printf("Failed to create Mickey queue file.\n"); 

     /* system logging */ 
     setlogmask (LOG_UPTO (LOG_NOTICE)); 
     openlog ("MickeyDonald", LOG_CONS | LOG_PID | LOG_NDELAY, LOG_LOCAL1); 
     syslog (LOG_NOTICE, "Failed to create Mickey queue file."); 
     closelog(); 
     exit(91); 
    } 

    fDonaldpointer = fopen(MQDONALD, "a"); 
    if(fDonaldpointer == NULL) 
    { 
     printf("Failed to create Donald queue file.\r\n"); 

     /* system logging */ 
     setlogmask (LOG_UPTO (LOG_NOTICE)); 
     openlog ("MickeyDonald", LOG_CONS | LOG_PID | LOG_NDELAY, LOG_LOCAL1); 
     syslog (LOG_NOTICE, "Failed to create Donald queue file."); 
     closelog(); 
     exit(92); 
    } 

    ipcMickey = 0; 
    ipcDonald = 0; 

    /* make sure we assign DIFFERENT ipc values */ 
    while(ipcMickey == ipcDonald) 
    { 
     ipcMickey = ftok(MQMICKEY, 63); 
     if (-1==ipcMickey) 
     { 
      printf("ipcMickey does not exist.\r\n"); 

      /* system logging */ 
      setlogmask (LOG_UPTO (LOG_NOTICE)); 
      openlog ("MickeyDonald", LOG_CONS | LOG_PID | LOG_NDELAY, LOG_LOCAL1); 
      syslog (LOG_NOTICE, "Failed to ftok Mickey file."); 
      closelog(); 
      exit(93); 
     } 

     ipcDonald = ftok(MQDONALD, 69); 
     if (-1==ipcDonald) 
     { 
      printf("ipcDonald does not exist.\r\n"); 

      /* system logging */ 
      setlogmask (LOG_UPTO (LOG_NOTICE)); 
      openlog ("MickeyDonald", LOG_CONS | LOG_PID | LOG_NDELAY, LOG_LOCAL1); 
      syslog (LOG_NOTICE, "Failed to ftok Donald file."); 
      closelog(); 
      exit(94); 
     } 
    } 

    mqMickeyid = msgget(ipcMickey, IPC_CREAT | 0666); 
    mqDonaldid = msgget(ipcDonald, IPC_CREAT | 0666); 

    pthread_create(&t1, NULL, MICKEYthreadFunc, "Mickey sends\n"); 
    pthread_create(&t2, NULL, MINNIEthreadFunc, "Minnie replies\n"); 
    pthread_create(&t3, NULL, DONALDthreadFunc, "Donald sends\n"); 
    pthread_create(&t4, NULL, PLUTOthreadFunc, "Pluto replies\n"); 

    long exitcounter = 0; 
    while(exitcounter < 100000) 
    { 
     exitcounter +=1;   
     usleep(100); 
    } 
    exit(0); 
}