0

我有一个程序可以在 UDP 套接字上接收数据包。

这是接收数据包的循环：

clientfd = bind(client_s,(const sockaddr*) &client_addr, sizeof(client_addr));

    /*---Forever... ---*/
    while (1)
    {   
        addrlen=sizeof(client_addr);
            bufferWithPacketData = new char [headerSizeTot+symbol_size];
        int n = recvfrom (client_s, bufferWithPacketData, symbol_size + headerSizeTot, 0,(struct sockaddr*)&addrSenderOfVideo, &fromlen);
        if (n >= 0  ) //n=-1 => nothing receieved.. 
        {
        pthread_create(&threadID , NULL, &ProcessDataOfPacketInThread, (void*) bufferWithPacketData );
        }

    } //end of while

处理数据包的线程代码：

void* ProcessDataOfPacketInThread (void* ptr) 
 {
 char* bufferWithPacketData = (char*) ptr;
    pthread_t threadID;

                 //todo remove fprint messages..in the listener thread..
        //first byte containing which fec-session,etcetera..

        //todo remove print outs here

        unsigned int curr_fec_session = extract_value_from_header (bufferWithPacketData,POSITION_OF_FEC_SESSION);
   unsigned int fecSessionModuloNr = curr_fec_session % SIZE_OF_MAX_FEC_SESSIONS_AT_SAME_TIME;
        assert ( fecSessionModuloNr < SIZE_OF_MAX_FEC_SESSIONS_AT_SAME_TIME );

        //todo remove print outs here
            //i can do modulo elsewhere also...easiest..
        //   int fecSessionNrModulo = curr_fec_session % SIZE_OF_MAX_FEC_SESSIONS_AT_SAME_TIME;
        //one more symbol for this fec-session...
        //todo remove this if we are missing packets..

                int nrOfPktReceivedForCurrentSession;
                int nrOfPacketLossesForCurrentSession;
                float currentLostRate;

                unsigned int curr_symbol_nr = extract_value_from_header (bufferWithPacketData,POSITION_OF_SYMBOL_NR);

                  pthread_mutex_lock (&mutexForAccessSymbolAndEncodingAndNrReceivedVector[fecSessionModuloNr]);          
                  vectorOfNumberOfReceivedSymbolsForFecUnit [fecSessionModuloNr]++;
          nrOfPktReceivedForCurrentSession = vectorOfNumberOfReceivedSymbolsForFecUnit [fecSessionModuloNr];
               pthread_mutex_unlock (&mutexForAccessSymbolAndEncodingAndNrReceivedVector[fecSessionModuloNr]);  
      //curr_fec_session == 0 &&
        //Do this in new threads all of the time to ensure that we are not blocked receiving new packets...

    //Run InitParameters-function, which will store parameters for the current fec-session in an array---necessary to use when decoding.Here we can also measure time...

           unsigned int session_sequence_nr = extract_value_from_header (bufferWithPacketData,POSITION_OF_SESSION_SEQUENCE_NR);

               fprintf (stderr, "\nProccessData; curr_fec_session %d; nrOfPktReceivedForCurrentSession %d; session_sequence_nr  %d; curr_symbol_nr %d",curr_fec_session,  nrOfPktReceivedForCurrentSession, session_sequence_nr, curr_symbol_nr  ); 


    //size vectorOfVetorWithEncodingSymbols == 0 in the beginning, so is curr_fec_session; check if curr_fec_session <
        //todo vectorOfvectorOfEncodingS0ymbolsTab ..decide order.
        //vectorOfvectorOfSymbolReceiveOrder contains receiver order..
        //i can have the same order for vectorOfvectorOfEncodingSymbolsTab ?
        //We need to expand!!! session=2, means we want size==3 to fit it..
        // curr_fec_session == 4 means we have expanded it ENOUGH!!



    //First packet of NEW SESSION!!!
                //only done one because of numberOfHighestFecSessionReceived =curr_fec_session below..
           if (curr_fec_session > numberOfHighestFecSessionReceived || ( nrOfPktReceivedForCurrentSession == 1  && curr_fec_session == 0 ) ) //First received packet. 
         {
                 pthread_mutex_lock (&mutexForAccessToHighestFecSessionNrReceived);
         numberOfHighestFecSessionReceived =curr_fec_session;
                 pthread_mutex_unlock (&mutexForAccessToHighestFecSessionNrReceived);
                 fprintf (stderr, "CALLING INIT-PARAM");
                 InitParametersAndArraysForNewFecSession (bufferWithPacketData );
            //Erase old for the fec-session three sessions before current..

            if (curr_fec_session >= SIZE_OF_MAX_FEC_SESSIONS_AT_SAME_TIME -1  )
            {
            EraseVectorsForFecSessionModuloNr ( curr_fec_session);
                        fprintf (stderr, "Calling EraseVectors");
            }

            //Received packet of new session-> time to decode old session..Check that all threads performing insertion-elements to vector are finished inside of decoding....       
            //We are decoding the previous session !!
                        if ( curr_fec_session != 0 )
                        {
            DecodeFECSession( curr_fec_session -1);     
                        fprintf (stderr, "Calling Decode FEC-session");
                        }
         } //end of if.

        //Will always take place afte we have inited arrays..
    //Do this for all of the packets received.  
    InsertElementAtCertainPositionOfEncodingSymbolsVector ( bufferWithPacketData );
        fprintf (stderr, "Calling Insert element");


//lastLostRateSentToReceiverFloat

 } //end of method--called for each packet received..

