rap.cc

Go to the documentation of this file.

/* -*-  Mode:C++; c-basic-offset:8; tab-width:8; indent-tabs-mode:t -*- */

/*
 * rap.cc
 * Copyright (C) 1997 by the University of Southern California
 * $Id: rap.cc,v 1.14 2005/09/18 23:33:34 tomh Exp $
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License,
 * version 2, as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License along
 * with this program; if not, write to the Free Software Foundation, Inc.,
 * 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA.
 *
 *
 * The copyright of this module includes the following
 * linking-with-specific-other-licenses addition:
 *
 * In addition, as a special exception, the copyright holders of
 * this module give you permission to combine (via static or
 * dynamic linking) this module with free software programs or
 * libraries that are released under the GNU LGPL and with code
 * included in the standard release of ns-2 under the Apache 2.0
 * license or under otherwise-compatible licenses with advertising
 * requirements (or modified versions of such code, with unchanged
 * license).  You may copy and distribute such a system following the
 * terms of the GNU GPL for this module and the licenses of the
 * other code concerned, provided that you include the source code of
 * that other code when and as the GNU GPL requires distribution of
 * source code.
 *
 * Note that people who make modified versions of this module
 * are not obligated to grant this special exception for their
 * modified versions; it is their choice whether to do so.  The GNU
 * General Public License gives permission to release a modified
 * version without this exception; this exception also makes it
 * possible to release a modified version which carries forward this
 * exception.
 *
 */

//
// rap.cc 
//      Code for the 'RAP Source' Agent Class
//
// Author: 
//   Mohit Talwar (mohit@catarina.usc.edu)
//
// $Header: /nfs/jade/vint/CVSROOT/ns-2/rap/rap.cc,v 1.14 2005/09/18 23:33:34 tomh Exp $

#include "rap.h"

int hdr_rap::offset_; // static offset of RAP header

static class RapHeaderClass : public PacketHeaderClass 
{
public:
    RapHeaderClass() : PacketHeaderClass("PacketHeader/RAP", 
                         sizeof(hdr_rap)) {
        bind_offset(&hdr_rap::offset_);
    }
} class_raphdr;


static class RapClass : public TclClass 
{
public:
    RapClass() : TclClass("Agent/RAP") {}
    TclObject* create(int, const char*const*) 
        { return (new RapAgent()); }
} class_rap;


void IpgTimer::expire(Event *)
{ 
    a_->timeout(RAP_IPG_TIMEOUT); 
}

void RttTimer::expire(Event *)
{ 
    a_->timeout(RAP_RTT_TIMEOUT); 
}

//----------------------------------------------------------------------
// EqualSeqno
//      Compare TransHistory Entries on the seqno field.
//
//      "i1", "i2" are the TransHistory entries to be compared.
//----------------------------------------------------------------------
 
int EqualSeqno(void *i1, void *i2)
{ 
    return (((TransHistoryEntry *) i1)->seqno == 
        ((TransHistoryEntry *) i2)->seqno);
}

//----------------------------------------------------------------------
// RapAgent::RapAgent
//  Initialize the RAP agent.
//      Bind variables which have to be accessed in both Tcl and C++.
//      Initializes time values.
//----------------------------------------------------------------------

