поток получает из очереди IPC-сообщение других потоков (Linux)

ProgramBox

поток получает из очереди IPC-сообщение других потоков (Linux)

Post author:admin
Запись опубликована:29 мая, 2023
Post category:Вопросы по программированию

#c #linux #ipc #mq

Вопрос:

Когда у меня есть две разные очереди сообщений IPC в Linux, иногда перехватываются сообщения из неправильной очереди.

Следующая игрушечная программа отображает проблему, может повторяться на разных процессорах.

Любая помощь с благодарностью!

Берт

 /*
    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, amp;myDonaldmsg, sizeof(myDonaldmsg), 0);
        printf("Donald sent a message to Pluto.rn");
        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, amp;myPlutomsg, sizeof(myPlutomsg), 0, 0);
        printf("Pluto received: %srnrn", 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, amp;myMickeymsg, sizeof(myMickeymsg), 0);
        printf("Mickey sent a message to Minnie.rn");
        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, amp;myMinniemsg, sizeof(myMinniemsg), 0, 0);
        printf("Minnie received: %srnrn", 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(amp;t1, NULL, MICKEYthreadFunc, "Mickey sendsn");
    pthread_create(amp;t2, NULL, MINNIEthreadFunc, "Minnie repliesn");
    pthread_create(amp;t3, NULL, DONALDthreadFunc, "Donald sendsn");
    pthread_create(amp;t4, NULL, PLUTOthreadFunc, "Pluto repliesn");

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

Ответ №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()000) ):

 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

1. Также маловероятно, но далеко в пределах возможного, что ftok возвращает один и тот же ключ для обоих имен файлов. В нем просто слишком мало битов, чтобы избежать случайных сговоров. OP следует рассмотреть возможность изучения очередей сообщений POSIX ( mq_* ), в которых отсутствует этот фундаментальный недостаток. В качестве альтернативы, вы могли бы создавать временные файлы в цикле, пока не получите уникальные ключи, а затем предоставлять ключи всем потокам / процессам, которым они нужны, вместо повторного вызова ftok .

Ответ №2:

БОЛЬШОЕ вам спасибо 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, amp;myDonaldmsg, sizeof(myDonaldmsg), 0);
        printf("Donald sent a message to Pluto.rn"); 
    }

    /* 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, amp;myPlutomsg, sizeof(myPlutomsg), 0, 0);
        printf("Pluto received: %srn", 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, amp;myMickeymsg, sizeof(myMickeymsg), 0);
        printf("Mickey sent a message to Minnie.rn");
    }

    /* 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, amp;myMinniemsg, sizeof(myMinniemsg), 0, 0);
        printf(" Minnie received: %srn", 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.rn");

        /* 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.rn");

            /* 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.rn");

            /* 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(amp;t1, NULL, MICKEYthreadFunc, "Mickey sendsn");
    pthread_create(amp;t2, NULL, MINNIEthreadFunc, "Minnie repliesn");
    pthread_create(amp;t3, NULL, DONALDthreadFunc, "Donald sendsn");
    pthread_create(amp;t4, NULL, PLUTOthreadFunc, "Pluto repliesn");

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

Ответ №3:

Вот слегка измененная версия вашей программы :

 /*
    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, amp;myDonaldmsg, sizeof(myDonaldmsg), 0);
        printf("Donald sent a message to Pluto.rn");
        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, amp;myPlutomsg, sizeof(myPlutomsg), 0, 0);
        printf("Pluto received: %srnPluto uses MQ with id = %drn", 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, amp;myMickeymsg, sizeof(myMickeymsg), 0);
        printf("Mickey sent a message to Minnie.rn");
        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, amp;myMinniemsg, sizeof(myMinniemsg), 0, 0);
        printf(" Minnie received: %srnMinnie uses MQ with id = %drn", myMinniemsg.text,mqMinnieid);
        sleep(3);
    }

    return (void *) strlen(s);
}

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

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

    pthread_create(amp;t1, NULL, MICKEYthreadFunc, "Mickey sendsn");
    pthread_create(amp;t2, NULL, MINNIEthreadFunc, "Minnie repliesn");
    pthread_create(amp;t3, NULL, DONALDthreadFunc, "Donald sendsn");
    pthread_create(amp;t4, NULL, PLUTOthreadFunc, "Pluto repliesn");

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

Я обнаружил, что по какой-то причине msgget происходит сбой, и оба MQS имеют одинаковый идентификатор, тогда это гонка, какой поток запускается первым.

С помощью этой модифицированной версии вашей программы сделайте это :

 touch mqmickey
touch mqdonald

чтобы сначала создать файлы.

И вы должны увидеть, что сообщения отправляются правильным получателям.