srr.cc

Go to the documentation of this file.

/*
 * Copyright (c) 1991-1994 Regents of the University of California.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. All advertising materials mentioning features or use of this software
 *    must display the following acknowledgement:
 *      This product includes software developed by the University of
 *      California, Berkeley and the Network Research Group at
 *      Lawrence Berkeley Laboratory.
 * 4. Neither the name of the University nor of the Laboratory may be used
 *    to endorse or promote products derived from this software without
 *    specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 */

 /*
 * SRR means Smoothed Round Robin. 
 * as compared with the DRR provided in NS, 
 * this version now can:
 * 1, assign multi flows to a queue
 * 2, assign same quantum to different queue
 * 3, use the flowid information in the packet to disdinguish 
 *    different packets.
 * 4, it has  a Weight Spread Sequence. the WSS with MAXWSSORDER is stored statically in the memory
 * 5, it has the Weight Matrix. the WM can be adjusted in the processing of SRR
 *
 * for details about SRR, see:  
 * " SRR: An O(1) Time Complexity Scheduler for Flows in Multi-Service Packet Networks", Chuanxiong Guo, sigcomm'01.
 */
// Ported by xuanc, 1/20/02

#include "config.h"   // for string.h
#include <stdlib.h>
#include "queue.h"
#include <math.h>
#include <assert.h>

#define MAXFLOW 100  // this version supports up to 1024 flows and queues.
#define MAXQUEUE 100 // if you want more, you can change it. 
#define MAXWSSORDER 16 // a 16th WSS needs pow(2, 16) space


//for debug 
#ifndef DEBUG_SRR
//#define DEBUG_SRR
#endif // DEBUG_SRR

// this struct is the basic element for the Weight Matrix
struct wm_node
{
        int queueid; // the queue the node belongs to 
        int weight;  // the weight of the node
        struct wm_node *next;
        struct wm_node *prev;
        struct wm_node *sibling;
};

#include "wss.h" 

class PacketSRR {
        PacketSRR(): pkts(0),bcount(0),deficitCounter(0),turn(0),head_(0), tail_(0), len_(0) {}
        friend class SRR;
protected: 
        int pkts;   //total packets in this Queue
        int bcount; //count of bytes in each flow to find the max flow;
        int deficitCounter; // concept borrowed from DRR 
        int turn;    // whether the queue is on service or not 
        int queueid; // mark the id of this packet queue

        Packet *head_, *tail_;
        int len_;

        inline void enque(Packet* p) {

                if (!tail_) head_= tail_= p;
                else {
                        tail_->next_= p;
                        tail_= p;
                }
                tail_->next_= 0;
                ++len_;
        }
        inline Packet* deque() {
                if (!head_) return 0;
                Packet* p = head_;
                head_= p->next_; // 0 if p == tail_
                if (p == tail_) head_= tail_= 0;
                --len_;
                return p;
        }
        Packet* lookup(int n) {
                for (Packet* p = head_; p != 0; p = p->next_) {
                        if (--n < 0)
                                return (p);
                }
                return (0);
        }
};

class SRR : public Queue {
public :
        SRR();
        virtual int command(int argc, const char*const* argv);
        Packet *deque(void);
        void enque(Packet *pkt);
        void clear();

public:
        int maxqueuenumber_ ;   //total number of flows allowed
        int blimit_;                    //total number of bytes allowed across all flows
        int bytecnt;                    //cumulative sum of bytes across all flows
        int pktcnt;                             // cumulative sum of packets across all flows
        int flwcnt;                             //total number of active flows

        int last_queueid;               // the id of the previously served queue
        int last_size;                  // the length of the last send packet
        
        PacketSRR srr[MAXFLOW]; //pointer to the entire srr struct

        int f2q[MAXFLOW];               // flow to queue mapping
        int private_rate[MAXFLOW];      // bandwidth for each queue
        int mtu_;                                       // it is used to mark the quantum adding to 
                                                                // each queue
        int granularity_;                       // the min rate that can be allocated to a queue
        int maxRate;
        int minRate; 

