q.c

Go to the documentation of this file.

/** \file q.c

  \brief   Queueing code

*/

#include "q.h"

/**AutomaticStart*************************************************************/

/*---------------------------------------------------------------------------*/
/* Static function prototypes                                                */
/*---------------------------------------------------------------------------*/

static int initScaleFactor (double **scaleFactorArrayPtr,
                util_timing_t * resetThresholdPtr,
                double subSampleFraction);

/**AutomaticEnd***************************************************************/

/*---------------------------------------------------------------------------*/
/* Definition of exported functions                                          */
/*---------------------------------------------------------------------------*/


/**
  * \brief  Allocate and initialize the class of service structure
  * 
  */

Q_Cos_t *
Q_CosAlloc (int N)
{
  int i;
  Q_Cos_t *result;

  result = (Q_Cos_t *) malloc (sizeof (Q_Cos_t));
  result->numClasses = N;
  result->packetQueueArray = (lsList *) malloc (sizeof (lsList) * N);
  result->numPktsInQueue = 0;

  for (i = 0; i < N; i++)
    {
      result->packetQueueArray[i] = lsCreate ();
    }

  return result;
}


/**
  * \brief  Free the class of service structure.
  * 
  */

int
Q_CosFree (Q_Cos_t * aCosStructure)
{
  int i, N;

  for (i = 0; i < N; i++)
    {
      lsList packetQueue = aCosStructure->packetQueueArray[i];
      lsDestroy (packetQueue, 0);
    }
  free (aCosStructure);
  return 0;
}


/**
  * \brief  Allocate and initialize the Drr structure.  Pass
  * in the function used to hash the packet.
  *
  * \sa Q_SrcIpHash
  * \sa Q_DestIpHash
  * \sa Q_DestIpHash
  * \sa Q_SrcDestIpHash
  * \sa Q_SrcDestTcpIpHash
  * 
  */

Q_Drr_t *
Q_DrrAlloc (int (*flowCmp) (Pkt_EthernetHdr_t *, Pkt_EthernetHdr_t *),
        int (*flowHash) (Pkt_EthernetHdr_t *, int))
{
  Q_Drr_t *result;

  result = (Q_Drr_t *) malloc (sizeof (Q_Drr_t));

  result->flowToQueueTable = Hash_InitTable ( ( int(*)() ) flowCmp,  ( int(*)() ) flowHash);

  result->qOfFlows = lsCreate ();
  result->numPktsInQueue = 0;
  result->numBytesInQueue = 0;

  return result;
}


/**
  * \brief  Free the Drr structure
  * 
  */

int
Q_DrrFree (Q_Drr_t * aDrrStruct)
{
  st_generator *stGen;
  int count;
  Q_Flow_t *flow;

  st_foreach_item (aDrrStruct->flowToQueueTable, stGen, (char **) &count,
           (char **) &flow)
  {
    lsDestroy (flow->queue, 0); // we aren't freeing the packets themselves, so del func is 0
    free (flow);
  }
  Hash_FreeTable (aDrrStruct->flowToQueueTable);
  lsDestroy (aDrrStruct->qOfFlows, 0);

  free (aDrrStruct);
  return 0;
}


/**
  * \brief  Insert an Ethernet frame into the Cos structure Return 1 on succ,
  * 0 on failure
  */

int
Q_CosInsertPpBegin (Q_Cos_t * cosStruct,
            Pkt_ProcessPkt_t * anEthFrame, int classIndex)
{
  // 1 to insert at end
  return Q_CosInsertPpBeginEnd (cosStruct, anEthFrame, classIndex, 0);
}

/**
  * \brief  Insert an Ethernet frame into the Cos structure Return 1 on succ,
  * 0 on failure
  */

int
Q_CosInsertPpEnd (Q_Cos_t * cosStruct,
          Pkt_ProcessPkt_t * anEthFrame, int classIndex)
{
  // 1 to insert at end
  return Q_CosInsertPpBeginEnd (cosStruct, anEthFrame, classIndex, 1);
}


/**
  * \brief  Insert an Ethernet frame into the Cos structure Return 1 on succ,
  * 0 on failure
  * 
  * The packet
  * is specified via its pointer. We need its class of service,
  * with 0 being the highest.
  *
  * We can overload this operation to insert into a nonCOS queue, specifically
  * a queue which is used logically to provide protection for one set of flows
  * against another.
  *
  * 
  * 
  */

