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

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

通讯编程

开发平台：

Visual C++

pgm-agent.cc：源码内容
  2. /*
  3.  * pgm-agent.cc
  4.  * Copyright (C) 2001 by the University of Southern California
  5.  * $Id: pgm-agent.cc,v 1.10 2006/02/21 15:20:19 mahrenho Exp $
  6.  *
  7.  * This program is free software; you can redistribute it and/or
  8.  * modify it under the terms of the GNU General Public License,
  9.  * version 2, as published by the Free Software Foundation.
  10.  *
  11.  * This program is distributed in the hope that it will be useful,
  12.  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13.  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14.  * GNU General Public License for more details.
  15.  *
  16.  * You should have received a copy of the GNU General Public License along
  17.  * with this program; if not, write to the Free Software Foundation, Inc.,
  18.  * 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA.
  19.  *
  20.  *
  21.  * The copyright of this module includes the following
  22.  * linking-with-specific-other-licenses addition:
  23.  *
  24.  * In addition, as a special exception, the copyright holders of
  25.  * this module give you permission to combine (via static or
  26.  * dynamic linking) this module with free software programs or
  27.  * libraries that are released under the GNU LGPL and with code
  28.  * included in the standard release of ns-2 under the Apache 2.0
  29.  * license or under otherwise-compatible licenses with advertising
  30.  * requirements (or modified versions of such code, with unchanged
  31.  * license).  You may copy and distribute such a system following the
  32.  * terms of the GNU GPL for this module and the licenses of the
  33.  * other code concerned, provided that you include the source code of
  34.  * that other code when and as the GNU GPL requires distribution of
  35.  * source code.
  36.  *
  37.  * Note that people who make modified versions of this module
  38.  * are not obligated to grant this special exception for their
  39.  * modified versions; it is their choice whether to do so.  The GNU
  40.  * General Public License gives permission to release a modified
  41.  * version without this exception; this exception also makes it
  42.  * possible to release a modified version which carries forward this
  43.  * exception.
  44.  *
  45.  */
  47. /*
  48.  * Pragmatic General Multicast (PGM), Reliable Multicast
  49.  *
  50.  * pgm-agent.cc
  51.  *
  52.  * This implements the network element PGM agent, "Agent/PGM".
  53.  *
  54.  * Ryan S. Barnett, 2001
  55.  * rbarnett@catarina.usc.edu
  56.  */
  58. #include "config.h"
  59. #ifdef HAVE_STL
  61. #include <stdlib.h>
  62. #include <stdio.h>
  64. #include <map>
  65. #include <list>
  66. #include <algorithm>
  68. #include "config.h"
  69. #include "tclcl.h"
  70. #include "agent.h"
  71. #include "packet.h"
  72. #include "ip.h"
  73. #include "random.h"
  74. #include "basetrace.h"
  76. #include "pgm.h"
  78. // ************************************************************
  79. // Register the PGM packet headers.
  80. // ************************************************************
  82. // Declare the static header offsets.
  83. int hdr_pgm::offset_;
  84. int hdr_pgm_spm::offset_;
  85. int hdr_pgm_nak::offset_;
  87. // Register the hdr_pgm with the packet header manager.
  88. class PGMHeaderClass : public PacketHeaderClass {
  89. public:
  90.   PGMHeaderClass() : PacketHeaderClass("PacketHeader/PGM", sizeof(hdr_pgm)) {
  91.     bind_offset(&hdr_pgm::offset_);
  92.   }
  94. } class_pgmhdr;
  96. // Register the hdr_pgm_spm with the packet header manager.
  97. class PGM_SPMHeaderClass : public PacketHeaderClass {
  98. public:
  99.   PGM_SPMHeaderClass() : PacketHeaderClass("PacketHeader/PGM_SPM",
  100.    sizeof(hdr_pgm_spm)) {
  101.     bind_offset(&hdr_pgm_spm::offset_);
  102.   }
  104. } class_pgm_spmhdr;
  106. // Register the hdr_pgm_nak with the packet header manager.
  107. class PGM_NAKHeaderClass : public PacketHeaderClass {
  108. public:
  109.   PGM_NAKHeaderClass() : PacketHeaderClass("PacketHeader/PGM_NAK",
  110.    sizeof(hdr_pgm_nak)) {
  111.     bind_offset(&hdr_pgm_nak::offset_);
  112.   }
  114. } class_pgm_nakhdr;
  117. // ************************************************************
  118. // Define the PGM Agent Timer Class
  119. // ************************************************************
  120. class PgmAgent;
  122. // Different timer types.
  123. enum {
  124.   TIMER_NAK_RETRANS = 0,
  125.   TIMER_NAK_RPT = 1,
  126.   TIMER_NAK_RDATA = 2,
  127.   TIMER_NAK_ELIM = 3
  128. };
  130. class PgmAgentTimer : public TimerHandler {
  131. public:
  132.   PgmAgentTimer(PgmAgent *a, int type) : TimerHandler(), data_(NULL) {
  133.     a_ = a;
  134.     type_ = type;
  135.   }
  137.   void * &data() { return data_; }
  139. protected:
  140.   virtual void expire(Event *e);
  141.   PgmAgent *a_;
  142.   int type_;
  143.   void *data_;
  144. };
  146. // ************************************************************
  147. // Define the Repair State control block.
  148. // ************************************************************
  150. // Different repair states.
  151. enum {
  152.   NAK_PENDING = 0,
  153.   NAK_CONFIRMED = 1
  154. };
  156. class StateInfo;
  158. class RepairState {
  159. public:
  160.   RepairState(PgmAgent *a, StateInfo *sinfo, int seqno, ns_addr_t &source,
  161.       ns_addr_t &group) : seqno_(seqno), source_(source),
  162.   group_(group),
  163.         nak_state_(NAK_PENDING), nak_elimination_(true),
  164. nak_retrans_timer_(a, TIMER_NAK_RETRANS),
  165. nak_rpt_timer_(a, TIMER_NAK_RPT),
  166. nak_rdata_timer_(a, TIMER_NAK_RDATA),
  167. nak_elim_timer_(a, TIMER_NAK_ELIM),
  168.         sinfo_(sinfo)
  169.   { }
  171.   int & seqno() { return seqno_; }
  172.   ns_addr_t & source() { return source_; }
  173.   ns_addr_t & group() { return group_; }
  175.   int & nak_state() { return nak_state_; }
  176.   bool & nak_elimination() { return nak_elimination_; }
  177.   PgmAgentTimer & nak_retrans_timer() { return nak_retrans_timer_; }
  178.   PgmAgentTimer & nak_rpt_timer() { return nak_rpt_timer_; }
  179.   PgmAgentTimer & nak_rdata_timer() { return nak_rdata_timer_; }
  180.   PgmAgentTimer & nak_elim_timer() { return nak_elim_timer_; }
  181.   list<int> & iface_list() { return iface_list_; }
  182.   list<NsObject *> & agent_list() { return agent_list_; }
  183.   StateInfo * & sinfo() { return sinfo_; }
  185. protected:
  187.   // Which sequence number is being requested for repair.
  188.   int seqno_;
  190.   ns_addr_t source_; // Original source of ODATA for the repair.
  191.   ns_addr_t group_; // The multicast group.
  193.   int nak_state_; // Present repair block state.
  195.   // Indicates whether or not we are to discard incoming NAK packets
  196.   // once a previous NAK is outstanding (got NCF, waiting for RDATA). 
  197.   // (See 7.4 of PGM specification) By default we do. When nak_elim_timer_
  198.   // expires, then we do not.
  199.   bool nak_elimination_;
  201.   // This timer controls sending retransmissions of NAK packets.
  202.   PgmAgentTimer nak_retrans_timer_;
  204.   // Timer that measures how long we can repeat NAK packets while waiting
  205.   // for NCF confirmation. Once expires, the repair state is removed.
  206.   PgmAgentTimer nak_rpt_timer_;
  208.   // Timer that is triggered waiting for RDATA for a given NAK seqno,
  209.   // provided that NAK has been confirmed by an NCF.  Only gets set
  210.   // once NCF is received.
  211.   PgmAgentTimer nak_rdata_timer_;
  213.   // Timer that is triggered when we disable nak_elimination_, allowing
  214.   // a duplicate NAK to be processed. This occurs after a previous
  215.   // NAK has been confirmed with an NCF, but before the RDATA has been
  216.   // received.  This timer gets set when we receive an NCF for a pending
  217.   // NAK.
  218.   PgmAgentTimer nak_elim_timer_;
  220.   // List of interfaces upon which the RDATA will be sent to.
  221.   list<int> iface_list_;
  223.   // List of agents upon which the RDATA will be sent to.
  224.   list<NsObject *> agent_list_;
  226.   // Back-pointer to the state information block that is holding this
  227.   // repair data.  We use this so we can get the upstream_path and the TSI.
  228.   StateInfo *sinfo_;
  230. };
  232. // ************************************************************
  233. // Define the TSI State control block.
  234. // ************************************************************
  236. class StateInfo {
  237. public:
  239.   StateInfo(ns_addr_t tsi) : tsi_(tsi), spm_seqno_(-1) { }
  241.   // Only used if the container holding StateInfo's will be in sorted order.
  242.   int operator<(const StateInfo &right) const {
  243.     return ((tsi_.addr_ < right.tsi_.addr_) || ( (tsi_.addr_==right.tsi_.addr_) && (tsi_.port_ < right.tsi_.port_)));
  244.   }
  246.   ns_addr_t & tsi() { return tsi_; }
  247.   int & spm_seqno() { return spm_seqno_; }
  248.   ns_addr_t & upstream_node() { return upstream_node_; }
  249.   int & upstream_iface() { return upstream_iface_; }
  250.   map<int, RepairState> & repair() { return repair_; }
  252. protected:
  254.   ns_addr_t tsi_; // Transport Session ID
  255.   int spm_seqno_; // Most recent SPM sequence number.
  256.   ns_addr_t upstream_node_; // Upstream node address.
  257.   int upstream_iface_; // Upstream interface number.
  259.   // Map between a NAK sequence number and the corresponding repair state
  260.   // for that sequence number.
  261.   map<int, RepairState> repair_;
  263. };
  266. // ************************************************************
  267. // Define the PGM Agent Class
  268. // ************************************************************
  270. // Structure to hold statistical information for PGM Agent.
  271. struct Stats {
  272.   int num_unsolicited_ncf_;
  273.   int num_unsolicited_rdata_;
  274.   int num_suppressed_naks_;
  275.   int num_naks_transmitted_;
  276. };
  278. // Used to count number of unique pgm agents.
  279. static int pgm_agent_uid_ = 0;
  281. class PgmAgent : public Agent {
  282. public:
  283.   PgmAgent();
  284.   virtual void recv(Packet *, Handler *);
  285.   virtual void timeout(int type, void *data);
  286.   virtual int command(int argc, const char*const* argv);
  288. protected:
  290.   void handle_spm(Packet *pkt);
  291.   void handle_odata(Packet *pkt);
  292.   void handle_rdata(Packet *pkt);
  293.   void handle_nak(Packet *pkt);
  294.   void handle_ncf(Packet *pkt);
  296.   void send_nak(ns_addr_t &upstream_node, ns_addr_t &tsi, int seqno, ns_addr_t &source, ns_addr_t &group);
  298.   void timeout_nak_retrans(RepairState *rstate);
  299.   void timeout_nak_rpt(RepairState *rstate);
  300.   void timeout_nak_rdata(RepairState *rstate);
  301.   void timeout_nak_elim(RepairState *rstate);
  303.   void remove_repair_state(RepairState *rstate);
  305.   void print_stats();
  307.   void trace_event(char *evType, double evTime);
  309. #ifdef PGM_DEBUG
  310.   void display_packet(Packet *pkt);
  311. #endif
  313.   NsObject* iface2link(int iface);
  314.   NsObject* pkt2agent (Packet *pkt);
  316.   StateInfo * find_TSI(ns_addr_t &tsi);
  317.   StateInfo * insert_TSI(ns_addr_t &tsi);
  319.   EventTrace * et_; //Trace Object for custom event trace
  321.   int pgm_enabled_; // Is this agent enabled? Default is YES.
  323.   char uname_[16]; // Agent's unique name.
  325.   Stats stats_; // Statistical information.
  327.   // TSI-indexed state control block list.
  328.   list<StateInfo> state_list_;
  330.   // Number of seconds to wait between retransmitting a NAK that is waiting
  331.   // for a NCF packet.
  332.   double nak_retrans_ival_;
  334.   // The length of time for which a network element will continue to repeat
  335.   // NAKs while waiting for a corresponding NCF.  Once this time expires and
  336.   // no NCF is received, then we remove the entire repair state.
  337.   double nak_rpt_ival_;
  339.   // The length of time for which a network element will wait for the
  340.   // corresponding RDATA before removing the entire repair state.
  341.   double nak_rdata_ival_;
  343.   // Once a NAK has been confirmed, network elements must discard all
  344.   // further NAKs for up to this length of time.  Should be a fraction
  345.   // of nak_rdata_ival_.
  346.   double nak_elim_ival_;
  348. };
  350. static class PgmClass : public TclClass {
  351. public:
  352.   PgmClass() : TclClass("Agent/PGM") {}
  353.   TclObject * create(int argc, const char * const * argv) {
  354.     return (new PgmAgent());
  355.   }
  356. } class_pgm_agent;
  358. void PgmAgentTimer::expire(Event *e) {
  359.   a_->timeout(type_, data_);
  360. }
  362. // Constructor.
  363. PgmAgent::PgmAgent() : Agent(PT_PGM), pgm_enabled_(1)
  364. {
  365.   // Set the unique identifier.
  366.   sprintf (uname_, "pgmAgent-%d", pgm_agent_uid_++);
  368.   // Initialize statistics.
  369.   stats_.num_unsolicited_ncf_ = 0;
  370.   stats_.num_unsolicited_rdata_ = 0;
  371.   stats_.num_suppressed_naks_ = 0;
  372.   stats_.num_naks_transmitted_ = 0;
  374.   bind("pgm_enabled_", &pgm_enabled_);
  375.   bind_time("nak_retrans_ival_", &nak_retrans_ival_);
  376.   bind_time("nak_rpt_ival_", &nak_rpt_ival_);
  377.   bind_time("nak_rdata_ival_", &nak_rdata_ival_);
  378.   bind_time("nak_elim_ival_", &nak_elim_ival_);
  380.   et_ = (EventTrace *) NULL;
  381. }
  383. // Code to execute when a packet is received.
  384. void PgmAgent::recv(Packet* pkt, Handler*)
  385. {
  386.   hdr_pgm *hp = HDR_PGM(pkt);
  388.   if (!pgm_enabled_) {
  389.     target_->recv(pkt);
  390.     return;
  391.   }
  393.   hdr_cmn *hc = HDR_CMN(pkt);
  395.   if (hc->ptype_ != PT_PGM) {
  396.     printf("%s ERROR (PgmAgent::recv): received non PGM pkt type %d, discarding.n", uname_, hc->ptype_);
  397.     Packet::free(pkt);
  398.     return;
  399.   }
  401. #ifdef PGM_DEBUG
  402.   display_packet(pkt);
  403. #endif
  405.   // Note, each handle function will free the packet itself or modify the
  406.   // headers and pass it on to the next NS object.
  408.   switch(hp->type_) {
  409.   case PGM_SPM:
  410.     handle_spm(pkt);
  411.     break;
  412.   case PGM_ODATA:
  413.     handle_odata(pkt);
  414.     break;
  415.   case PGM_RDATA:
  416.     handle_rdata(pkt);
  417.     break;
  418.   case PGM_NAK:
  419.     handle_nak(pkt);
  420.     break;
  421.   case PGM_NCF:
  422.     handle_ncf(pkt);
  423.     break;
  424.   default:
  425.     printf("ERROR (PgmAgent::recv): Received PGM packet with unknown type %d.n", hp->type_);
  427.     Packet::free(pkt);
  429.     break;
  430.   }
  432. }
  434. // Code to execute when a timeout occurs.
  435. void PgmAgent::timeout(int type, void *data)
  436. {
  438.   switch(type) {
  439.   case TIMER_NAK_RETRANS:
  440.     timeout_nak_retrans((RepairState *) data);
  441.     break;
  442.   case TIMER_NAK_RPT:
  443.     timeout_nak_rpt((RepairState *) data);
  444.     break;
  445.   case TIMER_NAK_RDATA:
  446.     timeout_nak_rdata((RepairState *) data);
  447.     break;
  448.   case TIMER_NAK_ELIM:
  449.     timeout_nak_elim((RepairState *) data);
  450.     break;
  451.   default:
  452.     printf("ERROR (PgmAgent::timeout()): Invalid timeout type %d.n", type);
  453.     break;
  454.   }
  456. }
  458. // Code to execute when a TCL command is issued to the PGM Agent object.
  459. int PgmAgent::command (int argc, const char*const* argv)
  460. {
  461.   //  Tcl& tcl = Tcl::instance();
  463.   if (argc == 2) {
  464.     if (strcmp(argv[1], "print-stats") == 0) {
  465.       print_stats();
  466.       return (TCL_OK);
  467.     }
  468.   }
  469.   else if (argc == 3) { //Set the Event Trace handle if Event Tracing is on
  470.     if (strcmp(argv[1], "eventtrace") == 0) {
  471.       et_ = (EventTrace *)TclObject::lookup(argv[2]);
  472.       return (TCL_OK);
  473.     }
  474.   }
  476.   return (Agent::command(argc, argv));
  477. }      
  479. void PgmAgent::trace_event(char *evType, double evTime) {
  481.   if (et_ == NULL) return;
  482.   char *wrk = et_->buffer();
  483.   char *nwrk = et_->nbuffer();
  485.   if (wrk != NULL) {
  486.     sprintf(wrk, "E "TIME_FORMAT" %d %d PGM %s "TIME_FORMAT, 
  487.             et_->round(Scheduler::instance().clock()),   
  488.             addr(),                    
  489.             0,                   
  490.             evType,                  
  491. evTime);
  492.   if (nwrk != 0)
  493.     sprintf(nwrk,
  494. "E -t "TIME_FORMAT" -o PGM -e %s -s %d.%d -d %d.%d",
  495. et_->round(Scheduler::instance().clock()),   // time
  496. evType,                    // event type
  497. addr(),                       // owner (src) node id
  498. port(),                       // owner (src) port id
  499. 0,                      // dst node id
  500. 0                       // dst port id
  501. );
  502. et_->dump();
  503.   }
  505. }
  507. void PgmAgent::handle_spm(Packet *pkt)
  508. {
  510.   hdr_pgm *hp = HDR_PGM(pkt);
  511.   hdr_pgm_spm *hs = HDR_PGM_SPM(pkt);
  512.   hdr_cmn *hc = HDR_CMN(pkt);
  514.   // Use the TSI from the SPM packet and locate the proper state block.
  515.   StateInfo *state = find_TSI(hp->tsi_);
  517.   if (state == NULL) {
  518.     // There is no state block for this TSI. Create new state.
  519.     state = insert_TSI(hp->tsi_);
  521.     // Set the sequence number.
  522.     state->spm_seqno() = hp->seqno_;
  523.     // Set the upstream path.
  524.     state->upstream_node() = hs->spm_path_;
  525.     state->upstream_iface() = hc->iface();
  526.   }
  527.   else {
  528.     // State already exists for this TSI. Check if the sequence number is
  529.     // newer than the last recorded sequence number.
  531.     if ( state->spm_seqno() < hp->seqno_ ) {
  532.       // Update the SPM sequence number.
  533.       state->spm_seqno() = hp->seqno_;
  534.       // Set the upstream path.
  535.       state->upstream_node() = hs->spm_path_;
  536.       state->upstream_iface() = hc->iface();
  537.     }
  538.     else {
  539.       printf("%s received an old SPM seqno, discarding.n", uname_);
  540.       Packet::free(pkt);
  541.       return;
  542.     }
  543.   }
  545.   // Modify the SPM packet and set the upstream path to be equal to
  546.   // the address of this agent.
  547.   hs->spm_path_ = here_;
  549.   // Send the modified packet off to the rest of the multicast group.
  550.   send(pkt, 0);
  551. }
  553. void PgmAgent::handle_odata(Packet *pkt)
  554. {
  555.   // Pass the ODATA along to the rest of the multicast group.  ODATA
  556.   // does not cancel NAK forwarding.
  558.   //hdr_cmn *hc = HDR_CMN(pkt);
  560.   send(pkt, 0);
  561. }
  563. void PgmAgent::handle_rdata(Packet *pkt)
  564. {
  566.   // Look for the TSI for this RDATA packet.
  567.   hdr_pgm *hp = HDR_PGM(pkt);
  568.   //  hdr_ip *hip = HDR_IP(pkt);
  570.   StateInfo *state = find_TSI(hp->tsi_);
  571.   if (state == NULL) {
  572.     printf("%s received RDATA for which no SPM state is established, discarding.n", uname_);
  573.     stats_.num_unsolicited_rdata_++;
  574.     Packet::free(pkt);
  575.     return;
  576.   }
  578.   // Look for the repair state for this RDATA packet.
  579.   map<int, RepairState>::iterator result = state->repair().find(hp->seqno_);
  581.   if (result == state->repair().end()) {
  582.     // No repair state present for this RDATA packet.
  583.     printf("%s received RDATA for which no repair state is present, discarding.n", uname_);
  584.     stats_.num_unsolicited_rdata_++;
  585.     Packet::free(pkt);
  586.     return;
  587.   }
  589.   RepairState *rstate = &((*result).second);
  591.   // Get the interface list for the repair state. For each interface, send
  592.   // out the RDATA packet.  Similarly for each agent that is also receiving
  593.   // RDATA attached to this node.
  595.   if (rstate->iface_list().empty() && rstate->agent_list().empty()) {
  596.     printf("%s received RDATA but repair state has no interfaces for it, discarding.n", uname_);
  598.     stats_.num_unsolicited_rdata_++;
  600.     Packet::free(pkt);
  601.   }
  603.   NsObject *tgt;
  604.   Packet *new_pkt;
  605.   int flag = 0;
  607.   trace_event("SEND RDATA", 0); //Repair is being forwarded
  609.   //  hdr_cmn *hc = HDR_CMN(pkt);
  611.   if (!rstate->iface_list().empty()) {
  612.     list<int>::iterator iter = rstate->iface_list().begin();
  614.     while (iter != rstate->iface_list().end()) {
  615.       if (!flag) {
  616. tgt = iface2link(*iter);
  617. if (tgt == NULL) {
  618.   printf("ERROR (PgmAgent::handle_rdata): iface2link returned NULL.n");
  619.   abort();
  620. }
  621. tgt->recv(pkt);
  622. flag = 1;
  623.       }
  624.       else {
  625. // Make a copy of each packet before sending it, so we don't free()
  626. // the same packet at the different receivers causing a deallocation
  627. // problem.
  628. new_pkt = pkt->copy();
  629. tgt = iface2link(*iter);
  630. if (tgt == NULL) {
  631.   printf("ERROR (PgmAgent::handle_rdata): iface2link returned NULL.n");
  632.   abort();
  633. }
  634. tgt->recv(new_pkt);
  635.       }
  637.       iter++;
  638.     }
  639.   }
  641.   if (!rstate->agent_list().empty()) {
  642.     list<NsObject *>::iterator iter = rstate->agent_list().begin();
  644.     while (iter != rstate->agent_list().end()) {
  645.       if (!flag) {
  646. (*iter)->recv(pkt);
  647. flag = 1;
  648.       }
  649.       else {
  650. // Make a copy of each packet before sending it, so we don't free()
  651. // the same packet at the different receivers causing a deallocation
  652. // problem.
  653. new_pkt = pkt->copy();
  654. (*iter)->recv(new_pkt);
  655.       }
  657.       iter++;
  658.     }
  659.   }
  661.   // Remove the repair state for this RDATA sequence number, since we sent
  662.   // out the repairs.
  663.   remove_repair_state(&((*result).second));
  665. }
  667. void PgmAgent::handle_nak(Packet *pkt)
  668. {
  670.   hdr_pgm *hp = HDR_PGM(pkt);
  671.   hdr_pgm_nak *hpn = HDR_PGM_NAK(pkt);
  672.   hdr_cmn *hc = HDR_CMN(pkt);
  673.   //  hdr_ip *hip = HDR_IP(pkt);
  675.   // Check to see if there is a state control block for the given TSI.
  676.   StateInfo *state = find_TSI(hp->tsi_);
  678.   if (state == NULL) {
  679.     printf("PGM Agent received NAK for which no SPM state is established, discarding.n");
  680.     Packet::free(pkt);
  681.     return;
  682.   }
  684.   // Create an NCF packet in response to the NAK packet.
  685.   Packet *ncf_pkt = allocpkt();
  686.   hdr_cmn *ncf_hc = HDR_CMN(ncf_pkt);
  687.   ncf_hc->size_ = sizeof(hdr_pgm); // Size of NCF packet.
  688.   ncf_hc->ptype_ = PT_PGM;
  689.   hdr_pgm *ncf_hp = HDR_PGM(ncf_pkt);
  690.   ncf_hp->type_ = PGM_NCF;
  691.   ncf_hp->tsi_ = hp->tsi_;
  692.   ncf_hp->seqno_ = hp->seqno_;
  694.   // Change the source of the NCF packet to be the original ODATA source.
  695.   hdr_ip *ncf_ip = HDR_IP(ncf_pkt);
  696.   ncf_ip->src() = hpn->source_;
  697.   // Set the destination to be the multicast group.
  698.   ncf_ip->dst() = hpn->group_;
  700.   // Set the color of NCF packets in nam to be green.
  701.   ncf_ip->fid_ = 6;
  703.   // Send out the NCF to the interface (or agent) for which the NAK was
  704.   // received.
  705.   NsObject *tgt;
  707.   if (hc->iface() < 0) {
  708.     tgt = pkt2agent(pkt);
  709.     if (tgt == NULL) {
  710.       printf("ERROR: (PgmAgent::handle_nak) pkt2agent returned NULL.n");
  711.       abort();
  712.     }
  713.     tgt->recv(ncf_pkt);
  714.   }
  715.   else {
  716.     tgt = iface2link(hc->iface());
  717.     if (tgt == NULL) {
  718.       printf("ERROR: (PgmAgent::handle_nak) iface2link returned NULL.n");
  719.       abort();
  720.     }
  721.     tgt->recv(ncf_pkt);
  722.   }
  724.   // Create repair state for the NAK query. Associate the sequence number
  725.   // of the NAK packet with the interface where the packet was received.
  726.   pair<map<int, RepairState>::iterator, bool> result;
  728.   result = state->repair().insert(pair<int, RepairState>(hp->seqno_, RepairState(this, state, hp->seqno_, hpn->source_, hpn->group_)));
  730.   RepairState *rstate = &(result.first->second);
  732.   if (result.second == true) {
  733.     // There was no previous repair state for the given NAK seqno.
  734.     // This must be a new NAK.
  736.     // Set the data fields of the timer.
  737.     rstate->nak_retrans_timer().data() = rstate;
  738.     rstate->nak_rpt_timer().data() = rstate;
  739.     rstate->nak_rdata_timer().data() = rstate;
  740.     rstate->nak_elim_timer().data() = rstate;
  742.     // Add the interface (or agent) to the interface list.
  743.     if (hc->iface() < 0) {
  744.       rstate->agent_list().push_back(pkt2agent(pkt));
  745.     }
  746.     else {
  747.       rstate->iface_list().push_back(hc->iface());
  748.     }
  750.     // Start the nak retransmission cycle time.
  751.     rstate->nak_retrans_timer().resched(nak_retrans_ival_);
  753.     // Set the nak repeat interval.
  754.     rstate->nak_rpt_timer().resched(nak_rpt_ival_);
  756.     trace_event("SEND NACK", nak_rpt_ival_); //Nack being Sent, Nack will refire after ival
  758.     // Don't set the RDATA timer until the NCF is received.
  759.     // Don't set the elimintation timer until the NCF is received.
  761.     // We're now in the NAK_PENDING state.
  762.   }
  763.   else {
  764.     // There was previous repair state for the given NAK seqno.
  766.     if (hc->iface() < 0) {
  767.       // Scan the agent list to see if the agent is in the list already
  768.       // for this repair state.
  769.       list<NsObject *> *agent_list = &(rstate->agent_list());
  770.       list<NsObject *>::iterator res = find(agent_list->begin(), agent_list->end(), pkt2agent(pkt));
  772.       if (res == agent_list->end()) {
  773. agent_list->push_back(pkt2agent(pkt));
  774.       }
  776.     }
  777.     else {
  779.       // Scan the interface list to see if the interface is in the list
  780.       // already for this repair state.
  781.       list<int> *iface_list = &(rstate->iface_list());
  782.       list<int>::iterator res = find(iface_list->begin(), iface_list->end(), hc->iface());
  784.       if (res == iface_list->end()) {
  785. // Interface not found in iface list for this NAK, add it.
  786. iface_list->push_back(hc->iface());
  787.       }
  788.     }
  790.     // If the NAK elimination timer has expired, then we are allowed to
  791.     // send another NAK to our upstream.
  792.     if (rstate->nak_elimination() == false) {
  793.       rstate->nak_state() = NAK_PENDING;
  794.       // Start the nak retransmission cycle time.
  795.       rstate->nak_retrans_timer().resched(nak_retrans_ival_);
  796.       rstate->nak_rpt_timer().resched(nak_rpt_ival_);
  798.       // Disable the rdata and elim timer if they were previously running.
  799.       rstate->nak_rdata_timer().force_cancel();
  800.       rstate->nak_elim_timer().force_cancel();
  802.       rstate->nak_elimination() = true;
  804. #ifdef PGM_DEBUG
  805.       printf("%s: Got NAK for seqno %d with previous NAK state, accepted.n",
  806.      uname_, hp->seqno_);
  807. #endif
  808.     }
  809.     else {
  810.       // NAK elimination requires us to not act on this duplicate NAK packet.
  811. #ifdef PGM_DEBUG
  812.       printf("%s: Got NAK for seqno %d but have previous NAK state, discarding.n", uname_, hp->seqno_);
  813. #endif
  814.       stats_.num_suppressed_naks_++;
  815.       Packet::free(pkt);
  816.       return;
  817.     }
  819.   }
  821.   stats_.num_naks_transmitted_++;
  823.   // Send the NAK packet to our upstream
  824.   send_nak(state->upstream_node(), hp->tsi_, hp->seqno_, hpn->source_, hpn->group_);
  826.   Packet::free(pkt);
  827. }
  829. void PgmAgent::handle_ncf(Packet *pkt)
  830. {
  832.   hdr_pgm *hp = HDR_PGM(pkt);
  833.   hdr_cmn *hc = HDR_CMN(pkt);
  834.   hdr_ip *hip = HDR_IP(pkt);
  836.   // Locate the state control block for this TSI.
  837.   StateInfo *state = find_TSI(hp->tsi_);
  839.   if (state == NULL) {
  840.     printf("%s received NCF for which no SPM state is established, discarding.n", uname_);
  841.     stats_.num_unsolicited_ncf_++;
  842.     Packet::free(pkt);
  843.     return;
  844.   }
  846.   if (hc->iface() != state->upstream_iface()) {
  847.     printf("%s received NCF from non-upstream interface, discarding.n", uname_);
  848.     stats_.num_unsolicited_ncf_++;
  849.     Packet::free(pkt);
  850.     return;
  851.   }
  853.   trace_event("SEND NCF", 0);
  855.   // Look for the repair state for this NCF packet.
  856.   map<int, RepairState>::iterator result = state->repair().find(hp->seqno_);
  857.   RepairState *rstate;
  859.   if (result == state->repair().end()) {
  860.     // No repair state present for this NCF packet.
  862.     // Since the interface for this NCF packet comes from the same interface
  863.     // used to reach our upstream node, we can create empty repair state.
  864.     // This is NAK Anticipation (see 7.5 in PGM specification).
  866.     pair<map<int, RepairState>::iterator, bool> res;
  867.     res = state->repair().insert(pair<int, RepairState>(hp->seqno_, RepairState(this, state, hp->seqno_, hip->src(), hip->dst())));
  869.     rstate = &(res.first->second);
  871.     // Set the data field of the timers.
  872.     rstate->nak_retrans_timer().data() = rstate;
  873.     rstate->nak_rpt_timer().data() = rstate;
  874.     rstate->nak_rdata_timer().data() = rstate;
  875.     rstate->nak_elim_timer().data() = rstate;
  877.     stats_.num_unsolicited_ncf_++;
  878.   }
  879.   else {
  880.     rstate = &((*result).second);
  882.     // Disable either of the retransmission or repeat interval timers since
  883.     // the NAK is confirmed.
  884.     rstate->nak_retrans_timer().force_cancel();
  885.     rstate->nak_rpt_timer().force_cancel();
  886.   }
  888.   rstate->nak_state() = NAK_CONFIRMED;
  889.   // Set/reset the rdata and elim timer to expire at the appropriate time.
  890.   rstate->nak_rdata_timer().resched(nak_rdata_ival_);
  891.   rstate->nak_elim_timer().resched(nak_elim_ival_);
  893.   Packet::free(pkt);
  894. }
  896. // Create and send a nak packet to the upstream path.
  897. void PgmAgent::send_nak(ns_addr_t &upstream_node, ns_addr_t &tsi, int seqno, ns_addr_t &source, ns_addr_t &group)
  898. {
  900. #ifdef PGM_DEBUG
  901.   double now = Scheduler::instance().clock();
  902.   printf("at %f %s sending NAK for seqno %d upstream.n", now, uname_, seqno);
  903. #endif
  905.   Packet *nak_pkt = allocpkt();
  906.   // Set the simulated size of the NAK packet.
  907.   hdr_cmn *nak_hc = HDR_CMN(nak_pkt);
  908.   nak_hc->size_ = sizeof(hdr_pgm) + sizeof(hdr_pgm_nak);
  909.   nak_hc->ptype_ = PT_PGM;
  911.   // Set the destination address to be our upstream node.
  912.   hdr_ip *nak_hip = HDR_IP(nak_pkt);
  913.   nak_hip->dst() = upstream_node;
  915.   // Set the color of NAK packet to be black in nam.
  916.   nak_hip->fid_ = 8;
  918.   // Fill in the PGM header for the NAK packet.
  919.   hdr_pgm *nak_hp = HDR_PGM(nak_pkt);
  920.   nak_hp->type_ = PGM_NAK;
  921.   nak_hp->tsi_ = tsi;
  922.   nak_hp->seqno_ = seqno;
  924.   // Fill in the PGM NAK header for the NAK packet.
  925.   hdr_pgm_nak *nak_hpn = HDR_PGM_NAK(nak_pkt);
  926.   nak_hpn->source_ = source;
  927.   nak_hpn->group_ = group;
  929.   // Send out the packet.
  930.   send(nak_pkt, 0);
  931. }
  933. // Code that is executed when the nak retransmission timer expires.
  934. void PgmAgent::timeout_nak_retrans(RepairState *rstate)
  935. {
  937.   stats_.num_naks_transmitted_++;
  939.   // Send out a new NAK packet.
  940.   send_nak(rstate->sinfo()->upstream_node(), rstate->sinfo()->tsi(), rstate->seqno(), rstate->source(), rstate->group());
  942.   // Reset the retransmission timer.
  943.   rstate->nak_retrans_timer().resched(nak_retrans_ival_);
  945. }
  947. // Code that is executed when a repair state NAK RPT timer expires.
  948. void PgmAgent::timeout_nak_rpt(RepairState *rstate)
  949. {
  951.   printf("%s: timeout_nak_rpt expired, removing repair state.n", uname_);
  953.   // We never got a confirmation for our NAK packet. We must now
  954.   // remove the repair state entirely.
  955.   remove_repair_state(rstate);
  957. }
  959. // Code that is executed when a repair state NAK RDATA timer expires.
  960. void PgmAgent::timeout_nak_rdata(RepairState *rstate)
  961. {
  963.   printf("%s: timeout_nak_rdata expired, removing repair state.n", uname_);
  965.   // We never got the RDATA for our NAK. We must now remove the repair
  966.   // state entirely.
  967.   remove_repair_state(rstate);
  969. }
  971. // Code that is executed when a repair state NAK elimination timer expires.
  972. void PgmAgent::timeout_nak_elim(RepairState *rstate)
  973. {
  975.   // Allow one duplicate NAK to come in to be processed and forwarded.
  976.   rstate->nak_elimination() = false;
  978. }
  980. void PgmAgent::remove_repair_state(RepairState *rstate)
  981. {
  983.   // Cancel all timers.
  984.   rstate->nak_retrans_timer().force_cancel();
  985.   rstate->nak_rpt_timer().force_cancel();
  986.   rstate->nak_rdata_timer().force_cancel();
  987.   rstate->nak_elim_timer().force_cancel();
  989.   // Erase the repair state from the TSI repair map.
  990.   if (!rstate->sinfo()->repair().erase(rstate->seqno())) {
  991.     printf("ERROR (PgmAgent::remove_repair_state): Did not erase seqno from map.n");
  992.   }
  994. }
  996. NsObject* PgmAgent::iface2link (int iface)
  997. {
  998. //      Tcl::instance().evalf("%s get-outlink %d", name(), iface);
  999. //      char* ni = Tcl::instance().result();
  1000.         Tcl&    tcl = Tcl::instance();
  1001.         char    wrk[64];
  1003. if (iface == -1) {
  1004.   return NULL;
  1005. }
  1007.         sprintf (wrk, "[%s set node_] ifaceGetOutLink %d", name (), iface);
  1008.         tcl.evalc (wrk);
  1009.         const char* result = tcl.result ();
  1010. #ifdef PGM_DEBUG
  1011. printf ("[iface2link] agent %sn", result);
  1012. #endif
  1013.         NsObject* obj = (NsObject*)TclObject::lookup(result);
  1014.         return (obj);
  1015. }
  1017. NsObject* PgmAgent::pkt2agent (Packet *pkt)
  1018. {
  1019.         Tcl&            tcl = Tcl::instance();
  1020.         char            wrk[64];
  1021.         const char     *result;
  1022.         int             port;
  1023.         NsObject*       agent;
  1024.         hdr_ip*         ih = HDR_IP(pkt);
  1025.         //nsaddr_t        src = ih->saddr();
  1027.         port = ih->sport();
  1029.         sprintf (wrk, "[%s set node_] agent %d", name (), port);
  1030.         tcl.evalc (wrk);
  1031.         result = tcl.result ();
  1033. #ifdef PGM_DEBUG
  1034. printf ("[pkt2agent] port %d, agent %sn", port, result);
  1035. #endif
  1037.         agent = (NsObject*)TclObject::lookup (result);
  1038.         return (agent);
  1039. }
  1041. // Find the state control block given a TSI. Returns NULL if not found.
  1042. StateInfo * PgmAgent::find_TSI(ns_addr_t &tsi)
  1043. {
  1045.   // Use the TSI from the SPM packet and locate the proper state block.
  1046.   list<StateInfo>::iterator iter = state_list_.begin();
  1047.   while(iter != state_list_.end()) {
  1048.     if ( (*iter).tsi().isEqual (tsi) ) {
  1049.       return &(*iter);
  1050.     }
  1051.     iter++;
  1052.   }
  1054.   return NULL;
  1055. }
  1057. // Insert a new state control block for the given TSI, and return a pointer
  1058. // to the control block.
  1059. StateInfo * PgmAgent::insert_TSI(ns_addr_t &tsi)
  1060. {
  1061.   state_list_.push_back(StateInfo(tsi));
  1063.   return &(state_list_.back());
  1064. }
  1066. void PgmAgent::print_stats()
  1067. {
  1068.   printf("%s:n", uname_);
  1069.   printf("tNAKs Transmitted: t%dn", stats_.num_naks_transmitted_);
  1070.   printf("tNAKs Suppressed: t%dn", stats_.num_suppressed_naks_);
  1071.   printf("tUnsolicited NCFs: t%dn", stats_.num_unsolicited_ncf_);
  1072.   printf("tUnsolicited RDATA: t%dn", stats_.num_unsolicited_rdata_);
  1073. }
  1075. #ifdef PGM_DEBUG
  1076. void PgmAgent::display_packet(Packet *pkt)
  1077. {
  1079.   double now = Scheduler::instance().clock();
  1081.   hdr_ip *hip = HDR_IP(pkt);
  1082.   hdr_cmn *hc = HDR_CMN(pkt);
  1084.   printf("at %f %s received packet type ", now, uname_);
  1086.   hdr_pgm *hp = HDR_PGM(pkt);
  1087.   
  1088.   hdr_pgm_spm *hps;
  1089.   hdr_pgm_nak *hpn;
  1091.   switch(hp->type_) {
  1092.   case PGM_SPM:
  1093.     hps = HDR_PGM_SPM(pkt);
  1095.     printf("SPM (TSI %d:%d) from %d:%d to %d:%d iface %d, size %d, seqno %d, spm_path %d:%dn", hp->tsi_.addr_, hp->tsi_.port_, hip->saddr(), hip->sport(), hip->daddr(), hip->dport(), hc->iface(), hc->size(), hp->seqno_, hps->spm_path_.addr_, hps->spm_path_.port_);
  1097.     break;
  1098.   case PGM_ODATA:
  1099.     printf("ODATA (TSI %d:%d) from %d:%d to %d:%d iface %d, size %d, seqno %dn", hp->tsi_.addr_, hp->tsi_.port_, hip->saddr(), hip->sport(), hip->daddr(), hip->dport(), hc->iface(), hc->size(), hp->seqno_);
  1101.     break;
  1102.   case PGM_RDATA:
  1103.     printf("RDATA (TSI %d:%d) from %d:%d to %d:%d iface %d, size %d, seqno %dn", hp->tsi_.addr_, hp->tsi_.port_, hip->saddr(), hip->sport(), hip->daddr(), hip->dport(), hc->iface(), hc->size(), hp->seqno_);
  1105.     break;
  1106.   case PGM_NAK:
  1107.     hpn = HDR_PGM_NAK(pkt);
  1109.     printf("NAK (TSI %d:%d) from %d:%d to %d:%d iface %d, size %d, seqno %d, source %d:%d, group %d:%dn", hp->tsi_.addr_, hp->tsi_.port_, hip->saddr(), hip->sport(), hip->daddr(), hip->dport(), hc->iface(), hc->size(), hp->seqno_, hpn->source_.addr_, hpn->source_.port_, hpn->group_.addr_, hpn->group_.port_);
  1111.     break;
  1112.   case PGM_NCF:
  1113.     printf("NCF (TSI %d:%d) from %d:%d to %d:%d iface %d, size %d, seqno %dn", hp->tsi_.addr_, hp->tsi_.port_, hip->saddr(), hip->sport(), hip->daddr(), hip->dport(), hc->iface(), hc->size(), hp->seqno_);
  1115.     break;
  1116.   default:
  1117.     printf("UNKNOWN (TSI %d:%d) from %d:%d to %d:%d iface %d, size %d, seqno %dn", hp->tsi_.addr_, hp->tsi_.port_, hip->saddr(), hip->sport(), hip->daddr(), hip->dport(), hc->iface(), hc->size(), hp->seqno_);
  1119.     break;
  1120.   }
  1122. }
  1123. #endif // PGM_DEBUG
  1125. #endif //HAVE_STL