ns-allinone-2.30/ns-2.30/rtp/session-rtp.cc

#include <stdlib.h>
#include "packet.h"
#include "ip.h"
#include "rtp.h"
#include <math.h>

#define NOW Scheduler::instance().clock()

static class RTPSourceClass : public TclClass {
public:
        RTPSourceClass() : TclClass("RTPSource") {}
        TclObject* create(int argc, const char*const* argv) {
                if (argc >= 5)
                        return (new RTPSource(atoi(argv[4])));

                return 0;
        }
} class_rtp_source;

static class RTPSessionClass : public TclClass {
public:
        RTPSessionClass() : TclClass("Session/RTP") {}
        TclObject* create(int, const char*const*) {
                return (new RTPSession());
        }
} class_rtp_session;

RTPSession::RTPSession()
        : allsrcs_(0), localsrc_(0), receivers_(0),last_np_(0), T_one_way_(0),rh_(0),
        we_sent(0),last_pkts_lost_(0),last_ehsr_(-1),tx_rate_(0), time_elapsed_(0)
{
        bind("enableFlowControl_", &enableFlowControl_);
        bind("rx_recv_",&rx_recv_);
        bind("jitter_",&jitter_);
        bind("RTT_",&RTT_);
        bind("smooth_loss_",&smooth_loss_);
        alpha = 1;  //we set alpha=1 in initialization 
        last_time_report_ = NOW;

        weight[0] = 1;
        weight[1] = 1;
        weight[2] = 1;
        weight[3] = 1;
        weight[4] = 0.8;
        weight[5] = 0.6;
        weight[6] = 0.4;
        weight[7] = 0.2;
}

RTPSession::~RTPSession()
{
        while (allsrcs_ != 0) {
                RTPSource* p = allsrcs_;
                allsrcs_ = allsrcs_->next;
                delete p;
        }
        delete localsrc_;
        while (receivers_ != 0) {
                RTPReceiver* p = receivers_;
                receivers_ = receivers_->next;
                delete p;
        }
        lst.~list<double>();
        
}

void RTPSession::localsrc_update(int)
{
        localsrc_->np(1);
        localsrc_->is_sender(true);
}

void RTPSession::localsrc_update_nbytes(int n)
{
        localsrc_->nbytes(n);
        
}

#define RTCP_HDRSIZE 8
#define RTCP_SR_SIZE 20
#define RTCP_RR_SIZE 48

int RTPSession::build_report(int bye)
{
        rh_ = new hdr_rtp;
        rh_->rr_ = 0;
        rh_->sr_ = 0;
        int nsrc = 0;
        int nrr = 0;
        int len = RTCP_HDRSIZE;
        we_sent = 0;
        double fraction = 0.0;
        time_elapsed_ = NOW - last_time_report_;
        last_time_report_ = NOW;
        if (localsrc_->np() != last_np_) {
                last_np_ = localsrc_->np();
                we_sent = 1;
                len += RTCP_SR_SIZE;
                //add sender report
                sender_report* sr;
                //fill the report
                sr = new sender_report;
                sr->sender_srcid()= localsrc_->srcid();
                sr->pkts_sent() = localsrc_->np();
                sr->octets_sent() = localsrc_->nbytes();
                sr->rcvrs_ = receivers_;
                //store the report
                rh_->sr_ = sr;
        }
        for (RTPSource* sp = allsrcs_; sp != 0; sp = sp->next) {
                ++nsrc;
                int received = sp->np() - sp->snp();

                if (received == 0) {

                        continue;
                }
                if(localsrc_->srcid() != sp->srcid()) {
                int bytes = received * (sp->ps());
                rx_recv_ = ( 8 * (double)bytes)/time_elapsed_;
                sp->snp(sp->np());
                len += RTCP_RR_SIZE;
                // calculate loss fraction since previous report
                int expected_interval  = sp->ehsr() - last_ehsr_;
                last_ehsr_ = sp->ehsr();
                int lost_interval = expected_interval -  received;


                if (lost_interval <= 0 || expected_interval == 0 ) {
                         fraction = 0;

                } else fraction = ((double)lost_interval  / (double)expected_interval);
                calculate_RTT();
                // Update the R_tcp according to fraction value
                if( sp->np() >= 1) {// I have already received RTP packets from the source
                        if(fraction == 0) {
                                increase_rate(sp->ps());

                        }       else /*if(fraction != 0)*/ {
                                        measure_smooth_loss(fraction);
                                        calculateR_tcp(sp->ps());
                                }
                }
                //add receiver report
                receiver_report* rr;
                rr = new receiver_report;
                //fill the report
                rr->cum_pkts_lost() = sp->cum_pkts_lost();
                rr->LSR() = sp->LSR();
                rr->DLSR()= NOW - sp->SRT();
                rr->R_tcp() = tx_rate_;
                rr->jitter() = sp->jitter();
                if(bye) {
                printf ("TIME for BYE %f \n", NOW);
                rr->bye()= 1;
                } else rr->bye() = 0;
                rh_->rr_ = rr;

                if (++nrr >= 31)
                        break;
                } // end if

        }// end for loop


        if (bye) {
                len += build_bye();
                }
        else
                len += build_sdes();

        Tcl::instance().evalf("%s adapt-timer %d %d %d", name(),
                              nsrc, nrr, we_sent);
        Tcl::instance().evalf("%s sample-size %d", name(), len);

        return (len);
}

