如何完整地接收高速UDP数据包？

Question

我有一个连续发送数据的UDP服务器。我要接收服务器发送的所有数据包。

在服务器端，我有两个线程。一个线程从文件中连续读取数据并放入Deque。另一个线程从deque读取数据并不断发送到UDP客户端。客户端代码不断地从服务器接收数据。

#define MAX_BUFFER_SIZE 1400

typedef struct
{
    T_UCHAR buffer[MAX_BUFFER_SIZE];
    DWORD buf_size;
}RAWDATA_LOG;

deque<RAWDATA_LOG>  m_RawdataLog;   

void TransmitContinuous()
{
    if (m_sock_type_tcp == SOCK_UDP)
    {
        fileReadComplete=false;
        //start data transmission thread
        pWin_thread=AfxBeginThread(StartDataTransmitThread, (LPVOID) this);
        ReadFromFile();
    }   

}


void ReadFromFile()
{
    int bytesRead=0;
    m_no_of_bytes = MAX_BUFFER_SIZE;
    BYTE input_buf[MAX_BUFFER_SIZE]={'\0'};


    GetDlgItemText(IDEBC_FILENAME,m_fileInput);


    m_InputFile=NULL;
    /*opening the file to read*/
    m_InputFile = _tfopen(m_fileInput,L"rb");
    if(m_InputFile == NULL)
    {
        AfxMessageBox(L"Unable to open the Input file");
    }
    else
    {

        while(!feof(m_InputFile))
        {
            bytesRead=fread(input_buf,1,m_no_of_bytes,m_InputFile);

            writeRawdataToDeque(input_buf,m_no_of_bytes);
            noofBytesReadfromFile+=bytesRead;

        }
        fileReadComplete=true;
    }
}


void writeRawdataToDeque(T_UCHAR *buffer,T_S32 size)
{

    T_S32 temp_size = size;
    T_S32 size_counter = 0;
    RAWDATA_LOG temp_rawDataStruct;

    while(temp_size>0)
    {
        if(temp_size <= MAX_BUFFER_SIZE)
        {
            memcpy(temp_rawDataStruct.buffer,&buffer[size_counter],temp_size);
            temp_rawDataStruct.buf_size = temp_size;
            noofBytesWrittentoDeque+=temp_size;

        }
        else
        {
            memcpy(temp_rawDataStruct.buffer,&buffer[size_counter],sizeof(temp_rawDataStruct.buffer));
            temp_rawDataStruct.buf_size = MAX_BUFFER_SIZE;
            noofBytesWrittentoDeque+=MAX_BUFFER_SIZE;
        }

        CSingleLock datalock(&m_Cs_RawDataLog);
        datalock.Lock();
        m_RawdataLog.push_back(temp_rawDataStruct);
        datalock.Unlock();


        memset(&temp_rawDataStruct,0,sizeof(temp_rawDataStruct));
        size_counter += MAX_BUFFER_SIZE;
        temp_size = temp_size - MAX_BUFFER_SIZE;
    }                                               
}

unsigned int StartDataTransmitThread (LPVOID param)
{

    RAWDATA_LOG temp_rawDataBuf;
    int byesWritten=0;
    CString tmpStr;

    while(1)
    {

        if(!m_RawdataLog.empty())
        {
            CSingleLock datalock(&m_Cs_RawDataLog);
            datalock.Lock();
            temp_rawDataBuf = m_RawdataLog.front();
            m_RawdataLog.pop_front();
            datalock.Unlock();

            //transmit the data through socket  
            byesWritten=WritetoClient(clientIp,clientPort,(const LPBYTE)&temp_rawDataBuf, MAX_BUFFER_SIZE);

            noofBytesTransmitted+=byesWritten;

        }
        else
        {
            if(fileReadComplete == true)
            {

                break;
            }
        }       
    }

    return true;
}



bool CreateServer(char ipaddr[],int port)
{
    sockaddr_in ServerSockAddr;
    WORD m_wVersionRequested;
    WSADATA m_wsaData;
    int m_wsaErr;

    ServerSockAddr.sin_addr.s_addr=inet_addr("192.168.11.80");
    ServerSockAddr.sin_family = AF_INET;
    ServerSockAddr.sin_port = htons(2011);


     m_wVersionRequested = MAKEWORD(2, 2);          
     m_wsaErr = WSAStartup(m_wVersionRequested, &m_wsaData);
     if (m_wsaErr != 0) 
     {
        /* Tell the user that we could not find a usable */
        /* Winsock DLL.                                  */
        MessageBox(L"WSAStartup failed with error:" + m_wsaErr);
        return 1;
     }

    SOCKET sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_IP);
    if (INVALID_SOCKET != sock)
    {


        if ( SOCKET_ERROR == bind(sock,(struct sockaddr *) & ServerSockAddr, sizeof(ServerSockAddr)))
        {
            int b= GetLastError();
            closesocket( sock );
            return false;
        }
    }
    else
    {
        closesocket( sock );
        return false;
    }

    m_hComm = (HANDLE) sock;
}