RapAgent::RapAgent() : Agent(PT_RAP_DATA), ipgTimer_(this), rttTimer_(this),
    seqno_(0), sessionLossCount_(0), curseq_(0), ipg_(2.0), srtt_(2.0), 
    timeout_(2.0), lastRecv_(0), lastMiss_(0), prevRecv_(0), dctr_(0), 
    flags_(0), fixIpg_(0)
{
    bind("packetSize_", &size_);    // Default 512
    bind("seqno_", &seqno_);    // Default 0
    bind("sessionLossCount_", &sessionLossCount_); // Default 0

    bind("ipg_", &ipg_);        // Default 2 seconds
    bind("beta_", &beta_);  // Default 0.5
    bind("alpha_", &alpha_);    // Default 1.0

    bind("srtt_", &srtt_);  // Default 2 seconds
    bind("variance_", &variance_);// Default 0
    bind("delta_", &delta_);    // Default 0.5
    bind("mu_", &mu_);      // Default 1.2
    bind("phi_", &phi_);        // Default 4

    bind("timeout_", &timeout_);    // Default 2 seconds

    bind("overhead_", &overhead_); // Default 0

    bind("useFineGrain_", &useFineGrain_); // Default FALSE
    bind("kfrtt_", &kfrtt_);    // Default 0.9
    bind("kxrtt_", &kxrtt_);    // Default 0.01

    bind("debugEnable_", &debugEnable_);    // Default FALSE

    bind("rap_base_hdr_size_", &rap_base_hdr_size_);

    bind("dpthresh_", &dpthresh_);

    frtt_ = xrtt_ = srtt_;
}

// Cancel all our timers before we quit
RapAgent::~RapAgent()
{
//      fprintf(stderr, "%g: rap agent %s(%d) stops.\n", 
//          Scheduler::instance().clock(), name(), addr());
//      Tcl::instance().eval("[Simulator instance] flush-trace");
    stop();
}

//----------------------------------------------------------------------
// RapAgent::UpdateTimeValues
//      Update the values for srtt_, variance_ and timeout_ based on
//      the "sampleRtt". Use Jacobson/Karl's algorithm.
//      
//  "sampleRtt" is the sample  round trip time obtained from the 
//      current ack packet.
//----------------------------------------------------------------------

void RapAgent::UpdateTimeValues(double sampleRtt)
{ 
    double diff;
    static int initial = TRUE;

    if (initial) {
        frtt_ = xrtt_ = srtt_ = sampleRtt; // First sample, no history
        variance_ = 0;
        initial = FALSE;
    }

    diff = sampleRtt - srtt_;
    srtt_ += delta_ * diff;
      
    diff = (diff < 0) ? diff * -1 : diff; // Take mod
    variance_ += delta_ * (diff - variance_);

    timeout_ = mu_ * srtt_ + phi_ * variance_;

    if (useFineGrain_) {
        frtt_ = (1 - kfrtt_) * frtt_ + kfrtt_ * sampleRtt;
        xrtt_ = (1 - kxrtt_) * xrtt_ + kxrtt_ * sampleRtt;
    }

    double debugSrtt = srtt_;   // $%#& stoopid compiler
    Debug(debugEnable_, logfile_, 
          "- srtt updated to %f\n", debugSrtt);
}

void RapAgent::start()
{
    if (debugEnable_)
        logfile_ = DebugEnable(this->addr() >> 
                       Address::instance().NodeShift_[1]);
    else
        // Should initialize it regardless of whether it'll be used.
        logfile_ = NULL;
    Debug(debugEnable_, logfile_, "%.3f: RAP start.\n", 
          Scheduler::instance().clock());

    flags_ = flags_ & ~RF_STOP;
    startTime_ = Scheduler::instance().clock();
    RttTimeout();       // Decreases initial IPG
    IpgTimeout();
}

// Used by a sink to listen to incoming packets
void RapAgent::listen()
{
    if (debugEnable_)
        logfile_ = DebugEnable(this->addr() >> 
                       Address::instance().NodeShift_[1]);
}

void RapAgent::stop()
{
    Debug(debugEnable_, logfile_, 
          "%.3f: RAP stop.\n", Scheduler::instance().clock());
            
    // Cancel the timer only when there is one
    if (ipgTimer_.status() == TIMER_PENDING)
        ipgTimer_.cancel();  
    if (rttTimer_.status() == TIMER_PENDING)
        rttTimer_.cancel();

    stopTime_ = Scheduler::instance().clock();
    int debugSeqno = seqno_;
    Debug(debugEnable_, logfile_, 
          "- numPackets %d, totalTime %f\n", 
          debugSeqno, stopTime_ - startTime_);
    flags_ |= RF_STOP;
}

