开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > Windows编程 > 通讯编程 > 查看源码
pushback.cc
资源名称：ns-allinone-2.33.tar.gz　[点击查看]
上传用户：rrhhcc
上传日期：2015-12-11
资源大小：54129k
文件大小：34k
源码类别：

通讯编程

开发平台：

Visual C++

pushback.cc：源码内容
  1. /* -*-  Mode:C++; c-basic-offset:4; tab-width:4; indent-tabs-mode:t -*- */
  2. /*
  3.  * Copyright (c) 2000  International Computer Science Institute
  4.  * All rights reserved.
  5.  *
  6.  * Redistribution and use in source and binary forms, with or without
  7.  * modification, are permitted provided that the following conditions
  8.  * are met:
  9.  * 1. Redistributions of source code must retain the above copyright
  10.  *    notice, this list of conditions and the following disclaimer.
  11.  * 2. Redistributions in binary form must reproduce the above copyright
  12.  *    notice, this list of conditions and the following disclaimer in the
  13.  *    documentation and/or other materials provided with the distribution.
  14.  * 3. All advertising materials mentioning features or use of this software
  15.  *    must display the following acknowledgement:
  16.  * This product includes software developed by ACIRI, the AT&T 
  17.  *      Center for Internet Research at ICSI (the International Computer
  18.  *      Science Institute).
  19.  * 4. Neither the name of ACIRI nor of ICSI may be used
  20.  *    to endorse or promote products derived from this software without
  21.  *    specific prior written permission.
  22.  *
  23.  * THIS SOFTWARE IS PROVIDED BY ICSI AND CONTRIBUTORS ``AS IS'' AND
  24.  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  25.  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  26.  * ARE DISCLAIMED.  IN NO EVENT SHALL ICSI OR CONTRIBUTORS BE LIABLE
  27.  * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  28.  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  29.  * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  30.  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  31.  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  32.  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  33.  * SUCH DAMAGE.
  34.  *
  35.  */
  37. #include "pushback.h"
  39. #include "ident-tree.h"
  40. #include "pushback-queue.h"
  41. #include "rate-limit.h"
  42. #include "pushback-message.h"
  44. //#define DEBUG  
  46. int hdr_pushback::offset_;
  48. static class PushbackHeaderClass : public PacketHeaderClass {
  49. public:
  50.   PushbackHeaderClass() : PacketHeaderClass("PacketHeader/Pushback", 
  51.     sizeof(hdr_pushback)) {
  52.     bind_offset(&hdr_pushback::offset_);
  53.   }
  54. } class_Pushback_hdr;
  57. static class PushbackClass : public TclClass {
  58. public:
  59.   PushbackClass() : TclClass("Agent/Pushback") {}
  60.   TclObject* create(int, const char*const*) {
  61.     return (new PushbackAgent());
  62.   }
  63. } class_Pushback;
  65. PushbackAgent::PushbackAgent() : Agent(PT_PUSHBACK), last_index_(0), intResult_(-1) {
  67.   bind("last_index_", &last_index_);
  68.   bind("intResult_", &intResult_);
  69.   bind_bool("enable_pushback_", &enable_pushback_);
  70.   bind_bool("verbose_", &verbose_);
  71.   timer_ = new PushbackTimer(this);
  72.   debugLevel = 3;
  73.   //  debugLevel = 0;
  74. }
  76. int 
  77. PushbackAgent::command(int argc, const char*const* argv) {
  78.   
  79.   Tcl& tcl = Tcl::instance();
  80.   if (argc == 4 ) {
  81.     if (strcmp(argv[1], "initialize") == 0) {
  82.       //get the node and routeLogic object
  83.       node_ = (Node *)TclObject::lookup(argv[2]);
  84.       rtLogic_ = (RouteLogic *)TclObject::lookup(argv[3]);
  86.       if (node_ == NULL || rtLogic_ == NULL) {
  87. if (verbose_) printf("Improper Initialization for Pushback Agentn");
  88. return(TCL_ERROR);
  89.       }
  90.       
  91.       sprintf(prnMsg, "node=%s rtLogic=%s id=%d address=%dn", node_->name(), 
  92.      rtLogic_->name(), node_->nodeid(), node_->address());
  93.       printMsg(prnMsg,0);
  95.       return(TCL_OK);
  96.     } 
  97.   }  
  98.   else if (argc == 3) { 
  99.     //$pba add-queue $queue
  100.     if (strcmp(argv[1], "add-queue") == 0) {
  101.       if (last_index_==MAX_QUEUES) {
  102. printf("queue list size exhausted - recompile with a bigger MAX_QUEUESn");
  103. exit(-1);
  104.       }
  105.       PushbackQueue * queue = (PushbackQueue *) TclObject::lookup(argv[2]);
  106.       if (queue == NULL) {
  107. printf("NULL queue passed n");
  108. exit(-1);
  109.       }
  111.       int index = last_index_++;
  112.       queue_list_[index].pbq_ = queue;
  113.       queue_list_[index].idTree_ = new IdentStruct();
  114.       
  115.       tcl.resultf("%d", index);
  116.       return (TCL_OK);
  117.     }
  118.   }
  119.   return (Agent::command(argc, argv));
  120. }
  122. void 
  123. PushbackAgent::reportDrop(int qid, Packet * p) {
  125.   if (!checkQID(qid)) {
  126.     sprintf(prnMsg,"Got invalid qid %dn", qid);
  127.     printMsg(prnMsg,0);
  128.     exit(-1);
  129.   }
  130.   
  131.   hdr_ip * iph = hdr_ip::access(p);
  132.   ns_addr_t src = iph->src();
  133.   ns_addr_t dst = iph->dst();
  134.   int fid = iph->flowid();
  136.   sprintf(prnMsg,"DropDetails from queue %d: %d.%d -> %d.%d (%d)n", qid, 
  137.    src.addr_, src.port_, dst.addr_, dst.port_, fid);
  138.   printMsg(prnMsg, 5);
  139.   
  140.   queue_list_[qid].idTree_->registerDrop(p);
  141. }
  143. void 
  144. PushbackAgent::calculateLowerBound(int qid, double arrRate) {
  146.   if (!checkQID(qid)) {
  147.     sprintf(prnMsg, "Got invalid id from queue in identifyAggregaten");
  148.     printMsg(prnMsg,0);
  149.     exit(-1);
  150.   }
  152.   AggReturn * aggReturn = queue_list_[qid].idTree_->calculateLowerBound();
  153.   if (aggReturn == NULL) {
  154.   //not sure what to do here.
  155.   //maybe lower bound should be left as it is
  156.   
  157.   return;
  158.   }
  160.   double lowerBound = 0;
  161.   int i = 0;
  162.   for (; i <= aggReturn->finalIndex_; i++) {
  163.       cluster currCluster = aggReturn->clusterList_[i];
  164.       AggSpec * aggSpec = new AggSpec(1, currCluster.prefix_, currCluster.bits_);
  165.       RateLimitSession * rls1 = 
  166.       queue_list_[qid].pbq_->rlsList_->containsLocalAggSpec(aggSpec, node_->nodeid());
  167.       if (rls1 !=NULL) continue;
  168.       lowerBound = (currCluster.count_)*(arrRate/aggReturn->totalCount_);
  169.       sprintf(prnMsg, "LB: count: %d totalCount_: %d arrRate: %gn", currCluster.count_, aggReturn->totalCount_, arrRate);
  170.       printMsg(prnMsg,0);
  171.       break;
  172.   }
  173.   
  174.   if (i == aggReturn->finalIndex_+1) {
  175.     sprintf(prnMsg, "Warning: All clusters being rate limitedn");
  176.     printMsg(prnMsg,0);
  177.     //exit(-1);
  178.   }
  180.   queue_list_[qid].idTree_->setLowerBound(lowerBound, 1);
  181.   
  182.   delete(aggReturn);
  183. }
  185. void
  186. PushbackAgent::identifyAggregate(int qid, double arrRate, double linkBW) {
  188.   //set up refresh timer for this queue, if this is the firstime you come here.
  189.   if (!timer_->containsRefresh(qid)) {
  190.     PushbackEvent * event = new PushbackEvent(PUSHBACK_CYCLE_TIME, PUSHBACK_REFRESH_EVENT, qid);
  191.     timer_->insert(event);
  192.   }
  194.   // if (debug_) 
  195.   sprintf(prnMsg, "identifyAggregate for %dn",  qid);
  196.   printMsg(prnMsg,0);
  198.   if (!checkQID(qid)) {
  199.     sprintf(prnMsg, "Got invalid id from queue in identifyAggregaten");
  200.     printMsg(prnMsg,0);
  201.     exit(-1);
  202.   }
  203.   if (verbose_) queue_list_[qid].idTree_->traverse();
  205.   //this is a quick way of achieving this.
  206.   //but it can be justified on some grounds. will do a check with Sally later.
  207.   int noSessions = queue_list_[qid].pbq_->rlsList_->noMySessions(node_->nodeid());
  208. //   if (noSessions >= MAX_SESSIONS) {
  209. //    sprintf(prnMsg, "My hands are fulln");
  210. //    printMsg(prnMsg,0); 
  211. //    return;
  212. //   }
  213.   
  214.   AggReturn * aggReturn = queue_list_[qid].idTree_->identifyAggregate(arrRate, linkBW);
  215.   if (aggReturn == NULL) return;
  217.   for (int i=0; i<=aggReturn->finalIndex_; i++) {
  218.     cluster currCluster = aggReturn->clusterList_[i];
  219.     AggSpec * aggSpec = new AggSpec(1, currCluster.prefix_, currCluster.bits_);
  220.     
  221.     //don't insert the same aggregate again.
  222.     RateLimitSession * rls1 = 
  223.       queue_list_[qid].pbq_->rlsList_->containsLocalAggSpec(aggSpec, node_->nodeid());
  224.     if (rls1 != NULL) {
  225.       sprintf(prnMsg, "got subset aggregate. Lowerbound = %g. agg = ", aggReturn->limit_);
  226.       printMsg(prnMsg,0);
  227.       aggSpec->print(); fflush(stdout);
  228.       delete(aggSpec);
  229.       //this could keep the lowerbound unnecessarily down.
  230.       //but don't be sympathetic with aggregates, which have been identified again.
  231.       if (aggReturn->limit_ < rls1->lowerBound_) {
  232. rls1->lowerBound_ = aggReturn->limit_;
  233.       }
  234.       //set the last misbehavior signal.
  235.       rls1->refreshed();
  236.       continue;
  237.     }
  238.     
  239.     double estimate = (currCluster.count_)*(arrRate/aggReturn->totalCount_);
  240.     
  241.     if (noSessions >= MAX_SESSIONS) {
  242. int rank = queue_list_[qid].pbq_->rlsList_->rankRate(node_->nodeid(), estimate);
  243. if (rank >= MAX_SESSIONS) {
  244.     sprintf(prnMsg, "got rate <= minRate. agg = ");
  245.     printMsg(prnMsg,0);aggSpec->print(); fflush(stdout);
  246.     delete(aggSpec);     
  247.     continue;
  248. } 
  249.     }
  250.     
  251.     sprintf(prnMsg, "starting rate-limiting lower=%g estimate=%g agg ",  
  252.     aggReturn->limit_, estimate);
  253.     printMsg(prnMsg,0);
  254.     aggSpec->print();  fflush(stdout);
  255.     
  256.     double initialLimit = estimate; //*(1 - ambientDropRate);
  257.     RateLimitSession * rls = new RateLimitSession(aggSpec, estimate, 1, initialLimit, 
  258.   node_->nodeid(), qid, 
  259.   RATE_LIMIT_TIME_DEFAULT, aggReturn->limit_,
  260.   node_, rtLogic_);
  261.     queue_list_[qid].pbq_->rlsList_->insert(rls);
  262.       
  263.     PushbackEvent * event = new PushbackEvent(INITIAL_UPDATE_TIME, INITIAL_UPDATE_EVENT, rls);
  264.     timer_->insert(event);
  265.     //    }
  267.     noSessions++;
  268.   }
  270.   queue_list_[qid].idTree_->setLowerBound(aggReturn->limit_, 0);
  271.   delete(aggReturn);
  272. }
  274. void
  275. PushbackAgent::resetDropLog(int qid) {
  277.   sprintf(prnMsg, " drop log reset for qid %dn",  qid);
  278.   printMsg(prnMsg,5);
  279.   
  280.   if (!checkQID(qid)) {
  281.     printf("Got invalid id from queue in resetDropLogn");
  282.     exit(-1);
  283.   }
  285.   queue_list_[qid].idTree_->reset();
  286. }
  288. void 
  289. PushbackAgent::timeout(PushbackEvent * event) {
  290.   
  291.   sprintf(prnMsg, "  %s event for qid %dn",  PushbackEvent::type(event), event->qid_);
  292.   printMsg(prnMsg,0);
  293.   switch (event->eventID_) {
  294.   case PUSHBACK_CHECK_EVENT: pushbackCheck(event->rls_);
  295.     break;
  296.   case PUSHBACK_REFRESH_EVENT: pushbackRefresh(event->qid_);
  297.     break;
  298.   case PUSHBACK_STATUS_EVENT: pushbackStatus(event->rls_);
  299.     break;
  300.   case INITIAL_UPDATE_EVENT: initialUpdate(event->rls_);
  301.     break;
  302.   default: sprintf(prnMsg, " Unrecognized event %dn",  event->eventID_);
  303.     printMsg(prnMsg,0);
  304.     break;
  305.   }
  307. }
  309. void 
  310. PushbackAgent::initialUpdate(RateLimitSession * rls) {
  311.   
  312.   if ( !rls->initialPhase_ ) {
  313.     sprintf(prnMsg, " Error: Update when not in initialphasen");
  314.      printMsg(prnMsg,0);
  315.      exit(-1);
  316.   }
  318.   double qdrop = queue_list_[rls->localQID_].pbq_->getDropRate();
  319.   double dropRate = rls->getDropRate();
  320.   double arrRate = rls->getArrivalRateForStatus();
  321.   double newLimit = arrRate*(1 - 2*(dropRate+qdrop));
  322.   
  323.   sprintf(prnMsg,"Initial-Update: qdrop=%g dr=%g newL=%g oldTarget=%g lowerBound=%g arr=%gn",
  324.   qdrop, dropRate, newLimit, rls->rlStrategy_->target_rate_, rls->lowerBound_, arrRate);
  325.   printMsg(prnMsg,0);
  327.   //cancel right now, if arrRate is significantly less than lower bound.
  328.   if (arrRate < 0.75*rls->lowerBound_) {
  329.       #ifdef DEBUG
  330.         double now = Scheduler::instance().clock();
  331.         printf("Cancel pushback A time: %5.3fn", now);
  332.       #endif
  333.       pushbackCancel(rls);
  334.       return;
  335.   }
  337.   if (newLimit > rls->lowerBound_) {
  338.     rls->setLimit(newLimit);
  339.     
  340.     PushbackEvent * event = new PushbackEvent(INITIAL_UPDATE_TIME, INITIAL_UPDATE_EVENT, rls);
  341.     timer_->insert(event);
  342.   }
  343.   else {
  344.     rls->setLimit(rls->lowerBound_);
  345.     rls->initialPhase_ = 0;
  346.     
  347.     if (rls->logData_->count_!=0 && enable_pushback_) {
  348.       PushbackEvent * event = new PushbackEvent(PUSHBACK_CHECK_TIME, PUSHBACK_CHECK_EVENT, rls);
  349.       timer_->insert(event);
  350.     }
  351.   }
  352. }
  355. void 
  356. PushbackAgent::pushbackCheck(RateLimitSession * rls) {
  357.   
  358.   double dropRate = rls->getDropRate();
  360.   if (dropRate >= DROP_RATE_FOR_PUSHBACK) {
  361.     rls->pushbackOn();
  362.     rls->heightInPTree_++;
  363.     
  364.     double totalRate =  rls->rlStrategy_->target_rate_;
  365.     int count = rls->logData_->count_;
  366.     double fairShare = totalRate/count;
  367.     int done = count;
  368.     
  369.     //max-min allocation of limit.
  370.     while (done != 0) {
  371.       LoggingDataStructNode * lgdsNode = rls->logData_->first_;
  372.       int countThisRound=0;
  373.       while (lgdsNode != NULL) {
  374. double rate = lgdsNode->rateEstimator_->estRate_;
  375. if (rate <= fairShare && !lgdsNode->pushbackSent_) {
  376.   AggSpec * aggSpec = rls->aggSpec_->clone();
  377.   PushbackMessage * msg;
  378.   if (rate < fairShare/2.0) {
  379.     msg = new PushbackRequestMessage(node_->nodeid(), lgdsNode->nid_, rls->localQID_, 
  380.      rls->localID_, aggSpec, INFINITE_LIMIT, 
  381.      rls->depthInPTree_);
  382.     lgdsNode->pushbackSent(INFINITE_LIMIT, rate);
  383.   }   
  384.   else {
  385.     msg = new PushbackRequestMessage(node_->nodeid(), lgdsNode->nid_, rls->localQID_, 
  386.      rls->localID_, aggSpec, rate, rls->depthInPTree_);
  387.     lgdsNode->pushbackSent(rate, rate);
  388.   }
  389.   sendMsg(msg);
  390.   countThisRound++;
  391.   done--;
  392.   totalRate -= rate;
  393. }
  394. lgdsNode = lgdsNode->next_;
  395.       }
  396.       if (done == 0) break;
  397.       if (countThisRound==0) {
  398. //allocate fairshare to everyone and end.
  399. LoggingDataStructNode * lgdsNode = rls->logData_->first_;
  400. while (lgdsNode != NULL) {
  401.   if (!lgdsNode->pushbackSent_) {
  402.     AggSpec * aggSpec = rls->aggSpec_->clone();
  403.     PushbackMessage * msg = new PushbackRequestMessage(node_->nodeid(), lgdsNode->nid_, 
  404.        rls->localQID_, rls->localID_, 
  405.        aggSpec, fairShare, 
  406.        rls->depthInPTree_);
  407.     lgdsNode->pushbackSent(fairShare,lgdsNode->rateEstimator_->estRate_);
  408.     sendMsg(msg);
  409.     done--;
  410.     totalRate-=fairShare;
  411.   }
  412.   lgdsNode = lgdsNode->next_;
  413. }
  414.       }
  415.       else {
  416. fairShare= totalRate/done;
  417.       }
  418.     }
  419.     
  420.   }
  421.   else {
  422.     //set up pushback check for later.
  423.     PushbackEvent * event = new PushbackEvent(PUSHBACK_CHECK_TIME, PUSHBACK_CHECK_EVENT, rls);
  424.     timer_->insert(event);
  425.   }
  426. }
  428. void 
  429. PushbackAgent::pushbackStatus(RateLimitSession * rls) {
  431.   if (rls->pushbackON_) {
  432.     sprintf(prnMsg, " Warning: status timer expired for non-leaf noden");
  433.     printMsg(prnMsg,0);
  434.      //exit(-1);
  435.   }
  436.   double rate = rls->getArrivalRateForStatus();
  437.   rls->logData_->resetStatus();
  438.   
  439.   PushbackMessage * msg = new PushbackStatusMessage(node_->nodeid(), rls->origin_, 
  440.     rls->remoteQID_, rls->remoteID_, 
  441.     rate, rls->heightInPTree_);
  442.   sendMsg(msg);
  443. }
  445. void 
  446. PushbackAgent::pushbackRefresh(int qid) {
  447.    
  448.   PushbackQueue * pbq = queue_list_[qid].pbq_;
  449.   int oldSessions = pbq->rlsList_->noMySessions(node_->nodeid());
  450.   if (!oldSessions) {
  451.     //set up refresh timers for a later time and return.
  452. // PushbackEvent * event = new PushbackEvent(PUSHBACK_CYCLE_TIME, PUSHBACK_REFRESH_EVENT, qid);
  453. // timer_->insert(event);
  454.     return;
  455.   }
  456.   
  457.   int noSessions = oldSessions;
  459.   if (MERGER_MODE == 1) {
  460.       pbq->rlsList_->mergeSessions(node_->nodeid());
  461.       noSessions = pbq->rlsList_->noMySessions(node_->nodeid());
  462.       
  463.       if (noSessions!=oldSessions) {
  464.       sprintf(prnMsg, " Some sessions merged. old = %d new = %dn",  oldSessions, noSessions);
  465.       printMsg(prnMsg,0);
  467.   //get rid of merged RLS's
  468.   RateLimitSession * listItem = pbq->rlsList_->first_;
  469.   while (listItem != NULL) {
  470.   if (listItem->origin_ == node_->nodeid() && listItem->merged_) {
  471.   pushbackCancel(listItem);
  472.   listItem = listItem->next_;
  473.   
  474.   }
  475.   }
  476.   } else {
  477.       sprintf(prnMsg, " No sessions merged. number = %dn", noSessions);
  478.       printMsg(prnMsg,0);
  479.       }
  480.   } else {
  481.   sprintf(prnMsg, "Number of sessions = %dn", noSessions);
  482.   printMsg(prnMsg,0);
  483.   }
  485.   double now = Scheduler::instance().clock();
  486.   
  487.   //check if some sessions need to be discarded because of rate-limiting too many sessions
  488.   RateLimitSession * listItem1 = pbq->rlsList_->first_;
  489.   while (noSessions > MAX_SESSIONS && listItem1 != NULL) {
  490.       int rank = pbq->rlsList_->rankRate(node_->nodeid(), listItem1->getArrivalRateForStatus());
  491.       if (listItem1->origin_ == node_->nodeid() && 
  492.   rank >= MAX_SESSIONS && (now - listItem1->startTime_) >= EARLIEST_TIME_TO_FREE) {
  493.   sprintf(prnMsg,"Releasing because of too many being rate-limitedn");
  494.   printMsg(prnMsg,0);
  495.   if (LOWER_BOUND_MODE == 1 && 
  496.       queue_list_[qid].idTree_->lowerBound_ < listItem1->getArrivalRateForStatus()) {
  497.       queue_list_[qid].idTree_->lowerBound_ = listItem1->getArrivalRateForStatus();
  498.   }
  499.   pushbackCancel(listItem1);
  500.   noSessions--;
  501.       }
  502.       listItem1 = listItem1->next_;
  503.   }
  504.   
  505.   double linkBW = pbq->getBW();
  506.   double arrRate = pbq->getRate();
  507.   double targetRate = linkBW/(1 - TARGET_DROPRATE);
  508.   
  509.   double totalRateLimitedArrivalRate = 0;
  510.   double totalLimit=0;
  511.   double lowerBound=-1;
  512.   RateLimitSession * listItem = pbq->rlsList_->first_;
  513.   while (listItem != NULL) {
  514.     if (listItem->origin_ == node_->nodeid() && !listItem->merged_) {
  515.       double sessionArrRate = listItem->getArrivalRateForStatus();
  516.       double sessionLimit = listItem->rlStrategy_->target_rate_;
  517.       totalRateLimitedArrivalRate+= sessionArrRate;
  518.       totalLimit+= (sessionArrRate > sessionLimit)? sessionLimit: sessionArrRate;
  519.       if (listItem->lowerBound_ < lowerBound || lowerBound == -1) {
  520.   lowerBound = listItem->lowerBound_;
  521.       }
  522.     }
  523.     listItem = listItem->next_;
  524.   }
  526.   if (LOWER_BOUND_MODE == 1) {
  527.   lowerBound = queue_list_[qid].idTree_->lowerBound_;
  528.   }
  530.   double excessRate = (arrRate - totalLimit + totalRateLimitedArrivalRate) - targetRate;
  531.   
  532.   sprintf(prnMsg,"arr=%g totalLimit=%g totalRateLimit=%g excess=%gn",  arrRate, totalLimit, 
  533.   totalRateLimitedArrivalRate, excessRate);
  534.   printMsg(prnMsg,0);
  535.   
  536.   if (excessRate < 0) {
  537.   sprintf(prnMsg, "Negative Excess Rate. Things maybe fine now.n");
  538.   printMsg(prnMsg,0);
  539.   //this would make all sessions go away after a while.
  540. #ifdef DEBUG
  541.   printf("Negative Excess Rate - time: %5.3fn", now);
  542. #endif
  543.   requiredLimit_ = 2*totalRateLimitedArrivalRate;
  544.   } else {
  545.   //Should we allow such an abrupt increase when the number of sessions 
  546.   // changes?
  547.   // How about: Let L be the requiredLimit.
  548.   // We need Sum (session arrival rate - L ) = excessRate
  549.   requiredLimit_ = (totalRateLimitedArrivalRate - excessRate)/noSessions;
  550.   if (requiredLimit_ < lowerBound) {
  551.   requiredLimit_ = lowerBound;
  552.   }
  553. #ifdef DEBUG
  554.       printf("New requiredLimit - time: %5.3f limit: %5.3f lowerBound:%5.3f n", now, requiredLimit_, lowerBound);
  555. #endif
  556.   }
  558.   sprintf(prnMsg,"Refresh. target=%g limit=%g floor=%gn", targetRate, requiredLimit_,
  559.   lowerBound);
  560.   printMsg(prnMsg,0);
  562.   //consider all sessions in ascending order of their arrival rate
  563.   for (int i=0; i<noSessions; i++) {
  564.   listItem = pbq->rlsList_->first_;
  565.   while (listItem != NULL ) {
  566.   if (listItem->origin_ == node_->nodeid() && 
  567.   pbq->rlsList_->rankSession(node_->nodeid(),listItem) == i) 
  568.   break;
  569.  listItem = listItem->next_;
  570.   }
  571.   if (listItem == NULL) {
  572.   printf("Error: Rank %d not foundn", i);
  573.   exit(0);
  574.   }
  575.   
  576.   double oldLimit = listItem->rlStrategy_->target_rate_;
  577.   double sendRate = listItem->getArrivalRateForStatus();
  578. #ifdef DEBUG
  579.   printf("time: %5.3f ID: %d sendRate %5.3f oldLimit %5.3f requiredLimit %5.3fn", now,
  580.  listItem->localID_, sendRate, oldLimit, requiredLimit_);
  581. #endif
  582.   //Session sending less than the limit.
  583.   if (sendRate < requiredLimit_) {
  584.   //if it has been sending less for "some" time.
  585.   if (now - listItem->refreshTime_ >= MIN_TIME_TO_FREE) {
  586. #ifdef DEBUG
  587. printf("time: %5.3f ID: %d refreshTime %5.3f MIN %d Cancel pushback B n", 
  588.    now, listItem->localID_, listItem->refreshTime_, MIN_TIME_TO_FREE);
  589. #endif
  590. pushbackCancel(listItem);       //cancel rate-limiting
  591. requiredLimit_+= (requiredLimit_ - sendRate)/(noSessions - i - 1);
  592. i--; noSessions--;
  593.   } 
  594.   else {
  595.   //refresh upstream with double of max(sending rate, old limit)
  596.   //just using sending rate, limits the amount an aggregate can grow till next refresh
  597.   //using just old limit is tricky when different aggregates have different limits.
  598.   //at the same time, we would prefer not to loosen the hold too much in one step.
  599. #ifdef DEBUG
  600.   printf("time: %5.3f ID: %d double limitn", now, listItem->localID_);
  601. #endif
  602.   double maxR = sendRate>oldLimit? sendRate: oldLimit;
  603.   if (now - listItem->refreshTime_ <= PRIMARY_WAITING_ZONE) {
  604.   sprintf(prnMsg,"Waiting Zone 1: sendRate=%g oldLimit=%gn", sendRate, oldLimit);
  605.   printMsg(prnMsg,0);
  606.   }
  607.   else {
  608.   sprintf(prnMsg,"Waiting Zone 2: sendRate=%g oldLimit=%gn", sendRate, oldLimit);
  609.   printMsg(prnMsg,0);
  610.   maxR *= 1.5;
  611.   }
  612.   if (maxR < requiredLimit_) {
  613.   listItem->setLimit(maxR);
  614.   requiredLimit_ += (requiredLimit_ - maxR)/(noSessions - i - 1);
  615.   } 
  616.   else {
  617.   listItem->setLimit(requiredLimit_);
  618.   }
  619.   
  620.   if (listItem->pushbackON_) 
  621.   refreshUpstreamLimits(listItem);
  622.   }
  623.   }
  624.       else {
  625.   //change the rate limit most half way.
  626.   double newLimit;
  627.   if (oldLimit > 1.25 * requiredLimit_ || oldLimit ==0) 
  628.   newLimit = requiredLimit_;
  629.   else 
  630.   newLimit = 0.5*requiredLimit_ + 0.5*oldLimit;
  631.   
  632.   if (newLimit < lowerBound) 
  633.   newLimit = lowerBound;
  634.   
  635.   listItem->refreshed();
  636.   listItem->setLimit(newLimit);
  637.   if (listItem->pushbackON_) 
  638.   refreshUpstreamLimits(listItem);
  639. #ifdef DEBUG
  640.   printf("time: %5.3f ID: %d newLimit %5.3f oldLimit %5.3f requiredLimit %5.3fn", 
  641.  now, listItem->localID_, newLimit, oldLimit, requiredLimit_);
  642. #endif
  643.       }
  644.   }    
  645.   
  646.   //setup refresh timer again
  647.   noSessions = pbq->rlsList_->noMySessions(node_->nodeid());
  648.   if (noSessions) {
  649.     PushbackEvent * event = new PushbackEvent(PUSHBACK_CYCLE_TIME, PUSHBACK_REFRESH_EVENT, qid);
  650.     timer_->insert(event);
  651.   }
  652. }
  655. void
  656. PushbackAgent::pushbackCancel(RateLimitSession * rls) {
  657.  
  658.   sprintf(prnMsg,"Stopping rate-limiting for aggregate: ");
  659.   printMsg(prnMsg,0);
  660.   rls->aggSpec_->print();
  661.   fflush(stdout);
  663.   #ifdef DEBUG
  664.     double now = Scheduler::instance().clock();
  665.     printf("time: %5.3f ID: %d Cancel pushback Cn", now, rls->localID_);
  666.   #endif
  668.   if (rls->pushbackON_) {
  669.     LoggingDataStructNode * lgdsNode = rls->logData_->first_;
  670.     while (lgdsNode != NULL) {
  671.       PushbackMessage * msg = new PushbackCancelMessage(node_->nodeid(), lgdsNode->nid_, 
  672. rls->localQID_, rls->localID_);
  673.       sendMsg(msg);
  674.       lgdsNode = lgdsNode->next_;
  675.     }
  676.   }
  677.   
  678.   //remove all events that point to this rls.
  679.   timer_->removeEvents(rls);
  680.   //local cancellation here.
  681.   queue_list_[rls->localQID_].pbq_->rlsList_->endSession(rls);
  682.   
  683. }
  685. //######################## Message Receiving Code #####################
  687. void 
  688. PushbackAgent::recv(Packet * pkt, Handler * h) {
  689.   
  690.   hdr_pushback * hdr_push = ((hdr_pushback*)pkt)->access(pkt);
  691.   PushbackMessage * msg = hdr_push->msg_;
  692.   
  693.   sprintf(prnMsg, " %s msg from %dn",  PushbackMessage::type(msg), msg->senderID_);
  694.   printMsg(prnMsg,0);
  695.  
  696.   switch (msg->msgID_) {
  697.   case PUSHBACK_REQUEST_MSG : processPushbackRequest((PushbackRequestMessage *)msg);
  698.     break;
  699.   case PUSHBACK_STATUS_MSG : processPushbackStatus((PushbackStatusMessage *) msg);
  700.     break;
  701.   case PUSHBACK_REFRESH_MSG : processPushbackRefresh((PushbackRefreshMessage *) msg);
  702.     break;
  703.   case PUSHBACK_CANCEL_MSG : processPushbackCancel((PushbackCancelMessage *) msg);
  704.     break;
  705.   default: fprintf(stderr,"PBA: %s Undefined Message ID %dn", name(),msg->msgID_);
  706.   } 
  707.   
  708.   delete(msg);
  709. }
  711. void 
  712. PushbackAgent::processPushbackRequest(PushbackRequestMessage * msg) {
  713.   
  714.   int qid = getQID(msg->senderID_);
  715.   sprintf(prnMsg, " pushback request from %d for qid=%d limit=%gn", msg->senderID_, 
  716.  qid, msg->limit_);
  717.   printMsg(prnMsg,0);
  718.  
  719.   AggSpec * aggSpec = msg->aggSpec_;
  720.   if (queue_list_[qid].pbq_->rlsList_->containsAggSpec(aggSpec)) {
  721.   fprintf(stdout,"PBA: %s got a pushback req for agg I already rate-limit. 
  722. Feature not yet Implementedn",name()); 
  723.   exit(-1);
  724.   }
  725.   
  726.   RateLimitSession * rls = new RateLimitSession(aggSpec, msg->limit_, msg->senderID_, qid, 
  727. msg->qid_, msg->rlsID_, msg->depth_+1, 
  728. RATE_LIMIT_TIME_DEFAULT, -1, node_, rtLogic_);
  729.   queue_list_[qid].pbq_->rlsList_->insert(rls);
  731.   //pushback propagation check if there are valid upstream neighbors && enable_pushback_
  732.   if (rls->logData_->count_ && enable_pushback_) {
  733.     PushbackEvent * event = new PushbackEvent(PUSHBACK_CHECK_TIME, PUSHBACK_CHECK_EVENT, rls);
  734.     timer_->insert(event);
  735.   }
  736. }
  739. void 
  740. PushbackAgent::processPushbackStatus(PushbackStatusMessage * msg) {
  742.   int qid = msg->qid_;
  744.   if (!checkQID(qid)) {
  745.     sprintf(prnMsg, " Got invalid qid from %d in status messagen",  msg->senderID_);
  746.     printMsg(prnMsg,0);
  747.     exit(-1);
  748.   }
  750.   RateLimitSession * rls = queue_list_[qid].pbq_->rlsList_->getSessionByLocalID(msg->rlsID_);
  751.   
  752.   if (rls == NULL) {
  753.     sprintf(prnMsg, " session %d not foundn",  msg->rlsID_);
  754.     printMsg(prnMsg,0);
  755.     exit(-1);
  756.   }
  757.   
  758.   //increase your height if you need to.
  759.   if (msg->height_ + 1 > rls->heightInPTree_) {
  760.     rls->heightInPTree_ = msg->height_ + 1;
  761.     sprintf(prnMsg, " height increased to %dn",  rls->heightInPTree_);
  762.     printMsg(prnMsg,0);
  763.   }
  765.   rls->logData_->registerStatus(msg->senderID_, msg->arrivalRate_);
  766.   sprintf(prnMsg, " got rate %gn",  msg->arrivalRate_);
  767.   printMsg(prnMsg,0);
  769.   //send status if you are not root.
  770.   if (rls->origin_!= node_->nodeid()) {
  771.     // 1. check to see if status from all the upstream neighbors has arrived.
  772.     // 2. if yes, then send status downstream.
  773.     int gotAll = rls->logData_->consolidateStatus();
  774.     if (gotAll==1) {
  775.       //send status down
  776.       double rate = rls->logData_->statusArrivalRateAll_;
  777.       PushbackMessage * msg = new PushbackStatusMessage(node_->nodeid(), rls->origin_, 
  778. rls->remoteQID_, rls->remoteID_, 
  779.       rate, rls->heightInPTree_);
  780.       sendMsg(msg);
  781.       timer_->cancelStatus(rls);
  782.       //reset status arrivals.
  783.       rls->logData_->resetStatus();
  784.     }
  785.   }
  786. }
  788. void
  789. PushbackAgent::processPushbackRefresh(PushbackRefreshMessage *msg) {
  791.   int qid = getQID(msg->senderID_);
  792.   sprintf(prnMsg, " pushback refresh from %d for qid=%d with limit=%gn",  msg->senderID_, qid, msg->limit_);
  793.   printMsg(prnMsg,0);
  795.   RateLimitSession * rls = queue_list_[qid].pbq_->rlsList_->getSessionByRemoteID(msg->rlsID_);
  796.   
  797.   if (rls == NULL) {
  798.     sprintf(prnMsg, " session %d not foundn",  msg->rlsID_);
  799.     printMsg(prnMsg,0);
  800.     exit(-1);
  801.   }
  803.   //1. change your own rate limit
  804.   rls->setAggSpec(msg->aggSpec_);
  805.   delete(msg->aggSpec_);
  806.   double newLimit = msg->limit_;
  807.   rls->setLimit(newLimit);
  809.   //2. if pushback has been propagated send out refreshes upstream with new limits
  810.   if (rls->pushbackON_) {
  811.     refreshUpstreamLimits(rls);
  812.   }
  813.   
  814.   //3. set up status timer.
  815.   PushbackEvent * event = new PushbackEvent(PUSHBACK_CYCLE_TIME - 0.1*rls->depthInPTree_, 
  816.      PUSHBACK_STATUS_EVENT, rls);
  817.   timer_->insert(event);
  818. }
  820. void 
  821. PushbackAgent::processPushbackCancel(PushbackCancelMessage *msg) {
  822.  
  823.   int qid = getQID(msg->senderID_);
  824.   sprintf(prnMsg, " pushback cancel from %d for queue index %dn",  msg->senderID_, qid);
  825.   printMsg(prnMsg,0);
  827.   RateLimitSession * rls = queue_list_[qid].pbq_->rlsList_->getSessionByRemoteID(msg->rlsID_);
  828.   
  829.   if (rls == NULL) {
  830.     sprintf(prnMsg, " session %d not foundn",  msg->rlsID_);
  831.     printMsg(prnMsg,0);
  832.     exit(-1);
  833.   }
  834.   pushbackCancel(rls);
  836. }
  838. void 
  839. PushbackAgent::refreshUpstreamLimits(RateLimitSession * rls) {
  841.   double totalRate =  rls->rlStrategy_->target_rate_;
  842.   int count = rls->logData_->count_;
  843.   double fairShare = totalRate/count;
  844.   int done = count;
  845.   double arrRate = rls->getArrivalRateForStatus();
  846.   sprintf(prnMsg, "Sending refresh messages to %d nodes. Limit = %g arrRate = %gn", count, totalRate, arrRate);
  847.   printMsg(prnMsg,0); 
  848.   
  849.   int excess = 0;
  850.   if (totalRate > arrRate) {
  851.   excess = 1;
  852.   }
  854.   //max-min allocation of limit.
  855.   while (done != 0) {
  856.     LoggingDataStructNode * lgdsNode = rls->logData_->first_;
  857.     int countThisRound=0;
  858.     while (lgdsNode != NULL) {
  859.       double rate;
  860.       rate = lgdsNode->statusArrivalRate_;
  861.       if (rate <= fairShare && !lgdsNode->sentRefresh_) {
  862. AggSpec * aggSpec = rls->aggSpec_->clone();
  863. PushbackMessage * msg;
  864. if (rate < fairShare/2.0) {
  865.   msg = new PushbackRefreshMessage(node_->nodeid(), lgdsNode->nid_, rls->localQID_, 
  866.    rls->localID_, aggSpec, INFINITE_LIMIT);
  867.   lgdsNode->sentRefresh(INFINITE_LIMIT);
  868. }
  869. else if (!excess) {
  870.   msg = new PushbackRefreshMessage(node_->nodeid(), lgdsNode->nid_, rls->localQID_, 
  871.    rls->localID_, aggSpec, rate);
  872.   lgdsNode->sentRefresh(rate);
  873. }
  874. else {
  875.   msg = new PushbackRefreshMessage(node_->nodeid(), lgdsNode->nid_, rls->localQID_, 
  876.    rls->localID_, aggSpec, fairShare);
  877.   lgdsNode->sentRefresh(fairShare);
  878.   rate = fairShare;
  879. }
  880. sendMsg(msg);
  881. countThisRound++;
  882. done--;
  883. totalRate -= rate;
  884.       }
  885.       lgdsNode = lgdsNode->next_;
  886.     }
  887.     if (done == 0) break;
  888.     if (countThisRound==0) {
  889.       //allocate fairshare to everyone and end.
  890.       LoggingDataStructNode * lgdsNode = rls->logData_->first_;
  891.       while (lgdsNode != NULL) {
  892. if (!lgdsNode->sentRefresh_) {
  893.   AggSpec * aggSpec = rls->aggSpec_->clone();
  894.   PushbackMessage * msg = new PushbackRefreshMessage(node_->nodeid(), lgdsNode->nid_, 
  895.      rls->localQID_, rls->localID_, 
  896.      aggSpec, fairShare);
  897.     lgdsNode->sentRefresh(fairShare);
  898.     sendMsg(msg);
  899.     done--;
  900.     totalRate-=fairShare;
  901. }
  902. lgdsNode = lgdsNode->next_;
  903.       }
  904.     }
  905.     else {
  906.       fairShare = totalRate/done;
  907.     }
  908.   }
  909.   
  910.   //reset all the sentRefresh bits
  911.   LoggingDataStructNode * lgdsNode = rls->logData_->first_;
  912.   while (lgdsNode != NULL) {
  913.     lgdsNode->sentRefresh_ = 0;
  914.     lgdsNode = lgdsNode->next_;
  915.   }
  917. }
  919. int
  920. PushbackAgent::getQID(int sender) {
  921.   
  922.   Tcl& tcl = Tcl::instance();
  923.   intResult_ = -1;
  924.   int index = 0;
  925.   // there gotta be better ways of doing this;  todoLater.
  926.   // like make Tcl call you back and set a variable using command.
  927.   for (; index <last_index_; index++) {
  928.     tcl.evalf("%s set intResult_ [%s check-queue %d %d %s]", name(), name(), 
  929.       node_->nodeid(), sender , queue_list_[index].pbq_->name());
  930.     if (intResult_ == 1) break;
  931.   }
  932.   
  933.   if (index == last_index_) {
  934.     sprintf(prnMsg, " right queue not foundn");
  935.     printMsg(prnMsg,0);
  936.     exit(-1);
  937.   }
  939.   return index;
  940. }
  942. void
  943. PushbackAgent::sendMsg(PushbackMessage * msg) {
  944.   
  945.   Tcl& tcl = Tcl::instance();
  946.   
  947.   dst_.addr_ = msg->targetID_;
  948.   //this assumes that all pushback agents have port zero.
  949.   
  950.   tcl.evalf("%s set intResult_ [%s get-pba-port %d]", name(), name(),dst_.addr_ );
  952.   if ( intResult_ == -1 ) {
  953.     fprintf(stderr,"PBA: %s Pushback Agent not found on Node %dn", name(), dst_.addr_);
  954.     return;
  955.   }
  956.   dst_.port_ = intResult_;
  957.   Packet *pkt = allocpkt();
  958.   hdr_pushback * hdr_push = ((hdr_pushback*)pkt)->access(pkt);
  959.   hdr_push->msg_ = msg;
  960.     
  961.   sprintf(prnMsg, " sent %s message to %d.%dn", PushbackMessage::type(msg), dst_.addr_, dst_.port_);
  962.   printMsg(prnMsg,4);
  963.   send(pkt,0);
  964. }
  966. void 
  967. PushbackAgent::printMsg(char * msg, int msgLevel) {
  968.   
  969.   if (msgLevel < debugLevel) {
  970.     if (verbose_) printf("PBA:%d (%g) %s", node_->nodeid(), Scheduler::instance().clock(), msg);
  971.     fflush(stdout);
  972.   }
  973. }
  975. int 
  976. PushbackAgent::checkQID(int qid) {
  977.   if (qid < 0 || qid >= last_index_) 
  978.     return 0;
  979.   else 
  980.     return 1;
  981. }
  983. //decide whether to accept a merger involving "count" aggregates, 
  984. //the number of bits in the resultant aggregate would be "bits"
  985. //the aggregate is being broadended by "bitsDiff" (measured from shortest prefix)
  986. int
  987. PushbackAgent::mergerAccept(int count, int bits, int bitsDiff) {
  988.   
  989.   //todo: think of a smarter way.
  990.   //currently merge if bits < some value.
  991.   //return (bits <= MIN_BITS_FOR_MERGER);
  993.   return 0;
  994. }
  995.   
  996. // ############################### PushbackTimer Methods ############################
  998. void 
  999. PushbackTimer::expire(Event *e) {
  1000.  
  1001.   if (firstEvent_ == NULL) {
  1002.     printf("PushbackTimer: No event found on expiryn");
  1003.     exit(-1);
  1004.   }
  1005.   
  1006.   PushbackEvent * event = firstEvent_;
  1007.   firstEvent_= firstEvent_->next_;
  1008.   schedule();
  1010.   agent_->timeout(event);
  1011.   delete(event);
  1012. }
  1013.  
  1015. void
  1016. PushbackTimer::insert(PushbackEvent * event) {
  1018.   sprintf(agent_->prnMsg,"%s timer setn", PushbackEvent::type(event));
  1019.   agent_->printMsg(agent_->prnMsg,4);
  1020.   if (firstEvent_ == NULL) {
  1021.     firstEvent_ = event;
  1022.     schedule();
  1023.     return;
  1024.   }
  1026.   if (event->time_ < firstEvent_->time_) {
  1027.     event->setSucc(firstEvent_);
  1028.     firstEvent_=event;
  1029.     schedule();
  1030.     return;
  1031.   }
  1033.   PushbackEvent * listItem = firstEvent_;
  1034.   while (listItem->next_!=NULL && listItem->next_->time_ <= event->time_) {
  1035.     listItem = listItem->next_;
  1036.   }
  1037.   
  1038.   event->setSucc(listItem->next_);
  1039.   listItem->setSucc(event);
  1040.  
  1041.   //comment the sanity check out later
  1042.   sanityCheck();
  1043.   
  1044.   return;
  1045. }
  1047. void
  1048. PushbackTimer::removeEvents(RateLimitSession * rls) {
  1049.   if (firstEvent_==NULL) return;
  1050.   while (firstEvent_!= NULL && firstEvent_->rls_==rls) {
  1051.     cancel();
  1052.     PushbackEvent * event = firstEvent_;
  1053.     firstEvent_=firstEvent_->next_;
  1054.     delete(event);
  1055.     schedule();
  1056.   }
  1057.   if (firstEvent_==NULL) return;
  1059.   PushbackEvent * previous = firstEvent_;
  1060.   PushbackEvent * current = firstEvent_->next_;
  1061.   while (current!=NULL) {
  1062.     if (current->rls_==rls) {
  1063.       previous->next_=current->next_;
  1064.       delete(current);
  1065.       current = previous->next_;
  1066.       continue;
  1067.     }
  1068.     previous=current;
  1069.     current=current->next_;
  1070.   }
  1072. }
  1073.       
  1074. void 
  1075. PushbackTimer::schedule() {
  1077.   if (firstEvent_== NULL) {
  1078.     sprintf(agent_->prnMsg,"Timer: Nothing to schedulen");
  1079.     agent_->printMsg(agent_->prnMsg, 0);
  1080.     return;
  1081.   }
  1083.   resched(firstEvent_->time_ - Scheduler::instance().clock());
  1084. }
  1086. void 
  1087. PushbackTimer::cancelStatus(RateLimitSession * rls) {
  1088.   
  1089.   if (firstEvent_==NULL) {
  1090.     sprintf(agent_->prnMsg, " Error timer list emptyn");
  1091.     agent_->printMsg(agent_->prnMsg, 0);
  1092.     //return; 
  1093.     exit(-1);
  1094.   }
  1095.   
  1096.   if (firstEvent_->eventID_==PUSHBACK_STATUS_EVENT && firstEvent_->rls_==rls) {
  1097.     cancel();
  1098.     PushbackEvent * event = firstEvent_;
  1099.     firstEvent_=firstEvent_->next_;
  1100.     delete(event);
  1101.     schedule();
  1102.     return;
  1103.   }
  1104.   
  1105.   PushbackEvent * previous = firstEvent_;
  1106.   PushbackEvent * current = firstEvent_->next_;
  1107.   
  1108.   while (current!=NULL) {
  1109.     if (current->eventID_ == PUSHBACK_STATUS_EVENT && current->rls_==rls) {
  1110.       previous->next_=current->next_;
  1111.       delete(current);
  1112.       return;
  1113.     } 
  1114.     previous=current;
  1115.     current=current->next_;
  1116.   }
  1118.   sprintf(agent_->prnMsg, "Error status timer not foundn");
  1119.   agent_->printMsg(agent_->prnMsg, 0);
  1120.   exit(-1);
  1121. }
  1123. int 
  1124. PushbackTimer::containsRefresh(int qid) {
  1125.   PushbackEvent * listItem = firstEvent_;
  1126.   while (listItem!=NULL) {
  1127.     if (listItem->eventID_ == PUSHBACK_REFRESH_EVENT && listItem->qid_==qid) 
  1128.       return 1;
  1129.     listItem = listItem->next_;
  1130.   }
  1131.   return 0;
  1132. }
  1134. void 
  1135. PushbackTimer::sanityCheck() {
  1137.   if (firstEvent_==NULL || firstEvent_->next_ == NULL) return;
  1138.   
  1139.   PushbackEvent * listItem = firstEvent_;
  1140.   while (listItem->next_!=NULL) {
  1141.     if (listItem->time_ > listItem->next_->time_) {
  1142.       sprintf(agent_->prnMsg, "Sanity Check Failedn");
  1143.       agent_->printMsg(agent_->prnMsg, 0);
  1144.       exit(-1);
  1145.     }
  1146.     listItem = listItem->next_;
  1147.   }
  1149. }
  1150.