上述方法中调用的一种方法：

void InsertElementAtCertainPositionOfEncodingSymbolsVector (char* bufferWithDataAndHeader ) 
        {
        vector<char*>::iterator it;
        unsigned int curr_fec_session = extract_value_from_header (bufferWithDataAndHeader,POSITION_OF_FEC_SESSION);
       unsigned int fecSessionModuloNr = curr_fec_session % SIZE_OF_MAX_FEC_SESSIONS_AT_SAME_TIME;

        unsigned int session_sequence_nr = extract_value_from_header (bufferWithDataAndHeader,POSITION_OF_SESSION_SEQUENCE_NR);                             
        unsigned int curr_symbol_nr = extract_value_from_header (bufferWithDataAndHeader,POSITION_OF_SYMBOL_NR);

        unsigned int nrOfSrcSymbolsSent = extract_value_from_header (bufferWithDataAndHeader,POSITION_OF_NR_SRC_SYMBOLS);

        unsigned int nrOfRepairSymbolsSent = extract_value_from_header (bufferWithDataAndHeader,POSITION_OF_NR_REPAIR_SYMBOLS); 

        unsigned int totNrOfSymbols =  nrOfSrcSymbolsSent+nrOfRepairSymbolsSent;
        fprintf (stderr, "\nINSERT:curr_fec_session %d; session_sequence_nr %d",curr_fec_session, session_sequence_nr);
        //change the address of the buffer...
        bufferWithDataAndHeader += headerSizeTot;
        fprintf (stderr, "\nINSERT:first symbol of buffer data %d; second %d:", bufferWithDataAndHeader[0], bufferWithDataAndHeader[1] ) ;
        //strcpy (destination, source)         
        strcpy (charArrayofCharArrayOfEncodingSymbolsTab [fecSessionModuloNr][curr_symbol_nr ],
bufferWithDataAndHeader );
        fprintf (stderr, "after doing strcpy..........");
    //just commented out to do some testing how the code works without it...
      //   pthread_mutex_lock (&mutexForAccessSymbolAndEncodingAndNrReceivedVector[fecSessionModuloNr]); 

//          fprintf (stderr, "inside of mutex: adding symbol to SymbolReceiveOrder");

    //    vectorOfvectorOfSymbolReceiveOrder[fecSessionModuloNr].push_back (curr_symbol_nr);         
  //      fprintf (stderr, "releasing lock for encoding sectioN");
// pthread_mutex_unlock (&mutexForAccessSymbolAndEncodingAndNrReceivedVector[fecSessionModuloNr]);    
    //   fprintf (stderr, "INSERT:after releasing lock!!");      
 }

测试程序的运行之一：发送方发送超过 512 个数据包。接收器接收到大约 100 个第一个数据包。之后，大约有 200 个数据包丢失。额外收到 30 个数据包，其序列号彼此接近；然后程序终止。

修改代码以测试没有互斥体的行为，表明仍然有很多丢失的数据包。

该程序接收具有以下代码的所有数据包：

while (1)
{   
    addrlen=sizeof(client_addr);
        bufferWithPacketData = new char [headerSizeTot+symbol_size];
    int n = recvfrom (client_s, bufferWithPacketData, symbol_size + headerSizeTot, 0,(struct sockaddr*)&addrSenderOfVideo, &fromlen);
    if (n >= 0  ) //n=-1 => nothing receieved.. 
    {
    nrReceived++;
     unsigned int session_sequence_nr = extract_value_from_header (bufferWithPacketData,POSITION_OF_SESSION_SEQUENCE_NR);        
    fprintf (stderr, "\nsession seq %d; nrReceived %d", session_sequence_nr, nrReceived);
    }

丢失数据包的原因是什么，我该如何解决这个问题？我的猜测是程序没有recvfrom在每个实例中使用 -command 运行线程，套接字接收到新数据包，从而导致数据包丢失。

一些澄清： 我正在使用环回发送数据包，并且通道中没有数据包丢失。当我处理数据包的代码较少时，所有数据包都会收到。

4

2 回答 2

2

UDP 无法保证。如果您需要保证发送的数据包将被接收并且它们将按照发送的顺序被接收，您应该使用其他一些协议，例如 TCP。

于 2012-06-06T19:36:15.857 回答
1

sendto()尽管有 TCP，但如果接收方尚未收到数据包，UDP 套接字将不会阻塞。所以你以最大可能的速率发送数据包。

另一方面，当您创建线程时接收部分可能不会那么快（请注意，您为每条消息创建一个线程），因此由于 UDP 的不可靠特性，您会丢失数据包。当您仅解析和打印接收到的数据包时，这可能不会发生。为什么不尝试通过注入代码来降低发送速率 a sleep()or usleep()，只是为了确定。

此外，您不需要检查n等于 0 的情况。这在 TCP 套接字上意味着对等方正常关闭了连接。

于 2012-06-06T22:43:00.867 回答