int RTPSession::build_bye()
{
        
        return (8);
}
int RTPSession::build_sdes()
{

        /* XXX We'll get to this later... */
        return (20);
}
void RTPSession::recv(Packet* p, Handler*)
{
//      printf("time %f localsrc_ %d \n", Scheduler::instance().clock(),localsrc_->srcid());
//      printf("time %f localsrc_ %d \n", NOW,localsrc_->srcid());
        hdr_cmn* mh = hdr_cmn::access(p);
        hdr_rtp* rh = hdr_rtp::access(p);
    
        RTPSource* s = lookup(rh->srcid());

        if (s == 0) {
                if(rh->srcid()!=localsrc_->srcid()){
                        Tcl& tcl = Tcl::instance();
                        tcl.evalf("%s new-source %d", name(),rh->srcid());
                        s = (RTPSource*)TclObject::lookup(tcl.result());
                }
        }
        
        if(rh->srcid()!=localsrc_->srcid())
        {
                // calculate one-way trip time

                T_one_way_ = NOW - rh->timestamp();
                // count packet losses
                int pkts_lost = 0;
                int difference =  rh->seqno() - s->ehsr();  
                if (difference <= 1) { // to deal with the first RTP packet in which s->ehrs() = -1
                        pkts_lost = 0;
                        
                }
                
                if (difference > 1) {
                        pkts_lost = difference -1;
                        
                }
                // calculate delay jitter, see RFC 3550 Appendix A8
                double transit = NOW - rh->timestamp();
                double d = transit - s->transit();
                s->transit(transit);
                if (d < 0) d = -d;
                s->jitter( s->jitter() + (1./16.) * (d - s->jitter()));
                jitter_= s->jitter();
                s->np(1); // counts the packets that I have received from sender s
                s->cum_pkts_lost(pkts_lost);// counts the packets that I have lost from sender s
                s->ehsr(rh->seqno());// the first ehsr = -1
                s->nbytes(mh->size()); // counts the number of bytes that I have lost from sender s
                s->ps(mh->size());// the packet size in bytes
                
        }

        Packet::free(p);
}

void RTPSession::calculate_alpha(double value)
{
        if( value == 0) {
                alpha = 1;
        } else
        alpha = (value/T_one_way_) -1;
}


void RTPSession::calculate_RTT()
{
        double temp=(1+alpha) * T_one_way_;
        RTT_ = temp * 0.1 + 0.9 * RTT_;   //taken from RFC 3448
}