int WritetoClient(char ipaddr[],int port,BYTE buf[],int len)
{

     sockaddr_in clientSockAddr;
     int res=0;


      SOCKET s = (SOCKET) m_hComm;

      clientSockAddr.sin_addr.s_addr=inet_addr("192.168.11.80");
      clientSockAddr.sin_family = AF_INET;
      clientSockAddr.sin_port = htons(port);

      res = sendto( s, (const char *)buf, len, 0, (SOCKADDR  *) &clientSockAddr, sizeof(clientSockAddr));

      return res;
}

#include <stdlib.h>
#include <winsock2.h>
#include <ws2tcpip.h>
#pragma comment(lib, "ws2_32")

#define BUFSIZE 1000000

int _tmain(int argc, _TCHAR* argv[])
{
    SOCKET sockfd;
    int  portno, n;
    int serverlen;
    struct sockaddr_in serveraddr;
    struct hostent *server;
    char *hostname;
    char buf[BUFSIZE];
    int BytesReceived=0;
    int buff_size=1000000;

    WORD m_wVersionRequested;
    WSADATA m_wsaData;
    int m_wsaErr;

     m_wVersionRequested = MAKEWORD(2, 2);          
     m_wsaErr = WSAStartup(m_wVersionRequested, &m_wsaData);
     if (m_wsaErr != 0) 
         {
        /* Tell the user that we could not find a usable */
        /* Winsock DLL.                                  */
        printf("WSAStartup failed with error:");
        return 1;
     }

    /* socket: create the socket */
    sockfd = socket(AF_INET, SOCK_DGRAM, IPPROTO_IP);
    if (sockfd < 0) 
    {

        printf("ERROR opening socket");
    }


    else
    {
        /* build the server's Internet address */

        serveraddr.sin_family = AF_INET;

        serveraddr.sin_port = htons(2010);
        serveraddr.sin_addr.s_addr=inet_addr("192.168.11.80");



        /* send the message to the server */
        serverlen = sizeof(serveraddr);

        setsockopt(sockfd,SOL_SOCKET,SO_RCVBUF,(char *)(&buff_size), sizeof(buff_size));

        if ( SOCKET_ERROR == bind(sockfd,(struct sockaddr *) & serveraddr, sizeof(serveraddr)))
        {
            printf("Bind Error");
            int a = GetLastError();
            printf("Error ID:%d",a);
        }
        else
        {

            printf("Reading From Server:\n");

            while(1)
            {


                n = recvfrom(sockfd, buf, sizeof(buf), 0,(struct sockaddr *) &serveraddr, &serverlen);

                if (n < 0) 
                {
                    printf("ERROR in recvfrom\n");
                    int b = GetLastError();
                    printf("Error ID:%d\n",b);
                }
                else
                {

                    BytesReceived+=n;

                    TRACE("\nTotal Bytes Received:%d\n",BytesReceived);
                }

            }
        }
    }
    getchar();
    return 0;



}

Answer 1

我发现你的代码有几个问题。

在服务器端，您要将服务器的套接字绑定到192.168.11.80，并且还要将数据包发送到192.168.11.80，而不是将数据包发送到WriteToclient()函数的iPaddr参数中指定的任何IP地址。

在客户端，您将bind()'将客户端的套接字与服务器的IP地址连接起来。

由于您（可能）没有正确绑定客户端的套接字，和/或（可能）没有发送到正确的IP，您很可能从其他地方接收到不期望的数据。这可以解释你看到的巨大差异。我建议让您的客户机输出它实际接收到的数据，这样您就可以确保它接收到您所期望的数据，例如：

n = recvfrom(sockfd, buf, sizeof(buf), 0, (struct sockaddr *) &serveraddr, &serverlen);
if (n == SOCKET_ERROR) 
{
    int b = WSAGetLastError();
    printf("ERROR in recvfrom\nError ID:%d\n",b);
}
else
{
    BytesReceived += n;
    TRACE("\nMessage Received from %s:%hu: '%.*s'", inet_ntoa(serveraddr.sin_addr), ntohs(serveraddr.sin_port), n, buf);
    TRACE("\nTotal Bytes Received: %d\n", BytesReceived);
}