int
Q_CosInsertPpBeginEnd (Q_Cos_t * cosStruct,
               Pkt_ProcessPkt_t * anEthFrame,
               int classIndex, int insertAtEnd)
{
  int numClasses;

  numClasses = Q_CosNumClasses (cosStruct);
  if (classIndex < 0 || classIndex >= numClasses)
    {
      printf ("Panic: class passed in is not valid\n");
      assert (0);
    }

  lsList packetQueue = cosStruct->packetQueueArray[classIndex];

  if (insertAtEnd)
    {
      lsNewEnd (packetQueue, (lsGeneric) anEthFrame, 0);
    }
  else
    {
      lsNewBegin (packetQueue, (lsGeneric) anEthFrame, 0);
    }
  cosStruct->numPktsInQueue++;

  return 1;
}


/**
  * \brief  Read an ethernet frame from the Cos structure
  * 
  * The packet
  * returned is NIL if there are no packets; otherwise it's
  * 
  */

Pkt_ProcessPkt_t *
Q_CosReadPp (Q_Cos_t * cosStruct)
{
  Pkt_ProcessPkt_t *result = NIL (Pkt_ProcessPkt_t);

  int i;
  for (i = 0; i < cosStruct->numClasses; i++)
    {
      if (lsDelBegin (cosStruct->packetQueueArray[i], (lsGeneric *) & result)
      == LS_NOMORE)
    {
      continue;
    }
      else
    {
      break;
    }
    }
  if (result != NIL (Pkt_ProcessPkt_t))
    {
      cosStruct->numPktsInQueue--;
    }
  return result;
}


/**
  * \brief  Read the head of the queue - queue IS NOT updated
  *
  */

Pkt_ProcessPkt_t *
Q_CosReadHead (Q_Cos_t * cosStruct, int *classPtr)
{
  Pkt_ProcessPkt_t *result = NIL (Pkt_ProcessPkt_t);

  int i;
  for (i = 0; i < cosStruct->numClasses; i++)
    {
      if (lsFirstItem
      (cosStruct->packetQueueArray[i], (lsGeneric *) & result,
       0) == LS_NOMORE)
    {
      continue;
    }
      else
    {
      if (classPtr != NIL (int))
        {
          *classPtr = i;
        }
      return result;
    }
    }
  return NIL (Pkt_ProcessPkt_t);
}


/**
  * \brief  Return the head of the queue - queue IS updated
  *
  * \todo this function and Q_CosReadPp are redundant
  */

Pkt_ProcessPkt_t *
Q_CosHead (Q_Cos_t * cosStruct, int *classPtr)
{
  Pkt_ProcessPkt_t *result = NIL (Pkt_ProcessPkt_t);

  int i;
  for (i = 0; i < cosStruct->numClasses; i++)
    {
      if (lsDelBegin (cosStruct->packetQueueArray[i], (lsGeneric *) & result)
      == LS_NOMORE)
    {
      continue;
    }
      else
    {
      cosStruct->numPktsInQueue--;
      if (classPtr != NIL (int))
        {
          *classPtr = i;
        }
      return result;
    }
    }
  return NIL (Pkt_ProcessPkt_t);
}


/**
  * \brief Return the head of the i-th queue - queue is updated
  *
  * Useful when we have a set of independent queues.
  */

Pkt_ProcessPkt_t *
Q_CosHeadClass (Q_Cos_t * cosStruct, int classIndex )
{
  Pkt_ProcessPkt_t *result = NIL (Pkt_ProcessPkt_t);
  assert ((0 <= classIndex ) && (classIndex  < cosStruct->numClasses));

  if (lsDelBegin (cosStruct->packetQueueArray[classIndex], (lsGeneric *) & result)
      == LS_NOMORE)
    {
      return NIL (Pkt_ProcessPkt_t);
    }
  return result;
}


/** \brief  Read the number of classes in a cosStruct.  */

int
Q_CosNumClasses (Q_Cos_t * cosStruct)
{
  return cosStruct->numClasses;
}


/** \brief  Return 1 if the queue is emtpy, 0 otherwise */

int
Q_DrrTestIsEmpty (Q_Drr_t * flowQueue)
{
  return (flowQueue->numPktsInQueue > 0) ? 0 : 1;
}


/** \brief  Return 1 if the queue is emtpy, 0 otherwise */

int
Q_CosTestIsEmpty (Q_Cos_t * cosStruct)
{
  return (cosStruct->numPktsInQueue > 0) ? 0 : 1;
}


/** \brief  Insert a Pkt_ProcessPkt_t * into a drrStruct, add to beginning */

