integrated_agent.cc

obs网络试验平台

    hdr_ip* hdrip = hdr_ip::access(pkt);
    hdr_IPKT* hdr = hdr_IPKT::access( pkt );
    hdr_cmn* ch = hdr_cmn::access(pkt);

    hdr_cmn* tcpch;
    int npkts = hdr->npkts();

    Packet **p = (Packet**)pkt->accessdata();

    while ( npkts--) {
        tcpch = hdr_cmn::access(*p);

        if (tcpch->ptype() == PT_TCP) //GMG -- added TCPRCV, ACKRCV, and UDPRCV
                                      // statistics collection
        {
           StatCollector &sc = StatCollector::instance();
           sc.updateEntry( "TCPRCV", sc.getValue( "TCPRCV" ) + 1.0 );
           sc.updateEntry( "TCPBYTESRCV", sc.getValue( "TCPBYTESRCV" ) + tcpch->size() );
           sc.updateEntry( "TCPDLY", sc.getValue( "TCPDLY" ) + Scheduler::instance().clock() - tcpch->timestamp() );

        }
        else if (tcpch->ptype() == PT_UDP)
        {
           StatCollector &sc = StatCollector::instance();
           sc.updateEntry( "UDPRCV", sc.getValue( "UDPRCV" ) + 1.0 );
           sc.updateEntry( "UDPBYTESRCV", sc.getValue( "UDPBYTESRCV" ) + tcpch->size() );
           sc.updateEntry( "UDPDLY", sc.getValue( "UDPDLY" ) + Scheduler::instance().clock() - tcpch->timestamp() );
        }
        else if (tcpch->ptype() == PT_ACK)
        {
           StatCollector &sc = StatCollector::instance();
           sc.updateEntry( "ACKRCV", sc.getValue( "ACKRCV" ) + 1.0 );
        }
        else if (tcpch->ptype() == PT_CBR)
        {
           StatCollector &sc = StatCollector::instance();
           sc.updateEntry( "CBRRCV", sc.getValue( "CBRRCV" ) + 1.0 );
           sc.updateEntry( "CBRBYTESRCV", sc.getValue( "CBRBYTESRCV" ) + tcpch->size() );
           sc.updateEntry( "CBRDLY", sc.getValue( "CBRDLY" ) + Scheduler::instance().clock() - tcpch->timestamp() );

        }
        else if (tcpch->ptype() == PT_EXP)
        {
           StatCollector &sc = StatCollector::instance();
           sc.updateEntry( "EXPRCV", sc.getValue( "EXPRCV" ) + 1.0 );
           sc.updateEntry( "EXPBYTESRCV", sc.getValue( "EXPBYTESRCV" ) + tcpch->size() );
           sc.updateEntry( "EXPDLY", sc.getValue( "EXPDLY" ) + Scheduler::instance().clock() - tcpch->timestamp() );
        }
        else if (tcpch->ptype() == PT_PARETO)
        {
           StatCollector &sc = StatCollector::instance();
           sc.updateEntry( "PARRCV", sc.getValue( "PARRCV" ) + 1.0 );
           sc.updateEntry( "PARBYTESRCV", sc.getValue( "PARBYTESRCV" ) + tcpch->size() );
           sc.updateEntry( "PARDLY", sc.getValue( "PARDLY" ) + Scheduler::instance().clock() - tcpch->timestamp() );
        }
        else if (tcpch->ptype() == PT_SELFSIM)
        {
           StatCollector &sc = StatCollector::instance();
           sc.updateEntry( "SSIMRCV", sc.getValue( "SSIMRCV" ) + 1.0 );
           sc.updateEntry( "SSIMBYTESRCV", sc.getValue( "SSIMBYTESRCV" ) + tcpch->size() );
           sc.updateEntry( "SSIMDLY", sc.getValue( "SSIMDLY" ) + Scheduler::instance().clock() - tcpch->timestamp() );
        }

        send(*p,0);
        p++;
    }
    Packet::free(pkt);
}