//----------------------------------------------------------------------
// RapAgent::command
//      Called when a Tcl command for the RAP agent is executed.
//      Two commands are supported
//         $rapsource start
//         $rapsource stop
//----------------------------------------------------------------------
int RapAgent::command(int argc, const char*const* argv)
{
    if (argc == 2) {
        if (strcmp(argv[1], "start") == 0) {
            start();
            // return TCL_OK, so the calling function knows that 
            // the command has been processed
            return (TCL_OK);
        } else if (strcmp(argv[1], "stop") == 0) {
            stop();
            return (TCL_OK);
        } else if (strcmp(argv[1], "listen") == 0) {
            listen();
            return (TCL_OK);
        }
    } else if (argc == 3) {
                if (strcmp(argv[1], "advanceby") == 0) {
                        advanceby(atoi(argv[2]));
                        return (TCL_OK);
                }
    }


    // If the command hasn't been processed by RapAgent()::command,
    // call the command() function for the base class
    return (Agent::command(argc, argv));
}

//----------------------------------------------------------------------
// RapAgent::SendPacket
//      Called in IpgTimeout().
//      Create a packet, increase seqno_, send the packet out.
//----------------------------------------------------------------------

void RapAgent::SendPacket(int nbytes, AppData *data)
{
    TransHistoryEntry *pktInfo;
    Packet *pkt;

    type_ = PT_RAP_DATA;
    if (data)
        pkt = allocpkt(data->size()); 
    else 
        pkt = allocpkt();

    // Fill in RAP headers
    hdr_rap* hdr = hdr_rap::access(pkt);
    hdr->seqno() = ++seqno_;    // Start counting from 1;
    hdr->lastRecv = hdr->lastMiss = hdr->prevRecv = 0; // Ignore @ sender
    hdr->flags() = RH_DATA;
    if (data) {
        hdr->size() = data->size();
        pkt->setdata(data);
    } else {
        hdr->size() = size_;
    }
    // XXX Simply set packet size to the given ADU's nominal size. 
    // Make sure that the size is reasonable!!
    hdr_cmn *ch = hdr_cmn::access(pkt);
    ch->size() = nbytes; 

    send(pkt, 0);
    pktInfo = new TransHistoryEntry(seqno_);
    transmissionHistory_.SetInsert((void *) pktInfo, EqualSeqno);
    int debugSeqno = seqno_;
    Debug(debugEnable_, logfile_, 
          "- packet %d sent\n", debugSeqno);
}

//----------------------------------------------------------------------
// RapAgent::recv
//      Called when a packet is received.
//      Should be of type PT_RAP_ACK.
//----------------------------------------------------------------------

void RapAgent::recv(Packet* pkt, Handler*)
{
    Debug(debugEnable_, logfile_, 
          "%.3f: RAP packet received.\n", Scheduler::instance().clock());

    hdr_rap* hdr = hdr_rap::access(pkt); // Access RAP header

    switch (hdr->flags()) { 
    case RH_DATA:
        UpdateLastHole(hdr->seqno());
        SendAck(hdr->seqno());
        if ((pkt->datalen() > 0) && app_) 
            // We do have user data, process it
            app_->process_data(pkt->datalen(), pkt->userdata());
        break;
    case RH_ACK:
        RecvAck(hdr);
        break;
    default:
        fprintf(stderr, 
            "RAP agent %s received a packet with unknown flags %x",
            name(), hdr->flags());
        break;
    } 
    Packet::free(pkt);      // Discard the packet
}

//----------------------------------------------------------------------
// RapAgent::RecvAck
//      Called when an Ack is received.
//      
//      "header" is the RAP header of the Ack.
//----------------------------------------------------------------------