        class WSS wss;    // wss deals with the Weight Spread Sequence.
                                
        /*
         * wmHead[0] according to the least important weight,
         * wmHead[maxColumn] according to the most important weight
         */
        struct wm_node wmHead[MAXWSSORDER];             // each column of WM has a head
        struct wm_node wmTail[MAXWSSORDER];             // each column of WM has a tail 
        struct wm_node *pwmCurr;                                // points to the current scanning position

        struct wm_node *pRowHead[MAXFLOW];
        struct wm_node *pRowTail[MAXFLOW];


        int wmEmptyFlag;                                                // 0: WM is not empty, 1:WM is empty

        int maxColumn;          // the max column number of the WM, so there are total 
                                                // maxColumn+1 columns, the column number is numbered from 0 to 
                                                // maxColumn from left to right.
        int currMaxColumn;      // the current max number of the column
        
        int min_quantum;

        inline PacketSRR *getMaxflow () { //returns flow with max bytes 
                int j,i;
                PacketSRR *tmp=0;
                PacketSRR *maxflow=0;
                

                int bcount=0;

                for (i=0;i<maxqueuenumber_;i++) {
                        tmp=&srr[i];
                        if (tmp->bcount > bcount){
                                        bcount= tmp->bcount;
                                        maxflow=tmp;j=i;
                        }
                }
                if(maxflow==0){
                        fprintf(stderr, "getMaxflow, err");
                        exit(1);
                }
                return maxflow;
        }
  
        //returns queuelength in packets
        inline int length () {
                return pktcnt;
        }

        //returns queuelength in bytes
        inline int blength () {
                return bytecnt;
        }

        int add_to_WM(int queueid, int weight);
        int del_from_WM(int queueid, int weight);
        struct wm_node * getNextNode( ); 
};

static class SRRClass : public TclClass {
public:
        SRRClass() : TclClass("Queue/SRR") {}
        TclObject* create(int, const char*const*) {
                return (new SRR);
        }
} class_srr;

SRR::SRR()
{       int i;

        maxqueuenumber_ = 10;
        blimit_ = 25000;
        mtu_= 1000;             //1000 bytes quantum at default setting
        granularity_ = 1000;    //default to 1K bit/s  

        last_queueid = -1; // -1 means that SRR does not have a previous deque operation        
        last_size = 0;

        flwcnt = 0;  // init 
        bytecnt = 0;
        pktcnt = 0;
        min_quantum = 1000;

        pwmCurr = 0; // at first, pwmCurr points to NULL

        for(i=0; i<MAXFLOW; i++)
        {
                private_rate[i] = granularity_; //default quantum value for each flow
                f2q[i]=0;                // default queue id for all the flow
                                         // or it will not works right at the default config
        }

        maxRate = 100000000; //100Mbps
        minRate = 1000; //1kbps 

        // init the WM double queues here too.
        for(i=0;i<MAXWSSORDER;i++){
                wmHead[i].prev=NULL;
                wmHead[i].next=&wmTail[i];
                wmTail[i].prev=&wmHead[i];
                wmTail[i].next=NULL;
                wmHead[i].queueid=wmTail[i].queueid=-1; // 
                wmHead[i].weight=wmHead[i].weight=i;
        }

        for (i=0;i<MAXFLOW; i++)        
                pRowHead[i]= pRowTail[i] = NULL;
        
        wmEmptyFlag=1; // it is empty at first
        wss.init(MAXWSSORDER); // create the Weight Spread Sequence

        currMaxColumn = -1;
        maxColumn=0;
        
        // allow the TCL scripts to change the following values
        bind("maxqueuenumber_",&maxqueuenumber_);  //it is the max queuenumber set in the TCL script
        bind("mtu_", &mtu_);      // set the Max Transfer Unit of the output link
        bind("granularity_", &granularity_); // set the rate allocation granularity of the 
                                // all the flows. 
        bind("blimit_",&blimit_);
}


