dest_queue.cc

Go to the documentation of this file.

/* -*-  Mode:C++; c-basic-offset:8; tab-width:8; indent-tabs-mode:t -*- */
/*
 * Copyright (c) 1997 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 Computer Systems
 *      Engineering 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.
 */
/* Ported from CMU/Monarch's code*/

/* 
   dest_queue.cc
   $Id: dest_queue.cc,v 1.2 1999/08/12 21:17:11 yaxu Exp $
   
   implement a group of resequencing queues.  one for each source of packets.
   the name destination queue is a misnomer.
   */

#include <imep/dest_queue.h>
#ifdef TEST_ONLY
#include <stdio.h>
#else
#include <imep/imep.h>
#endif
#ifndef TEST_ONLY
#define CURRENT_TIME Scheduler::instance().clock()
#else
extern double CURRENT_TIME;
#endif


static const int verbose = 0;

// Transmission Queue Entry
txent::txent(double e, u_int32_t s, Packet *p)
{
    expire_ = e;
    seqno_ = s;
    pkt_ = p;
}

// Destination Queue Entry
dstent::dstent(nsaddr_t index)
{
    LIST_INIT(&txentHead);
    ipaddr_ = index;
    seqno_ = ILLEGAL_SEQ;
}

static int
SEQ_GT(u_int8_t a, u_int8_t b)
{
  int8_t reg = (int8_t)a - (int8_t) b;
  return reg > 0;
}
    
void
dstent::addEntry(double e, u_int32_t s, Packet *p)
{
    txent *t, *u, *v = 0;

    if((t = findEntry(s)) == 0) {
        t = new txent(e, s, p);
        assert(t);

        for(u = txentHead.lh_first; u; u = u->link.le_next) {
            if(SEQ_GT(u->seqno(), s))
              break;
            v = u;
        }

        if(u == 0 && v == 0) {
            LIST_INSERT_HEAD(&txentHead, t, link);
        } else if(u) {
            LIST_INSERT_BEFORE(u, t, link);
        } else {
            assert(v);
            LIST_INSERT_AFTER(v, t, link);
        }

    } else {
        Packet::free(p);
        // already have a copy of this packet
    }

#ifdef  DEBUG
    // verify that I did not fuck up...
    u_int32_t max = 0;
    for(t = txentHead.lh_first; t; t = t->link.le_next) {
        if(max == 0)
            max = t->seqno();
        else {
            assert(t->seqno() > max);
            max = t->seqno();
        }
    } 
#endif
}            

void
dstent::delEntry(txent *t)
{
    LIST_REMOVE(t, link);
    delete t;
}

txent*
dstent::findEntry(u_int32_t s)
{
    txent *t;

    for(t = txentHead.lh_first; t; t = t->link.le_next) {
        if(t->seqno() == s)
            return t;
    }
    return 0;
}

txent*
dstent::findFirstEntry(void)
{
    return txentHead.lh_first;
    // this gives the minimum sequence number for the destination
}

// Destination Queue
dstQueue::dstQueue(imepAgent *a, nsaddr_t index) : agent_(a), ipaddr_(index)
{
    LIST_INIT(&dstentHead);
}

void
dstQueue::addEntry(nsaddr_t dst, double e, u_int32_t s, Packet *p)
{
    dstent *t;

    if((t = findEntry(dst)) == 0) {
        t = new dstent(dst);
        assert(t);
        LIST_INSERT_HEAD(&dstentHead, t, link);
    }

    if (NULL == t->txentHead.lh_first) agent_->stats.num_holes_created++;
    t->addEntry(e, s, p);
}

    
dstent*
dstQueue::findEntry(nsaddr_t dst)
{
    dstent *t;

    for(t = dstentHead.lh_first; t; t = t->link.le_next) {
        if(t->ipaddr() == dst)
            return t;
    }
    return 0;
}


Packet*
dstQueue::getPacket(nsaddr_t dst, u_int32_t seqno)
{
    dstent *d;
    txent *t;

    for(d = dstentHead.lh_first; d; d = d->link.le_next) {
        if(d->ipaddr() == dst)
            break;
    }

    if(d && (t = d->findEntry(seqno))) {
        Packet *p = t->pkt();
        d->delEntry(t);

        // make sure packets come out in increasing order only
        //  int8_t reg = (int8_t) seqno - (int8_t) d->seqno();
        //assert(reg > 0); // SEQ_GT(seqno, d->seqno())
        //d->seqno() = seqno; 

        return p;
    }
    return 0;
}

