/* * pgm-sender.cc * Copyright (C) 2001 by the University of Southern California * $Id: pgm-sender.cc,v 1.11 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-sender.cc * * This implements the Sending PGM agent, "Agent/PGM/Sender". * * Ryan S. Barnett, 2001 * rbarnett@catarina.usc.edu */ #include "config.h" #ifdef HAVE_STL #include #include /* Standard Template Library */ #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" // ************************************************************ // Define the PGM Sender Timer Class // ************************************************************ class PgmSender; /* Define the different timer types. */ enum { TIMER_SPM = 0, TIMER_RDATA = 1 }; class PgmSenderTimer : public TimerHandler { public: PgmSenderTimer(PgmSender *a, int type) : TimerHandler(), data_(NULL) { a_ = a; type_ = type; } int & type() { return type_; } void * &data() { return data_; } protected: virtual void expire(Event *e); PgmSender *a_; int type_; void *data_; }; // Bundles an RDATA packet with a timer (for sending the RDATA packet), // and a list of interfaces for which the packet should be sent to. class RdataItem { public: RdataItem(int seqno, PgmSender *a, Packet *rdata_pkt = NULL) : seqno_(seqno), rdata_pkt_(rdata_pkt), rdata_timer_(a, TIMER_RDATA) { } // Allow direct access to the private data. int & seqno() { return seqno_; } Packet * & rdata_pkt() { return rdata_pkt_; } PgmSenderTimer & rdata_timer() { return rdata_timer_; } list & iface_list() { return iface_list_; } list & agent_list() { return agent_list_; } protected: // The sequence number of this RDATA item. int seqno_; // The RDATA packet itself. Packet *rdata_pkt_; // The timer responsible for sending out this RDATA packet. PgmSenderTimer rdata_timer_; // The list of interfaces for which this packet must be sent to. list iface_list_; // The list of agents for which this packet must be sent to. list agent_list_; }; // A class used to keep track of duplicate reply requests. class ReplyItem { public: ReplyItem(int seqno) : seqno_(seqno), retransmissions_(0) { } int & seqno() { return seqno_; } int & retransmissions() { return retransmissions_; } protected: int seqno_; int retransmissions_; }; // Largest size we will allow the reply list to grow. const int MAX_REPLY_LIST_SIZE = 100; // Miscellaneous statistical information gathered during simulation. struct Stats { int num_naks_received_; int num_rdata_sent_; int max_num_repeated_rdata_; }; // ************************************************************ // Define the PGM Sender Class // ************************************************************ static int pgm_snd_uid_ = 0; class PgmSender: public Agent { public: PgmSender(); virtual void recv(Packet *, Handler *); virtual void timeout(int type, void *data); virtual int command(int argc, const char*const* argv); virtual void sendmsg(int nbytes, const char *flags = 0); protected: virtual void start(); // Starts the SPM heartbeats. virtual void stop(); // Stops the SPM heartbeats. virtual void handle_nak(Packet *pkt); // Process a NAK packet. virtual void send_spm(); // Sends an SPM packet to the multicast group. virtual void send_rdata(RdataItem *pkt); // Sends the given RDATA packet. NsObject* iface2link(int iface); NsObject* pkt2agent(Packet *pkt); void print_stats(); void display_packet(Packet *pkt); // For debugging. void PgmSender::trace_event(char *evType, nsaddr_t daddr, double evTime); EventTrace * et_; //Trace Object for custom Event Traces Stats stats_; // Keep track of various statistics. char uname_[16]; // Unique PGM sender name. // Map the sequence number of a NAK (requested RDATA) with an item // that represents the RDATA packet including which interfaces the // RDATA should be sent to, along with a timer that is used to trigger // sending of the RDATA packet. map pending_rdata_; // A list to keep track of the number of retransmissions for a given // RDATA reply. The max size this will grow is MAX_REPLY_LIST_SIZE. list reply_; // The 'typicial' size of a data packet including header. This should // get set automatically from the application calling sendmsg(). We // make a simplifying assumption that all packet sizes in a session are // of equal size. int pktSize_; PgmSenderTimer spm_heartbeat_; // Timer for sending out SPM packets. int spm_running_; // Whether the heartbeats are running or not. double spm_interval_; // Time between SPM packets (in seconds). // Time to delay sending out an RDATA in response to a NAK packet, this // is to allow slow NAKs to get processed at one time, so we don't send // out duplicate RDATA. double rdata_delay_; int odata_seqno_; // Current ODATA sequence number. int spm_seqno_; // Current SPM sequence number. // nsaddr_t group_; // The multicast group we send to. }; void PgmSenderTimer::expire(Event *e) { a_->timeout(type_, data_); } static class PgmSenderClass : public TclClass { public: PgmSenderClass() : TclClass("Agent/PGM/Sender") {} TclObject * create(int argc, const char * const * argv) { return (new PgmSender()); } } class_pgm_sender; // Constructor. PgmSender::PgmSender() : Agent(PT_PGM), pktSize_(0), spm_heartbeat_(this, TIMER_SPM), spm_running_(0), odata_seqno_(-1), spm_seqno_(-1) { stats_.num_naks_received_ = 0; stats_.num_rdata_sent_ = 0; stats_.max_num_repeated_rdata_ = 0; sprintf(uname_, "pgmSender-%d", pgm_snd_uid_++); bind_time("spm_interval_", &spm_interval_); bind_time("rdata_delay_", &rdata_delay_); et_ = (EventTrace *) NULL; } // Code that is called when a packet is received. void PgmSender::recv(Packet *pkt, Handler *) { hdr_cmn* hc = HDR_CMN(pkt); if (hc->ptype_ == PT_PGM) { #ifdef PGM_DEBUG display_packet(pkt); #endif // Identify the type of PGM packet, if it is a NAK process it, otherwise // throw an error. hdr_pgm *hp = HDR_PGM(pkt); if (hp->type_ == PGM_NAK) { handle_nak(pkt); } else { printf("ERROR (PgmSender::handle_pgm_pkt): received unexpected PGM packet type %d, discarding.\n", hp->type_); } } else { printf ("%s ERROR (PgmSender::recv): received non PGM pkt type %d, discarding.\n", uname_, hc->ptype_); } // Free all packets that this agent receives. Packet::free(pkt); } // Code that is called when a timer expires. void PgmSender::timeout(int type, void *data) { switch(type) { case TIMER_SPM: if (spm_running_) { send_spm(); spm_heartbeat_.resched(spm_interval_); } break; case TIMER_RDATA: send_rdata((RdataItem *)data); break; default: printf("ERROR (PgmSender::timeout): invalid timeout type.\n"); break; } } // Code that is called when a TCL command is issued to the PGM Sender object. int PgmSender::command(int argc, const char*const* argv) { if (argc == 2) { if (strcmp(argv[1], "start-SPM") == 0) { start(); return (TCL_OK); } if (strcmp(argv[1], "stop-SPM") == 0) { stop(); return (TCL_OK); } if (strcmp(argv[1], "print-stats") == 0) { print_stats(); return (TCL_OK); } } else if (argc == 3) { //If Event Trace is on, set the Event trace handle if (strcmp(argv[1], "eventtrace") == 0) { et_ = (EventTrace *)TclObject::lookup(argv[2]); return (TCL_OK); } } return (Agent::command(argc, argv)); } void PgmSender::trace_event(char *evType, nsaddr_t daddr, 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(), daddr, 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 daddr, // dst node id 0 // dst port id ); et_->dump(); } } // The application calls this function to send out new ODATA (original DATA). void PgmSender::sendmsg(int nbytes, const char *flags /* = 0 */) { odata_seqno_++; #ifdef PGM_DEBUG double now = Scheduler::instance().clock(); printf("at %f %s sending ODATA seqno %d\n", now, uname_, odata_seqno_); #endif // Create a packet with the given ODATA. Packet *pkt = allocpkt(); // Set the simulated size of the packet to the indicated nbytes. hdr_cmn *hc = HDR_CMN(pkt); pktSize_ = nbytes + sizeof(hdr_pgm); hc->size_ = pktSize_; hc->ptype_ = PT_PGM; // Fill in the PGM header. hdr_pgm *hp = HDR_PGM(pkt); hp->type_ = PGM_ODATA; hp->tsi_ = here_; // Set transport session ID to addr/port of this agent. hp->seqno_ = odata_seqno_; hdr_ip *hip = HDR_IP(pkt); // Set the color for ODATA packets. hip->fid_ = 1; // Send out the packet. send(pkt, 0); } void PgmSender::start() { spm_running_ = 1; send_spm(); spm_heartbeat_.resched(spm_interval_); } void PgmSender::stop() { spm_heartbeat_.cancel(); spm_running_ = 0; Tcl::instance().evalf("%s done", this->name()); } // Process a NAK packet. void PgmSender::handle_nak(Packet *pkt) { hdr_cmn *hc = HDR_CMN(pkt); hdr_pgm *hp = HDR_PGM(pkt); //hdr_pgm_nak *pnak = HDR_PGM_NAK(pkt); if (!(hp->tsi_.isEqual (here_))) { printf("%s received NAK with wrong TSI, discarding.\n", uname_); return; } stats_.num_naks_received_++; // Create the NCF packet. Packet *ncf_pkt = allocpkt(); // Set the simulated size of the NCF packet. hdr_cmn *ncf_hc = HDR_CMN(ncf_pkt); ncf_hc->size_ = sizeof(hdr_pgm); ncf_hc->ptype_ = PT_PGM; // Fill in the PGM header for the NCF packet. hdr_pgm *ncf_hp = HDR_PGM(ncf_pkt); ncf_hp->type_ = PGM_NCF; ncf_hp->tsi_ = here_; ncf_hp->seqno_ = hp->seqno_; hdr_ip *ncf_hip = HDR_IP(ncf_pkt); // Set the color for NCF packets in nam. ncf_hip->fid_ = 6; // Immediately send the NCF packet to the interface where the NAK // packet was received. If the packet came from another agent attached // to this node, then send the packet to that agent. NsObject *tgt; if (hc->iface() < 0) { tgt = pkt2agent(pkt); } else { tgt = iface2link(hc->iface()); } if (tgt == NULL) { printf("ERROR (PgmSender::handle_nak): iface2link returned NULL.\n"); abort(); } tgt->recv(ncf_pkt); // Queue up an RDATA packet to be transferred to the requestor on the // appropriate interface. // Attempt to locate this NAK sequence number on the pending RDATA map. pair::iterator, bool> result; result = pending_rdata_.insert(pair(hp->seqno_, RdataItem(hp->seqno_, this))); RdataItem *ritem = &(result.first->second); if (result.second == true) { // The entry was added to the map. // Set the data field of the timer. ritem->rdata_timer().data() = ritem; // There is NO pending RDATA for the indicated sequence number. // Create the RDATA packet. Packet *rdata_pkt = allocpkt(); // Set the simulated size of the RDATA packet to the typicial data size. hdr_cmn *rdata_hc = HDR_CMN(rdata_pkt); rdata_hc->size_ = pktSize_; rdata_hc->ptype_ = PT_PGM; // Fill in the PGM header for RDATA packet. hdr_pgm *rdata_hp = HDR_PGM(rdata_pkt); rdata_hp->type_ = PGM_RDATA; rdata_hp->tsi_ = here_; // Set transport session ID to addr/port of this agent. rdata_hp->seqno_ = hp->seqno_; hdr_ip *rdata_hip = HDR_IP(rdata_pkt); // Set the color for RDATA packets in nam. rdata_hip->fid_ = 3; // Place the new packet into the RdataItem in the map. ritem->rdata_pkt() = rdata_pkt; // Set the timer to go off at rdata_delay_ seconds from now. ritem->rdata_timer().resched(rdata_delay_); //Output Event Trace, Repair will be sent after rdata_delay_ trace_event("REPAIR BACKOFF", rdata_hip->daddr(), rdata_delay_); if (hc->iface() < 0) { // The NAK was sent from a local agent attached to this node. Keep // track of which agent this is. ritem->agent_list().push_back(pkt2agent(pkt)); } else { // Set the interface number for this RdataItem. ritem->iface_list().push_back(hc->iface()); } } else { // Seqno entry already exists in the map. // The RDATA is already pending. if (hc->iface() < 0) { // Scan the agent list to see if the agent is already registered // for this RDATA. list *agent_list = &(ritem->agent_list()); list::iterator res = find(agent_list->begin(), agent_list->end(), pkt2agent(pkt)); if (res == agent_list->end()) { // Agent not found in agent list for this RDATA, add it. agent_list->push_back(pkt2agent(pkt)); } else { printf("%s: NAK received and already had NAK state for that same agent.\n", uname_); } } else { // Scan the interface list to see if the interface is already registered // for this RDATA. list *iface_list = &(ritem->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 RDATA, add it. iface_list->push_back(hc->iface()); } else { // Interface already present in the iface list for this RDATA, // therefore this NAK is a duplicate. printf("%s: NAK received and already had NAK state for that same interface.\n", uname_); } } } } // Send out a new SPM packet to the multicast group. void PgmSender::send_spm() { spm_seqno_++; // Create a packet with the given ODATA. Packet *pkt = allocpkt(); // Set the simulated size of the packet to the indicated nbytes. hdr_cmn *hc = HDR_CMN(pkt); hc->size_ = sizeof(hdr_pgm) + sizeof(hdr_pgm_spm); hc->ptype_ = PT_PGM; hdr_ip *hip = HDR_IP(pkt); // Set the color for SPM packets in nam. hip->fid_ = 7; // hip->daddr() = group_; // Fill in the PGM header. hdr_pgm *hp = HDR_PGM(pkt); hp->type_ = PGM_SPM; hp->tsi_ = here_; // Set transport session ID to addr/port of this agent. hp->seqno_ = spm_seqno_; // Fill in SPM header. hdr_pgm_spm *hs = HDR_PGM_SPM(pkt); hs->spm_path_ = here_; // Set current path to the source agent. #ifdef PGM_DEBUG double now = Scheduler::instance().clock(); printf("at %f %s sending SPM, from %d:%d (here = %d:%d) to %d:%d, TSI %d:%d, type %d\n", now, uname_, hip->saddr(), hip->sport(), addr(), port(), hip->daddr(), hip->dport(), hp->tsi_.addr_, hp->tsi_.port_, hp->type_); #endif // Send out the packet. send(pkt, 0); } // Send out the given RDATA packet. The packet should be already created // and headers filled in. This is triggered when the timer expires. void PgmSender::send_rdata(RdataItem *item) { // Locate the sequence number of this rdata in the list of sent // retransmissions. int count = 0; list::iterator iter = reply_.begin(); while (iter != reply_.end()) { if ((*iter).seqno() == item->seqno()) { (*iter).retransmissions() += 1; if ((*iter).retransmissions() > stats_.max_num_repeated_rdata_) { stats_.max_num_repeated_rdata_ = (*iter).retransmissions(); } break; } count++; iter++; } if (iter == reply_.end()) { // This is the first time we're sending out this RDATA. Append it to // the back of the reply list. if (count >= MAX_REPLY_LIST_SIZE) { // Pop off the front-most element if we've reached the max size of // the reply list. reply_.pop_front(); } reply_.push_back(ReplyItem(item->seqno())); } stats_.num_rdata_sent_++; // Send the packet to each of the interfaces. if (!item->iface_list().empty()) { list::iterator iter = item->iface_list().begin(); int flag = 0; // Used to determine when we need to make additional copies of the packet. while (iter != item->iface_list().end()) { NsObject *tgt; tgt = iface2link(*iter); Packet *pkt = item->rdata_pkt(); if (flag) { // 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. pkt = pkt->copy(); } else { trace_event("SEND RDATA", HDR_IP(pkt)->daddr(), 0); flag = 1; } tgt->recv(pkt); iter++; } } if (!item->agent_list().empty()) { list::iterator iter = item->agent_list().begin(); int flag = 0; while (iter != item->agent_list().end()) { Packet *pkt = item->rdata_pkt(); if (flag) { pkt = pkt->copy (); } else { flag = 1; } (*iter)->recv(pkt); iter++; } } hdr_pgm *hp = HDR_PGM(item->rdata_pkt()); // Remove this sequence number from the pending RDATA list, since // we have now sent that RDATA. if (!pending_rdata_.erase(hp->seqno_)) { printf("ERROR (PgmSender::send_rdata): Did not erase RdataItem from map.\n"); } } NsObject* PgmSender::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* PgmSender::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); } void PgmSender::print_stats() { printf("%s\n", uname_); printf("\tLast ODATA seqno: %d\n", odata_seqno_); printf("\tLast SPM seqno: %d\n", spm_seqno_); printf("\tNumber of NAKs received: %d\n", stats_.num_naks_received_); printf("\tNumber of RDATA transmitted: %d\n", stats_.num_rdata_sent_); printf("\tMax retransmission count for a single RDATA: %d\n", stats_.max_num_repeated_rdata_); } #ifdef PGM_DEBUG void PgmSender::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