void IPKTAgent::sendBurst( Packet** pBurst, int bid, int npkts,
    int burstsize, int destnode, double offsettime, int pcntguard,
    double delta, int priority )
{

    char s[100];

    // This block sends a burst header packet BHP //
    Packet* pkt_bhp = allocpkt();
    hdr_ip* hdr_ip_bhp = hdr_ip::access(pkt_bhp);
    hdr_IPKT *hdr_IPKT_bhp = hdr_IPKT::access(pkt_bhp);
    hdr_cmn* cmn_bhp = hdr_cmn::access(pkt_bhp);

    hdr_ip_bhp->daddr() = destnode;
    hdr_ip_bhp->prio_ = 1;  // BHP
    hdr_ip_bhp->fid_ = 1;  // BHP; GMG added --  put into IP header flow ID
       // for use with trace file (trace file prints flow id but not priority;
       // need to have indication in trace file of whether IPKT is BHP or DB)

    // set the values into the individual bhp elements
    hdr_IPKT_bhp->C_burst_id_   = bid;
    hdr_IPKT_bhp->C_burst_size_ = size_ + burstsize;
    hdr_IPKT_bhp->offset_time_   = offsettime;
    hdr_IPKT_bhp->prio_ = priority;
    hdr_IPKT_bhp->delta_ = delta;
    hdr_IPKT_bhp->pcntguard_ = pcntguard;
    hdr_IPKT_bhp->end_end_delay_ = Scheduler::instance().clock();
    //GMG -- initialize the fdl_count_ field to zero
    hdr_IPKT_bhp->fdl_count_ = 0;
    //GMG -- update the IPKT sequence number, and put into the BHP
    hdr_IPKT_bhp->seqno = seqno_;
    send(pkt_bhp,0);   // send the BHP immediately
    StatCollector &sc = StatCollector::instance();
    sc.updateEntry( "BHPSND", sc.getValue( "BHPSND" ) + 1.0 );

    //char s[100];
    // This is to send the burst pkt //
    Packet* pkt_burst   = allocpkt();
    hdr_ip* hdr_ip_burst = hdr_ip::access(pkt_burst);
    hdr_IPKT* hdr_IPKT_burst = hdr_IPKT::access(pkt_burst);

    hdr_cmn* cmn_burst = hdr_cmn::access(pkt_burst);
    
    //copy the payload
    pkt_burst->allocdata(npkts*sizeof(Packet*));
    // memcpy could be heavy
    memcpy(pkt_burst->accessdata(),pBurst,npkts*sizeof(Packet*));
  
    hdr_ip_burst->daddr() = destnode;
    hdr_ip_burst->prio_ = 2; // Prio=2 indicated burst pkt sent
    hdr_ip_burst->fid_ = 2;  // BHP; GMG added --  put into IP header flow ID
       // for use with trace file (trace file prints flow id but not priority;
       // need to have indication in trace file of whether IPKT is BHP or DB)

    hdr_IPKT_burst->C_burst_id_   = bid;
    hdr_IPKT_burst->C_burst_size_ = burstsize;
    hdr_IPKT_burst->offset_time_   = offsettime;
    hdr_IPKT_burst->prio_   = priority;
    hdr_IPKT_burst->end_end_delay_ = Scheduler::instance().clock();

    hdr_IPKT_burst->npkts() = npkts;

    cmn_burst->size() +=  burstsize;
    sprintf( s, "Sending a DB to node : %d", destnode );
    // Debug::debug ( s );

    //GMG -- initialize the fdl_count_ field to zero (note that
    //       this field is not used in DB, but we set it to zero)
    hdr_IPKT_burst->fdl_count_ = 0;
    //GMG -- put the IPKT sequence number into the DB; note that the DB and
    //corresponding BHP have the same sequence number
    hdr_IPKT_burst->seqno = (seqno_++);
    send(pkt_burst,0);
    sc = StatCollector::instance();
    sc.updateEntry( "BURSTSND", sc.getValue( "BURSTSND" ) + 1.0 );
}

