dsscheduler.cc.svn-base

diff serve for ns-2, work for version 2.29

{ for (int i=0; i<MAX_QUEUES; i++) fs_[i].finish_t=0;
  idle=1;
}

void dsWFQ::EnqueEvent(Packet* pkt, int queue)
{
    double now=Scheduler::instance().clock();

    // virtual time update. Formula 10
	if(idle) {
		v_time=0;
		last_vt_update=now;
		idle=0;
	} else {
		v_time=v_time+(now-last_vt_update)/sum;
		last_vt_update=now;
	}

    // finish time computation. Formula 11 
    fs_[queue].finish_t = (max(fs_[queue].finish_t,v_time))+
                         ((double)hdr_cmn::access(pkt)->size())*8/(fs_[queue].weight_*LinkBandwith); 


    // update sum and B
    if( (fs_[queue].B++)==0 ) sum+=fs_[queue].weight_; 
    // insertion in both lists
    fs_[queue].GPS.push(fs_[queue].finish_t);
    fs_[queue].PGPS.push(sessionDelay+fs_[queue].finish_t);
	
    // schedule next departure in the GPS reference system
    if (wfq_event!=0) {
		Scheduler::instance().cancel(wfq_event);
		delete wfq_event;
    }      
    scheduleWFQ();   
}

void dsWFQ::handle(Event *e) 
{
	double now = Scheduler::instance().clock();
        //double fTime=fs_[GPSDeparture].GPS.front();
      
	//update virtual time
	v_time=v_time+(now-last_vt_update)/sum;
	last_vt_update=now;

	//extract packet in GPS system
	
   	if (GPSDeparture!=-1) {		        
		fs_[GPSDeparture].GPS.pop();
 	        if (--fs_[GPSDeparture].B == 0) sum-=fs_[GPSDeparture].weight_; 
		if ( fabs(sum) < 0.00001 ) sum=0; 
		if (sum==0) {Reset();            
			     sessionDelay=now;	
	  		     }  	
        } else printf("DEBUG: ERROR, no active sessions \n");
    
	// if GPS is not idle, schedule next GPS departure
	delete e;
	if(!idle) 
		scheduleWFQ();
	else 
		wfq_event=NULL;
}

void dsWFQ::scheduleWFQ() 
{
	wfq_event=new Event(); 
        double tmp;
	GPSDeparture=-1; 

  	double minFinishTime=MAXDOUBLE;
  	for (int i=0; i<NumQueues; i++)  
   	 	if (!fs_[i].GPS.empty()) {
			if  (fs_[i].GPS.front()<minFinishTime) {
			                GPSDeparture=i; 
			                minFinishTime=fs_[GPSDeparture].GPS.front();
                        }			
                }
              
        if (GPSDeparture!=-1) { 
		tmp=(minFinishTime-v_time)*sum;
		// following line is there to recover errors due to finite precision	
		if (tmp<0) tmp=0; 
		Scheduler::instance().schedule((Handler *)this,wfq_event,tmp);
	} else printf("DEDUG: ERROR no event to schedule \n");	
}

int dsWFQ::DequeEvent()
{ 
  int qToDq=-1; 
  double minFinishTime=MAXDOUBLE;
  for (int i=0; i<NumQueues; i++) 
	if (!fs_[i].PGPS.empty()) 
	     if (fs_[i].PGPS.front()<minFinishTime) {qToDq=i; minFinishTime=fs_[qToDq].PGPS.front();}
	  
  
  if (qToDq!=-1) fs_[qToDq].PGPS.pop(); 
  return qToDq;	
}

dsSCFQ::dsSCFQ(int NQ, double LBw):dsScheduler(NQ, LBw) 
{	
	NumQueues=NQ;	
	LinkBandwith=LBw;
	for (int i=0; i<MAX_QUEUES; i++) {
		session[i].weight=1;
	}
	Reset();
}

void dsSCFQ::Reset()
{ for (int i=0; i<MAX_QUEUES; i++) {
		 session[i].label=0;
   }
	tlabel=0;
}

void dsSCFQ::EnqueEvent(Packet* pkt, int queue)
{   session[queue].label = (max(session[queue].label, tlabel))+
                           ((double)hdr_cmn::access(pkt)->size())/session[queue].weight/(LinkBandwith/8.0);
    session[queue].SessionQueue.push(session[queue].label); 
}

int dsSCFQ::DequeEvent()
{ 
  int qToDq=-1; 
  double minFinishTime=MAXDOUBLE;
  for (int i=0; i<NumQueues; i++) 
	if ((!session[i].SessionQueue.empty())&&(session[i].SessionQueue.front()<minFinishTime)) {
		qToDq=i; 
		minFinishTime=session[qToDq].SessionQueue.front();
	}     
  if (qToDq!=-1) { tlabel=minFinishTime; 
	           session[qToDq].SessionQueue.pop();}
  return qToDq;
}


dsSFQ::dsSFQ(int NQ, double LBw):dsScheduler(NQ, LBw) 
{	
	NumQueues=NQ;	
	LinkBandwith=LBw; 
	for (int i=0; i<MAX_QUEUES; i++) {
		flow[i].weight=1;
	}
	idle=1; 
        MaxFinishTag=0.0;
	Reset();
}

void dsSFQ::Reset()
{   	for (int i=0; i<MAX_QUEUES; i++) {
		 flow[i].LastFinishTag=0;
   	}
	V=0;
}

