/* * pgm-agent.cc * Copyright (C) 2001 by the University of Southern California * $Id: pgm-agent.cc,v 1.9 2005/08/25 18:58:10 johnh Exp $ * * This program is free software; you can redistribute it and/or * modify it under the terms of the GNU General Public License, * version 2, as published by the Free Software Foundation. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License along * with this program; if not, write to the Free Software Foundation, Inc., * 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA. * * * The copyright of this module includes the following * linking-with-specific-other-licenses addition: * * In addition, as a special exception, the copyright holders of * this module give you permission to combine (via static or * dynamic linking) this module with free software programs or * libraries that are released under the GNU LGPL and with code * included in the standard release of ns-2 under the Apache 2.0 * license or under otherwise-compatible licenses with advertising * requirements (or modified versions of such code, with unchanged * license). You may copy and distribute such a system following the * terms of the GNU GPL for this module and the licenses of the * other code concerned, provided that you include the source code of * that other code when and as the GNU GPL requires distribution of * source code. * * Note that people who make modified versions of this module * are not obligated to grant this special exception for their * modified versions; it is their choice whether to do so. The GNU * General Public License gives permission to release a modified * version without this exception; this exception also makes it * possible to release a modified version which carries forward this * exception. * */ /* * Pragmatic General Multicast (PGM), Reliable Multicast * * pgm-agent.cc * * This implements the network element PGM agent, "Agent/PGM". * * Ryan S. Barnett, 2001 * rbarnett@catarina.usc.edu */ #include "config.h" #ifdef HAVE_STL #include #include #include #include #include #include "config.h" #include "tclcl.h" #include "agent.h" #include "packet.h" #include "ip.h" #include "random.h" #include "basetrace.h" #include "pgm.h" // ************************************************************ // Register the PGM packet headers. // ************************************************************ // Declare the static header offsets. int hdr_pgm::offset_; int hdr_pgm_spm::offset_; int hdr_pgm_nak::offset_; // Register the hdr_pgm with the packet header manager. class PGMHeaderClass : public PacketHeaderClass { public: PGMHeaderClass() : PacketHeaderClass("PacketHeader/PGM", sizeof(hdr_pgm)) { bind_offset(&hdr_pgm::offset_); } } class_pgmhdr; // Register the hdr_pgm_spm with the packet header manager. class PGM_SPMHeaderClass : public PacketHeaderClass { public: PGM_SPMHeaderClass() : PacketHeaderClass("PacketHeader/PGM_SPM", sizeof(hdr_pgm_spm)) { bind_offset(&hdr_pgm_spm::offset_); } } class_pgm_spmhdr; // Register the hdr_pgm_nak with the packet header manager. class PGM_NAKHeaderClass : public PacketHeaderClass { public: PGM_NAKHeaderClass() : PacketHeaderClass("PacketHeader/PGM_NAK", sizeof(hdr_pgm_nak)) { bind_offset(&hdr_pgm_nak::offset_); } } class_pgm_nakhdr; // ************************************************************ // Define the PGM Agent Timer Class // ************************************************************ class PgmAgent; // Different timer types. enum { TIMER_NAK_RETRANS = 0, TIMER_NAK_RPT = 1, TIMER_NAK_RDATA = 2, TIMER_NAK_ELIM = 3 }; class PgmAgentTimer : public TimerHandler { public: PgmAgentTimer(PgmAgent *a, int type) : TimerHandler(), data_(NULL) { a_ = a; type_ = type; } void * &data() { return data_; } protected: virtual void expire(Event *e); PgmAgent *a_; int type_; void *data_; }; // ************************************************************ // Define the Repair State control block. // ************************************************************ // Different repair states. enum { NAK_PENDING = 0, NAK_CONFIRMED = 1 }; class StateInfo; class RepairState { public: RepairState(PgmAgent *a, StateInfo *sinfo, int seqno, ns_addr_t &source, ns_addr_t &group) : seqno_(seqno), source_(source), group_(group), nak_state_(NAK_PENDING), nak_elimination_(true), nak_retrans_timer_(a, TIMER_NAK_RETRANS), nak_rpt_timer_(a, TIMER_NAK_RPT), nak_rdata_timer_(a, TIMER_NAK_RDATA), nak_elim_timer_(a, TIMER_NAK_ELIM), sinfo_(sinfo) { } int & seqno() { return seqno_; } ns_addr_t & source() { return source_; } ns_addr_t & group() { return group_; } int & nak_state() { return nak_state_; } bool & nak_elimination() { return nak_elimination_; } PgmAgentTimer & nak_retrans_timer() { return nak_retrans_timer_; } PgmAgentTimer & nak_rpt_timer() { return nak_rpt_timer_; } PgmAgentTimer & nak_rdata_timer() { return nak_rdata_timer_; } PgmAgentTimer & nak_elim_timer() { return nak_elim_timer_; } list & iface_list() { return iface_list_; } list & agent_list() { return agent_list_; } StateInfo * & sinfo() { return sinfo_; } protected: // Which sequence number is being requested for repair. int seqno_; ns_addr_t source_; // Original source of ODATA for the repair. ns_addr_t group_; // The multicast group. int nak_state_; // Present repair block state. // Indicates whether or not we are to discard incoming NAK packets // once a previous NAK is outstanding (got NCF, waiting for RDATA). // (See 7.4 of PGM specification) By default we do. When nak_elim_timer_ // expires, then we do not. bool nak_elimination_; // This timer controls sending retransmissions of NAK packets. PgmAgentTimer nak_retrans_timer_; // Timer that measures how long we can repeat NAK packets while waiting // for NCF confirmation. Once expires, the repair state is removed. PgmAgentTimer nak_rpt_timer_; // Timer that is triggered waiting for RDATA for a given NAK seqno, // provided that NAK has been confirmed by an NCF. Only gets set // once NCF is received. PgmAgentTimer nak_rdata_timer_; // Timer that is triggered when we disable nak_elimination_, allowing // a duplicate NAK to be processed. This occurs after a previous // NAK has been confirmed with an NCF, but before the RDATA has been // received. This timer gets set when we receive an NCF for a pending // NAK. PgmAgentTimer nak_elim_timer_; // List of interfaces upon which the RDATA will be sent to. list iface_list_; // List of agents upon which the RDATA will be sent to. list agent_list_; // Back-pointer to the state information block that is holding this // repair data. We use this so we can get the upstream_path and the TSI. StateInfo *sinfo_; }; // ************************************************************ // Define the TSI State control block. // ************************************************************ class StateInfo { public: StateInfo(ns_addr_t tsi) : tsi_(tsi), spm_seqno_(-1) { } // Only used if the container holding StateInfo's will be in sorted order. int operator<(const StateInfo &right) const { return ((tsi_.addr_ < right.tsi_.addr_) || ( (tsi_.addr_==right.tsi_.addr_) && (tsi_.port_ < right.tsi_.port_))); } ns_addr_t & tsi() { return tsi_; } int & spm_seqno() { return spm_seqno_; } ns_addr_t & upstream_node() { return upstream_node_; } int & upstream_iface() { return upstream_iface_; } map & repair() { return repair_; } protected: ns_addr_t tsi_; // Transport Session ID int spm_seqno_; // Most recent SPM sequence number. ns_addr_t upstream_node_; // Upstream node address. int upstream_iface_; // Upstream interface number. // Map between a NAK sequence number and the corresponding repair state // for that sequence number. map repair_; }; // ************************************************************ // Define the PGM Agent Class // ************************************************************ // Structure to hold statistical information for PGM Agent. struct Stats { int num_unsolicited_ncf_; int num_unsolicited_rdata_; int num_suppressed_naks_; int num_naks_transmitted_; }; // Used to count number of unique pgm agents. static int pgm_agent_uid_ = 0; class PgmAgent : public Agent { public: PgmAgent(); virtual void recv(Packet *, Handler *); virtual void timeout(int type, void *data); virtual int command(int argc, const char*const* argv); protected: void handle_spm(Packet *pkt); void handle_odata(Packet *pkt); void handle_rdata(Packet *pkt); void handle_nak(Packet *pkt); void handle_ncf(Packet *pkt); void send_nak(ns_addr_t &upstream_node, ns_addr_t &tsi, int seqno, ns_addr_t &source, ns_addr_t &group); void timeout_nak_retrans(RepairState *rstate); void timeout_nak_rpt(RepairState *rstate); void timeout_nak_rdata(RepairState *rstate); void timeout_nak_elim(RepairState *rstate); void remove_repair_state(RepairState *rstate); void print_stats(); void PgmAgent::trace_event(char *evType, double evTime); #ifdef PGM_DEBUG void display_packet(Packet *pkt); #endif NsObject* iface2link(int iface); NsObject* pkt2agent (Packet *pkt); StateInfo * find_TSI(ns_addr_t &tsi); StateInfo * insert_TSI(ns_addr_t &tsi); EventTrace * et_; //Trace Object for custom event trace int pgm_enabled_; // Is this agent enabled? Default is YES. char uname_[16]; // Agent's unique name. Stats stats_; // Statistical information. // TSI-indexed state control block list. list state_list_; // Number of seconds to wait between retransmitting a NAK that is waiting // for a NCF packet. double nak_retrans_ival_; // The length of time for which a network element will continue to repeat // NAKs while waiting for a corresponding NCF. Once this time expires and // no NCF is received, then we remove the entire repair state. double nak_rpt_ival_; // The length of time for which a network element will wait for the // corresponding RDATA before removing the entire repair state. double nak_rdata_ival_; // Once a NAK has been confirmed, network elements must discard all // further NAKs for up to this length of time. Should be a fraction // of nak_rdata_ival_. double nak_elim_ival_; }; static class PgmClass : public TclClass { public: PgmClass() : TclClass("Agent/PGM") {} TclObject * create(int argc, const char * const * argv) { return (new PgmAgent()); } } class_pgm_agent; void PgmAgentTimer::expire(Event *e) { a_->timeout(type_, data_); } // Constructor. PgmAgent::PgmAgent() : Agent(PT_PGM), pgm_enabled_(1) { // Set the unique identifier. sprintf (uname_, "pgmAgent-%d", pgm_agent_uid_++); // Initialize statistics. stats_.num_unsolicited_ncf_ = 0; stats_.num_unsolicited_rdata_ = 0; stats_.num_suppressed_naks_ = 0; stats_.num_naks_transmitted_ = 0; bind("pgm_enabled_", &pgm_enabled_); bind_time("nak_retrans_ival_", &nak_retrans_ival_); bind_time("nak_rpt_ival_", &nak_rpt_ival_); bind_time("nak_rdata_ival_", &nak_rdata_ival_); bind_time("nak_elim_ival_", &nak_elim_ival_); et_ = (EventTrace *) NULL; } // Code to execute when a packet is received. void PgmAgent::recv(Packet* pkt, Handler*) { hdr_pgm *hp = HDR_PGM(pkt); if (!pgm_enabled_) { target_->recv(pkt); return; } hdr_cmn *hc = HDR_CMN(pkt); if (hc->ptype_ != PT_PGM) { printf("%s ERROR (PgmAgent::recv): received non PGM pkt type %d, discarding.\n", uname_, hc->ptype_); Packet::free(pkt); return; } #ifdef PGM_DEBUG display_packet(pkt); #endif // Note, each handle function will free the packet itself or modify the // headers and pass it on to the next NS object. switch(hp->type_) { case PGM_SPM: handle_spm(pkt); break; case PGM_ODATA: handle_odata(pkt); break; case PGM_RDATA: handle_rdata(pkt); break; case PGM_NAK: handle_nak(pkt); break; case PGM_NCF: handle_ncf(pkt); break; default: printf("ERROR (PgmAgent::recv): Received PGM packet with unknown type %d.\n", hp->type_); Packet::free(pkt); break; } } // Code to execute when a timeout occurs. void PgmAgent::timeout(int type, void *data) { switch(type) { case TIMER_NAK_RETRANS: timeout_nak_retrans((RepairState *) data); break; case TIMER_NAK_RPT: timeout_nak_rpt((RepairState *) data); break; case TIMER_NAK_RDATA: timeout_nak_rdata((RepairState *) data); break; case TIMER_NAK_ELIM: timeout_nak_elim((RepairState *) data); break; default: printf("ERROR (PgmAgent::timeout()): Invalid timeout type %d.\n", type); break; } } // Code to execute when a TCL command is issued to the PGM Agent object. int PgmAgent::command (int argc, const char*const* argv) { // Tcl& tcl = Tcl::instance(); if (argc == 2) { if (strcmp(argv[1], "print-stats") == 0) { print_stats(); return (TCL_OK); } } else if (argc == 3) { //Set the Event Trace handle if Event Tracing is on if (strcmp(argv[1], "eventtrace") == 0) { et_ = (EventTrace *)TclObject::lookup(argv[2]); return (TCL_OK); } } return (Agent::command(argc, argv)); } void PgmAgent::trace_event(char *evType, double evTime) { if (et_ == NULL) return; char *wrk = et_->buffer(); char *nwrk = et_->nbuffer(); if (wrk != NULL) { sprintf(wrk, "E "TIME_FORMAT" %d %d PGM %s "TIME_FORMAT, et_->round(Scheduler::instance().clock()), addr(), 0, evType, evTime); if (nwrk != 0) sprintf(nwrk, "E -t "TIME_FORMAT" -o PGM -e %s -s %d.%d -d %d.%d", et_->round(Scheduler::instance().clock()), // time evType, // event type addr(), // owner (src) node id port(), // owner (src) port id 0, // dst node id 0 // dst port id ); et_->dump(); } } void PgmAgent::handle_spm(Packet *pkt) { hdr_pgm *hp = HDR_PGM(pkt); hdr_pgm_spm *hs = HDR_PGM_SPM(pkt); hdr_cmn *hc = HDR_CMN(pkt); // Use the TSI from the SPM packet and locate the proper state block. StateInfo *state = find_TSI(hp->tsi_); if (state == NULL) { // There is no state block for this TSI. Create new state. state = insert_TSI(hp->tsi_); // Set the sequence number. state->spm_seqno() = hp->seqno_; // Set the upstream path. state->upstream_node() = hs->spm_path_; state->upstream_iface() = hc->iface(); } else { // State already exists for this TSI. Check if the sequence number is // newer than the last recorded sequence number. if ( state->spm_seqno() < hp->seqno_ ) { // Update the SPM sequence number. state->spm_seqno() = hp->seqno_; // Set the upstream path. state->upstream_node() = hs->spm_path_; state->upstream_iface() = hc->iface(); } else { printf("%s received an old SPM seqno, discarding.\n", uname_); Packet::free(pkt); return; } } // Modify the SPM packet and set the upstream path to be equal to // the address of this agent. hs->spm_path_ = here_; // Send the modified packet off to the rest of the multicast group. send(pkt, 0); } void PgmAgent::handle_odata(Packet *pkt) { // Pass the ODATA along to the rest of the multicast group. ODATA // does not cancel NAK forwarding. //hdr_cmn *hc = HDR_CMN(pkt); send(pkt, 0); } void PgmAgent::handle_rdata(Packet *pkt) { // Look for the TSI for this RDATA packet. hdr_pgm *hp = HDR_PGM(pkt); // hdr_ip *hip = HDR_IP(pkt); StateInfo *state = find_TSI(hp->tsi_); if (state == NULL) { printf("%s received RDATA for which no SPM state is established, discarding.\n", uname_); stats_.num_unsolicited_rdata_++; Packet::free(pkt); return; } // Look for the repair state for this RDATA packet. map::iterator result = state->repair().find(hp->seqno_); if (result == state->repair().end()) { // No repair state present for this RDATA packet. printf("%s received RDATA for which no repair state is present, discarding.\n", uname_); stats_.num_unsolicited_rdata_++; Packet::free(pkt); return; } RepairState *rstate = &((*result).second); // Get the interface list for the repair state. For each interface, send // out the RDATA packet. Similarly for each agent that is also receiving // RDATA attached to this node. if (rstate->iface_list().empty() && rstate->agent_list().empty()) { printf("%s received RDATA but repair state has no interfaces for it, discarding.\n", uname_); stats_.num_unsolicited_rdata_++; Packet::free(pkt); } NsObject *tgt; Packet *new_pkt; int flag = 0; trace_event("SEND RDATA", 0); //Repair is being forwarded // hdr_cmn *hc = HDR_CMN(pkt); if (!rstate->iface_list().empty()) { list::iterator iter = rstate->iface_list().begin(); while (iter != rstate->iface_list().end()) { if (!flag) { tgt = iface2link(*iter); if (tgt == NULL) { printf("ERROR (PgmAgent::handle_rdata): iface2link returned NULL.\n"); abort(); } tgt->recv(pkt); flag = 1; } else { // Make a copy of each packet before sending it, so we don't free() // the same packet at the different receivers causing a deallocation // problem. new_pkt = pkt->copy(); tgt = iface2link(*iter); if (tgt == NULL) { printf("ERROR (PgmAgent::handle_rdata): iface2link returned NULL.\n"); abort(); } tgt->recv(new_pkt); } iter++; } } if (!rstate->agent_list().empty()) { list::iterator iter = rstate->agent_list().begin(); while (iter != rstate->agent_list().end()) { if (!flag) { (*iter)->recv(pkt); flag = 1; } else { // Make a copy of each packet before sending it, so we don't free() // the same packet at the different receivers causing a deallocation // problem. new_pkt = pkt->copy(); (*iter)->recv(new_pkt); } iter++; } } // Remove the repair state for this RDATA sequence number, since we sent // out the repairs. remove_repair_state(&((*result).second)); } void PgmAgent::handle_nak(Packet *pkt) { hdr_pgm *hp = HDR_PGM(pkt); hdr_pgm_nak *hpn = HDR_PGM_NAK(pkt); hdr_cmn *hc = HDR_CMN(pkt); // hdr_ip *hip = HDR_IP(pkt); // Check to see if there is a state control block for the given TSI. StateInfo *state = find_TSI(hp->tsi_); if (state == NULL) { printf("PGM Agent received NAK for which no SPM state is established, discarding.\n"); Packet::free(pkt); return; } // Create an NCF packet in response to the NAK packet. Packet *ncf_pkt = allocpkt(); hdr_cmn *ncf_hc = HDR_CMN(ncf_pkt); ncf_hc->size_ = sizeof(hdr_pgm); // Size of NCF packet. ncf_hc->ptype_ = PT_PGM; hdr_pgm *ncf_hp = HDR_PGM(ncf_pkt); ncf_hp->type_ = PGM_NCF; ncf_hp->tsi_ = hp->tsi_; ncf_hp->seqno_ = hp->seqno_; // Change the source of the NCF packet to be the original ODATA source. hdr_ip *ncf_ip = HDR_IP(ncf_pkt); ncf_ip->src() = hpn->source_; // Set the destination to be the multicast group. ncf_ip->dst() = hpn->group_; // Set the color of NCF packets in nam to be green. ncf_ip->fid_ = 6; // Send out the NCF to the interface (or agent) for which the NAK was // received. NsObject *tgt; if (hc->iface() < 0) { tgt = pkt2agent(pkt); if (tgt == NULL) { printf("ERROR: (PgmAgent::handle_nak) pkt2agent returned NULL.\n"); abort(); } tgt->recv(ncf_pkt); } else { tgt = iface2link(hc->iface()); if (tgt == NULL) { printf("ERROR: (PgmAgent::handle_nak) iface2link returned NULL.\n"); abort(); } tgt->recv(ncf_pkt); } // Create repair state for the NAK query. Associate the sequence number // of the NAK packet with the interface where the packet was received. pair::iterator, bool> result; result = state->repair().insert(pair(hp->seqno_, RepairState(this, state, hp->seqno_, hpn->source_, hpn->group_))); RepairState *rstate = &(result.first->second); if (result.second == true) { // There was no previous repair state for the given NAK seqno. // This must be a new NAK. // Set the data fields of the timer. rstate->nak_retrans_timer().data() = rstate; rstate->nak_rpt_timer().data() = rstate; rstate->nak_rdata_timer().data() = rstate; rstate->nak_elim_timer().data() = rstate; // Add the interface (or agent) to the interface list. if (hc->iface() < 0) { rstate->agent_list().push_back(pkt2agent(pkt)); } else { rstate->iface_list().push_back(hc->iface()); } // Start the nak retransmission cycle time. rstate->nak_retrans_timer().resched(nak_retrans_ival_); // Set the nak repeat interval. rstate->nak_rpt_timer().resched(nak_rpt_ival_); trace_event("SEND NACK", nak_rpt_ival_); //Nack being Sent, Nack will refire after ival // Don't set the RDATA timer until the NCF is received. // Don't set the elimintation timer until the NCF is received. // We're now in the NAK_PENDING state. } else { // There was previous repair state for the given NAK seqno. if (hc->iface() < 0) { // Scan the agent list to see if the agent is in the list already // for this repair state. list *agent_list = &(rstate->agent_list()); list::iterator res = find(agent_list->begin(), agent_list->end(), pkt2agent(pkt)); if (res == agent_list->end()) { agent_list->push_back(pkt2agent(pkt)); } } else { // Scan the interface list to see if the interface is in the list // already for this repair state. list *iface_list = &(rstate->iface_list()); list::iterator res = find(iface_list->begin(), iface_list->end(), hc->iface()); if (res == iface_list->end()) { // Interface not found in iface list for this NAK, add it. iface_list->push_back(hc->iface()); } } // If the NAK elimination timer has expired, then we are allowed to // send another NAK to our upstream. if (rstate->nak_elimination() == false) { rstate->nak_state() = NAK_PENDING; // Start the nak retransmission cycle time. rstate->nak_retrans_timer().resched(nak_retrans_ival_); rstate->nak_rpt_timer().resched(nak_rpt_ival_); // Disable the rdata and elim timer if they were previously running. rstate->nak_rdata_timer().force_cancel(); rstate->nak_elim_timer().force_cancel(); rstate->nak_elimination() = true; #ifdef PGM_DEBUG printf("%s: Got NAK for seqno %d with previous NAK state, accepted.\n", uname_, hp->seqno_); #endif } else { // NAK elimination requires us to not act on this duplicate NAK packet. #ifdef PGM_DEBUG printf("%s: Got NAK for seqno %d but have previous NAK state, discarding.\n", uname_, hp->seqno_); #endif stats_.num_suppressed_naks_++; Packet::free(pkt); return; } } stats_.num_naks_transmitted_++; // Send the NAK packet to our upstream send_nak(state->upstream_node(), hp->tsi_, hp->seqno_, hpn->source_, hpn->group_); Packet::free(pkt); } void PgmAgent::handle_ncf(Packet *pkt) { hdr_pgm *hp = HDR_PGM(pkt); hdr_cmn *hc = HDR_CMN(pkt); hdr_ip *hip = HDR_IP(pkt); // Locate the state control block for this TSI. StateInfo *state = find_TSI(hp->tsi_); if (state == NULL) { printf("%s received NCF for which no SPM state is established, discarding.\n", uname_); stats_.num_unsolicited_ncf_++; Packet::free(pkt); return; } if (hc->iface() != state->upstream_iface()) { printf("%s received NCF from non-upstream interface, discarding.\n", uname_); stats_.num_unsolicited_ncf_++; Packet::free(pkt); return; } trace_event("SEND NCF", 0); // Look for the repair state for this NCF packet. map::iterator result = state->repair().find(hp->seqno_); RepairState *rstate; if (result == state->repair().end()) { // No repair state present for this NCF packet. // Since the interface for this NCF packet comes from the same interface // used to reach our upstream node, we can create empty repair state. // This is NAK Anticipation (see 7.5 in PGM specification). pair::iterator, bool> res; res = state->repair().insert(pair(hp->seqno_, RepairState(this, state, hp->seqno_, hip->src(), hip->dst()))); rstate = &(res.first->second); // Set the data field of the timers. rstate->nak_retrans_timer().data() = rstate; rstate->nak_rpt_timer().data() = rstate; rstate->nak_rdata_timer().data() = rstate; rstate->nak_elim_timer().data() = rstate; stats_.num_unsolicited_ncf_++; } else { rstate = &((*result).second); // Disable either of the retransmission or repeat interval timers since // the NAK is confirmed. rstate->nak_retrans_timer().force_cancel(); rstate->nak_rpt_timer().force_cancel(); } rstate->nak_state() = NAK_CONFIRMED; // Set/reset the rdata and elim timer to expire at the appropriate time. rstate->nak_rdata_timer().resched(nak_rdata_ival_); rstate->nak_elim_timer().resched(nak_elim_ival_); Packet::free(pkt); } // Create and send a nak packet to the upstream path. void PgmAgent::send_nak(ns_addr_t &upstream_node, ns_addr_t &tsi, int seqno, ns_addr_t &source, ns_addr_t &group) { #ifdef PGM_DEBUG double now = Scheduler::instance().clock(); printf("at %f %s sending NAK for seqno %d upstream.\n", now, uname_, seqno); #endif Packet *nak_pkt = allocpkt(); // Set the simulated size of the NAK packet. hdr_cmn *nak_hc = HDR_CMN(nak_pkt); nak_hc->size_ = sizeof(hdr_pgm) + sizeof(hdr_pgm_nak); nak_hc->ptype_ = PT_PGM; // Set the destination address to be our upstream node. hdr_ip *nak_hip = HDR_IP(nak_pkt); nak_hip->dst() = upstream_node; // Set the color of NAK packet to be black in nam. nak_hip->fid_ = 8; // Fill in the PGM header for the NAK packet. hdr_pgm *nak_hp = HDR_PGM(nak_pkt); nak_hp->type_ = PGM_NAK; nak_hp->tsi_ = tsi; nak_hp->seqno_ = seqno; // Fill in the PGM NAK header for the NAK packet. hdr_pgm_nak *nak_hpn = HDR_PGM_NAK(nak_pkt); nak_hpn->source_ = source; nak_hpn->group_ = group; // Send out the packet. send(nak_pkt, 0); } // Code that is executed when the nak retransmission timer expires. void PgmAgent::timeout_nak_retrans(RepairState *rstate) { stats_.num_naks_transmitted_++; // Send out a new NAK packet. send_nak(rstate->sinfo()->upstream_node(), rstate->sinfo()->tsi(), rstate->seqno(), rstate->source(), rstate->group()); // Reset the retransmission timer. rstate->nak_retrans_timer().resched(nak_retrans_ival_); } // Code that is executed when a repair state NAK RPT timer expires. void PgmAgent::timeout_nak_rpt(RepairState *rstate) { printf("%s: timeout_nak_rpt expired, removing repair state.\n", uname_); // We never got a confirmation for our NAK packet. We must now // remove the repair state entirely. remove_repair_state(rstate); } // Code that is executed when a repair state NAK RDATA timer expires. void PgmAgent::timeout_nak_rdata(RepairState *rstate) { printf("%s: timeout_nak_rdata expired, removing repair state.\n", uname_); // We never got the RDATA for our NAK. We must now remove the repair // state entirely. remove_repair_state(rstate); } // Code that is executed when a repair state NAK elimination timer expires. void PgmAgent::timeout_nak_elim(RepairState *rstate) { // Allow one duplicate NAK to come in to be processed and forwarded. rstate->nak_elimination() = false; } void PgmAgent::remove_repair_state(RepairState *rstate) { // Cancel all timers. rstate->nak_retrans_timer().force_cancel(); rstate->nak_rpt_timer().force_cancel(); rstate->nak_rdata_timer().force_cancel(); rstate->nak_elim_timer().force_cancel(); // Erase the repair state from the TSI repair map. if (!rstate->sinfo()->repair().erase(rstate->seqno())) { printf("ERROR (PgmAgent::remove_repair_state): Did not erase seqno from map.\n"); } } NsObject* PgmAgent::iface2link (int iface) { // Tcl::instance().evalf("%s get-outlink %d", name(), iface); // char* ni = Tcl::instance().result(); Tcl& tcl = Tcl::instance(); char wrk[64]; if (iface == -1) { return NULL; } sprintf (wrk, "[%s set node_] ifaceGetOutLink %d", name (), iface); tcl.evalc (wrk); const char* result = tcl.result (); #ifdef PGM_DEBUG printf ("[iface2link] agent %s\n", result); #endif NsObject* obj = (NsObject*)TclObject::lookup(result); return (obj); } NsObject* PgmAgent::pkt2agent (Packet *pkt) { Tcl& tcl = Tcl::instance(); char wrk[64]; const char *result; int port; NsObject* agent; hdr_ip* ih = HDR_IP(pkt); //nsaddr_t src = ih->saddr(); port = ih->sport(); sprintf (wrk, "[%s set node_] agent %d", name (), port); tcl.evalc (wrk); result = tcl.result (); #ifdef PGM_DEBUG printf ("[pkt2agent] port %d, agent %s\n", port, result); #endif agent = (NsObject*)TclObject::lookup (result); return (agent); } // Find the state control block given a TSI. Returns NULL if not found. StateInfo * PgmAgent::find_TSI(ns_addr_t &tsi) { // Use the TSI from the SPM packet and locate the proper state block. list::iterator iter = state_list_.begin(); while(iter != state_list_.end()) { if ( (*iter).tsi().isEqual (tsi) ) { return &(*iter); } iter++; } return NULL; } // Insert a new state control block for the given TSI, and return a pointer // to the control block. StateInfo * PgmAgent::insert_TSI(ns_addr_t &tsi) { state_list_.push_back(StateInfo(tsi)); return &(state_list_.back()); } void PgmAgent::print_stats() { printf("%s:\n", uname_); printf("\tNAKs Transmitted: \t%d\n", stats_.num_naks_transmitted_); printf("\tNAKs Suppressed: \t%d\n", stats_.num_suppressed_naks_); printf("\tUnsolicited NCFs: \t%d\n", stats_.num_unsolicited_ncf_); printf("\tUnsolicited RDATA: \t%d\n", stats_.num_unsolicited_rdata_); } #ifdef PGM_DEBUG void PgmAgent::display_packet(Packet *pkt) { double now = Scheduler::instance().clock(); hdr_ip *hip = HDR_IP(pkt); hdr_cmn *hc = HDR_CMN(pkt); printf("at %f %s received packet type ", now, uname_); hdr_pgm *hp = HDR_PGM(pkt); hdr_pgm_spm *hps; hdr_pgm_nak *hpn; switch(hp->type_) { case PGM_SPM: hps = HDR_PGM_SPM(pkt); printf("SPM (TSI %d:%d) from %d:%d to %d:%d iface %d, size %d, seqno %d, spm_path %d:%d\n", 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_); break; case PGM_ODATA: printf("ODATA (TSI %d:%d) from %d:%d to %d:%d iface %d, size %d, seqno %d\n", hp->tsi_.addr_, hp->tsi_.port_, hip->saddr(), hip->sport(), hip->daddr(), hip->dport(), hc->iface(), hc->size(), hp->seqno_); break; case PGM_RDATA: printf("RDATA (TSI %d:%d) from %d:%d to %d:%d iface %d, size %d, seqno %d\n", hp->tsi_.addr_, hp->tsi_.port_, hip->saddr(), hip->sport(), hip->daddr(), hip->dport(), hc->iface(), hc->size(), hp->seqno_); break; case PGM_NAK: hpn = HDR_PGM_NAK(pkt); printf("NAK (TSI %d:%d) from %d:%d to %d:%d iface %d, size %d, seqno %d, source %d:%d, group %d:%d\n", 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_); break; case PGM_NCF: printf("NCF (TSI %d:%d) from %d:%d to %d:%d iface %d, size %d, seqno %d\n", hp->tsi_.addr_, hp->tsi_.port_, hip->saddr(), hip->sport(), hip->daddr(), hip->dport(), hc->iface(), hc->size(), hp->seqno_); break; default: printf("UNKNOWN (TSI %d:%d) from %d:%d to %d:%d iface %d, size %d, seqno %d\n", hp->tsi_.addr_, hp->tsi_.port_, hip->saddr(), hip->sport(), hip->daddr(), hip->dport(), hc->iface(), hc->size(), hp->seqno_); break; } } #endif // PGM_DEBUG #endif //HAVE_STL