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..
上記のメソッドで呼び出される 1 つのメソッド:
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!!");
}
プログラムをテストする実行の 1 つ: 送信者は 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 を使用してスレッドを実行していないため、パケットが失われていると思われます。
明確化: ループバックを使用してパケットを送信していますが、チャネルにパケット損失はありません。パケットを処理するコードが少ない場合、すべてのパケットが受信されます。