// enque and deque are two interface functions that provided by 
// NS 
void SRR::enque(Packet* pkt)
{
        PacketSRR *q;
        //PacketSRR *remq;
        int flowid, queueid;
        int weight;

        hdr_cmn *ch=hdr_cmn::access(pkt);
        hdr_ip *iph = hdr_ip::access(pkt);

        flowid= iph->flowid();          //get the flowid
        queueid= f2q[flowid];   // get the corresponding queue id

        if(queueid > maxqueuenumber_)
        {
                fprintf(stderr, "queueid too big\n");
                exit(1);
        }

#ifdef DEBUG_SRR
        printf("   in enque\n"); fflush(0);
#endif
        

// we drop packets from the longest queue
/*      if( bytecnt+ ch->size() > blimit_){

                drop(pkt);
                return;
        }
*/
        
        q=&srr[queueid];

        if(q->pkts==qlim_){

                drop(pkt);
                return;
        }

        q->enque(pkt);

        ++q->pkts;
        ++pktcnt;
        q->bcount += ch->size();
        bytecnt +=ch->size();


        if (q->pkts==1)
        {
                        //add to the WM
                        // adjust currMaxColumn
                        // if it is a first active flow, put the pwmPtr
                        // all the above works are done in add_to_WM
                        weight= private_rate[queueid];
                        weight/=granularity_;
                        
                        add_to_WM(queueid, weight);
                        
                        q->deficitCounter=0;
        }

}

Packet *SRR::deque(void) 
{
        hdr_cmn *ch;
        hdr_ip *iph;
        Packet *pkt=0;
        int flowid;
        int queueid;
        PacketSRR *pP;
//      static int dcnt = 0;

//      if(dcnt <20){   
//      printf(" in dequeue: %lf \n",  Scheduler::instance().clock() ); 
//      dcnt++;
//      }
#ifdef DEBUG_SRR
printf("  in deque\n");
#endif

        if(last_queueid>=0){
//printf("last size=%d\n ", last_size);
                        srr[last_queueid].bcount -= last_size;
                        srr[last_queueid].pkts-=1;
                        --pktcnt;
                        bytecnt -= last_size;

                        if (srr[last_queueid].pkts == 0) {
                                srr[last_queueid].turn=0;
                                srr[last_queueid].deficitCounter=0;

                                // delete the queue from SRR
                                del_from_WM(last_queueid, private_rate[last_queueid]/granularity_ );
                        }

        }       

assert(pktcnt>=0);
        if (pktcnt==0) {
        //      fprintf (stderr,"No active flow\n");
                last_queueid=-1;
                return(0);
        }

        if(pwmCurr == NULL){
                printf("wrong, pwmCurr is NULL\n");
                exit(0);
        }

//printf("pktcnt=%d\n", pktcnt);

        while (!pkt) {

                if(pwmCurr->queueid==-1){ // it should never happen

                        fprintf(stderr,"pwmCurr points to head or tail\n"); 
                        fprintf(stderr, "weight:%d", pwmCurr->weight);
                        exit(0);
                }
                pP=&srr[pwmCurr->queueid]; 

                pkt=pP->lookup(0);  

                if(pkt==0){
                        fprintf(stderr, "wrong place, should never be here\n");
                        exit(2);
        
                }
        
                iph = hdr_ip::access(pkt);

                flowid=iph->flowid();
                queueid= f2q[flowid]; // get the corresponding queue id

assert(queueid== pwmCurr->queueid);
        
                if (!pP->turn) {
                        pP->deficitCounter+= mtu_; // each queue shares the same quantum!
                                        // this is the difference between DRR!
                        pP->turn=1;
                }

                ch=hdr_cmn::access(pkt);

                if (pP->deficitCounter >= ch->size()) {
                        pP->deficitCounter -= (ch->size());
                        pkt=pP->deque();
                        last_size=ch->size( );
                        
//                      pP->bcount -= ch->size();
//                      --pP->pkts;
//                      --pktcnt;
//                      bytecnt -= ch->size();
//                      if (pP->pkts == 0) {
//                              pP->turn=0;
//                              pP->deficitCounter=0;

                                // delete the queue from SRR
//                              del_from_WM(queueid, private_rate[queueid]/granularity_ );
                                //getNextNode();
//                      }
                
#ifdef DEBUG_SRR
printf("deque a packet, id=%d, size=%d\n", queueid, last_size);
#endif  
                        last_queueid=queueid;

                        return pkt;
                }
                else {
                        pP->turn=0; 

                        // pwmCurr should be adjusted.
                        getNextNode( );
                        pkt=0;
                }
        }
        return 0;    // not reached
}