void RTPSession::increase_rate(int ps)
{
        tx_rate_ = tx_rate_ +  (double)ps/RTT_;

}

 void RTPSession::measure_smooth_loss (double fraction)
{
        
        double I_tot0 = 0;
        double I_tot1 = 0;
        double W_tot = 0;
        for (int i=0; i<7; i++) {
        pkt_loss_history[i+1]= pkt_loss_history[i];
        }
        pkt_loss_history[0] = fraction;

        for (int i=0; i<7; i++) {
        I_tot0 = I_tot0+ (pkt_loss_history[i] * weight[i]);
        W_tot = W_tot +  weight[i];
        }

        for (int i=1; i<8; i++) {
        I_tot1 = I_tot1 + (pkt_loss_history[i] * weight[i-1]);
        }
        double I_tot =0;
        if( I_tot0 > I_tot1) {
        I_tot = I_tot0;
        } else {
        I_tot = I_tot1;
        }
        double  I_mean = I_tot/W_tot;

        smooth_loss_= I_mean;
        
        

}           

void RTPSession::calculateR_tcp(int ps)
{

        double min =0;
        double fraction = 3 * sqrt((3 * smooth_loss_) / 8);
        if(fraction <  1 ){
                min = fraction;
        }       else min =1;
        // TCP friendly BW share equation
         tx_rate_ = (double)ps / (RTT_ *  (sqrt(2 * smooth_loss_ /3)) + 4 * (RTT_ * min * smooth_loss_) * (1 + (32 * pow(smooth_loss_,2))));
         printf("Time:%f Receiver:%d smooth_loss_:%f \n", NOW,localsrc_->srcid(),smooth_loss_);

        
}

void RTPSession::recv_ctrl(Packet* p)
{
        hdr_cmn* mh = hdr_cmn::access(p);
        hdr_rtp* rh = hdr_rtp::access(p);
        u_int32_t local_src = localsrc_->srcid();
        
        if(rh->srcid() != local_src) {
        
                if (rh->sr_ != 0) { // get the Sender report
                        double temp = 0.0;
                        RTPSource* source = lookup(rh->sr_->sender_srcid());
                        if (source != 0 && source->srcid()!= local_src) {
                                source->LSR(rh->timestamp());
                                source->SRT(Scheduler::instance().clock());
                                for (RTPReceiver* p = rh->sr_->rcvrs_;   p != 0; p = p->next) {
 
                                        if(p->srcid() == localsrc_->srcid()) {
                                                temp = p->eff_rtt();
                                                calculate_alpha(temp);
                                        }
                                }
                        }
                }

                if(localsrc_->is_sender()){ // get the Receiver report
                                 
                                if(rh->rr_!= 0){
                                        
                                        RTPReceiver* s = lookup_rcv(rh->srcid());
                                                if (s == 0 ) {
                                                        s = new RTPReceiver(rh->srcid());
                                                        enter_rcv(s);
                                                        return;
                                                        }
                                                 
                                                if (s != 0) {
                                                        for (s = receivers_; s!= 0; s = s->next) {
                
                                                                if(rh->srcid() == s->srcid()) {
                                                                double eff_rtt = 0.0;
                                                                double alpha = Scheduler::instance().clock();
                                                                        if(rh->rr_->LSR()!=0) {
                                                                        eff_rtt = alpha - rh->rr_->LSR() - rh->rr_->DLSR();
                                                                        }
                                                                        s->eff_rtt(eff_rtt);
                                                                        s->rate(rh->rr_->R_tcp());
                                                                        if(rh->rr_->bye() == 1) remove_receiver(s);
                                                                }

                                                        }// end for loop

                                                } //end if
                                                if(enableFlowControl_ == 1) {
                                                update_rate();
                                                }
                                        }
                                }
                        }

        Tcl::instance().evalf("%s sample-size %d", name(), mh->size());
        Packet::free(p);
}


