[Linphone-developers] measuring end to end delay

[Iban Lopetegi Zinkunegi]

Hi all,

I am not sure whether this is the right place to ask this, so apologise in advance if this is the wrong place.

I am trying to measure the end to end delay or RoundTrip time for RTP packets. So I ve got computer A with a RTP_Sender that sends packets to another computer B which simply Receives the packet and sends it back. Both computers have a sender and a receiver programed with threads. Now, if i leave the computer to handle the threads, the scheduling of them appears to be faulty, meaning faulty that spends too long periods on a thread, causing many packets to be lost. To avoid this, I use a jumping system where each time I receive a packet I send a packet. In order to do that, i create a queue where all received payloads are copied and taken by the receiver and sender respectively.   

The problem is that in a 10Mbps ethernet I am getting end to end delays of about 2 seconds which is far too much, and it is obvious that I am some programming is wrong. I have tried changing rtp_session_set_scheduling_mode and rtp_session_set_blocking_mode by putting the to 0 but there is no difference.

The pseudo code for the proxy is below. The code has been checked with Valgrind for memory allocation faults, so ignore variables that i havent posted here. My main, concern lies in "my_rtp_session_recv_with_ts_v2" which is virtually a "rtp_session_recv_with_ts " function but adapted so that i can get the sequence number.This is important for me here so that i can keep a track of the packets.

Thanks


/*------------Proxy pseudo code-----------*/
/*----Global Variables---*/
volatile sig_atomic_t runcond;

pthread_mutex_t token_mutex;
pthread_cond_t ReceiveData_cond;//=PTHREAD_COND_INITIALIZER;
int breceive;
pthread_cond_t SendData_cond;//=PTHREAD_COND_INITIALIZER;
int bsend;
pthread_cond_t Start_cond;

struct _cdn_buffer{
        unsigned char * buf_f;
    uint16_t cseq_f;
};
typedef struct     _cdn_buffer cdn_buffer;

/*----END Global Variables---*/
int main (int argc, char ** argv)
{

breceive=0;
bsend=0;
pthread_cond_init(&ReceiveData_cond,NULL);
pthread_cond_init(&SendData_cond,NULL);
pthread_cond_init(&Start_cond,NULL);
...
runcond=1;
pthread_create (&pid_rtp_sendera, NULL, &send_rtpa, (void *) &channels);
pthread_create (&pid_rtp_receiver, NULL, &receive_rtp_ntwcdn, (void *) &channels);

pthread_join (pid_rtp_sendera,NULL);
pthread_join (pid_rtp_receiver,NULL);
...
}


int my_rtp_session_recv_with_ts_v2 (uint16_t * cseq_tmp,RtpSession * session, uint8_t * buffer,
                   int len, uint32_t ts, int * have_more){
    mblk_t *mp=NULL;
    int plen=0,blen=0;
    *have_more=0;
    rtp_header_t *hdr;

    while(1){
        if (session->pending){
            mp=session->pending;
            session->pending=NULL;
        }else {
            mp=rtp_session_recvm_with_ts(session,ts);
            
            if (mp!=NULL) {
                    hdr=(rtp_header_t*)mp->b_rptr;
                    rtp_get_payload(mp,&mp->b_rptr);
            }
        }
        if (mp){
            plen=mp->b_wptr-mp->b_rptr;
            if (plen<len){
                memcpy(buffer,mp->b_rptr,plen);
                buffer+=plen;
                blen+=plen;
                len-=plen;
                freemsg(mp);
                mp=NULL;
            }else{
                memcpy(buffer,mp->b_rptr,len);
                *cseq_tmp=(hdr->seq_number);
                //if (DEBUG){printf("Tcsq %d",*cseq_tmp);fflush(stdout);}
                mp->b_rptr+=len;
                buffer+=len;
                blen+=len;
                len=0;
                session->pending=mp;
                *have_more=1;
                break;
            }
        }else    
            break;
    }
    return blen;
}