void SRR::clear()
{
        PacketSRR *q =srr;
        int i = maxqueuenumber_;

        if (!q)
                return;
        while (i--) {
                if (q->pkts) {
                        fprintf(stderr, "Changing non-empty flow from srr\n");
                        exit(1);
                }
                ++q;
        }
}


// weight is the queues rate/granularity.
int SRR::add_to_WM(int queueid, int weight)
{
        struct wm_node *pNode;
        int i;
        //int j=maxColumn;
        //int temp=weight;
        //int flag=0;

        int old_colno = currMaxColumn;

        if(weight==0)
        {
                fprintf(stderr, "add_to_WM: weight should not be zero");
                exit(1);
        }

        if(weight > ( (1<<(maxColumn+1))-1) )
        {
                fprintf(stderr, "add_to_WM: weight too big");
                exit(1);
        } 

        // add to the WM
        // adjust currMaxColumn
        // if it is a first active flow, put the pwmPtr

        
        for(i=maxColumn; i>=0; i--)
        {

                if (weight & (1<<i) )
                {  
                        // 
                        // add to queueid= i; wmHead[queueid], wmTail[queueid]
                        pNode=(struct wm_node*)malloc(sizeof(struct wm_node));

                        if(pNode==NULL)
                        {
                                fprintf(stderr, "no memeory to create WM node");
                                exit(2);
                        }


                        pNode->queueid = queueid;
                        pNode->weight = i;
                        pNode->sibling = NULL;  
                        
                        if(pRowTail[queueid] == NULL){
                                pRowHead[queueid]= pRowTail[queueid] = pNode;
                        }else{
                                pRowTail[queueid]->sibling = pNode;
                                pRowTail[queueid] = pNode;
                        }

                        if( pwmCurr && (pwmCurr->weight == i) ){
                                pNode-> prev =  pwmCurr->prev;
                                pNode-> next =  pwmCurr;
                                (pwmCurr->prev)->next = pNode;
                                pwmCurr->prev = pNode;

                        }else {

                                pNode->prev = wmTail[i].prev;
                                pNode->next = &wmTail[i];
                                (wmTail[i].prev)->next = pNode;
                                wmTail[i].prev = pNode;
                        }

                        if(currMaxColumn < i)
                                currMaxColumn = i; // adjust the current max column number
                        
                        if(wmEmptyFlag == 1)
                        {
                                wmEmptyFlag=0;
                                if(pwmCurr == NULL) // we should let it points to the correct place.
                                        pwmCurr=pNode;
                        }

                }
        }
        
        if (  old_colno < currMaxColumn )
        {
                if(old_colno >= 0){
        //      if(old_colno > 0){
                        int pc = wss.get_ptr () + 1;
                        pc = pc << (currMaxColumn - old_colno);
                        wss.set_ptr ( pc -1);
                //      printf("set_ptr in add_to_wm: ptr:%d\n", pc-1);
                //      printf("old column no: %d %d\n", old_colno, currMaxColumn);
                }
        }

        //printf("in add_to_wm: k:%d, j:%d\n", currMaxColumn, old_colno);       
        ++flwcnt;
        return 0;
}