int
Q_DrrInsertPpBegin (Q_Drr_t * drrStruct, Pkt_ProcessPkt_t * pp)
{
  return Q_DrrInsertPpBeginEnd (drrStruct, pp, 1);
}


/** \brief  Insert a Pkt_ProcessPkt_t * into a drrStruct, add to end */

int
Q_DrrInsertPpEnd (Q_Drr_t * drrStruct, Pkt_ProcessPkt_t * pp)
{
  return Q_DrrInsertPpBeginEnd (drrStruct, pp, 0);
}


/** \brief  Insert a Pkt_ProcessPkt_t * into a drrStruct.  */

int
Q_DrrInsertPpBeginEnd (Q_Drr_t * drrStruct,
               Pkt_ProcessPkt_t * pp, int insertAtEnd)
{
  Q_Flow_t *aQ;

  st_table *flowToQueueTable = Q_DrrReadFlowToQueueTable (drrStruct);

  if (!Hash_Lookup (flowToQueueTable, (char *) pp->pkt, (char **) &aQ))
    {
      // no queue exists for this flowid, so let's allocate one 
      aQ = Q_AllocFlow ();

      // cant use pp->pkt as the key as once the packet
      // is freed it will become garbage

      Pkt_IpHdr_t *tmpPkt = Pkt_EthernetExtractIp (pp->pkt);
      int ipHdrLength = tmpPkt->ihl * 4;

      int length =
    sizeof (Pkt_EthernetHdr_t) + ipHdrLength + sizeof (Pkt_TcpHdr_t);
      char *flowId = (char *) malloc (length);
      memcpy (flowId, pp->pkt, length); // dest, src, length

      Hash_Insert (flowToQueueTable, (char *) flowId, (char *) aQ);
      // was missing originally:
      lsNewEnd (drrStruct->qOfFlows, (lsGeneric) aQ->queue, 0);
    }
  if (insertAtEnd)
    {
      lsNewEnd (aQ->queue, (lsGeneric) pp, 0);
    }
  else
    {
      lsNewBegin (aQ->queue, (lsGeneric) pp, 0);
    }

  aQ->numPkts++;
  aQ->numBytes += Pkt_EthernetPktHdrReadLength (pp->pkt);

  drrStruct->numPktsInQueue++;

  return 1;
}


/** \brief Get flow associated with pkt */

Q_Flow_t *
Q_DrrReadFlow (Q_Drr_t * drrStruct, Pkt_EthernetHdr_t * pkt)
{
  Q_Flow_t *aQ;
  if (!Hash_Lookup (drrStruct->flowToQueueTable, (char *) pkt, (char **) &aQ))
    {
      return NIL (Q_Flow_t);
    }
  else
    {
      return aQ;
    }
}

/**
  * \brief  Get the head from the Drr structure
  * 
  * The packet returned is NIL if there are no packets.  For now, we're 
  * simply implementing plain round robin, not deficit round robin.
  *
  * We keep a queue of the next flow to service - head is current
  * high priority, and is moved to the tail if it's nonempty.  
  * If it is empty we remove the 
  * corresponding entry from the hash table, then free the queue.
  *
  * This is only RR, and to implement DRR, the qOfFlow's becomes
  * a heap, with the heap value being the current deficit, calculated
  * based on packet lengths.
  * 
  */

Pkt_ProcessPkt_t *
Q_DrrHead (Q_Drr_t * drrStruct)
{
  Pkt_ProcessPkt_t *result;
  lsList aPktQueue;

  lsList qOfFlows = drrStruct->qOfFlows;

  if ((lsLength (qOfFlows) == 0))
    {
      return NIL (Pkt_ProcessPkt_t);
    }

  int status;
  status = lsDelBegin (qOfFlows, (lsGeneric *) & aPktQueue);
  assert (status == LS_OK);

  status = lsDelBegin (aPktQueue, (lsGeneric *) & result);
  assert (status == LS_OK);

  drrStruct->numPktsInQueue--;

  Q_Flow_t *aFlow;
  int succ =
    Hash_Lookup (drrStruct->flowToQueueTable, (char *) result->pkt,
           (char **) &aFlow);
  assert (succ);
  // check if the packet queue is now empty
  if (lsLength (aPktQueue) == 0)
    {
      // int flowid = Pkt_TcpIpFlowHash( Pkt_EthernetExtractIp( result->pkt ) );
      assert (aFlow->numPkts == 1); // since we just took out the last packet
      Pkt_EthernetHdr_t *tmp = result->pkt;
      Hash_Delete (drrStruct->flowToQueueTable, (char **) &tmp, 0);
      /// \todo free tmp?
      lsDestroy (aPktQueue, NULL);
    }
  else
    {
      // move this flow's queue back to the end of the queue
      lsNewEnd (qOfFlows, (lsGeneric) aPktQueue, 0);
      // update numPkts, numBytes
      aFlow->numPkts--;
      aFlow->numBytes -= Pkt_EthernetPktHdrReadLength (result->pkt);
    }

  return result;
}


