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

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

通讯编程

开发平台：

Visual C++

rmst_filter.cc：源码内容
  1. //
  2. // rmst_filter.cc  : RmstFilter Class Methods
  3. // authors         : Fred Stann
  4. //
  5. // Copyright (C) 2003 by the University of Southern California
  6. // $Id: rmst_filter.cc,v 1.3 2005/09/13 04:53:48 tomh Exp $
  7. //
  8. // This program is free software; you can redistribute it and/or
  9. // modify it under the terms of the GNU General Public License,
  10. // version 2, as published by the Free Software Foundation.
  11. //
  12. // This program is distributed in the hope that it will be useful,
  13. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  14. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15. // GNU General Public License for more details.
  16. //
  17. // You should have received a copy of the GNU General Public License along
  18. // with this program; if not, write to the Free Software Foundation, Inc.,
  19. // 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA.
  20. //
  21. // Linking this file statically or dynamically with other modules is making
  22. // a combined work based on this file.  Thus, the terms and conditions of
  23. // the GNU General Public License cover the whole combination.
  24. //
  25. // In addition, as a special exception, the copyright holders of this file
  26. // give you permission to combine this file with free software programs or
  27. // libraries that are released under the GNU LGPL and with code included in
  28. // the standard release of ns-2 under the Apache 2.0 license or under
  29. // otherwise-compatible licenses with advertising requirements (or modified
  30. // versions of such code, with unchanged license).  You may copy and
  31. // distribute such a system following the terms of the GNU GPL for this
  32. // file and the licenses of the other code concerned, provided that you
  33. // include the source code of that other code when and as the GNU GPL
  34. // requires distribution of source code.
  35. //
  36. // Note that people who make modified versions of this file are not
  37. // obligated to grant this special exception for their modified versions;
  38. // it is their choice whether to do so.  The GNU General Public License
  39. // gives permission to release a modified version without this exception;
  40. // this exception also makes it possible to release a modified version
  41. // which carries forward this exception.
  43. #include "rmst_filter.hh"
  45. char *rmstmsg_types[] = {"INTEREST", "POSITIVE REINFORCEMENT",
  46.                      "NEGATIVE REINFORCEMENT", "DATA",
  47.                      "EXPLORATORY DATA", "PUSH EXPLORATORY DATA",
  48.                      "CONTROL", "REDIRECT"};
  50. #ifdef NS_DIFFUSION
  51. class DiffAppAgent;
  52. #endif // NS_DIFFUSION
  54. #ifdef NS_DIFFUSION
  55. static class RmstFilterClass : public TclClass {
  56. public:
  57.   RmstFilterClass() : TclClass("Application/DiffApp/RmstFilter") {}
  58.   TclObject* create(int argc, const char*const* argv) {
  59.     return(new RmstFilter());
  60.   }
  61. } class_rmst_filter;
  63. int RmstFilter::command(int argc, const char*const* argv) {
  64.   //Tcl& tcl =  Tcl::instance();
  65.   if (argc == 2) {
  66.     if (strcmp(argv[1], "start") == 0) {
  67.       run();
  68.       return (TCL_OK);
  69.     }
  70.   }
  71.   return (DiffApp::command(argc, argv));
  72. }
  73. #endif // NS_DIFFUSION
  75. class ReinfMessage {
  76. public:
  77.   int32_t rdm_id_;
  78.   int32_t pkt_num_;
  79. };
  82. // RmstFilterCallback::recv
  83. // Called by diffusion core when a message is available for this filter.
  84. // RmstFilterCallback is derived from the abstract class FilterCallback.
  85. // A pointer to the FilterCallback class is required in the API method "addFilter."
  87. void RmstFilterCallback::recv(Message *msg, handle h)
  88. {
  89.   app_->recv(msg, h);
  90. }
  93. // RmstFilter::recv
  94. //
  95. // Called by the Callback::recv method.
  97. void RmstFilter::recv(Message *msg, handle h)
  98. {
  99.   // Process the message handed to us by the core.
  100.   // If true is returned we forward the message. Otherwise it dies here.
  101.   if(processMessage(msg))
  102.     ((DiffusionRouting *)dr_)->sendMessage(msg, h);
  103. }
  105. // RmstFilter::processMessage
  106. //
  107. // Called by the RmstFilter::recv method when this filter gets a message.
  109. bool RmstFilter::processMessage(Message *msg)
  110. {
  111.   NRSimpleAttribute<int> *rmst_id_attr = NULL;
  112.   NRSimpleAttribute<int> *frag_attr = NULL;
  113.   NRSimpleAttribute<int> *pkts_sent_attr = NULL;
  114.   NRSimpleAttribute<int> *tsprt_ctl_attr = NULL;
  115.   NRSimpleAttribute<void *> *reinf_attr = NULL;
  116.   NRSimpleAttribute<int> *nrscope = NULL;
  117.   NRSimpleAttribute<int> *nr_class = NULL;
  118.   NRAttrVec *data;
  120.   Key2ExpLog::iterator exp_iterator;
  121.   Int2Rmst::iterator rmst_iterator;
  122.   int rmst_no;
  123.   int frag_no;
  124.   int class_type;
  125.   int rmst_ctl_type;
  126.   union LlToInt key;
  127.   Rmst *rmst_ptr;
  129.   // If this is a message that uses the transport layer, we process it.
  130.   // Otherwise we send it back to the core (by returning true).
  131.   tsprt_ctl_attr = RmstTsprtCtlAttr.find(msg->msg_attr_vec_);
  132.   if (!tsprt_ctl_attr){
  133.     DiffPrint(DEBUG_SOME_DETAILS, "RmstFilter got non-transport messagen");
  134.     return true;
  135.   }
  136.   rmst_ctl_type = tsprt_ctl_attr->getVal();
  138.   DiffPrint(DEBUG_IMPORTANT, "RmstFilter::processMessage got a");
  140.   if (msg->new_message_)
  141.     DiffPrint(DEBUG_IMPORTANT, " new (%d) ", msg->msg_type_);
  142.   else
  143.     DiffPrint(DEBUG_IMPORTANT, "n old (%d) ", msg->msg_type_);
  145.   if (msg->last_hop_ != LOCALHOST_ADDR)
  146.     DiffPrint(DEBUG_IMPORTANT, "%s message from %d to %dn",
  147.       rmstmsg_types[msg->msg_type_],
  148.       msg->last_hop_, msg->next_hop_);
  149.   else
  150.     DiffPrint(DEBUG_IMPORTANT, "%s message from local agentn",
  151.       rmstmsg_types[msg->msg_type_]);
  153.   // We only care about messages we haven't seen before,
  154.   // but we generally let other filters get them (because they may need them).
  155.   // However, if this is an old DATA message arriving at a sink, the sink may
  156.   // negatively reinforce a reinforced path.  This is because we withold the 
  157.   // new messages until we get the entire blob.  The old message is the result
  158.   // of a lost ACK when using SMAC with ARQ.
  159.   if (!msg->new_message_ && msg->msg_type_ == DATA
  160.       && rmst_ctl_type == RMST_RESP){
  161.     DiffPrint(DEBUG_SOME_DETAILS, 
  162.       "  Sink got an old DATA message from node %dn", msg->last_hop_);
  163.     data = msg->msg_attr_vec_;
  164.     rmst_id_attr = RmstIdAttr.find(data);
  165.     if (!rmst_id_attr){
  166.       DiffPrint(DEBUG_SOME_DETAILS,
  167.         "  Filter received a bad transport packet!n");
  168.       return false;
  169.     }
  170.     rmst_no = rmst_id_attr->getVal();
  171.     // Find the rmst.
  172.     rmst_iterator = rmst_map_.find(rmst_no);
  173.     if(rmst_iterator == rmst_map_.end()){
  174.       DiffPrint(DEBUG_IMPORTANT,
  175.         "  couldn't find DB entry for Rmst %dn", rmst_no);
  176.       return false;
  177.     }
  178.     else{
  179.       rmst_ptr = (*rmst_iterator).second;
  180.       if ( (local_sink_) && (msg->last_hop_ == rmst_ptr->last_hop_) ){
  181.         // This is the case where SMAC sent the same DATA message twice to a sink.
  182.         // We suppress this message so we don't kill our reinforced path.
  183.         DiffPrint(DEBUG_IMPORTANT,
  184.           "  We suppress old DATA message from smac retransmission!n");
  185.         return false;
  186.       }
  187.       else
  188.         return true;
  189.     }
  190.   }
  191.   else if (!msg->new_message_)
  192.     return true;
  194.   // When we get Rmst Fragments we must sync the local cache!
  195.   if ( (rmst_ctl_type == RMST_RESP) && 
  196.        ((msg->msg_type_ == DATA) || (msg->msg_type_ == EXPLORATORY_DATA)) ){
  197.     rmst_ptr = syncLocalCache(msg);
  198.     // syncLocalCache will return NULL if the 
  199.     // attribute set doesn't make sense.
  200.     if (rmst_ptr == NULL)
  201.       return false;
  202.     rmst_no = rmst_ptr->rmst_no_;
  203.     // Mark the time we got some kind of data.
  204.     GetTime (&last_data_rec_);
  205.   }
  207.   //  New exploratory messages are entered into the exp_map_,
  208.   //  so that we can find the last hop if it gets reinforced.
  209.   //  Positive reinforcement messages are used to find the 
  210.   //  corresponding message in the exp_map_, so we know the
  211.   //  current reinforced path to the source of an rmst.
  212.   switch (msg->msg_type_){
  214.     case(EXPLORATORY_DATA):
  215.       ExpLog exp_msg;
  217.       DiffPrint(DEBUG_LOTS_DETAILS,
  218.         "  Exploratory_Msg: ptk_num = %x, rdm_id_ = %x, last_hop = %dn",
  219.         msg->pkt_num_, msg->rdm_id_, msg->last_hop_);
  221.       // Put the ID for this Exploratory message, along with its last hop,
  222.       // into the exp_map_.  If this message gets reinforced, we will be
  223.       // able to identify the next hop in the back channel.
  224.       DiffPrint(DEBUG_SOME_DETAILS,
  225.         "  Exploratory message for Reliable transport Id = %dn", rmst_no);
  226.       key.int_val_[0] = msg->pkt_num_;
  227.       key.int_val_[1] = msg->rdm_id_;
  228.       DiffPrint(DEBUG_LOTS_DETAILS, "  Key = %llxn", key.ll_val_);
  229.       exp_msg.rmst_no_ = rmst_no;
  230.       exp_msg.last_hop_ = msg->last_hop_;
  231.       exp_map_.insert(Key2ExpLog::value_type(key.ll_val_, exp_msg));
  233.       // If this is a new exploratory message arriving at a sink,
  234.       // we assume that this path will get reinforced by the 
  235.       // gradient filter. Sinks don't get positive reinforcement
  236.       // messages, so we must record last_hop_ now.
  237.       if (local_sink_){
  238.         rmst_ptr->last_hop_ = msg->last_hop_; 
  239.         if (rmst_ptr->reinf_){
  240.           DiffPrint(DEBUG_IMPORTANT, "  got a new path exploratory msg at sink.n");
  241.           rmst_ptr->wait_for_new_path_ = true;
  242.         }
  243.         else{
  244.           rmst_ptr->reinf_ = true;
  245.           DiffPrint(DEBUG_IMPORTANT, "  got an initial exploratory msg at sink.n");
  246.         }
  247.         DiffPrint(DEBUG_IMPORTANT, "  set last_hop for rmst %d to %dn",
  248.           rmst_no, rmst_ptr->last_hop_);
  249.         rmst_ptr->pkts_rec_ = 0; 
  250.         rmst_ptr->last_hop_pkts_sent_ = 0; 
  251.       }
  252.       else{
  253.         // If this is not a sink we reset the base fragment that
  254.         // we look for holes from.
  255.         DiffPrint(DEBUG_LOTS_DETAILS,
  256.           "  intermediate node resets sync_base_ and reinf_.n");
  257.         frag_attr = RmstFragAttr.find(msg->msg_attr_vec_);
  258.         frag_no = frag_attr->getVal();
  259.         rmst_ptr->sync_base_ = frag_no;
  260.         if(rmst_ptr->reinf_)
  261.           rmst_ptr->reinf_ = false;
  262.         rmst_ptr->last_hop_ = 0;
  263.         rmst_ptr->pkts_sent_ = 0;
  264.         rmst_ptr->pkts_rec_ = 0;
  265.         rmst_ptr->last_hop_pkts_sent_ = 0; 
  266.         rmst_ptr->naks_rec_ = 0; 
  267.       }
  269.       // If this is not a sink and a watchdog timer is active, we cancel
  270.       // it because we may not end up on the new reinforced path. We
  271.       // don't want to look for fragments that will never arrive.
  272.       if ((rmst_ptr->watchdog_active_) && (!local_sink_) 
  273.         && (!rmst_ptr->local_source_)){
  274.         rmst_ptr->cancel_watchdog_ = true;
  275.         rmst_ptr->cleanHoleMap();
  276.       }
  278.       // We always forward exploratory data.
  279.       return(true);
  280.       break;
  282.     case(DATA):
  284.       if (rmst_ctl_type != RMST_RESP){
  285.         processCtrlMessage(msg);
  286.         // We don't let Rmst control messages go to the gradient or other filters.
  287.         return false;
  288.       }
  290.       // We have a normal DATA packet.
  291.       rmst_ptr->pkts_rec_++;
  293.       // If we got the upstream send count - update it in Rmst.
  294.       pkts_sent_attr = RmstPktsSentAttr.find(msg->msg_attr_vec_);
  295.       if (pkts_sent_attr){
  296.         rmst_ptr->last_hop_pkts_sent_ = pkts_sent_attr->getVal();
  297.         DiffPrint(DEBUG_SOME_DETAILS,
  298.           "processMessage:: got last_hop_pkts_sent_ = %d packetsn",
  299.           rmst_ptr->last_hop_pkts_sent_);
  300.         if ( (rmst_ptr->last_hop_pkts_sent_ > 20) && 
  301.              (rmst_ptr->pkts_rec_ < (rmst_ptr->last_hop_pkts_sent_ * BLACKLIST_THRESHOLD)) ){
  302.           Blacklist::iterator black_list_iterator;
  303.           black_list_iterator = black_list_.begin();
  304.           while(black_list_iterator != black_list_.end()){
  305.             if(*black_list_iterator == rmst_ptr->last_hop_)
  306.             break;
  307.             black_list_iterator++;
  308.           }
  309.           if(black_list_iterator == black_list_.end()){
  310.             DiffPrint(DEBUG_IMPORTANT, "Adding node %d to black_list_ !!n",
  311.               rmst_ptr->last_hop_);
  312.             black_list_.push_front(rmst_ptr->last_hop_);
  313.             ((DiffusionRouting *)dr_)->addToBlacklist(rmst_ptr->last_hop_);
  314.             // Now send an EXP_REQ!
  315.             sendExpReqUpstream(rmst_ptr);
  316.             rmst_ptr->sent_exp_req_ = true;
  317.             GetTime(&rmst_ptr->exp_req_time_);
  318.             // We need to send a negative reinforcement on blacklisted link!
  319.             Message *neg_reinf_msg;
  320.             neg_reinf_msg = new Message(DIFFUSION_VERSION, NEGATIVE_REINFORCEMENT,
  321.               0, 0, interest_attrs_->size(), pkt_count_, rdm_id_, 
  322.               rmst_ptr->last_hop_, LOCALHOST_ADDR);
  323.             neg_reinf_msg->msg_attr_vec_ = CopyAttrs(interest_attrs_);
  324.             ((DiffusionRouting *)dr_)->sendMessage(neg_reinf_msg, filter_handle_, 1);
  325.             pkt_count_++;
  326.             delete neg_reinf_msg;
  327.           }
  328.         }
  329.       }
  331.       // We suppress new DATA messages that don't arrive on the 
  332.       // reinforced path. 
  333.       if ( msg->last_hop_ != rmst_ptr->last_hop_ ){
  334.         DiffPrint(DEBUG_IMPORTANT,
  335.           "  We suppress new DATA message on non-backchannel path!; backchannel = %dn",
  336.           rmst_ptr->last_hop_);
  337.         msg->new_message_ = 0;
  338.         return true;
  339.       }
  341.       if (rmst_ptr->wait_for_new_path_){
  342.         rmst_ptr->wait_for_new_path_ = false;
  343.         DiffPrint(DEBUG_SOME_DETAILS, "  node resets wait_for_new_path_.n");
  344.       }
  346.       if (local_sink_ && rmst_ptr->sent_exp_req_){
  347.         DiffPrint(DEBUG_SOME_DETAILS,
  348.           "  source got a new path, set sent_exp_req_ false.n");
  349.         rmst_ptr->sent_exp_req_ = false;
  350.       }
  352.       // We forward DATA if we aren't a source or a sink.
  353.       // Sources collect all fragments and send them from a timer.
  354.       // Sinks collect all fragments and send them to the app when they
  355.       // have all arrived.
  356.       if(rmst_ptr->local_source_ || local_sink_)
  357.         return false;
  358.       else{
  359.         rmst_ptr->pkts_sent_++;
  360.         // We need to alter the RmstPktsSentAttr to reflect this node!
  361.         if(pkts_sent_attr)
  362.           pkts_sent_attr->setVal(rmst_ptr->pkts_sent_);
  363.         return true;
  364.       }
  365.       break;
  367.     case(INTEREST):
  368.       data = msg->msg_attr_vec_;
  369.       nr_class = NRClassAttr.find(data);
  370.       if (nr_class){
  371.         class_type = nr_class->getVal();
  372.         if (class_type == NRAttribute::DISINTEREST_CLASS)
  373.           DiffPrint(DEBUG_SOME_DETAILS, "  DISINTEREST_CLASSn");
  374.       }
  376.       nrscope = NRScopeAttr.find(msg->msg_attr_vec_);
  377.       if(nrscope->getVal() == NRAttribute::NODE_LOCAL_SCOPE)
  378.         DiffPrint(DEBUG_SOME_DETAILS, "  rmst LOCAL_SCOPE Interest Messagen");
  379.       else if (msg->last_hop_ == LOCALHOST_ADDR){
  380.         DiffPrint(DEBUG_SOME_DETAILS, "  rmst Interest Message from local SINKn");
  381.         local_sink_ = true;
  382.         local_sink_port_ = msg->source_port_;
  383.         GetTime (&last_sink_time_);
  384.         if (interest_attrs_ == NULL)
  385.           interest_attrs_ = CopyAttrs(msg->msg_attr_vec_);
  386.       }
  387.       else{
  388.         DiffPrint(DEBUG_SOME_DETAILS, "  rmst Interest Message from non-local noden");
  389.           if (interest_attrs_ == NULL)
  390.             interest_attrs_ = CopyAttrs(msg->msg_attr_vec_);
  391.       }
  392.       break;
  394.     case(POSITIVE_REINFORCEMENT):
  395.       ReinfMessage *reinf_msg;
  396.       ExpLog exp_log;
  398.       DiffPrint(DEBUG_IMPORTANT, "  Positive Reinf arrivedn");
  399.       reinf_attr = ReinforcementAttr.find(msg->msg_attr_vec_);
  400.       reinf_msg = (ReinfMessage*)reinf_attr->getVal();
  401.       DiffPrint(DEBUG_LOTS_DETAILS, "  Pos_Reinf: ptk_num = %x, rdm_id_ = %xn",
  402.         reinf_msg->pkt_num_, reinf_msg->rdm_id_);
  404.       key.int_val_[0] = reinf_msg->pkt_num_;
  405.       key.int_val_[1] = reinf_msg->rdm_id_;
  406.       exp_iterator = exp_map_.find(key.ll_val_);
  407.       if(exp_iterator != exp_map_.end()){
  408.         exp_log = (*exp_iterator).second;
  409.         DiffPrint(DEBUG_SOME_DETAILS, "  Reinforcement for rmst_no = %d, last_hop_ = %dn",
  410.           exp_log.rmst_no_, exp_log.last_hop_);
  412.         // Here is where we must update the rmst with back-channel
  413.         // last hop.
  414.         rmst_no = exp_log.rmst_no_;
  415.         rmst_iterator = rmst_map_.find(rmst_no);
  416.         if(rmst_iterator != rmst_map_.end()){
  417.           rmst_ptr = (*rmst_iterator).second;
  418.           rmst_ptr->last_hop_ = exp_log.last_hop_;
  419.           rmst_ptr->fwd_hop_ = msg->last_hop_;
  420.           DiffPrint(DEBUG_SOME_DETAILS, "  Setting rmst_no %d last_hop_ = %d, fwd_hop_ = %dn",
  421.             rmst_no, rmst_ptr->last_hop_, rmst_ptr->fwd_hop_);
  422.           if(!rmst_ptr->reinf_){
  423.             rmst_ptr->reinf_ = true;
  424.             if(rmst_ptr->local_source_)
  425.               DiffPrint(DEBUG_LOTS_DETAILS, "  Local source got a Reinfn");
  426.           }
  427.         }
  428.         else{
  429.           DiffPrint(DEBUG_IMPORTANT, "  Reinforcement cant't find rmst_non");
  430.           break;
  431.         }
  432.         // We are on the reinforced path, so we must start a timer
  433.         // if one hasn't already been started. Sinks don't get
  434.         // reinforced, so they start a WATCHDOG in syncLocalCache.
  435.         if( (rmst_ptr->watchdog_active_ == false) && (caching_mode_) ){
  436.           TimerCallback *rmst_timer;
  437.           DiffPrint(DEBUG_IMPORTANT,
  438.             "  Set a WATCHDOG_TIMER at caching node for reinforced rmst_no %dn",
  439.             rmst_no);
  440.           rmst_timer = new RmstTimeout(this, rmst_no, WATCHDOG_TIMER);
  441.           // We check on things every 10 seconds.
  442.           rmst_ptr->watchdog_handle_ = 
  443.             ((DiffusionRouting *)dr_)->addTimer(WATCHDOG_INTERVAL, rmst_timer);
  444.           rmst_ptr->watchdog_active_ = true;
  445.         }
  446.         if (rmst_ptr->wait_for_new_path_){
  447.           DiffPrint(DEBUG_SOME_DETAILS, "  Resetting wait_for_new_path_ for rmst_no %dn", rmst_no);
  448.           rmst_ptr->wait_for_new_path_ = false;
  449.         }
  450.         if (rmst_ptr->sent_exp_req_){
  451.           DiffPrint(DEBUG_SOME_DETAILS,
  452.             "  intermediate node got a new path, set sent_exp_req_ false.n");
  453.           rmst_ptr->sent_exp_req_ = false;
  454.         }
  455.       }
  456.       else{
  457.         if(!rmst_ptr->local_source_)
  458.           DiffPrint(DEBUG_IMPORTANT, "  Reinforcement matches no Exploratory msgn");
  459.       }
  460.       break;
  462.     case(NEGATIVE_REINFORCEMENT):
  463.       bool ret_val;
  464.       if (tsprt_ctl_attr){
  465.         DiffPrint(DEBUG_SOME_DETAILS,
  466.             "  NEGATIVE_REINFORCEMENT, last_hop_ = %d, rmst_ctl_type = %dn", 
  467.             msg->last_hop_, rmst_ctl_type);
  468.       }
  469.       // We need to check if we got a NEGATIVE REINFORCEMENT from a node that is the
  470.       // next node in the forward direction (downstream).  If so, and we are the source
  471.       // we must send a new EXPLORATORY message;  else if we are not the source,
  472.       // we must send and exp request upstream.
  473.       ret_val = true;
  474.       rmst_iterator = rmst_map_.begin();
  475.       while(rmst_iterator != rmst_map_.end()){
  476.         rmst_ptr = (*rmst_iterator).second;
  477.         DiffPrint(DEBUG_SOME_DETAILS,
  478.             "  searching rmsts - rmst_no_ %d: fwd_hop_ = %d, reinf_ = %d, acked = %dn",
  479.             rmst_ptr->rmst_no_, rmst_ptr->fwd_hop_, rmst_ptr->reinf_, rmst_ptr->acked_);
  480.         if (rmst_ptr->local_source_ && rmst_ptr->reinf_
  481.             && (rmst_ptr->fwd_hop_ == msg->last_hop_)
  482.             && !rmst_ptr->acked_){
  483.           // If we are reinforced then we never got and EXP_REQ!!
  484.           DiffPrint(DEBUG_SOME_DETAILS, "  local source sees NEG_REINFn");
  485.           processExpReq(rmst_ptr, rmst_ptr->max_frag_sent_);
  486.         }
  487.         else if (!rmst_ptr->local_source_ && (rmst_ptr->fwd_hop_ == msg->last_hop_)
  488.                  && rmst_ptr->reinf_ && !rmst_ptr->acked_){
  489.           DiffPrint(DEBUG_SOME_DETAILS, "  intermediate node sees NEG_REINF from reinforced noden");
  490.           DiffPrint(DEBUG_SOME_DETAILS, "  send Exp Request upstream!n");
  491.           ret_val = false;
  492.           sendExpReqUpstream(rmst_ptr);
  493.         }
  494.         else{
  495.           DiffPrint(DEBUG_SOME_DETAILS,
  496.             "  node sees NEG_REINF from non-reinforced node - let routing layer see itn");
  497.   ret_val = true;
  498. }
  499.         rmst_iterator++;
  500.       }
  501.       if (!ret_val)
  502.         return false;
  503.       break;
  505.     default:
  506.       break;
  507.   } // switch msg->type
  508.   return true;
  509. }
  512. // RmstFilter::syncLocalCache
  513. //
  514. // This routine adds new transport data messages to the local data base.
  516. Rmst* RmstFilter::syncLocalCache (Message *msg)
  517. {
  518.   NRSimpleAttribute<int> *rmst_id_attr = NULL;
  519.   NRSimpleAttribute<int> *frag_attr = NULL;
  520.   NRSimpleAttribute<int> *max_frag_attr = NULL;
  521.   NRSimpleAttribute<void *> *data_buf_attr = NULL;
  522.   NRAttrVec *data = msg->msg_attr_vec_;
  523.   Int2Rmst::iterator rmst_iterator;
  524.   int rmst_no;
  525.   int frag_no;
  526.   int max_frag_no;
  527.   void *blob_ptr;
  528.   int blob_len;
  529.   void *tmp_frag_ptr;
  530.   Rmst *rmst_ptr;
  532.   rmst_id_attr = RmstIdAttr.find(data);
  533.   frag_attr = RmstFragAttr.find(data);
  534.   max_frag_attr = RmstMaxFragAttr.find(data);
  535.   data_buf_attr = RmstDataAttr.find(data);
  537.   if (! (rmst_id_attr && frag_attr && data_buf_attr) ){
  538.     DiffPrint(DEBUG_IMPORTANT, "  Filter received a BAD transport packet!n");
  539.     return NULL;
  540.   }
  542.   rmst_no = rmst_id_attr->getVal();
  543.   frag_no = frag_attr->getVal();
  544.   if(max_frag_attr)
  545.     max_frag_no = max_frag_attr->getVal();
  546.   else
  547.     max_frag_no = 0;
  548.   blob_ptr = data_buf_attr->getVal();
  549.   blob_len = data_buf_attr->getLen();
  551.   // Here is where I consuslt the Data Base and possibly add a new map,
  552.   // or add to an existing map.
  554.   rmst_iterator = rmst_map_.find(rmst_no);
  555.   if(rmst_iterator == rmst_map_.end()){
  556.     DiffPrint(DEBUG_IMPORTANT, "  creating a new DB entry for Rmst %dn", rmst_no);
  557.     DiffPrint(DEBUG_SOME_DETAILS, "  Max Fragment number = %dn", max_frag_no);
  558.     rmst_ptr = new Rmst(rmst_no);
  559.     rmst_ptr->max_frag_ = max_frag_no;
  560.     rmst_map_.insert(Int2Rmst::value_type(rmst_no, rmst_ptr));
  562.     // Artificially initialize last_nak_time_ so it's older
  563.     // than any Naks we may get.
  564.     GetTime(&rmst_ptr->last_nak_time_);
  566.     // Several decisions in this routine relate to messages that emanated
  567.     // from a local source.
  568.     if (msg->last_hop_ == LOCALHOST_ADDR) {
  569.       rmst_ptr->local_source_ = true;
  570.       rmst_ptr->local_source_port_ = msg->source_port_;
  571.       // This is the first fragment of an rmst from a local source.
  572.       // The message will be marked exploratory by this filter.
  573.       // We need to mark the last hop as LOCALHOST_ADDR.
  574.       rmst_ptr->last_hop_ = LOCALHOST_ADDR;
  575.     }
  576.     // We need to capture the RmstTargetAttr for concantenation on sendMessage.
  577.     if((rmst_ptr->local_source_)||(local_sink_)||(caching_mode_)){
  578.       NRSimpleAttribute<char *> *rmst_tgt_attr = NULL;
  579.       rmst_tgt_attr = RmstTargetAttr.find(msg->msg_attr_vec_);
  580.       if (rmst_tgt_attr){
  581.         char *tmp_str = rmst_tgt_attr->getVal();
  582.         rmst_ptr->target_str_ = new char[strlen(tmp_str)+1];
  583.         strcpy (rmst_ptr->target_str_, tmp_str);
  584.         DiffPrint(DEBUG_IMPORTANT, "  RmstTargetAttr = %sn", rmst_ptr->target_str_);
  585.       }
  586.       else
  587.         DiffPrint(DEBUG_IMPORTANT, "  no RmstTargetAttr Rmst %d !n", rmst_no);
  588.     }
  589.   }
  590.   else
  591.     rmst_ptr = (*rmst_iterator).second;
  593.   if(!rmst_ptr->local_source_)
  594.     DiffPrint(DEBUG_IMPORTANT, "  Got a blob, rmstId = %d, frag_no = %dn", rmst_no, frag_no);
  596.   // Update the time we last saw data for this Rmst.
  597.   GetTime(&rmst_ptr->last_data_time_);
  599.   // We cache the fragment at the sink and source,
  600.   // or in caching mode at each node that receives it.
  601.   if((rmst_ptr->local_source_)||(local_sink_)||(caching_mode_)){
  602.     tmp_frag_ptr = rmst_ptr->getFrag(frag_no);
  603.     if (tmp_frag_ptr == NULL){
  604.       if(!rmst_ptr->local_source_)
  605.         DiffPrint(DEBUG_SOME_DETAILS, "  creating a new frag %d entry for Rmst %dn",
  606.           frag_no, rmst_no);
  607.       if (frag_no == rmst_ptr->max_frag_)
  608.       rmst_ptr->max_frag_len_ = blob_len;
  609.       tmp_frag_ptr = new char[blob_len];
  610.       memcpy(tmp_frag_ptr, blob_ptr, blob_len);
  611.       rmst_ptr->putFrag(frag_no, tmp_frag_ptr);
  613.       // Check to see if this fragment was NAKed.
  614.       // If so, delete from the hole map.
  615.       if(!rmst_ptr->local_source_){
  616.         if ( rmst_ptr->inHoleMap(frag_no) ){
  617.           // We need to see if we sent a NAK for this frag.
  618.           NakData *nak_ptr = rmst_ptr->getHole(frag_no);
  619.           if(nak_ptr->nak_sent_)
  620.             DiffPrint(DEBUG_SOME_DETAILS, "  We sent a NAK_REQ for this fragment.n");
  621.           DiffPrint(DEBUG_SOME_DETAILS, "  filter removing hole %d from hole_map_n",frag_no);
  622.           rmst_ptr->delHole(frag_no);
  623.         }
  624.       }
  625.       // We start a WATCHDOG for an rmst here if: this is a local sink,
  626.       // we haven't started a timer, and this is not the initial fragment.
  627.       // Intermediate nodes (in caching mode) start a timer if they are on the
  628.       // reinforced path, which is checked in processMessage.
  629.       if((!rmst_ptr->local_source_)&&(local_sink_)&&(rmst_ptr->watchdog_active_ == false)
  630.         && (frag_no>0)){
  631.         TimerCallback *rmst_timer;
  632.         DiffPrint(DEBUG_IMPORTANT, "  Set a WATCHDOG_TIMER at sink for rmst_no %dn", rmst_no);
  633.         rmst_timer = new RmstTimeout(this, rmst_no, WATCHDOG_TIMER);
  634.         // We check on things every 10 seconds.
  635.         rmst_ptr->watchdog_handle_ = ((DiffusionRouting *)dr_)->addTimer(WATCHDOG_INTERVAL,
  636.           rmst_timer);
  637.         rmst_ptr->watchdog_active_ = true;
  638.       }
  639.     }
  640.     else
  641.       DiffPrint(DEBUG_SOME_DETAILS, "  got a duplicate frag %d for blob %dn",
  642.         frag_no, rmst_no);
  644.     // If we have still have a hole in the fragment map, update the hole map.
  646.     if ((!rmst_ptr->local_source_) && (rmst_ptr->holeInFragMap()))
  647.       rmst_ptr->syncHoleMap();
  649.     // If the Rmst is complete, cancell timer, stop timer,
  650.     // send Ack, give to local sinks.
  651.     if(rmst_ptr->rmstComplete()){
  652.       if ((rmst_ptr->watchdog_active_) && (!rmst_ptr->local_source_)){
  653.         DiffPrint(DEBUG_SOME_DETAILS, 
  654.           "  Rmst #%d is complete set cancel_watchdog_ to stop WATCHDOGn",
  655.           rmst_no);
  656.         rmst_ptr->cancel_watchdog_ = true;
  657.       }
  659.       // Send this Rmst to any local sink
  660.       if(local_sink_ && !(rmst_ptr->acked_)){
  661.         sendRmstToSink(rmst_ptr);
  662.         // We mark the rmst acked at the sink so it will clean up.
  663.         rmst_ptr->acked_ = true;
  664.       }
  666.       // If this is a source, we only send out the fragments when we've got
  667.       // them all from the application. If the Rmst is complete we add the
  668.       // Rmst to the send_list_, and if there is no send timer we start one.
  669.       if(rmst_ptr->local_source_){
  670.         SendMsgData new_send_msg;
  671.         // The Rmst is complete and this is a source - put in send list.
  672.         new_send_msg.rmst_no_ = rmst_no;
  673.         new_send_msg.last_frag_sent_ = -1;
  674.         new_send_msg.exp_base_ = 0;
  675.         send_list_.push_back(new_send_msg);
  676.         if(!send_timer_active_){
  677.           TimerCallback *send_timer;
  678.           // Now add a timer to send this and any NAKS.
  679.           DiffPrint(DEBUG_SOME_DETAILS,
  680.             "  Rmst %d ready to send - Set a SEND_TIMERn", rmst_no);
  681.           send_timer = new RmstTimeout(this, -1, SEND_TIMER);
  682.           // We check on things every second.
  683.           send_timer_handle_ = 
  684.             ((DiffusionRouting *)dr_)->addTimer(SEND_INTERVAL, send_timer);
  685.           send_timer_active_ = true;
  686.         }
  687.       }
  688.       else
  689.         // We must let upstream nodes know that we got the whole blob.
  690.         sendAckToSource(rmst_ptr);
  691.     }
  692.   }
  693.   else{
  694.     rmst_ptr->max_frag_rec_ = frag_no;
  695.     DiffPrint(DEBUG_LOTS_DETAILS, "  Not caching frag %d entry for Rmst %dn", frag_no, rmst_no);
  696.   }
  698.   return rmst_ptr;
  699. }
  701. void RmstFilter::processCtrlMessage(Message *msg)
  702. {
  703.   NRSimpleAttribute<int> *tsprt_ctl_attr = NULL;
  704.   NRSimpleAttribute<int> *rmst_id_attr = NULL;
  705.   NRSimpleAttribute<int> *frag_attr = NULL;
  706.   NRAttrVec *data;
  707.   NRAttrVec attrs;
  708.   Int2Rmst::iterator rmst_iterator;
  709.   int rmst_no; 
  710.   int frag_no;
  711.   int rmst_ctl_type;
  712.   Rmst *rmst_ptr;
  713.   void *frag_ptr;
  714.   Message *nak_msg;
  715.   NRAttrVec::iterator place;
  716.   bool forwarding_nak = false;
  718.   data = msg->msg_attr_vec_;
  720.   tsprt_ctl_attr = RmstTsprtCtlAttr.find(data);
  721.   rmst_ctl_type = tsprt_ctl_attr->getVal();
  723.   rmst_id_attr = RmstIdAttr.find(data);
  724.   if(!rmst_id_attr) {
  725.     DiffPrint(DEBUG_SOME_DETAILS, "  Node got a bad Rmst control msg - no RmstIdAttr!n");
  726.     return;
  727.   }
  728.   rmst_no = rmst_id_attr->getVal();
  730.   // Let's make sure we have this rmst
  731.   rmst_iterator = rmst_map_.begin();
  732.   rmst_iterator = rmst_map_.find(rmst_no);
  733.   if(rmst_iterator != rmst_map_.end())
  734.     rmst_ptr = (*rmst_iterator).second;
  735.   else{
  736.     DiffPrint(DEBUG_IMPORTANT, "  Filter can't find Rmst %d for Rmst control msgn", rmst_no);
  737.     return;
  738.   }
  740.   switch (rmst_ctl_type){
  742.   case(ACK_RESP):
  743.     DiffPrint(DEBUG_IMPORTANT, "  Got an ACK_RESPn");
  744.     // For now we automatically forward ACKs if we're not the source.
  745.     rmst_ptr->acked_ = true;
  746.     if(!rmst_ptr->local_source_){
  747.       Message *ack_msg;
  748.       // If we got an ACK and we aren't the source, we must be an
  749.       // intermediate node (Sinks don't get ACKs, they send them).
  750.       // We need to forward ACK toward source if possible.
  751.       if (rmst_ptr->reinf_) {
  752.         DiffPrint(DEBUG_SOME_DETAILS, "  forwarding ACK to %dn", rmst_ptr->last_hop_);
  753.         attrs.push_back(RmstTsprtCtlAttr.make(NRAttribute::IS, ACK_RESP));
  754.         attrs.push_back(RmstIdAttr.make(NRAttribute::IS, rmst_no));
  755.         ack_msg = new Message(DIFFUSION_VERSION, DATA, 0, 0, attrs.size(),
  756.           pkt_count_, rdm_id_, rmst_ptr->last_hop_, LOCALHOST_ADDR);
  757.         ack_msg->msg_attr_vec_ = CopyAttrs(&attrs);
  758.         ((DiffusionRouting *)dr_)->sendMessage(ack_msg, filter_handle_, 1);
  759.         pkt_count_++;
  760.         delete ack_msg;
  761.         ClearAttrs(&attrs);
  762.       }
  763.       else
  764.         DiffPrint(DEBUG_IMPORTANT, "  intermediate node can't forward ACK for Rmst %dn", rmst_no);
  765.     }
  766.     else{
  767.       DiffPrint(DEBUG_IMPORTANT, "  Source got ACK for Rmst %dn", rmst_no);
  768.       sendContToSource(rmst_ptr);
  769.     }
  770.     break;
  772.   case(NAK_REQ):
  774.     // Mark the time we got this NAK for cleanup timer.
  775.     GetTime(&rmst_ptr->last_nak_time_);
  776.     rmst_ptr->naks_rec_++;
  777.     DiffPrint(DEBUG_IMPORTANT, "  Got a NAK_REQ; number = %dn", rmst_ptr->naks_rec_);
  779.     if ((rmst_ptr->naks_rec_ > 10) && (rmst_ptr->naks_rec_ > (.30 * rmst_ptr->max_frag_)) &&
  780.             rmst_ptr->local_source_){
  781.       DiffPrint(DEBUG_IMPORTANT, "  Too many NAKs - send an EXPLORATORY msg!n");
  782.       processExpReq(rmst_ptr, rmst_ptr->max_frag_sent_);
  783.       return;
  784.     }
  786.     // If we sent an exp request more than 30 seconds ago,
  787.     // we send it again.
  788.     if (rmst_ptr->sent_exp_req_){
  789.       int exp_time = rmst_ptr->last_nak_time_.tv_sec - rmst_ptr->exp_req_time_.tv_sec;
  790.       DiffPrint(DEBUG_SOME_DETAILS, 
  791.         "  Node that sent an EXP_REQ got a NAK: time since last exp = %dn", exp_time);
  792.       if( (rmst_ptr->last_nak_time_.tv_sec - rmst_ptr->exp_req_time_.tv_sec) > 30){
  793.         // Resend an EXP_REQ!!!
  794.         DiffPrint(DEBUG_IMPORTANT, "  Node resends EXP_REQ up blacklisted stream!n");
  795.         sendExpReqUpstream(rmst_ptr);
  796.         GetTime(&rmst_ptr->exp_req_time_);
  797.         // Send another negative reinforcement on blacklisted link!
  798.         Message *neg_reinf_msg;
  799.         neg_reinf_msg = new Message(DIFFUSION_VERSION, NEGATIVE_REINFORCEMENT,
  800.           0, 0, interest_attrs_->size(), pkt_count_, rdm_id_, 
  801.           rmst_ptr->last_hop_, LOCALHOST_ADDR);
  802.         neg_reinf_msg->msg_attr_vec_ = CopyAttrs(interest_attrs_);
  803.         ((DiffusionRouting *)dr_)->sendMessage(neg_reinf_msg, filter_handle_, 1);
  804.         pkt_count_++;
  805.         delete neg_reinf_msg;
  806.       }
  807.       return;
  808.     }
  810.     // We need to send the naked fragments if we are the source,
  811.     // or a caching node.
  812.     place = data->begin();
  813.     for(;;){
  814.       frag_attr = RmstFragAttr.find_from(data, place, &place);
  815.       if (!frag_attr)
  816.         break;
  817.       frag_no = frag_attr->getVal();
  818.       DiffPrint(DEBUG_IMPORTANT, "  Filter received a NAK_REQ for Rmst %d, frag %dn",
  819.         rmst_no, frag_no);
  821.       // Check if we have this fragment.
  822.       // If not forward NAK toward source if possible.
  823.       frag_ptr =  rmst_ptr->getFrag(frag_no);
  824.       if (frag_ptr == NULL){
  825.         DiffPrint(DEBUG_SOME_DETAILS, "  Filter can't find frag %d of Rmst %d for NAKn",
  826.         frag_no, rmst_no);
  827.         // We need to forward NAK toward source if possible.
  828.         if ( (rmst_ptr->reinf_) && (rmst_ptr->last_hop_ != LOCALHOST_ADDR) ){
  829.           forwarding_nak = true;
  830.           DiffPrint(DEBUG_IMPORTANT, "  forwarding NAK to %dn", rmst_ptr->last_hop_);
  831.           attrs.push_back(RmstFragAttr.make(NRAttribute::IS, frag_no));
  833.           if(caching_mode_){
  834.             // We need to add this fragment to our hole map!
  835.             rmst_ptr->putHole(frag_no);
  836.             NakData *nak_ptr = rmst_ptr->getHole(frag_no);
  837.             // Artificially age this hole so it gets
  838.             // naked immediately.
  839.             nak_ptr->tmv.tv_sec -= 1;
  840.           }
  841.         }
  842.         else
  843.           DiffPrint(DEBUG_IMPORTANT, "  not forwarding NAK! - no place to send it!n");
  844.       }
  845.       else{
  846.         // We have this fragment so add it to the NakList for sending.
  847.         NakMsgData nak_msg_data;
  848.         NakList::iterator nak_list_iterator;
  849.         nak_list_iterator = nak_list_.begin();
  850.         while(nak_list_iterator != nak_list_.end()){
  851.           if((nak_list_iterator->rmst_no_ == rmst_no) &&
  852.             (nak_list_iterator->frag_no_ == frag_no))
  853.             break;
  854.           nak_list_iterator++;
  855.         }
  856.         if(nak_list_iterator == nak_list_.end()){
  857.           DiffPrint(DEBUG_SOME_DETAILS,
  858.             "  adding NAK for rmst %d frag %d to nak_list_n", rmst_no, frag_no);
  859.           nak_msg_data.rmst_no_ = rmst_no;
  860.           nak_msg_data.frag_no_ = frag_no;
  861.           nak_list_.push_back(nak_msg_data);
  862.           if(!send_timer_active_){
  863.             TimerCallback *send_timer;
  864.             // Now add a timer to send this and any NAKS.
  865.             DiffPrint(DEBUG_LOTS_DETAILS, 
  866.               "  Set a SEND_TIMER for reinforced rmst_no %dn", rmst_no);
  867.             send_timer = new RmstTimeout(this, rmst_no, SEND_TIMER);
  868.             // We check on things every second.
  869.             send_timer_handle_ = 
  870.               ((DiffusionRouting *)dr_)->addTimer(SEND_INTERVAL, send_timer);
  871.             send_timer_active_ = true;
  872.           }
  873.         }
  874.       }
  875.       place++;
  876.     }
  878.     if (forwarding_nak){
  879.       attrs.push_back(RmstTsprtCtlAttr.make(NRAttribute::IS, NAK_REQ));
  880.       attrs.push_back(RmstIdAttr.make(NRAttribute::IS, rmst_no));
  881.       nak_msg = new Message(DIFFUSION_VERSION, DATA, 0, 0,
  882.       attrs.size(), pkt_count_, rdm_id_, rmst_ptr->last_hop_,
  883.         LOCALHOST_ADDR);
  884.       nak_msg->msg_attr_vec_ = CopyAttrs(&attrs);
  885.       ((DiffusionRouting *)dr_)->sendMessage(nak_msg, filter_handle_, 1);
  886.       pkt_count_++;
  887.       delete nak_msg;
  888.       ClearAttrs(&attrs);
  889.     }
  890.     break;
  892.   case(EXP_REQ):
  893.     DiffPrint(DEBUG_IMPORTANT, "  Got an EXP_REQn");
  894.     if(!rmst_ptr->local_source_){
  895.       DiffPrint(DEBUG_SOME_DETAILS, "  Filter forwarding EXP_REQ for Rmst %dn", rmst_no);
  896.       // We need to forward EXP_REQ toward source if possible.
  897.       if (rmst_ptr->reinf_)
  898.         sendExpReqUpstream(rmst_ptr);
  899.     }
  900.     else{
  901.       // We need to call a routine that will clean the NAK list of 
  902.       // outstanding NAK responses for this Rmst, put a new expBase
  903.       // in the send list (lowest of nak or send Lists), and set this
  904.       // rmst as not reinforced.
  905.       frag_attr = RmstFragAttr.find(msg->msg_attr_vec_);
  906.       frag_no = frag_attr->getVal();
  907.       DiffPrint(DEBUG_IMPORTANT, "  Source got EXP request for Rmst %dn", rmst_no);
  908.       if (rmst_ptr->reinf_)
  909.           processExpReq(rmst_ptr, frag_no);
  910.       else
  911.           DiffPrint(DEBUG_IMPORTANT, "  EXP request for non-reinforced Rmst %dn", rmst_no);
  912.     }
  913.     break;
  915.   default:
  916.     break;
  917.   }  // switch (rmst_ctl_type)
  918.   return;
  919. }
  921. void RmstFilter::setupNak(int rmst_id)
  922. {
  923.   NRAttrVec attrs;
  924.   int frag_id;
  925.   NakData *nak_ptr;
  926.   Rmst *rmst_ptr;
  927.   int nak_count = 0;
  929.   Int2Rmst::iterator rmst_iterator = rmst_map_.find(rmst_id);
  930.   if(rmst_iterator != rmst_map_.end())
  931.     rmst_ptr = (*rmst_iterator).second;
  932.   else{
  933.     DiffPrint(DEBUG_IMPORTANT, "setupNak - can't find Rmst %dn", rmst_id);
  934.     return;
  935.   }
  937.   Int2Nak::iterator hole_iter = rmst_ptr->hole_map_.begin();
  938.   bool send_new_nak = false;
  939.   timeval cur_time;
  941.   // We now have an iterator to look at each hole (hole_iter),
  942.   // a Rmst Id (rmst_id), a fragment Id ((*hole_iter).first),
  943.   // a NakData pointer ((*hole_iter).second), and an Rmst
  944.   // pointer (rmst_ptr).
  946.   GetTime (&cur_time);
  948.   // The first pass finds holes that haven't been NAKed and should be.
  949.   while(hole_iter != rmst_ptr->hole_map_.end()){
  950.     frag_id = (*hole_iter).first;
  951.     nak_ptr = (*hole_iter).second;
  952.     DiffPrint(DEBUG_SOME_DETAILS,
  953.       "  setupNak - found hole rmst_id %d, frag %dn", rmst_id, frag_id);
  955.     // If we never NAKed this fragment and it's past due,
  956.     // mark it so it gets NAKed.
  957.     if (!nak_ptr->nak_sent_){
  958.       if ( (cur_time.tv_sec - nak_ptr->tmv.tv_sec) > 3 ){
  959.         nak_ptr->nak_sent_ = true;
  960.         nak_ptr->send_nak_ = true;
  961.         send_new_nak = true;
  962.       }
  963.       else
  964.         DiffPrint(DEBUG_SOME_DETAILS,
  965.           "  setupNak - hole %d not old enough to NAKn", frag_id);
  966.     }
  967.     // If we NAKed this fragment and the NAK response is past due,
  968.     // NAK it again.
  969.     else if ( (cur_time.tv_sec - nak_ptr->tmv.tv_sec) > NAK_RESPONSE_WAIT ){
  970.       DiffPrint(DEBUG_SOME_DETAILS, "  setupNak - hole %d has an overdue NAKn", frag_id);
  971.       nak_ptr->send_nak_ = true;
  972.       send_new_nak = true;
  973.     }
  974.     hole_iter++;
  975.   }
  977.   if (send_new_nak){
  978.     Message *nak_msg;
  979.     if ( rmst_ptr->last_hop_ == LOCALHOST_ADDR ){
  980.       DiffPrint(DEBUG_IMPORTANT, "  can't send NAK, no last_hop_!n");
  981.       return;
  982.     }
  983.     // The second pass adds all holes that should be NAKed to vector.
  984.     hole_iter = rmst_ptr->hole_map_.begin();
  985.     while( (hole_iter != rmst_ptr->hole_map_.end()) && (nak_count <= 10) ){
  986.       frag_id = (*hole_iter).first;
  987.       nak_ptr = (*hole_iter).second;
  989.       if ( nak_ptr->send_nak_ ){
  990.         nak_ptr->send_nak_ = false;
  991.         DiffPrint(DEBUG_SOME_DETAILS,
  992.           "  setupNak - adding a NAK for frag_id %d to attrsn", frag_id);
  993.         attrs.push_back(RmstFragAttr.make(NRAttribute::IS, frag_id));
  994.         GetTime(&(nak_ptr->tmv));
  995.         nak_count++;
  996.       }
  997.       hole_iter++;
  998.     }
  999.     attrs.push_back(RmstTsprtCtlAttr.make(NRAttribute::IS, NAK_REQ));
  1000.     attrs.push_back(RmstIdAttr.make(NRAttribute::IS, rmst_id));
  1002.     // Code to send a message to last_hop_
  1003.     nak_msg = new Message(DIFFUSION_VERSION, DATA, 0, 0, attrs.size(),
  1004.       pkt_count_, rdm_id_, rmst_ptr->last_hop_, LOCALHOST_ADDR);
  1005.     pkt_count_++;
  1006.     nak_msg->msg_attr_vec_ = CopyAttrs(&attrs);
  1008.     DiffPrint(DEBUG_IMPORTANT, "  Sending NAK_REQ to node %dn", rmst_ptr->last_hop_);
  1009.     ((DiffusionRouting *)dr_)->sendMessage(nak_msg, filter_handle_, 1);
  1011.     delete nak_msg;
  1012.     ClearAttrs(&attrs);
  1014.     // Mark the time we sent this NAK for cleanup timer.
  1015.     GetTime(&rmst_ptr->last_nak_time_);
  1017.   }
  1018.   else
  1019.     DiffPrint(DEBUG_SOME_DETAILS, "  setupNak - no need for a new NAK for rmst_id %dn", rmst_id);
  1020.   return;
  1021. }
  1023. void RmstFilter::processExpReq(Rmst *rmst_ptr, int frag_no)
  1024. {
  1025.   NakList::iterator nak_list_iterator;
  1026.   SendList::iterator send_list_iterator;
  1027.   int rmst_no = rmst_ptr->rmst_no_;
  1029.   DiffPrint(DEBUG_IMPORTANT, "  processExpReq called for rmstId %d, frag_no %dn", rmst_no, frag_no);
  1031.   // Indicate that Rmst is not reinforced.
  1032.   rmst_ptr->reinf_ = false;
  1033.   rmst_ptr->pkts_sent_ = 0;
  1035.   // If we have an ACK_TIMER active cancel it. We want to resend some packets.
  1036.   if(rmst_ptr->ack_timer_active_){
  1037.     ((DiffusionRouting *)dr_)->removeTimer(rmst_ptr->ack_timer_handle_);
  1038.     rmst_ptr->ack_timer_active_ = false;
  1039.   }
  1041.   // Erase any NAKs. We are about to establish a new path.
  1042.   nak_list_iterator = nak_list_.begin();
  1043.   while (nak_list_iterator != nak_list_.end()){
  1044.     if (nak_list_iterator->rmst_no_ == rmst_no){
  1045.       DiffPrint(DEBUG_SOME_DETAILS,
  1046.         "  processExpReq erasing frag_no %d from nak_list_n", nak_list_iterator->frag_no_);
  1047.       nak_list_iterator = nak_list_.erase(nak_list_iterator);
  1048.     }
  1049.     else
  1050.       nak_list_iterator++;
  1051.   }
  1053.   DiffPrint(DEBUG_LOTS_DETAILS, "  processExpReq done with nak_list_ for rmstId %dn", rmst_no);
  1055.   // If we are being told to start by resending the last packet, back up by one.
  1056.   // When a sink gets an exploratory message, they don't start re-NAKing until they
  1057.   // know they are reinforced. Sinks only know they are reinforced when they get DATA.
  1058.   if ( (frag_no == rmst_ptr->max_frag_) && (rmst_ptr->max_frag_ > 0) ){
  1059.     frag_no--;
  1060.     DiffPrint(DEBUG_IMPORTANT, "  processExpReq decrements frag_no to %dn", frag_no);
  1061.   }
  1063.   // Update send_list_ entry or add one.
  1064.   send_list_iterator = send_list_.begin();
  1065.   while (send_list_iterator != send_list_.end()){
  1066.     if (send_list_iterator->rmst_no_ == rmst_no)
  1067.       break;
  1068.     send_list_iterator++;
  1069.   }
  1070.   if (send_list_iterator != send_list_.end()){
  1071.     send_list_iterator->exp_base_ = frag_no;
  1072.     DiffPrint(DEBUG_SOME_DETAILS, "  processExpReq sets send_list_ expBase to %dn", frag_no);
  1073.     send_list_iterator->last_frag_sent_ = frag_no-1;
  1074.   }
  1075.   else{
  1076.     SendMsgData new_send_msg;
  1077.     DiffPrint(DEBUG_SOME_DETAILS,
  1078.       "  processExpReq creating new send_list_ entry for rmstId %dn", rmst_no);
  1079.     DiffPrint(DEBUG_SOME_DETAILS,
  1080.       "  processExpReq sets send_list_ expBase to %dn", frag_no);
  1081.     new_send_msg.rmst_no_ = rmst_no;
  1082.     new_send_msg.exp_base_ = frag_no;
  1083.     new_send_msg.last_frag_sent_ = frag_no-1;
  1084.     send_list_.push_front(new_send_msg);
  1085.     if(!send_timer_active_){
  1086.       TimerCallback *send_timer;
  1087.       // Now add a timer to send this and any NAKS.
  1088.       DiffPrint(DEBUG_SOME_DETAILS, "  Set a SEND_TIMER for reinforced rmst_no %dn", rmst_no);
  1089.       send_timer = new RmstTimeout(this, rmst_no, SEND_TIMER);
  1090.       // We check on things every second.
  1091.       send_timer_handle_ = ((DiffusionRouting *)dr_)->addTimer(SEND_INTERVAL, send_timer);
  1092.       send_timer_active_ = true;
  1093.     }
  1094.   }
  1095. }
  1097. handle RmstFilter::setupFilter()
  1098. {
  1099.   NRAttrVec attrs;
  1100.   handle h;
  1102.   // This is a dummy attribute for filtering that matches everything
  1103.   attrs.push_back(NRClassAttr.make(NRAttribute::IS,
  1104.     NRAttribute::INTEREST_CLASS));
  1105.   h = ((DiffusionRouting *)dr_)->addFilter(&attrs, RMST_FILTER_PRIORITY, fcb_);
  1106.   ClearAttrs(&attrs);
  1107.   return h;
  1108. }
  1110. void RmstFilter::run()
  1111. {
  1112. #ifdef NS_DIFFUSION
  1113.   TimerCallback *stat_timer;
  1114.   filter_handle_ = setupFilter();
  1115.   DiffPrint(DEBUG_LOTS_DETAILS, "RmstFilter:: subscribed to all, received handle %dn",
  1116.     (int)filter_handle_);
  1117.   
  1118.   DiffPrint(DEBUG_LOTS_DETAILS, "RmstFilter constructor: start cleanup timern");
  1119.   stat_timer = new RmstTimeout(this, -1, CLEANUP_TIMER);
  1120.   stat_timer_handle_ = ((DiffusionRouting *)dr_)->addTimer(CLEANUP_INTERVAL, stat_timer);
  1121. #else
  1122.   // Doesn't do anything
  1123.   while(1){
  1124.       sleep(1000);
  1125.   }
  1126. #endif // NS_DIFFUSION
  1127. }
  1129. RmstTimeout::RmstTimeout(RmstFilter *rmst_flt, int no, int type)
  1130. {
  1131.   filter_ = rmst_flt;
  1132.   rmst_no_ = no;
  1133.   timer_type_ = type;
  1134. }
  1136. int RmstTimeout::expire()
  1137. {
  1138.   int retval;
  1140.   retval = filter_->processTimer(rmst_no_, timer_type_);
  1141.   if(retval == -1)
  1142.     delete this;
  1144.   return retval;
  1145. }
  1147. int RmstFilter::processTimer(int rmst_no, int timer_type)
  1148. {
  1149.   Rmst *rmst_ptr;
  1150.   void *frag_ptr;
  1151.   int frag_no;
  1152.   Int2Rmst::iterator rmst_iterator;
  1153.   timeval cur_time;
  1155.   GetTime (&cur_time);
  1157.   switch (timer_type){
  1159.   case SEND_TIMER:
  1160.     DiffPrint(DEBUG_SOME_DETAILS, "RmstFilter::processTimer SEND_TIMER");
  1161.     PrintTime(&cur_time);
  1162.     // If we haven't got any NAKs pending, send the next fragment of the Rmst
  1163.     // in progress. If we've sent all fragments of the current Rmst, we cancel
  1164.     // ourself. When we get a NAK, if this timer is active we send NAK responses
  1165.     // here. Otherwise we can send them directly from the NAK response routine.
  1166.     if (!nak_list_.empty()){
  1167.       Message *nak_resp;
  1168.       NRAttrVec nak_data_attrs;
  1169.       NakMsgData nak_msg_data = nak_list_.front();
  1170.       rmst_no = nak_msg_data.rmst_no_;
  1171.       frag_no = nak_msg_data.frag_no_;
  1173.       rmst_iterator = rmst_map_.find(rmst_no);
  1174.       if(rmst_iterator != rmst_map_.end()){
  1175.         rmst_ptr = (*rmst_iterator).second;
  1176.         // We have the fragment, set the last_data_time_ so that we defer
  1177.         // the cleanup of this Rmst, then we send the fragment to last hop.
  1178.         if (rmst_ptr->reinf_){
  1179.           GetTime(&rmst_ptr->last_data_time_);
  1180.           rmst_ptr->pkts_sent_++;
  1181.           nak_data_attrs.push_back(RmstTargetAttr.make(NRAttribute::IS,
  1182.             rmst_ptr->target_str_));
  1183.           nak_data_attrs.push_back(RmstTsprtCtlAttr.make(NRAttribute::IS,
  1184.             RMST_RESP));
  1185.           nak_data_attrs.push_back(RmstFragAttr.make(NRAttribute::IS, frag_no));
  1186.           // We routinely send the packet sent count on NAKs.
  1187.           nak_data_attrs.push_back(RmstPktsSentAttr.make(NRAttribute::IS,
  1188.             rmst_ptr->pkts_sent_));
  1189.           nak_data_attrs.push_back(RmstIdAttr.make(NRAttribute::IS, rmst_no));
  1190.           // Add the actual data; the length depends on if it's the last
  1191.           // fragment or not.
  1192.           frag_ptr =  rmst_ptr->getFrag(frag_no);
  1193.           if (frag_no == rmst_ptr->max_frag_)
  1194.             nak_data_attrs.push_back(RmstDataAttr.make(NRAttribute::IS,
  1195.               frag_ptr, rmst_ptr->max_frag_len_));
  1196.           else
  1197.             nak_data_attrs.push_back(RmstDataAttr.make(NRAttribute::IS,
  1198.               frag_ptr, MAX_FRAG_SIZE));
  1200.           DiffPrint(DEBUG_IMPORTANT, "  Filter sending Data for NAKed frag %d of Rmst %dn",
  1201.             frag_no, rmst_no);
  1203.           nak_resp = new Message(DIFFUSION_VERSION, DATA, 0, 0,
  1204.             nak_data_attrs.size(), pkt_count_, rdm_id_,
  1205.             LOCALHOST_ADDR, LOCALHOST_ADDR);
  1206.           nak_resp->msg_attr_vec_ = CopyAttrs(&nak_data_attrs);
  1207.           ((DiffusionRouting *)dr_)->sendMessage(nak_resp, filter_handle_);
  1208.           pkt_count_++;
  1209.           delete nak_resp;
  1210.           ClearAttrs(&nak_data_attrs);
  1211.         }
  1212.         else
  1213.           DiffPrint(DEBUG_IMPORTANT, 
  1214.             "RmstFilter::processTimer sees non-reinforced path for NAK on rmst %d!n",
  1215.             rmst_no);
  1216.       }
  1217.       else
  1218.         DiffPrint(DEBUG_IMPORTANT, "RmstFilter::processTimer can't find Rmst %d for NAK!n",
  1219.           rmst_no);
  1220.       nak_list_.pop_front();
  1221.     }
  1223.     else if (!send_list_.empty()){
  1224.       int8_t msg_type;
  1225.       int action = DO_NOTHING;
  1226.       // Get the rmst and frag_no that is in progress.
  1227.       NRAttrVec data_attrs;
  1229.       SendMsgData send_data = send_list_.front();
  1230.       rmst_no = send_data.rmst_no_;
  1231.       rmst_iterator = rmst_map_.find(rmst_no);
  1232.       if(rmst_iterator == rmst_map_.end())
  1233.         action = DELETE_FROM_QUEUE;
  1234.       else{
  1235.         rmst_ptr = (*rmst_iterator).second;
  1236.         if ((send_data.last_frag_sent_ == rmst_ptr->max_frag_) && rmst_ptr->reinf_)
  1237.           action = DELETE_FROM_QUEUE;
  1238.         else if ( (send_data.last_frag_sent_ == send_data.exp_base_) &&
  1239.           (!rmst_ptr->reinf_) && (exp_gap_ < 10) ){
  1240.           action = DO_NOTHING;
  1241.           exp_gap_++;
  1242.         }
  1243.         else
  1244.           action = SEND_NEXT_FRAG;
  1245.       }
  1247.       switch (action){
  1249.       case(DELETE_FROM_QUEUE):
  1250.         // Delete message data from front. 
  1251.         send_list_.pop_front();
  1252.         break;
  1254.       case(SEND_NEXT_FRAG):
  1255.         send_list_.pop_front();
  1256.         if (rmst_ptr->reinf_){
  1257.           send_data.last_frag_sent_++;
  1258.           frag_no = send_data.last_frag_sent_;
  1259.         }
  1260.         else{
  1261.           frag_no = send_data.exp_base_;
  1262.           send_data.last_frag_sent_ = frag_no;
  1263.         }
  1264.         send_list_.push_front(send_data);
  1265.         rmst_ptr->max_frag_sent_ = frag_no;
  1267.         DiffPrint(DEBUG_IMPORTANT, "  Source Filter sending frag %d of Rmst %dn",
  1268.           frag_no, rmst_no);
  1269.         
  1270.         // Now make a message.
  1271.         data_attrs.push_back(RmstTargetAttr.make(NRAttribute::IS, rmst_ptr->target_str_));
  1272.         data_attrs.push_back(RmstTsprtCtlAttr.make(NRAttribute::IS, RMST_RESP));
  1273.         data_attrs.push_back(RmstFragAttr.make(NRAttribute::IS, frag_no));
  1275.         // We send the MaxFragAttr on the first Exploratory packet,
  1276.         // and the PktsSentAttr on the last packet.
  1277.         if(frag_no == 0)
  1278.           data_attrs.push_back(RmstMaxFragAttr.make(NRAttribute::IS,
  1279.             rmst_ptr->max_frag_));
  1280.         else if (frag_no == rmst_ptr->max_frag_){
  1281.           rmst_ptr->pkts_sent_++;
  1282.           data_attrs.push_back(RmstPktsSentAttr.make(NRAttribute::IS,
  1283.             rmst_ptr->pkts_sent_));
  1284.         }
  1285.         else
  1286.           rmst_ptr->pkts_sent_++;
  1288.         data_attrs.push_back(RmstIdAttr.make(NRAttribute::IS, rmst_no));
  1290.         // Add the actual data; the length depends on if it's the last
  1291.         // fragment or not.
  1292.         frag_ptr =  rmst_ptr->getFrag(frag_no);
  1293.         if (rmst_ptr->max_frag_ == frag_no)
  1294.           data_attrs.push_back(RmstDataAttr.make(NRAttribute::IS, 
  1295.             frag_ptr, rmst_ptr->max_frag_len_));
  1296.         else
  1297.           data_attrs.push_back(RmstDataAttr.make(NRAttribute::IS,
  1298.             frag_ptr, MAX_FRAG_SIZE));
  1300.         if (frag_no == send_data.exp_base_){
  1301.           ExpLog exp_msg;
  1302.           union LlToInt key;
  1303.           // Insert this Exploratory message in exp_map_.
  1304.           // When we get a reinforcement we'll know what rmst it's for.
  1305.           key.int_val_[0] = pkt_count_;
  1306.           key.int_val_[1] = rdm_id_;
  1307.           DiffPrint(DEBUG_LOTS_DETAILS, "  Key = %llxn", key.ll_val_);
  1308.           exp_msg.rmst_no_ = rmst_no;
  1309.           exp_msg.last_hop_ = LOCALHOST_ADDR;
  1310.           exp_map_.insert(Key2ExpLog::value_type(key.ll_val_, exp_msg));
  1311.           msg_type = EXPLORATORY_DATA;
  1312.           rmst_ptr->reinf_ = false;
  1313.           rmst_ptr->pkts_sent_ = 0;
  1314.           rmst_ptr->naks_rec_ = 0;
  1315.           exp_gap_ = 0;
  1316.           DiffPrint(DEBUG_IMPORTANT,
  1317.             "  Source Filter sending EXPLORATORY frag %d of Rmst %dn",
  1318.             frag_no, rmst_no);
  1319.         }
  1320.         else
  1321.           msg_type = DATA;
  1323.         Message *new_frag;
  1324.         new_frag = new Message(DIFFUSION_VERSION, msg_type, 0, 0,
  1325.           data_attrs.size(), pkt_count_, rdm_id_,
  1326.           LOCALHOST_ADDR, LOCALHOST_ADDR);
  1327.         new_frag->msg_attr_vec_ = CopyAttrs(&data_attrs);
  1328.         ((DiffusionRouting *)dr_)->sendMessage(new_frag, filter_handle_);
  1329.         pkt_count_++;
  1330.         delete new_frag;
  1331.         ClearAttrs(&data_attrs);
  1332.         // We sent a fragment, set the last_data_time_ for the cleanup timer.
  1333.         GetTime(&rmst_ptr->last_data_time_);
  1335.         // If this is the last frag, we start an ACK_TIMER.
  1336.         if ( (rmst_ptr->max_frag_ == frag_no) &&
  1337.           (rmst_ptr->ack_timer_active_ == false) && (rmst_ptr->local_source_) ){
  1338.           TimerCallback *rmst_timer;
  1339.           DiffPrint(DEBUG_SOME_DETAILS, "  Set an ACK_TIMER at source for rmst_no %dn",
  1340.             rmst_no);
  1341.           rmst_timer = new RmstTimeout(this, rmst_no, ACK_TIMER);
  1342.           // We check on things every 20 seconds.
  1343.           rmst_ptr->ack_timer_handle_ = 
  1344.             ((DiffusionRouting *)dr_)->addTimer(ACK_INTERVAL, rmst_timer);
  1345.           rmst_ptr->ack_timer_active_ = true;
  1346.         }
  1347.         break;
  1349.       case(DO_NOTHING):
  1350.         DiffPrint(DEBUG_LOTS_DETAILS, "  Nothing to don");
  1351.         break;
  1352.       } // Switch on Action
  1353.     }
  1355.     if (nak_list_.empty() && send_list_.empty()){
  1356.       DiffPrint(DEBUG_LOTS_DETAILS, "   Cancelling SEND_TIMER, no NAKS or data to sendn");
  1357.       send_timer_active_ = false;
  1358.       return -1;
  1359.     }
  1360.     else
  1361.       return 0;
  1362.     break;
  1364.   case WATCHDOG_TIMER:
  1365.     DiffPrint(DEBUG_SOME_DETAILS, "RmstFilter::processTimer WATCHDOG_TIMER for Rmst %d", rmst_no);
  1366.     PrintTime(&cur_time);
  1367.     rmst_iterator = rmst_map_.find(rmst_no);
  1368.     if(rmst_iterator != rmst_map_.end())
  1369.       rmst_ptr = (*rmst_iterator).second;
  1370.     else{
  1371.       DiffPrint(DEBUG_IMPORTANT,
  1372.         "RmstFilter::processTimer can't find Rmst %d for WATCHDOG, cancell timer!n",
  1373.         rmst_no);
  1374.       return -1;
  1375.     }
  1377.     if(rmst_ptr->cancel_watchdog_){
  1378.       DiffPrint(DEBUG_SOME_DETAILS,
  1379.         "  processTimer cancelling WATCHDOG_TIMER for Rmst %dn", rmst_no);
  1380.       rmst_ptr->watchdog_active_ = false;
  1381.       rmst_ptr->cancel_watchdog_ = false;
  1382.       return -1;
  1383.     }
  1385.     if (rmst_ptr->wait_for_new_path_){
  1386.       DiffPrint (DEBUG_IMPORTANT, "  WATCHDOG_TIMER sees wait_for_new_path_ - suspend NAKsn");
  1387.       return 0;
  1388.     }
  1390.     // If we sent an exp request more than 20 seconds ago,
  1391.     // we send it again.
  1392.     if (rmst_ptr->sent_exp_req_){
  1393.       int exp_time;
  1394.       exp_time = cur_time.tv_sec - rmst_ptr->exp_req_time_.tv_sec;
  1395.       DiffPrint(DEBUG_SOME_DETAILS,
  1396.         "  Node sent an EXP_REQ: time since last exp = %dn", exp_time);
  1397.       if( exp_time > 20){
  1398.         // Resend an EXP_REQ!!!
  1399.         DiffPrint(DEBUG_IMPORTANT, "  Node resends EXP_REQ up blacklisted stream!n");
  1400.         sendExpReqUpstream(rmst_ptr);
  1401.         GetTime(&rmst_ptr->exp_req_time_);
  1402.       }
  1403.       else
  1404.         DiffPrint(DEBUG_LOTS_DETAILS, "  Node waits to send another EXP_REQn");
  1405.       return 0;
  1406.     }
  1408.     if (rmst_ptr->local_source_){
  1409.       if(rmst_ptr->acked_){
  1410.         DiffPrint(DEBUG_IMPORTANT, 
  1411.           "  WATCHDOG_TIMER Local Source sees acked state - cancel timern");
  1412.         return -1;
  1413.       }
  1414.       else{
  1415.         DiffPrint(DEBUG_LOTS_DETAILS, "  WATCHDOG_TIMER Local Source sees rmst not ackedn");
  1416.         return 0;
  1417.       }
  1418.     }
  1420.     // Check if we have waited too long for next fragment.
  1421.     if( ((cur_time.tv_sec - rmst_ptr->last_data_time_.tv_sec) > NEXT_FRAG_WAIT) &&
  1422.       (!rmst_ptr->rmstComplete()) ){
  1423.       int newHole = (rmst_ptr->max_frag_rec_)+1;
  1424.       if ( (newHole <= rmst_ptr->max_frag_) && (!rmst_ptr->inHoleMap(newHole)) ){
  1425.         DiffPrint(DEBUG_SOME_DETAILS, "  WATCHDOG_TIMER adds new hole, frag %dn",
  1426.           newHole);
  1427.         rmst_ptr->putHole(newHole);
  1428.         NakData *nak_ptr = rmst_ptr->getHole(newHole);
  1429.         // Artificially age this hole so it gets naked immediately
  1430.         nak_ptr->tmv.tv_sec -= 4;
  1431.       }
  1432.     }
  1434.     if(rmst_ptr->holeMapEmpty()){
  1435.       // There aren't any holes!
  1436.       DiffPrint(DEBUG_SOME_DETAILS, "  WATCHDOG_TIMER sees No holesn");
  1437.       return 0;
  1438.     }
  1439.     else{
  1440.       // The WATCHDOG_TIMER expired and we have a hole,
  1441.       //   so we may need to construct a NAK from the hole map
  1442.       //   If we've fallen off the reinforced path, we must
  1443.       //   stop adding NAKs to the NakList.
  1444.       DiffPrint(DEBUG_SOME_DETAILS, "  WATCHDOG_TIMER sees holes - check times.n");
  1445.       if (rmst_ptr->reinf_)
  1446.         setupNak(rmst_no);
  1447.       // Reschedule timer with same value.
  1448.       return 0;
  1449.     }
  1450.     break;
  1452.   case ACK_TIMER:
  1453.     DiffPrint(DEBUG_SOME_DETAILS, "RmstFilter::processTimer ACK_TIMER for Rmst %d", rmst_no);
  1454.     PrintTime(&cur_time);
  1455.     rmst_iterator = rmst_map_.find(rmst_no);
  1456.     if(rmst_iterator != rmst_map_.end())
  1457.       rmst_ptr = (*rmst_iterator).second;
  1458.     else{
  1459.       DiffPrint(DEBUG_IMPORTANT,
  1460.         "RmstFilter::processTimer can't find Rmst %d for ACK_TIMER, cancell timer!n",
  1461.            rmst_no);
  1462.       return -1;
  1463.     }
  1465.     DiffPrint(DEBUG_SOME_DETAILS, "RmstFilter::processTimer ACK_TIMER, at source for rmst_no %dn",
  1466.       rmst_no);
  1468.     if (rmst_ptr->acked_){
  1469.       DiffPrint(DEBUG_IMPORTANT, "RmstFilter::processTimer cancel ACK_TIMER, Rmst %d ACKedn",
  1470.         rmst_no);
  1471.       rmst_ptr->ack_timer_active_ = false;
  1472.       return -1;
  1473.     }
  1475.     // If there has been no data sent for 30 seconds like a NAK response, we need to resend a packet.
  1476.     if( (cur_time.tv_sec - rmst_ptr->last_data_time_.tv_sec) > ACK_WAIT ){
  1477.       NRAttrVec attrs;
  1478.       int8_t msg_type;
  1479.       DiffPrint(DEBUG_IMPORTANT, 
  1480.         "RmstFilter::processTimer ACK_TIMER, waited too long for Rmst %d ACK!n", rmst_no);
  1481.       if(rmst_ptr->reinf_  && !rmst_ptr->resent_last_data_){
  1482.         // We should send the last frag again as an DATA packet.
  1483.         DiffPrint(DEBUG_SOME_DETAILS, 
  1484.           "RmstFilter::processTimer ACK_TIMER, resend last packet as DATAn");
  1485.         msg_type = DATA;
  1486.         rmst_ptr->resent_last_data_ = true;
  1487.       }
  1488.       else if(rmst_ptr->resent_last_data_ && !rmst_ptr->resent_last_exp_){
  1489.         ExpLog exp_msg;
  1490.         union LlToInt key;
  1491.         // We tried resending last frag as data and it didn't work, try as EXP
  1492.         DiffPrint(DEBUG_IMPORTANT,
  1493.           "RmstFilter::processTimer ACK_TIMER, resend last packet as EXPLORATORY_DATAn");
  1494.         // Insert this Exploratory message in exp_map_.
  1495.         // When we get a reinforcement we'll know what rmst it's for.
  1496.         key.int_val_[0] = pkt_count_;
  1497.         key.int_val_[1] = rdm_id_;
  1498.         DiffPrint(DEBUG_LOTS_DETAILS, "  Key = %llxn", key.ll_val_);
  1499.         exp_msg.rmst_no_ = rmst_no;
  1500.         exp_msg.last_hop_ = LOCALHOST_ADDR;
  1501.         exp_map_.insert(Key2ExpLog::value_type(key.ll_val_, exp_msg));
  1502.         msg_type = EXPLORATORY_DATA;
  1503.         rmst_ptr->reinf_ = false;
  1504.         rmst_ptr->naks_rec_ = 0;
  1505.         rmst_ptr->pkts_sent_ = 0;
  1506.         rmst_ptr->resent_last_exp_ = true;
  1507.       }
  1508.       else if(rmst_ptr->resent_last_data_ && rmst_ptr->resent_last_exp_ && rmst_ptr->reinf_){
  1509.         // We should send the last frag again as an DATA packet.
  1510.         DiffPrint(DEBUG_IMPORTANT,
  1511.           "RmstFilter::processTimer ACK_TIMER, resend last packet on new reinf path as DATAn");
  1512.         msg_type = DATA;
  1513.         rmst_ptr->resent_last_data_ = false;
  1514.         rmst_ptr->resent_last_exp_ = false;
  1515.       }
  1516.       else{
  1517.         ExpLog exp_msg;
  1518.         union LlToInt key;
  1519.         DiffPrint(DEBUG_IMPORTANT, 
  1520.           "RmstFilter::processTimer ACK_TIMER, resent last packet as EXP and no reinforced path, Try again!n");
  1521.         // Insert this Exploratory message in exp_map_.
  1522.         // When we get a reinforcement we'll know what rmst it's for.
  1523.         key.int_val_[0] = pkt_count_;
  1524.         key.int_val_[1] = rdm_id_;
  1525.         DiffPrint(DEBUG_LOTS_DETAILS, "  Key = %llxn", key.ll_val_);
  1526.         exp_msg.rmst_no_ = rmst_no;
  1527.         exp_msg.last_hop_ = LOCALHOST_ADDR;
  1528.         exp_map_.insert(Key2ExpLog::value_type(key.ll_val_, exp_msg));
  1529.         msg_type = EXPLORATORY_DATA;
  1530.       }
  1531.       attrs.push_back(RmstTargetAttr.make(NRAttribute::IS, rmst_ptr->target_str_));
  1532.       attrs.push_back(RmstTsprtCtlAttr.make(NRAttribute::IS, RMST_RESP));
  1533.       attrs.push_back(RmstFragAttr.make(NRAttribute::IS, rmst_ptr->max_frag_));
  1534.       attrs.push_back(RmstIdAttr.make(NRAttribute::IS, rmst_ptr->rmst_no_));
  1535.       frag_ptr =  rmst_ptr->getFrag(rmst_ptr->max_frag_);
  1536.       attrs.push_back(RmstDataAttr.make(NRAttribute::IS, 
  1537.         frag_ptr, rmst_ptr->max_frag_len_));
  1538.       Message *new_frag;
  1539.       new_frag = new Message(DIFFUSION_VERSION, msg_type, 0, 0, attrs.size(), pkt_count_,
  1540.         rdm_id_, LOCALHOST_ADDR, LOCALHOST_ADDR);
  1541.       new_frag->msg_attr_vec_ = CopyAttrs(&attrs);
  1542.       ((DiffusionRouting *)dr_)->sendMessage(new_frag, filter_handle_);
  1543.       pkt_count_++;
  1544.       delete new_frag;
  1545.       ClearAttrs(&attrs);
  1546.       // We sent a fragment, set the last_data_time_ for the cleanup timer.
  1547.       GetTime(&rmst_ptr->last_data_time_);
  1548.     }
  1549.     return 0;
  1550.     break;
  1552.   case CLEANUP_TIMER:
  1553.     DiffPrint(DEBUG_SOME_DETAILS, "RmstFilter::processTimer CLEANUP_TIMER");
  1554.     PrintTime(&cur_time);
  1556.     DiffPrint(DEBUG_IMPORTANT, "  CLEANUP_TIMER calledn");
  1557.     rmst_iterator = rmst_map_.begin();
  1558.     while(rmst_iterator != rmst_map_.end()){
  1559.       rmst_ptr = (*rmst_iterator).second;
  1561.       DiffPrint(DEBUG_SOME_DETAILS, 
  1562.         "  CLEANUP_TIMER:: rmst_no %d : pkts_sent_ = %d, pkts_rec_ = %d, last_hop_pkts_sent_ = %dn",
  1563.         rmst_ptr->rmst_no_, rmst_ptr->pkts_sent_, rmst_ptr->pkts_rec_, rmst_ptr->last_hop_pkts_sent_);
  1565.       if((!rmst_ptr->reinf_)&&(!rmst_ptr->acked_)&&(!rmst_ptr->local_source_)&&(!local_sink_)){
  1566.         if( (cur_time.tv_sec - rmst_ptr->last_data_time_.tv_sec) > LONG_CLEANUP_WAIT )
  1567.           cleanUpRmst(rmst_ptr);
  1568.       }
  1569.       else if (rmst_ptr->acked_){
  1570.         if ( ( (cur_time.tv_sec - rmst_ptr->last_data_time_.tv_sec) > SHORT_CLEANUP_WAIT ) &&
  1571.            ( (cur_time.tv_sec - rmst_ptr->last_nak_time_.tv_sec) > SHORT_CLEANUP_WAIT ) )
  1572.            cleanUpRmst(rmst_ptr);
  1573.       }
  1574.             
  1575.       rmst_iterator++;
  1576.     }
  1578.     // Check on the BlackList, in case network is partitioned.
  1579.     if (!black_list_.empty()){
  1580.       if ( (cur_time.tv_sec - last_data_rec_.tv_sec) > RMST_BLACKLIST_WAIT ){
  1581.         DiffPrint(DEBUG_IMPORTANT, "  clearing black_list_!n");
  1582.         ((DiffusionRouting *)dr_)->clearBlacklist();
  1583.         black_list_.clear();
  1584.       }
  1585.     }
  1587.     if (local_sink_){
  1588.       if ( (cur_time.tv_sec - last_sink_time_.tv_sec) > SINK_REFRESH_WAIT ){
  1589.         DiffPrint(DEBUG_IMPORTANT, "  local sink timed outn");
  1590.         local_sink_ = false;
  1591.       }
  1592.       else
  1593.         DiffPrint(DEBUG_IMPORTANT, "  local sink still alive.n");
  1594.     }
  1596.     return 0;
  1597.     break;
  1599.   default:
  1600.     break;
  1601.   }
  1602.   return -1;
  1603. }
  1605. void RmstFilter::sendRmstToSink(Rmst *rmst_ptr)
  1606. {
  1607.   NRAttrVec attrs;
  1608.   Message *rmst_msg;
  1609.   NRSimpleAttribute<void *> *rmst_data_attr;
  1610.   NRSimpleAttribute<int> *frag_number_attr;
  1611.   void *frag_ptr;
  1612.   int size, i;
  1614.   DiffPrint(DEBUG_IMPORTANT, "RmstFilter::sendRmstToSink - sending rmst %d to local sinkn",
  1615.     rmst_ptr->rmst_no_);
  1617.   // Prepare attribute vector
  1618.   attrs.push_back(RmstTargetAttr.make(NRAttribute::IS, rmst_ptr->target_str_));
  1619.   attrs.push_back(RmstTsprtCtlAttr.make(NRAttribute::IS, RMST_RESP));
  1620.   attrs.push_back(RmstIdAttr.make(NRAttribute::IS, rmst_ptr->rmst_no_));
  1621.   attrs.push_back(RmstMaxFragAttr.make(NRAttribute::IS, rmst_ptr->max_frag_));
  1622.   frag_number_attr = RmstFragAttr.make(NRAttribute::IS, 0);
  1623.   attrs.push_back(frag_number_attr);
  1624.   // Add the blob fragment
  1625.   if (rmst_ptr->max_frag_ == 0)
  1626.     size = rmst_ptr->max_frag_len_;
  1627.   else
  1628.     size = MAX_FRAG_SIZE;
  1629.   frag_ptr =  rmst_ptr->getFrag(0);
  1630.   rmst_data_attr = RmstDataAttr.make(NRAttribute::IS, frag_ptr, size);
  1631.   attrs.push_back(rmst_data_attr);
  1633.   // Prepare the message
  1634.   rmst_msg = new Message(DIFFUSION_VERSION, DATA, 0, 0, attrs.size(),
  1635.     pkt_count_, rdm_id_, LOCALHOST_ADDR, LOCALHOST_ADDR);
  1636.   rmst_msg->next_hop_ = LOCALHOST_ADDR;
  1637.   rmst_msg->next_port_ = local_sink_port_;
  1638.   rmst_msg->msg_attr_vec_= CopyAttrs(&attrs);
  1639.   ((DiffusionRouting *)dr_)->sendMessage(rmst_msg, filter_handle_, 1);
  1640.   delete rmst_msg;
  1641.   pkt_count_++;
  1643.   // Send all the fragments
  1644.   for (i=1; i <= (rmst_ptr->max_frag_); i++){
  1645.     frag_number_attr->setVal(i);
  1646.     frag_ptr =  rmst_ptr->getFrag(i);
  1647.     if(frag_ptr == NULL)
  1648.       DiffPrint(DEBUG_IMPORTANT, "RmstFilter::sendRmstToSink - got a null frag_ptr for frag!%dn",
  1649.       i);
  1650.     else{
  1651.       if (rmst_ptr->max_frag_ == i)
  1652.         rmst_data_attr->setVal(frag_ptr, rmst_ptr->max_frag_len_);
  1653.       else
  1654.         rmst_data_attr->setVal(frag_ptr, MAX_FRAG_SIZE);
  1655.     }
  1656.     rmst_msg = new Message(DIFFUSION_VERSION, DATA, 0, 0, attrs.size(),
  1657.       pkt_count_, rdm_id_, LOCALHOST_ADDR, LOCALHOST_ADDR);
  1658.     rmst_msg->next_hop_ = LOCALHOST_ADDR;
  1659.     rmst_msg->next_port_ = local_sink_port_;
  1660.     rmst_msg->msg_attr_vec_= CopyAttrs(&attrs);
  1661.     ((DiffusionRouting *)dr_)->sendMessage(rmst_msg, filter_handle_, 1);
  1662.     delete rmst_msg;
  1663.     pkt_count_++;
  1664.   }
  1666.   ClearAttrs(&attrs);
  1667. }
  1669. void RmstFilter::sendAckToSource(Rmst *rmst_ptr)
  1670. {
  1671.   NRAttrVec attrs;
  1672.   Message *ack_msg;
  1674.   attrs.push_back(RmstTsprtCtlAttr.make(NRAttribute::IS, ACK_RESP));
  1675.   attrs.push_back(RmstIdAttr.make(NRAttribute::IS, rmst_ptr->rmst_no_));
  1677.   // New code to send a message to last_hop_
  1678.   ack_msg = new Message(DIFFUSION_VERSION, DATA, 0, 0,
  1679.             attrs.size(), pkt_count_, rdm_id_, rmst_ptr->last_hop_,
  1680.             LOCALHOST_ADDR);
  1681.   ack_msg->msg_attr_vec_ = CopyAttrs(&attrs);
  1683.   DiffPrint(DEBUG_IMPORTANT, "  Sending ACK_RESP to node %dn", rmst_ptr->last_hop_);
  1684.   ((DiffusionRouting *)dr_)->sendMessage(ack_msg, filter_handle_, 1);
  1685.   pkt_count_++;
  1686.   delete ack_msg;
  1687.   ClearAttrs(&attrs);
  1688. }
  1690. void RmstFilter::sendExpReqUpstream(Rmst *rmst_ptr)
  1691. {
  1692.   NRAttrVec attrs;
  1693.   Message *exp_msg;
  1695.   attrs.push_back(RmstTsprtCtlAttr.make(NRAttribute::IS, EXP_REQ));
  1696.   attrs.push_back(RmstIdAttr.make(NRAttribute::IS, rmst_ptr->rmst_no_));
  1697.   attrs.push_back(RmstFragAttr.make(NRAttribute::IS, rmst_ptr->max_frag_rec_));
  1699.   exp_msg = new Message(DIFFUSION_VERSION, DATA, 0, 0,
  1700.             attrs.size(), pkt_count_, rdm_id_, rmst_ptr->last_hop_,
  1701.             LOCALHOST_ADDR);
  1702.   exp_msg->msg_attr_vec_ = CopyAttrs(&attrs);
  1704.   DiffPrint(DEBUG_IMPORTANT, "  Sending EXP_REQ to node %dn", rmst_ptr->last_hop_);
  1705.   ((DiffusionRouting *)dr_)->sendMessage(exp_msg, filter_handle_, 1);
  1706.   pkt_count_++;
  1707.   delete exp_msg;
  1708.   ClearAttrs(&attrs);
  1709. }
  1711. void RmstFilter::sendContToSource(Rmst *rmst_ptr)
  1712. {
  1713.   NRAttrVec attrs;
  1714.   Message *cont_msg;
  1715.   attrs.push_back(RmstTsprtCtlAttr.make(NRAttribute::EQ, RMST_CONT));
  1716.   attrs.push_back(NRClassAttr.make(NRAttribute::IS,
  1717.     NRAttribute::INTEREST_CLASS));
  1718.   DiffPrint(DEBUG_IMPORTANT, "  Sending a RMST_CONT to sourcen");
  1719.   cont_msg = new Message(DIFFUSION_VERSION, DATA, 0, 0, attrs.size(),
  1720.     pkt_count_, rdm_id_, LOCALHOST_ADDR, LOCALHOST_ADDR);
  1721.   cont_msg->msg_attr_vec_ = CopyAttrs(&attrs);
  1722.   cont_msg->next_hop_ = LOCALHOST_ADDR;
  1723.   cont_msg->next_port_ = rmst_ptr->local_source_port_;
  1724.   ((DiffusionRouting *)dr_)->sendMessage(cont_msg, filter_handle_, 1);
  1725.   pkt_count_++;
  1726.   delete cont_msg;
  1727.   ClearAttrs(&attrs);
  1728. }
  1730. void RmstFilter::cleanUpRmst(Rmst *rmst_ptr)
  1731. {
  1732.   int rmst_no = rmst_ptr->rmst_no_;
  1734.   Int2Rmst::iterator rmst_iterator;
  1735.   Key2ExpLog::iterator exp_iterator;
  1736.   ExpLog exp_msg;
  1737.   rmst_no = rmst_ptr->rmst_no_;
  1738.   DiffPrint(DEBUG_IMPORTANT, "  cleanUpRmst called to delete Rmst %dn", rmst_no);
  1740.   if(rmst_ptr->watchdog_active_)
  1741.     ((DiffusionRouting *)dr_)->removeTimer(rmst_ptr->watchdog_handle_);
  1742.   if(rmst_ptr->ack_timer_active_)
  1743.     ((DiffusionRouting *)dr_)->removeTimer(rmst_ptr->ack_timer_handle_);
  1745.   rmst_iterator = rmst_map_.find(rmst_no);
  1746.   if(rmst_iterator != rmst_map_.end()){
  1747.     rmst_map_.erase(rmst_iterator);
  1748.   }
  1749.   delete rmst_ptr;
  1751.   // clean up the exp_map_ of any entries base on this rmst
  1752.   exp_iterator = exp_map_.begin();
  1753.   while(exp_iterator != exp_map_.end()){
  1754.     exp_msg = (*exp_iterator).second;
  1755.     if(exp_msg.rmst_no_ == rmst_no){
  1756.       DiffPrint(DEBUG_LOTS_DETAILS, "  cleanUpRmst deleting exp_map_ entry for Rmst %dn", rmst_no);
  1757.       exp_map_.erase(exp_iterator);
  1758.     }
  1759.     exp_iterator++;
  1760.   }
  1761. }
  1763. #ifdef NS_DIFFUSION
  1764. RmstFilter::RmstFilter()
  1765. {
  1766. #else
  1767. RmstFilter::RmstFilter(int argc, char **argv)
  1769. {
  1770.   TimerCallback *stat_timer;
  1772.   parseCommandLine(argc, argv);
  1773.   dr_ = NR::createNR(diffusion_port_);
  1774. #endif // NS_DIFFUSION
  1776.   fcb_ = new RmstFilterCallback;
  1777.   fcb_->app_ = this;
  1778.   rdm_id_ = rand();
  1779.   pkt_count_ = rand();
  1780.   local_sink_ = false;
  1781.   caching_mode_ = false;
  1782.   send_timer_active_ = false;
  1784.   DiffPrint(DEBUG_ALWAYS, "RmstFilter constructor: rdm_id_ = %x, pkt_count_ = %xn",
  1785.     rdm_id_, pkt_count_);
  1787. #ifndef NS_DIFFUSION
  1788.   filter_handle_ = setupFilter();
  1789.   DiffPrint(DEBUG_LOTS_DETAILS, "RmstFilter:: subscribed to all, received handle %dn",
  1790.     (int)filter_handle_);
  1791.   
  1792.   DiffPrint(DEBUG_LOTS_DETAILS, "RmstFilter constructor: start cleanup timern");
  1793.   stat_timer = new RmstTimeout(this, -1, CLEANUP_TIMER);
  1794.   stat_timer_handle_ = ((DiffusionRouting *)dr_)->addTimer(CLEANUP_INTERVAL, stat_timer);
  1795. #endif // !NS_DIFFUSION
  1796. }
  1798. #ifndef NS_DIFFUSION
  1799. int main(int argc, char **argv)
  1800. {
  1801.   RmstFilter *app;
  1802.   app = new RmstFilter(argc, argv);
  1803.   app->run();
  1805.   return 0;
  1806. }
  1807. #endif // !NS_DIFFUSION