void dsSFQ::EnqueEvent(Packet* pkt, int queue)
{   PacketTags.StartTag=max(V,flow[queue].LastFinishTag);
    PacketTags.FinishTag=PacketTags.StartTag+((double)hdr_cmn::access(pkt)->size())/flow[queue].weight/(LinkBandwith/8.0);
    flow[queue].LastFinishTag=PacketTags.FinishTag;
    flow[queue].FlowQueue.push(PacketTags); 
    idle=0;	
}

int dsSFQ::DequeEvent()
{ int qToDq=-1; 
  double MinStartTag=MAXDOUBLE;
  
  for (int i=0; i<NumQueues; i++) {   
	PacketTags=flow[i].FlowQueue.front();
	if ((!flow[i].FlowQueue.empty())&&(PacketTags.StartTag<MinStartTag)) {
		qToDq=i; 
		MinStartTag=PacketTags.StartTag;
	}     
  }

  if (qToDq!=-1) { V=MinStartTag; 
	           MaxFinishTag=max(MaxFinishTag,PacketTags.FinishTag);
	           flow[qToDq].FlowQueue.pop();
		 }
  else if (idle==0) { MaxFinishTag=0;
		      Reset(); 	
	 	      idle=1;
                  }
  return qToDq;
}

dsWF2Qp::dsWF2Qp(int NQ, double LBw):dsScheduler(NQ, LBw) 
{	
     	NumQueues=NQ;	
	LinkBandwith=LBw; 
	/* initialize flow's structure */
  	for (int i = 0; i < MAX_QUEUES; ++i) {
    		flow[i].qcrtSize        = 0;
    		flow[i].weight          = 1;
    		flow[i].S               = 0;
    		flow[i].F               = 0;
  	}
  	V=0; lastTimeV=0;
	//Reset();
}	

void dsWF2Qp::Reset()
{ 
}

void dsWF2Qp::EnqueEvent(Packet* pkt, int flowId)
{
 
  int pktSize=hdr_cmn::access(pkt)->size();
  if (!flow[flowId].qcrtSize) {
      /* If flow queue is empty, calculate start and finish times 
       * paragraph 5.2 formula (23b) and (24)  			  
       */
      flow[flowId].S = max(V, flow[flowId].F);
      flow[flowId].F = flow[flowId].S+pktSize/(flow[flowId].weight*LinkBandwith/8);

      /* update system virtual clock  
       * paragraph 5.2 formula (22)
       */
      double minS = flow[flowId].S;
      for (int i = 0; i < NumQueues; ++i) 
	if ((flow[i].qcrtSize)&&(flow[i].S<minS)) minS=flow[i].S;
      V=max(minS,V);
    }
    flow[flowId].flowQueue.push(pktSize);
    flow[flowId].qcrtSize+=pktSize;
}

int dsWF2Qp::DequeEvent()
{  
  int     i;
  int     pktSize;
  double  minF   = MAXDOUBLE;
  int     flowId = -1;
  double  W = 0;

  
  /* look for the candidate flow with the earliest finish time */
  for (i = 0; i<NumQueues; i++){
    if (flow[i].qcrtSize) { 
                W+=flow[i].weight;
		if ((flow[i].S<=V)&&(flow[i].F<minF)) {
			flowId = i;
			minF   = flow[i].F;
    		}
    }
  }

  if (flowId!=-1) {
	pktSize=flow[flowId].flowQueue.front();
  	flow[flowId].qcrtSize-=pktSize;
 	flow[flowId].flowQueue.pop();	
  
  	/* Set start and finish times of the remaining packets in the queue */
  	if (!flow[flowId].flowQueue.empty()) {
    		flow[flowId].S = flow[flowId].F;
    		flow[flowId].F = flow[flowId].S + flow[flowId].flowQueue.front()/(flow[flowId].weight*LinkBandwith/8);
  	}

  	/* update the virtual clock */
  	/* looking for min service time of eligibles (=active) flows */
	double now=Scheduler::instance().clock();
  	double minS=MAXDOUBLE;
  	for (i = 0; i < NumQueues; ++i) 
    		if ((flow[i].qcrtSize)&&(flow[i].S<minS)) minS=flow[i].S;
	if (minS==MAXDOUBLE) minS=0;

  	/* provided service in the last period, this packet sent */

	W = (now-lastTimeV)/W;
  	V = max(minS,(V+W));
	lastTimeV=now;
  }
  return(flowId);
}

dsLLQ::dsLLQ(int NQ, double LBw, const char* PFQSchedType):dsScheduler(NQ, LBw, PFQSchedType) 
{	
	NumQueues=NQ;
	PFQAssignedBandwith=LBw; 
	if (strcmp(PFQSchedType, "WFQ")==0)
		xPFQ = new dsWFQ(NQ-1,PFQAssignedBandwith);
	else if (strcmp(PFQSchedType, "WF2Qp")==0)
		xPFQ = new dsWF2Qp(NQ-1,PFQAssignedBandwith);
	else if (strcmp(PFQSchedType, "SCFQ")==0)
		xPFQ = new dsSCFQ(NQ-1,PFQAssignedBandwith);
	else if (strcmp(PFQSchedType, "SFQ")==0)
		xPFQ = new dsSFQ(NQ-1,PFQAssignedBandwith);
	xPQ = new dsPQ(1,1);
//	Reset();
}

void dsLLQ::Reset()
{ 
}

void dsLLQ::EnqueEvent(Packet* pkt, int queue)
{   if (queue==0) xPQ->EnqueEvent(pkt, queue);
    else xPFQ->EnqueEvent(pkt, queue-1);	
}

int dsLLQ::DequeEvent()
{ int qToDq=xPQ->DequeEvent();
  if (qToDq==-1) { qToDq=xPFQ->DequeEvent();	
		   if (qToDq>-1) qToDq++;
	         }
  return qToDq;
}