int IPKTAgent::command(int argc, const char*const* argv) {
int j;  // GMG -- changed the invocation of offsettime() to be for each QoS class
    if( argc == 2 ) {
        /*
         * $iagent dumpburstdefaults
         */
        if( strcmp( argv[1], "dumpburstdefaults" ) == 0 ) {
            char s[100];
            sprintf( s, "maxburstsize: %d", BurstManager::maxburstsize() );
            Debug::debug( s );
            sprintf( s, "pcntguard: %d", BurstManager::pcntguard() );
            Debug::debug( s );
            sprintf( s, "burst-time-out: %lf", BurstManager::burst_timeout() );
            Debug::debug( s );
            for (j = 0; j < nqos_classes; j++)
            {
               sprintf( s, "QoS Class: %d     offset-time: %lf", j, BurstManager::offsettime(j) );
               Debug::debug( s );
            }
            sprintf( s, "delta: %lf", BurstManager::delta() );
            Debug::debug( s );
            return (TCL_OK);
        }
    }
    else if ( argc == 3 ) {  //GMG -- Changed to BM_ initialization to be over QoS
                             //       classes as well as other edge nodes
        /*
         * $iagent initiagent $numEdgenodes
         */
        if(strcasecmp(argv[1],"initiagent") == 0 ) {
            maxindx_ = atoi(argv[2]);

            for (j = 0; j < nqos_classes; j++)
               BM_[j] = new BurstManager[maxindx_];

            for ( int i = 0; i < maxindx_ ; i++ )
               for (j = 0; j < nqos_classes; j++)
                  BM_[j][i].init(this, i, j);
            return(TCL_OK);
        }
    }

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

// Expire event in the burst timer
void BurstTimer::expire(Event* e)
{
    bm_->timeout();
}

// Definition of the IPKT header class
static class IPKTHeaderClass :  public PacketHeaderClass
{
    public:
        IPKTHeaderClass() : PacketHeaderClass("PacketHeader/IPKT",
            sizeof(hdr_IPKT))

    {
        bind_offset(&hdr_IPKT::offset_ipkt_);
    }
}class_IPKThdr;

// Defintion of the IPKT Class
static class IPKTClass : public TclClass
{
public:
    IPKTClass() :  TclClass("Agent/IPKT"){}
    TclObject * create (int, const char*const*)    {
        return ( new IPKTAgent );
    }
}class_IPKT;

/*=====================================================================*
 *                                                                     *
 *           Implementation of the Burst-Manager class                 *
 *                                                                     *
 *    Definition and Implementation of the Burst manager class.        *
 *    derived from the TclClass so that a TCL interface (or TCL        *
 *    instantiation) of the BurstManager is available.                 *
 *    Note: the bind and method methods are overriden so as to         *
 *    provide tcl access to the static variables.                      *
 *    ref section 3.0 of the ns-2 manual on more details               *
 *                                                                     *
 *=====================================================================*/
static class BurstManagerClass : public TclClass
{
    public:
    BurstManagerClass() : TclClass( "BurstManager" ) {}
    TclObject *create( int, const char*const* ) {
        return ( new BurstManager );
    }

    virtual void bind();
    virtual int  method( int argc, const char*const* argv );		
}class_burstmanagerclass;

// bind
void BurstManagerClass::bind() {
		TclClass::bind();
  	add_method( "maxburstsize" );

    add_method( "pcntguard" );
    add_method( "offsettime" );
    add_method( "delta" );
    add_method( "bursttimeout" );

}

// method
int BurstManagerClass::method( int ac, const char*const* av ) {
	  Tcl& tcl = Tcl::instance();

    int argc = ac-2;
    const char*const* argv = av + 2;

    if( argc == 2 ) {
        if( strcmp( argv[1], "maxburstsize" ) == 0 ) {
		    tcl.resultf( "%d", BurstManager::maxburstsize() );
       		return (TCL_OK);	
        } else if( strcmp( argv[1], "pcntguard" ) == 0 ) {
            tcl.resultf( "%d", BurstManager::pcntguard() );
            return (TCL_OK);
        } else if( strcmp( argv[1], "delta" ) == 0 ) {
            tcl.resultf( "%lf", BurstManager::delta() );
            return (TCL_OK);
        } else if( strcmp( argv[1], "bursttimeout" ) == 0 ) {
            tcl.resultf( "%lf", BurstManager::burst_timeout() );
            return (TCL_OK);
        }
    } else if( argc == 3 ) {
        if( strcmp( argv[1], "maxburstsize" ) == 0 ) {
            BurstManager::setMaxburstsize( atoi( argv[2] ) );
            return (TCL_OK);
        } else if( strcmp( argv[1], "pcntguard" ) == 0 ) {
            BurstManager::setPcntguard( atoi( argv[2] ) );
            return (TCL_OK);
        } else if( strcmp( argv[1], "offsettime" ) == 0 ){ // for backward compatibility, set offset time for
                                                           // QoS class 0 if class not specified
            BurstManager::setOffsettime( 0, atof( argv[2]) );
            tcl.resultf( "%lf", BurstManager::offsettime( atoi(argv[2]) ) );
            return (TCL_OK);
        } else if( strcmp( argv[1], "delta" ) == 0 ) {
            BurstManager::setDelta( atof( argv[2] ) );
            return (TCL_OK);
        } else if( strcmp( argv[1], "bursttimeout" ) == 0 ) {
            BurstManager::setBursttimeout( atof( argv[2] ) );
            return (TCL_OK);
        }       	    				
    } else if( argc == 4 ) {
        if( strcmp( argv[1], "offsettime" ) == 0 ){
            BurstManager::setOffsettime( atoi(argv[2]), atof( argv[3]) );
            return (TCL_OK);
       }
    }
    return TclClass::method( ac, av );
}