Данные накапливаются (FIFO?) В широковещательном UDP-сокете

Question

У меня есть устройство, которое отправляет мне поток данных мониторинга (размер 58) каждую секунду в широковещательной рассылке.Я читаю этот поток с помощью программы на Си.Моя проблема в том, что если я не читаю этот поток в течение нескольких секунд, в следующий раз, когда я его читаю, у меня будет гораздо больше данных, которые мне нужны, например, если у меня был FIFO, который заполняется в течение времени, когда я не читаюstream.

Я пытался использовать разные методы чтения, например, использовать select, неблокирующий сокет ... всегда один и тот же.Может быть, есть только одна деталь, по которой я скучаю ... Вот пример кода:

#include <stdio.h>      /* for printf() and fprintf() */
#include <sys/socket.h> /* for socket(), connect(), sendto(), and recvfrom() */
#include <arpa/inet.h>  /* for sockaddr_in and inet_addr() */
#include <stdlib.h>     /* for atoi() and exit() */
#include <string.h>     /* for memset() */
#include <unistd.h>     /* for close() */

#define MAXRECVSTRING 58  /* Longest string to receive */


int main(int argc, char *argv[])
{
    int sock;                         /* Socket */
    struct sockaddr_in broadcastAddr; /* Broadcast Address */
    unsigned short broadcastPort;     /* Port */
    char recvString[MAXRECVSTRING+1]; /* Buffer for received string */
    int recvStringLen;                /* Length of received string */

    if (argc != 2)    /* Test for correct number of arguments */
    {
        fprintf(stderr,"Usage: %s <Broadcast Port>\n", argv[0]);
        exit(1);
    }

    broadcastPort = atoi(argv[1]);   /* First arg: broadcast port */

    /* Create a best-effort datagram socket using UDP */
    if ((sock = socket(PF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0)
        perror("socket() failed");

    /* Construct bind structure */
    memset(&broadcastAddr, 0, sizeof(broadcastAddr));   /* Zero out structure */
    broadcastAddr.sin_family = AF_INET;                 /* Internet address family */
    broadcastAddr.sin_addr.s_addr = htonl(INADDR_ANY);  /* Any incoming interface */
    broadcastAddr.sin_port = htons(broadcastPort);      /* Broadcast port */

    /* Bind to the broadcast port */
    if (bind(sock, (struct sockaddr *) &broadcastAddr, sizeof(broadcastAddr)) < 0)
        perror("bind() failed");

    /* Receive datagram from the server */
    int g=0;
    int time_to_sleep=10; 
    while(1)
    {
        if(g==10)
        {
            while(time_to_sleep)
            {
                printf("sleep for %d\n", time_to_sleep);
                time_to_sleep=sleep(time_to_sleep);
            }//after that my next read is 10 data stream of 58 in one !!
        }


        if ((recvStringLen = recvfrom(sock, recvString, MAXRECVSTRING, 0, NULL, 0)) < 0)
        {
             perror("recvfrom() failed");
        }
        //later will check the header but for now just the size is enough
        if (recvStringLen==58){ printf("Read okay bc Size = %d \n", recvStringLen); }


    }

    close(sock);
    exit(0);
}

Итак, я попытаюсь объяснить мою проблему: мое оборудование отправляет мне UDP-пакеты (размер 58) каждые секунды, которые япродолжайте читать постоянно.Через x раз (g == 10) я решаю спать в течение 10 секунд.В течение этих 10 секунд мое оборудование продолжает отправлять UDP-пакеты, но я их не читаю и не хочу.В 11 секунд, когда я просыпаюсь, я хочу прочитать мои (g == 10) + 11-ые пакеты, а не 11-й + 10, которые я не прочитал во время сна.К сожалению, когда я читаю 11-е с одним recvfrom, я получаю все 10 предыдущих ...

Я пробовал с и без коммутатора Ethernet в случае, но та же проблема.Я должен что-то неправильно понять в сокете ... вы можете мне помочь?

Кстати, мой код может быть просто катастрофой, я новичок.Не стесняйтесь меня поправлять!

Спасибо!

Answer 1

Я должен что-то неправильно понять в сокете ... Вы можете мне помочь?

В ядре есть буферы приема и отправки сокета.Ядро получает дейтаграммы для вашего слушающего сокета и сохраняет их в буфере приема сокета ядра.recvfrom call копирует самую старую дейтаграмму из буфера сокета ядра в ваш буфер пространства пользователя.Тот факт, что вы не вызываете recvfrom, не означает, что ядро ​​прекращает прием дейтаграмм (вам нужно закрыть сокет, чтобы ядро ​​прекратило получать данные).

Answer 2

Вы на самом деле не получаете 10 или 11 пакетов от одного recvfrom.Вы вызываете recvfrom в цикле, и каждая итерация цикла читает пакет.Таким образом, вам нужен какой-то способ узнать, что самый последний прочитанный вами пакет является «последним».

Способ сделать это - постоянно читать, пока нечего читать, тогда вы знаете, что самый последний - этопоследний.Однако, как ваш код записан в настоящий момент, если нет пакетов для чтения, recvfrom будет блокироваться, пока пакет не станет доступным.Вы можете изменить это, переведя сокет в неблокирующий режим:

int fd_flag;
if ((fdflag = fcntl(sock, F_GETFL)) == -1) {
    perror("Error getting socket descriptor flags");
    exit(1);
}
fdflag |= O_NONBLOCK;
if (fcntl(sock, F_SETFL, fdflag) == -1) {
    perror("Error setting non-blocking option");
    exit(1); 
}

Теперь, когда вы вызываете recvfrom, если нет доступных данных, он вернет -1 и установит errno в EAGAIN,Вы можете использовать это для проверки «последнего» пакета:

while(1)
{
    while(time_to_sleep)
    {
        printf("sleep for %d\n", time_to_sleep);
        time_to_sleep=sleep(time_to_sleep);
    }//after that my next read is 10 data stream of 58 in one !!

    int read_one = 0;
    while (1) {
        if ((recvStringLen = recvfrom(sock, recvString, MAXRECVSTRING, 0, NULL, 0)) < 0)
        {
            if (errno == EAGAIN) {
                // nothing more to read
                break;
            } else {
                perror("recvfrom() failed");
            }
        } else {
            read_one = 1;
            if (recvStringLen==58){ printf("Read okay bc Size = %d \n", recvStringLen); }
        }
    }

    if (read_one) {
        // process last packet read
    } else {
        // nothing read, so do nothing
    }
}