void RapAgent::RecvAck(hdr_rap *ackHeader)
{
    double sampleRtt;
    TransHistoryEntry *old, key(ackHeader->seqno_);

    assert(ackHeader->seqno_ > 0);

    Debug(debugEnable_, logfile_, 
          "- ack %d\n", ackHeader->seqno_);

    old = (TransHistoryEntry *) 
        transmissionHistory_.SetRemove((void *) &key, EqualSeqno);

    if (old != NULL) {
        Debug(debugEnable_, logfile_, 
              "- found in transmission history\n");
        assert((old->status == RAP_SENT) || (old->status == RAP_INACTIVE));

        // Get sample rtt       
        sampleRtt = key.departureTime - old->departureTime; 

        UpdateTimeValues(sampleRtt);
        
        delete old;
    }
  
    if (!anyack()) {
        flags_ |= RF_ANYACK;
        ipg_ = srtt_;
    }
  
    if (LossDetection(RAP_ACK_BASED, ackHeader))
        LossHandler();

    // XXX We only stop by sequence number when we are in 
    // "counting sequence number" mode.   -- haoboy
    if (counting_pkt() && (ackHeader->seqno_ >= curseq_)) 
        finish();
}

//----------------------------------------------------------------------
// RapAgent::timeout
//      Called when a timer fires.
//
//      "type" is the type of Timeout event
//----------------------------------------------------------------------

void RapAgent::timeout(int type)
{
    if (type == RAP_IPG_TIMEOUT)
        IpgTimeout();
    else if (type == RAP_RTT_TIMEOUT)
        RttTimeout();
    else
        assert(FALSE);
}

//----------------------------------------------------------------------
// RapAgent::IpgTimeout
//      Called when the ipgTimer_ fires.
//----------------------------------------------------------------------

void RapAgent::IpgTimeout()
{
    double waitPeriod;      // Time before next transmission

    Debug(debugEnable_, logfile_, 
          "%.3f: IPG Timeout.\n", Scheduler::instance().clock());

    if (LossDetection(RAP_TIMER_BASED))
        LossHandler();
    else if (!counting_pkt()) {
        if (app_) {
            int nbytes;
            AppData* data = app_->get_data(nbytes);
            // Missing data in application. What should we do??
            // For now, simply schedule the next SendPacket(). 
            // If the application has nothing to send, it'll stop 
            // the rap agent later on. 
            if (data != NULL) {
                SendPacket(nbytes, data);
                dctr_++;
            }
        } else {
            // If RAP doesn't have application, just go ahead and 
            // send packet
            SendPacket(size_);
            dctr_++;
        }
    } else if (seqno_ < curseq_) {
            SendPacket(size_);
            dctr_++;
    }

    // XXX If we only bound IPG in DecreaseIpg(), the thresholding will 
    // happen immediately because DecreaseIpg() isn't called immediately. 
    // So we do it here. 
    if (fixIpg_ != 0)
        ipg_ = fixIpg_;

    if (useFineGrain_)
        waitPeriod = frtt_ / xrtt_ * ipg_;
    else 
        waitPeriod = ipg_;
    // By this point, we may have been stopped by applications above
    // Thus, do not reschedule a timer if we are stopped. 
    if (!is_stopped())
        ipgTimer_.resched(waitPeriod + Random::uniform(overhead_));
}
  
//----------------------------------------------------------------------
// RapAgent::RttTimeout
//      Called when the rttTimer_ fires.
//      Decrease IPG. Restart rttTimer_.
//----------------------------------------------------------------------