void* receive_rtp_ntwcdn (void* unused)
{

int have_more;
int payloadsize=160;//PCM
RtpSession *session;
unsigned char * payload;
if (!(payload=malloc((payloadsize)*sizeof(char))))printf("malloc error for payload\n");
uint16_t diff_cseq,Lcseq,Rcseq,cseq_tmp;
diff_cseq=0;
Lcseq=0;
Rcseq=0;
cseq_tmp=0;
int clost=0;

...
recv_session_set=session_set_new();
session_rx=rtp_session_new(RTP_SESSION_RECVONLY);
rtp_session_set_scheduling_mode(session_rx,0);
rtp_session_set_blocking_mode(session_rx,0);
rtp_session_set_local_addr(session_rx,myip,nsrcport);
rtp_session_enable_adaptive_jitter_compensation(session_rx,adapt);
rtp_session_set_jitter_compensation(session_rx,jittcomp);
rtp_session_set_payload_type(session_rx,0);
rtp_session_signal_connect(session_rx,"ssrc_changed",(RtpCallback)ssrc_cb,0);
rtp_session_signal_connect(session_rx,"ssrc_changed",(RtpCallback)rtp_session_reset,0);    


...
unsined char buffer[160];
while (runcond){
    have_more=1;
    while ((have_more==1))
    {
        pthread_mutex_lock(&token_mutex);
        err=my_rtp_session_recv_with_ts_v2(&cseq_tmp,session,payload,payloadsize,ts,&have_more);         
        if (err>0){
            cdnbuffer.cseq_f=cseq_tmp;
            Rcseq=cdnbuffer.cseq_f;
            diff_cseq=Rcseq-Lcseq;
            if (diff_cseq>1){
                clost=clost+diff_cseq-1;
                printf("rXA Lost %d with Lcseq %d ",clost,Lcseq);fflush(stdout);
                Lcseq=Rcseq;
            }else Lcseq=Rcseq;

            /*Add to a queue*/
            strncpy(cdnbuffer.buf_f, payload, payloadsize);
            if (q_position_a[pair]>10)//to keep buffer small
                q_position_a[pair]=q_delete(queue_a[pair],q_position_a[pair],payloadsize);
            q_position_a[pair] =q_add(queue_a[pair],q_position_a[pair],&cdnbuffer,payloadsize);

            /*Jump to the sender*/
            bsend=1;
            pthread_cond_signal(&SendData_cond);
            breceive=0;        
            while (!breceive)
                pthread_cond_wait(&ReceiveData_cond,&token_mutex);
              pthread_mutex_unlock(&token_mutex);        
                                
            
        }    
        
        ts+=payloadsize;

    }
}
   

void* send_rtpa (void* unused)
{
int have_more;
int payloadsize=160;//PCM
RtpSession *session;
unsigned char * payload;
if (!(payload=malloc((payloadsize)*sizeof(char))))printf("malloc error for payload\n");
...

send_session_set=session_set_new();
session=rtp_session_new(RTP_SESSION_SENDONLY);
rtp_session_set_scheduling_mode(session,1);
rtp_session_set_blocking_mode(session,1);
rtp_session_set_symmetric_rtp(session,FALSE);
rtp_session_set_remote_addr(session,destip,ndestport);
rtp_session_set_local_addr (session, myip, nsrcport);
rtp_session_set_payload_type(session,0);

...

while (runcond){
    pthread_mutex_lock(&token_mutex);

        /*Wait until packet received*/
        while(!bsend)
            pthread_cond_wait(&SendData_cond,&token_mutex);
        if (q_position_a[pair]>0){
            while(q_position_a[pair]!=0){
                q_position_a[pair]=q_get(queue_a[pair], q_position_a[pair], & cdnbuffer_a, payloadsize);
                strncpy(buffer_ex, cdnbuffer_a.buf_f,payloadsize);
                add_to_payload_padding(buffer_ex,payloadsize,cdnbuffer_a.cseq_f,0);
                rtp_session_send_with_ts(session,buffer_ex,payloadsize,user_ts);
                user_ts+=payloadsize;
            }
            
        }
        
        breceive=1;
        pthread_cond_signal(&ReceiveData_cond);
        bsend=0;


    pthread_mutex_unlock(&token_mutex);
    }


}

;