void RTPSession::update_rate()
{
        RTPReceiver* s;
        for (s = receivers_; s!= 0; s = s->next) {
                if(s->rate() !=0) {
                        lst.push_front(s->rate());
                }
        }
        lst.sort();
        double inst_tx =lst.front();
        lst.clear();
        if(inst_tx != tx_rate_) {
                if(inst_tx < (tx_rate_/2)) {
                        tx_rate_ = tx_rate_/2;
                } else tx_rate_ = inst_tx;
        printf("Time:%f Sender:%d ...Rate change to:%f \n", NOW,localsrc_->srcid(),8*tx_rate_);
        Tcl::instance().evalf("%s session_bw %fb/s",name(),8*tx_rate_);
        Tcl::instance().evalf("%s transmit %fb/s",name(),8*tx_rate_);
        }
        
}

hdr_rtp* RTPSession::access_hdr_rtp()
{
        return rh_;
}

RTPSource* RTPSession::lookup(u_int32_t srcid)
{
        RTPSource *p;
        for (p = allsrcs_; p != 0; p = p->next)
        {       
                if (p->srcid() == srcid)
                        return (p);
        }
        return (0);
}

RTPReceiver* RTPSession::lookup_rcv(u_int32_t srcid)
{

        RTPReceiver *p;
        for (p = receivers_; p != 0; p = p->next)
        {
                if (p->srcid() == srcid)
                        return (p);
        }
        return (0);
}

void RTPSession::print_rcv( )
{

        RTPReceiver *p;
        for (p = receivers_; p != 0; p = p->next)
        {
            printf("Receiver:%d\n", p->srcid());
        }
}

void RTPSession::enter(RTPSource* s)
{
        if(s->srcid() != localsrc_->srcid() ) {
        printf("Time:%f Receiver:%d ...I add sender %d \n", NOW,localsrc_->srcid(), s->srcid());
        s->next = allsrcs_;
        allsrcs_ = s;
        }
}

void RTPSession::enter_rcv(RTPReceiver* s)
{
        printf("Time:%f Sender:%d ...I add receiver %d \n", NOW,localsrc_->srcid(), s->srcid());
        s->next = receivers_;
        receivers_ = s;
}

void RTPSession::remove_receiver(RTPReceiver* p)
{

        printf("Time:%f Sender:%d ...I remove receiver %d \n", NOW,localsrc_->srcid(),p->srcid());
        RTPReceiver* s;
        RTPReceiver* q;
        if (receivers_==p)
        {
                q = receivers_;
                receivers_=receivers_->next;
                delete q;
        }
        else
        {
                for (s = receivers_; s!=0; s=s->next)
                {
                        if (s->next==p)
                        {
                                q = s->next;
                                s->next=q->next;
                                delete q; 
                        }       
                }
        }

}

void RTPSession::initial_rate(double a)
{
        tx_rate_ =(double)a/8;
        
        
}

int RTPSession::command(int argc, const char*const* argv)
{
        if (argc == 3) {
                if (strcmp(argv[1], "enter") == 0) {
                        RTPSource* s = (RTPSource*)TclObject::lookup(argv[2]);
                        enter(s);
                        return (TCL_OK);
                }
                if (strcmp(argv[1], "localsrc") == 0) {
                        localsrc_ = (RTPSource*)TclObject::lookup(argv[2]);
                        enter(localsrc_);
                        return (TCL_OK);
                }
                if (strcmp(argv[1], "rate") == 0) {
                        initial_rate(atoi(argv[2]));
                        return(TCL_OK);
                }
        }

        return (TclObject::command(argc, argv));
}

RTPReceiver::RTPReceiver(u_int32_t srcid)
        : next(0),cum_pkts_lost_(0),eff_rtt_(0),rate_(0)
{
        srcid_ = srcid;
}

RTPSource::RTPSource(u_int32_t srcid)
        : next(0),np_(0), snp_(0), ehsr_(-1), nbytes_(0),cum_pkts_lost_(0),LSR_(0),SRT_(0),is_sender_(false),ps_(0),rate_(0),transit_(0),jitter_(0)
{
        bind("srcid_", (int*)&srcid_);
        srcid_ = srcid;
}

---