//remove the wm_node from the links and free the memory
int SRR::del_from_WM(int queueid, int weight)
{
        struct wm_node *pNode, *pNode2;
        int i;
        int wss_term; 
        int temp; 


        if(pwmCurr->queueid==queueid)  // we adjust pwmCurr before we delete the row
        { 
                if (pwmCurr->next != &wmTail[pwmCurr->weight])
                        pwmCurr = pwmCurr->next ;
                else pwmCurr = NULL;
        }

        /*
        for(i=0;i<=currMaxColumn;i++) // travel all double links, and delete the node whose id is queueid
        { 
                pNode=wmHead[i].next;

                while(pNode!=&wmTail[i]){
                        if(pNode->queueid==queueid){ //yes, we get one.
                                //remove it from the link, and free the node.
                                (pNode->prev)->next=pNode->next;
                                (pNode->next)->prev=pNode->prev;
                                free(pNode);

                                break;
                        }
                        else
                                pNode=pNode->next;

                }
                
        } */

        pNode = pRowHead[queueid];
        while (pNode){
                (pNode->prev)->next=pNode->next;
                (pNode->next)->prev=pNode->prev;
                
                pNode2= pNode;
                pNode = pNode->sibling;
                free(pNode2);
        }

        pRowHead[queueid] = pRowTail[queueid] = NULL;

// we should adjust currMaxColumn, 
//and if currMaxColumn becomes less, we should also adjust WSS's pointer
        
        int old_colno = currMaxColumn;

        for(i=currMaxColumn;i>=0;i--)
        {
                if(wmHead[i].next != &wmTail[i])
                {
                  currMaxColumn=i;
                  break;
                }
        }

        if(i<0)
        {
                // it is empty now.
#ifdef DEBUG_SRR
//printf("WM empty\n");
#endif
                wmEmptyFlag=1;
                currMaxColumn=-1;
                pwmCurr=NULL;
                last_queueid=-1;
                wss.set_ptr (0); // reset the WSS sequence pointer to 0;
                goto rr;
        } else if ( currMaxColumn < old_colno)
        {
                int pc = wss.get_ptr () +1;
                //printf("pc = %d \n", pc);
                int mul = 1 << (old_colno - currMaxColumn ); 
                int tmpc = pc / mul;
                if (pc % mul){
                        tmpc += 1;
                        if (tmpc > ((1<<(currMaxColumn+1) )- 1) )
                                tmpc = 1;

                }
                wss.set_ptr (tmpc -1);
                //printf("set_ptr in del_from__wm: ptr:%d\n", tmpc-1);
                //printf("k:%d j:%d\n", currMaxColumn, old_colno);

        }
        if (pwmCurr == NULL){
loop:
                wss_term=wss.get(currMaxColumn+1);
                wss.inc_ptr (currMaxColumn+1 );
                temp = currMaxColumn+1-wss_term;
                 
                // try to point pwmCurr to the right place.
                if( wmHead[temp].next != &wmTail[temp]) 
                {
                        pwmCurr=wmHead[temp].next;      

                }else
                        goto loop;
        }

rr:
        --flwcnt;
        return 0;
}


/* should be checked:
 * whether it returns nothing when the WM is empty
 */
struct wm_node *SRR::getNextNode(){
  //struct wm_node *pNode;
  int queueid;
  int weight;
  int wss_term;
  int temp;


        if(bytecnt==0){
                //      printf("getNextNode, pwmCurr = NULL, wmEmptyFlag=%d\n", wmEmptyFlag);
                return NULL;
        }

        queueid=pwmCurr->queueid;
        weight= pwmCurr->weight;