void RapAgent::RttTimeout()
{
    Debug(debugEnable_, logfile_, 
          "%.3f: RTT Timeout.\n", Scheduler::instance().clock());

    // During the past srtt_, we are supposed to send out srtt_/ipg_
    // packets. If we sent less than that, we may not increase rate
    if (100*dctr_*(ipg_/srtt_) >= dpthresh_)
        DecreaseIpg();      // Additive increase in rate
    else 
        Debug(debugEnable_, logfile_, 
              "- %f Cannot increase rate due to insufficient data.\n",
              Scheduler::instance().clock());
    dctr_ = 0;

    double debugIpg = ipg_ + overhead_ / 2;
    Debug(debugEnable_, logfile_, 
          "- ipg decreased at %.3f to %f\n", 
          Scheduler::instance().clock(), debugIpg);

    rttTimer_.resched(srtt_);
}

//----------------------------------------------------------------------
// RapAgent::LossDetection
//      Called in ipgTimeout (RAP_TIMER_BASED) 
//          or in RecvAck (RAP_ACK_BASED).
//
// Returns:
//      TRUE if loss detected, FALSE otherwise.
//
//      "ackHeader" is the RAP header of the received Ack (PT_RAP_ACK).
//----------------------------------------------------------------------

static double currentTime;
static hdr_rap *ackHdr;
static RapAgent *rapAgent; 
static int numLosses;

int EqualStatus(void *i1, void *i2)
{ 
    return (((TransHistoryEntry *) i1)->status == 
        ((TransHistoryEntry *) i2)->status);
}

void DestroyTransHistoryEntry(long item)
{
    TransHistoryEntry *entry = (TransHistoryEntry *) item;

    Debug(rapAgent->GetDebugFlag(), rapAgent->GetLogfile(),
          "- purged seq num %d\n", entry->seqno);

    delete entry;
}

void TimerLostPacket(long item)
{
    TransHistoryEntry *entry = (TransHistoryEntry *) item;

    if ((entry->departureTime + rapAgent->GetTimeout()) <= currentTime) {
        // ~ Packets lost in RAP session
        rapAgent->IncrementLossCount(); 

        // Ignore cluster losses
        if (entry->status != RAP_INACTIVE) {
            assert(entry->status == RAP_SENT);

            numLosses++;
            Debug(rapAgent->GetDebugFlag(), rapAgent->GetLogfile(),
                  "- timerlost seq num %d , last sent %d\n", 
                  entry->seqno, rapAgent->GetSeqno());
        }
        entry->status = RAP_PURGED; 
    }
}

void AckLostPacket(long item)
{
    TransHistoryEntry *entry = (TransHistoryEntry *) item;

    int seqno, lastRecv, lastMiss, prevRecv;

    seqno = entry->seqno;
    lastRecv = ackHdr->lastRecv;
    lastMiss = ackHdr->lastMiss;
    prevRecv = ackHdr->prevRecv;

    if (seqno <= lastRecv) {
        if ((seqno > lastMiss) || (seqno == prevRecv))
            entry->status = RAP_PURGED; // Was Received, now purge
        else if ((lastRecv - seqno) >= 3) {
            // ~ Packets lost in RAP session
            rapAgent->IncrementLossCount(); 
            
            if (entry->status != RAP_INACTIVE) {
                assert(entry->status == RAP_SENT);

                numLosses++;
                Debug(rapAgent->GetDebugFlag(), 
                      rapAgent->GetLogfile(),
                      "- acklost seqno %d , last sent %d\n", 
                      seqno, rapAgent->GetSeqno());
            }
            // Was Lost, purge from history
            entry->status = RAP_PURGED; 
        }
    }
}

int RapAgent::LossDetection(RapLossType type, hdr_rap *ackHeader)
{
    TransHistoryEntry key(0, RAP_PURGED);

    currentTime = key.departureTime;
    ackHdr = ackHeader;
    rapAgent = this;
    numLosses = 0;

    switch(type) {
    case RAP_TIMER_BASED:
        transmissionHistory_.Mapcar(TimerLostPacket);
        break;
        
    case RAP_ACK_BASED:
        transmissionHistory_.Mapcar(AckLostPacket);
        break;

    default:
        assert(FALSE);
    }

    Debug(debugEnable_, logfile_, 
          "- %d losses detected\n", numLosses); 

    Debug(debugEnable_, logfile_, 
          "- history size %d\n", transmissionHistory_.Size());
  
    transmissionHistory_.Purge((void *) &key, 
                   EqualStatus, // Purge PURGED packets
                   DestroyTransHistoryEntry);

    Debug(debugEnable_, logfile_, 
          "- history size %d\n", transmissionHistory_.Size());

    if (numLosses)
        return TRUE;
    else
        return FALSE;
}