/**
  * \brief  Read the head from the Drr structure. Does not change the state of the queue.
  * 
  */

Pkt_ProcessPkt_t *
Q_DrrReadHead (Q_Drr_t * drrStruct)
{
  Pkt_ProcessPkt_t *result;
  lsList aPktQueue;

  lsList qOfFlows = drrStruct->qOfFlows;

  if ((lsLength (qOfFlows) == 0))
    {
      return NIL (Pkt_ProcessPkt_t);
    }

  int status;
  status = lsFirstItem (qOfFlows, (lsGeneric *) & aPktQueue, 0);
  assert (status == LS_OK);

  status = lsFirstItem (aPktQueue, (lsGeneric *) & result, 0);
  assert (status == LS_OK);

  return result;
}


/**
  * \brief  Read the flowToQueue table for a given Drr structure.
  * 
  */

st_table *
Q_DrrReadFlowToQueueTable (Q_Drr_t * drrStruct)
{
  return drrStruct->flowToQueueTable;
}


/**
  * \brief  Return the number of flows currently in the drr struct
  * 
  */

int
Q_DrrNumFlows (Q_Drr_t * drrStruct)
{
  int count = st_count (drrStruct->flowToQueueTable);
  return count;
}


/**
  * \brief  Print elementary data about a COS structure.
  * 
  * Use more as a sanity check
  * then for detailed statistics.
  * 
  */

void
Q_CosPrint (Q_Cos_t * cosData)
{
  printf (">>> Printing COS data structure...\n");
  printf ("Num of classes = %d\n", cosData->numClasses);

  int i;
  for (i = 0; i < cosData->numClasses; i++)
    {
      printf ("COS %d has %d classes\n", i,
          lsLength (cosData->packetQueueArray[i]));
    }

  return;
}


/**
  * \brief  Print elementary data about a DRR structure
  * 
  */

void
Q_DrrPrint (Q_Drr_t * drrData)
{
  st_generator *stGen;
  int flowId;
  lsList aQ;

  printf ("Printing DRR structure:\n");

  if (st_count (drrData->flowToQueueTable) > 100)
    {
      printf
    ("This DRR struct has more than 100 indivudual queues, skipping!\n");
    }
  else
    {
      st_foreach_item (drrData->flowToQueueTable, stGen, (char **) &flowId,
               (char **) &aQ)
      {
    printf ("Flow id = %u and queue for this flow is has %d elements\n",
        flowId, lsLength (aQ));
      }
    }
  return;
}


/**
  * \brief Allocate a flow structure
  */

Q_Flow_t *
Q_AllocFlow ()
{
  Q_Flow_t *flow = (Q_Flow_t *) malloc (sizeof (Q_Flow_t));
  flow->queue = lsCreate ();
  flow->numBytes = 0;
  flow->numPkts = 0;
  flow->rate = 0.0;
  util_timing_set (&flow->startOfCurrentWindow);
  flow->numBytes = 0.0;

  return flow;
}


/**
  * \brief Get the new rate - uses a windowing approach
  */

double
Q_ComputeRate (util_timing_t startOfWindow,
           double timeWindowDuration,
           int numBytesExitedInWindow, int pktLength)
{
  double newRate;

  util_timing_t currentTime;
  util_timing_set (&currentTime);
  double delta =
    util_timing_secs (currentTime) - util_timing_secs (startOfWindow);
  // printf("delta is %f\n", delta );
  if (delta > timeWindowDuration)
    {
      newRate = 0.0;
    }
  else
    {
      newRate = (8.0 * (double) (numBytesExitedInWindow + pktLength)) / delta;
    }
  return newRate;
}


/**
  * \brief Get the new rate for this flow - uses an IIR averaging approach
  */