double
dstQueue::getNextExpire()
{
    dstent *t;
    Time min = 0.0;

    for(t = dstentHead.lh_first; t; t = t->link.le_next) {
        Time texp = t->expire(); // computed by traversing a list
        if(min == 0.0 || (texp && texp < min))
            min = texp;
    }
    if (verbose)
      agent_->trace("T %.9f _%d_ dest_queue -  getNextExpire is %.9f",
            CURRENT_TIME, ipaddr_, min);

    return min;
}

Packet*
dstQueue::getNextPacket(u_int32_t& s)
{
    dstent *d;

    for(d = dstentHead.lh_first; d; d = d->link.le_next) {
      txent *t = d->findFirstEntry();

      if (t == 0) 
        {
          d->seqno() = ILLEGAL_SEQ;
          continue; // no packets here
        }

      Time texp = d->expire();
      assert(texp);
#ifdef TEST_ONLY
      fprintf(stderr,
          "IN:\td->expire: %f, d->seqno: %d, t->expire: %f, t->seqno: %d\n",
          d->expire(), d->seqno(), t->expire(), t->seqno());
#endif
      if (texp > CURRENT_TIME && d->seqno() == ILLEGAL_SEQ) continue;

      if (t->expire() <= CURRENT_TIME)
        { // remember this seq as starting a chain we can pull off
          d->seqno() = t->seqno();
        }
      else if (d->seqno() != ILLEGAL_SEQ && t->seqno() != (u_int8_t) (d->seqno() + 1))
        { // the next pkt isn't part of the chain we were pulling off
          // stop pulling off packets
          d->seqno() = ILLEGAL_SEQ;
          continue;     
        }

        Packet *p = t->pkt();
        assert(p);

        s = t->seqno();

#ifdef TEST_ONLY
        fprintf(stderr, "\t%s: returning seqno: %d\n",
            __FUNCTION__, s);
#endif
        if (d->seqno() != ILLEGAL_SEQ) 
          { // advance d->seqno() along the chain
        d->seqno() = t->seqno();
          }
        
        d->delEntry(t);
#ifdef TEST_ONLY
      fprintf(stderr,
          "OUT:\td->expire: %f, d->seqno: %d, t->expire: %f, t->seqno: %d\n",
          d->expire(), d->seqno(), t->expire(), t->seqno());
#endif
        return p;
    }
    return 0;
}

void 
dstQueue::deleteDst(nsaddr_t dst)
{
  dstent *d;
  txent *t;
  
  if (verbose)
    agent_->trace("T %.9f _%d_ purge dstQ id %d",
          CURRENT_TIME, ipaddr_, dst);

  for(d = dstentHead.lh_first; d; d = d->link.le_next) 
    {
      if(d->ipaddr() == dst)
    break;
    }
  if (!d) return;

  while ((t = d->findFirstEntry()))
    {
      Packet *p = t->pkt();
      if (verbose)
    agent_->trace("T %.9f _%d_ dstQ id %d delete seq %d",
              CURRENT_TIME, ipaddr_, dst, t->seqno());
      Packet::free(p);
      d->delEntry(t); 
      agent_->stats.num_reseqq_drops++;
    }

  LIST_REMOVE(d,link);
  delete d;
}

void
dstQueue::dumpAll()
{
    dstent *t;

    for(t = dstentHead.lh_first; t; t = t->link.le_next) {
      if (verbose)
        agent_->trace("T %.9f _%d_ dest_queue - src %d expire %.9f seqno %d",
                  CURRENT_TIME, ipaddr_, t->ipaddr(), t->expire(), t->seqno());

        txent *u = t->findFirstEntry();
        for(;u; u = u->link.le_next) {
          if(verbose)
            agent_->trace("T %.9f _%d_ dest_queue - src %d seq %d expire %.9f",
                  CURRENT_TIME, ipaddr_, t->ipaddr(),
                  u->seqno(), u->expire());
        }
    }
}

---