//----------------------------------------------------------------------
// RapAgent::LossHandler
//      Called when loss detected.
//      Increase IPG. Mark packets INACTIVE. Reschedule rttTimer_.
//----------------------------------------------------------------------

void MarkInactive(long item)
{
    TransHistoryEntry *entry = (TransHistoryEntry *) item;

    entry->status = RAP_INACTIVE;
}

void RapAgent::LossHandler()
{
    IncreaseIpg();      // Multiplicative decrease in rate

    double debugIpg = ipg_ + overhead_ / 2;
    Debug(debugEnable_, logfile_, 
          "- ipg increased at %.3f to %f\n", 
          Scheduler::instance().clock(), debugIpg);

    transmissionHistory_.Mapcar(MarkInactive);
    Debug(debugEnable_, logfile_, 
          "- window full packets marked inactive\n");

    rttTimer_.resched(srtt_);
}

//----------------------------------------------------------------------
// RapAgent::SendAck
//      Create an ack packet, set fields, send the packet out.
//
//      "seqNum" is the sequence number of the packet being acked.
//----------------------------------------------------------------------

void RapAgent::SendAck(int seqNum)
{
    type_ = PT_RAP_ACK;
    Packet* pkt = allocpkt();   // Create a new packet
    hdr_rap* hdr = hdr_rap::access(pkt);   // Access header

    hdr->seqno() = seqNum;
    hdr->flags() = RH_ACK;

    hdr->lastRecv = lastRecv_;
    hdr->lastMiss = lastMiss_;
    hdr->prevRecv = prevRecv_;

    hdr_cmn *ch = hdr_cmn::access(pkt);
    ch->size() = rap_base_hdr_size_;

    send(pkt, 0);
    Debug(debugEnable_, logfile_, 
          "- ack sent %u [%u %u %u]\n", 
          seqNum, lastRecv_, lastMiss_, prevRecv_);
}

//----------------------------------------------------------------------
// RapSinkAgent::UpdateLastHole
//      Update the last hole in sequence number space at the receiver.
//      
//  "seqNum" is the sequence number of the data packet received.
//----------------------------------------------------------------------

void RapAgent::UpdateLastHole(int seqNum)
{
    assert(seqNum > 0);

    if (seqNum > (lastRecv_ + 1)) {
        prevRecv_ = lastRecv_;
        lastRecv_ = seqNum;
        lastMiss_ = seqNum - 1;
        return;
    }

    if (seqNum == (lastRecv_ + 1)) {
        lastRecv_ = seqNum;
        return;
    }
    
    if ((lastMiss_ < seqNum) && (seqNum <= lastRecv_)) // Duplicate
        return;

    if (seqNum == lastMiss_) {
        if ((prevRecv_ + 1) == seqNum) // Hole filled
            prevRecv_ = lastMiss_ = 0;
        else
            lastMiss_--;
        
        return;
    }
    
    if ((prevRecv_ < seqNum) && (seqNum < lastMiss_)) {
        prevRecv_ = seqNum;
        return;
    }

    assert(seqNum <= prevRecv_);    // Pretty late...
}


// take pkt count
void RapAgent::advanceby(int delta)
{
    flags_ |= RF_COUNTPKT;
        curseq_ = delta;
    start();
}

void RapAgent::finish()
{
    stop();
    Tcl::instance().evalf("%s done", this->name());
}

---