double
Q_ComputeRateOld (util_timing_t lastTime, Pkt_EthernetHdr_t * pkt)
{
  static int count = 0;
  util_timing_t currentTime;
  util_timing_set (&currentTime);
  util_timing_t delta = util_timing_diff (currentTime, lastTime);
  double deltaTime = util_timing_secs (delta);
  double result;

  const double tick = 0.00000001;
  static double *scaleFactor = NIL (double);
  static util_timing_t resetThreshold;

  if (scaleFactor == NIL (double))
    {
      initScaleFactor (&scaleFactor, &resetThreshold, tick);
    }

  if (util_timing_compare (resetThreshold, delta))
    {
      // delta is larger than resetThreshold
      result =
    8.0 * ((double) Pkt_EthernetPktHdrReadLength (pkt)) / deltaTime;
    }
  else
    {
      // currRate = (1-a)^k currRate + a . newRate
      // for a tick of 0.00000001
      double a = 0.999999931;
      result = pow (a, (deltaTime / tick)) +
    (1.0 -
     a) * (8.0 * (double) Pkt_EthernetPktHdrReadLength (pkt)) / deltaTime;

      // for tick of 0.001
      // double a = 0.9999996451;
      // result = scaleFactor[ (int) ( deltaTime  / subSampleFraction ) ] * ( (double) lastRate ) +
      //              (1.0-a) * ( 8.0 * (double) Pkt_EthernetPktHdrReadLength( pkt ) ) / deltaTime;
    }
  if (0 == (count % 1000))
    {
      printf ("count %d, rate %.8f\n", count, result);
    }
  count++;
  return result;
}


/**
  *     \brief  Return the drop prob array, returns length of array
  *
  */
static int
initScaleFactor (double **scaleFactorArrayPtr,
         util_timing_t * resetThresholdPtr, double subSampleFraction)
{
  // parameters used to update flow rate
  const double resetThresholdTime = 10.0;
  util_timing_t resetThreshold = util_timing_pair (resetThresholdTime);
  *resetThresholdPtr = resetThreshold;

  // sample every 0.512 microseconds, which is the time for a 64 byte packet
  const double sampleTime = 0.512 * 0.000001;

  // after 1 second, i.e., 1/(0.512*0.000001) cycles, the scale factor is 0.5
  const double alpha = 0.9999996451;

  const int subSampleRate = (int) (1.0 / subSampleFraction);
  int length = (int) ((resetThresholdTime / sampleTime) * subSampleFraction);
  static double *scaleFactor = NIL (double);
  scaleFactor = (double *) malloc (sizeof (double) * length);   // works out to 20000 entries

  double currProd = alpha;
  int loopBound = (int) (resetThresholdTime / sampleTime);

  int i;
  for (i = 0; i < loopBound; i++)
    {
      if ((i % subSampleRate) == 0)
    {
      scaleFactor[i % subSampleRate] = currProd;
    }
      currProd *= alpha;
    }
  *scaleFactorArrayPtr = scaleFactor;

  return length;
}


/**
  *     \brief  Allocate a Q_Q_t structure.
  *
  */

Q_Q_t *
Q_AllocQ (char *name)
{
  Q_Q_t *result = (Q_Q_t *) malloc (sizeof (Q_Q_t));

  // this make sense only for flow-based queueing, but put it in as a default
  result->flowType = Pkt_SrcIpFlow_c;
  result->flowMaxRate = 0;
  result->maxFlows = -1;    // -1 means there's no limit
  (result->lastEntryTime).high = 0;
  (result->lastEntryTime).low = 0;

  // actual queueing structure for flow-based queuing case
  result->flowQ = NIL (Q_Drr_t);

  result->name = name;
  result->type = Q_QueueTypeUndef_c;

  result->maxRate = INFINITY;
  result->currentRate = 0;
  result->flowMaxRate = INFINITY;
  result->maxBytesPerFlow = INFINITY;
  result->flowDropMethod = Q_DropTail_c;

  result->currentPriority = 0;
  result->assignedPriority = 0;

  result->numPktsInQueue = 0;
  result->numPktsProcessed = 0.0;   // double, use for stats gathering

  result->maxAllowedBytes = 0;  // means there's no limi
  result->numBytesInQueue = 0;
  result->numBytesEntered = 0.0;
  result->numBytesExited = 0.0;

  result->timeWindowDuration = 1.0; // default to 1 second
  result->numBytesExitedInWindow = 0.0;
  (result->startOfCurrentWindow).high = 0;
  (result->startOfCurrentWindow).low = 0;

  result->dropMethod = Q_DropTail_c;

  result->numClasses = 0;
  result->cosQ = NIL (Q_Cos_t);

  return result;
}