        if(pwmCurr->next != &wmTail[weight]){ // not the tail 
                pwmCurr=pwmCurr ->next;
        } else{
                if (currMaxColumn==-1){
                        fprintf(stderr, "getNextNode, empty WM");
                        exit(0);
                }

again:
                wss_term=wss.get(currMaxColumn+1);
                wss.inc_ptr (currMaxColumn+1 );

                temp = currMaxColumn+1-wss_term;
                
                // wss_term according to currMaxColumn+1-wss_term queue
                if( wmHead[temp].next != &wmTail[temp]) {
                        pwmCurr=wmHead[temp].next;      

                }else
                        goto again;


        }

        return pwmCurr;
}


int getOrder(int i){
        int order=0;

        while( (1 << order)<= i)
                order++;
        return (order-1);       
}


/*
 *Allows one to change 
 *      blimit_  
 *      maxqueuenumber_ 
 *      mtu_
 *      granularity_
 *      for a particular srrQ :
 *
 *
 */
int SRR::command(int argc, const char*const* argv)
{


        if (argc==3) {

                if (strcmp(argv[1], "blimit") == 0) {
                        blimit_ = atoi(argv[2]);
                        if (bytecnt > blimit_)
                                {
                                        fprintf (stderr,"More packets in buffer than the new limit");
                                        exit (1);
                                }
                        return (TCL_OK);
                }
                if (strcmp(argv[1], "maxqueuenumber") == 0) {
                        //clear();
                        maxqueuenumber_ = atoi(argv[2]);
                        return (TCL_OK);
                }
                if (strcmp(argv[1],"mtu")==0) {
                        mtu_= atoi(argv[2]);
                        return (TCL_OK);
                }
                if (strcmp(argv[1], "granularity")==0) {
                        granularity_ = atoi(argv[2]);
                        return (TCL_OK);
                }
        }

        if (argc == 4) {
                if(!strcmp(argv[1],"setrate"))  {
                        int rate;
                        int queue,success=0;
                        int temp;

                        success += sscanf(argv[2],"%d",&queue);
                        success += sscanf(argv[3],"%d",&rate);

                        if(success!=2){
                                fprintf(stderr, "SRR setrate ??"); exit(0);
                                exit(1);
                        }
                
                        if ( queue>MAXQUEUE ) {
                                        fprintf(stderr,"queue id out of range"); exit(0);
                        }
                        min_quantum= min_quantum<rate ? min_quantum:rate;
                        private_rate[queue]=rate;


                        if(private_rate[queue]<minRate)
                                private_rate[queue]=minRate;
                        if(private_rate[queue]>maxRate){
                                fprintf(stderr, "Rate too hight!\n");
                                exit(1);
                        }
                                
                                
                        success=private_rate[queue]/granularity_;
                        if(success==0) success=1;
                        /*if(private_rate[queue]%granularity_)
                                success++;*/
                        temp=getOrder(success);  //now we have the order of the WSS
                        if(maxColumn<temp){
                                maxColumn=temp;

                        }
        
#ifdef DEBUG_SRR
        printf("maxColumn=%d\n", maxColumn), fflush(0);
#endif
                        if(maxColumn>(MAXWSSORDER-1)){ //the order of max band flow is too big!
                                fprintf(stderr, "granularity too small or band too big!");      
                                exit(2);
                        }
        
                        return (TCL_OK);
                }
        
        else if(!strcmp(argv[1],"setqueue")) {
                        int queue,flow,success=0;
                        success += sscanf(argv[2],"%d",&flow);
                        success += sscanf(argv[3],"%d",&queue);
                        if(success==2) {
                                if ( !(queue<MAXQUEUE) ) {
                                        fprintf(stderr,"queue id out of range");exit(1);
                                }
                                if ( !(flow<MAXFLOW) ) {
                                        fprintf(stderr,"flow id out of range"); exit(1);
                                }

                                f2q[flow]=queue;

                                return TCL_OK;
                        }
                }
        }
        return (Queue::command(argc, argv));
}

---