lms-agent.cc

Go to the documentation of this file.

/*
 * Copyright (c) 2001 University of Southern California.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms are permitted
 * provided that the above copyright notice and this paragraph are
 * duplicated in all such forms and that any documentation, advertising
 * materials, and other materials related to such distribution and use
 * acknowledge that the software was developed by the University of
 * Southern California, Information Sciences Institute.  The name of the
 * University may not be used to endorse or promote products derived from
 * this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED "AS IS" AND WITHOUT ANY EXPRESS OR IMPLIED
 * WARRANTIES, INCLUDING, WITHOUT LIMITATION, THE IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE.
 *
 **
 * Light-Weight Multicast Services (LMS), Reliable Multicast
 *
 * lms-agent.cc
 *
 * This implements the network element LMS agent, "Agent/LMS".
 *
 * Christos Papadopoulos. 
 * christos@isi.edu
 */

#include <assert.h>

#include "agent.h"
#include "tclcl.h"
#include "packet.h"
#include "random.h"
#include "lms.h"
#include "ip.h"

static int lms_agent_uid_ = 0;  // each LMS agent has a uid

class LmsAgent : public Agent {
 public:
        LmsAgent ();
        int     command (int argc, const char*const* argv);
        void    recv (Packet*, Handler*);
 protected:
        //inline int    off_lms() { return off_lms_; };
        NsObject*       iface2link (int iface);
        NsObject*       pkt2agent (Packet *pkt);
        void            send_upstream (Packet *p);
        void            send_downstream (Packet *p);

        inline void
         send2replier (Packet *p)
                {
                hdr_ip* piph = HDR_IP(p);
                hdr_cmn* c = HDR_CMN(p);
#ifdef LMS_DEBUG
printf("%s send2replier: Got piph->saddr: %d, downstream_addr: %d. Replier: %d\n", uname_, piph->saddr(), downstream_lms_.addr_, replier_ );
#endif
                if (replier_ != NULL)
                        replier_->recv (p);
                else if ((downstream_lms_.addr_ != LMS_NOADDR) &&
                         (piph->saddr() != downstream_lms_.addr_) )
                        send_downstream(p);
                else    send_upstream (p); 
                };

        char    uname_[16];     // agent's unique name
        int     lms_enabled_;   // is this agent enabled? Default is YES
        NsObject* replier_;     // agent's replier (link or receiver)
        int     replier_iface_; // replier interface (-2 if replier is local)
        int     replier_cost_;  // replier cost
        int     replier_dist_;  // replier distance, in #of hops
        nsaddr_t upstream_lms_; // upstream LMS entity
        ns_addr_t downstream_lms_;// needed for incr.deployment
    
        int     upstream_iface_;// upstream interface (-2 if source is local)
        int     have_rcvr_;     // flag the presence of a receiver on this node

        // Header offsets
        // int  off_ip_;
        // int  off_lms_;
};

//
// Declaration of Agent/LMS
//
static class LmsClass : public TclClass {
public:
        LmsClass() : TclClass("Agent/LMS") {}
        TclObject* create(int, const char*const*) {
                return (new LmsAgent());
        }
} class_lms_agent;


//
// Declaration of PacketHeader/Lms
//

int hdr_lms::offset_;

static class LmsHeaderClass : public PacketHeaderClass {
public:
        LmsHeaderClass() : PacketHeaderClass("PacketHeader/Lms",
                                             sizeof(hdr_lms)) {
                bind_offset(&hdr_lms::offset_);
        }
} class_lmshdr;


//
// Initialization routine
//
LmsAgent::LmsAgent() : Agent(PT_LMS)
{
        sprintf (uname_, "lms%d", lms_agent_uid_++);
        replier_  = NULL;
        replier_iface_ = LMS_NOIFACE;
        replier_cost_  = LMS_INFINITY;
        replier_dist_  = 0;
        upstream_lms_  = LMS_NOADDR;
        downstream_lms_.addr_ = LMS_NOADDR;
        downstream_lms_.port_ = LMS_NOPORT;
        have_rcvr_ = 0;

        bind("lms_enabled_", &lms_enabled_);
        bind("packetSize_", &size_);
        // bind("off_ip_", &off_ip_);
        // bind("off_lms_", &off_lms_);
}

//
// Main LMS receive function
//
void LmsAgent::recv (Packet* pkt, Handler*)
{
    hdr_cmn* h = HDR_CMN(pkt);
    hdr_lms* lh = HDR_LMS(pkt);
    double now = Scheduler::instance().clock();
    
#ifdef LMS_DEBUG
int a1 = lh->from_; int a2 = lh->src_;
printf ("at %f %s received packet type %d from %d:%d src %d:%d group %x iface %d\n\n", now,uname_, lh->type(), a1>>8, a1&0xff, a2>>8, a2&0xff, lh->group_,h->iface());
#endif

        // if this agent is not enabled, simply pass on the packet
        if (!lms_enabled_)
                {
                target_->recv (pkt);
                return;
                }

        switch (lh->type())
        {
         case LMS_SRC_REFRESH:
                {
                // If we have a replier already, and the upstream adv cost
                // is less than our current cost and the upstream distance
                // is less than our current distance, then choose the
                // upstream link as the replier link.
                struct lms_ctl *c = (struct lms_ctl *)pkt->accessdata ();
                c->hop_cnt_++;
#ifdef LMS_DEBUG
printf ("%s LMS_SRC_REFRESH from iface %d, cost %d, ttl %d\n\n",
uname_, h->iface(), c->cost_, lh->ttl_);
#endif
                 assert (upstream_iface_ == h->iface());

                 if (replier_iface_ == h->iface())
                        {
                        if (replier_cost_ != c->cost_)
                                replier_cost_ = c->cost_;
                        if (replier_dist_ != c->hop_cnt_)
                                replier_dist_ = c->hop_cnt_;
                        }
                 else if (replier_cost_ >= c->cost_ && replier_dist_ > c->hop_cnt_)
                        {
#ifdef LMS_DEBUG
printf ("%s chose upstream replier\n\n", uname_);
#endif
                        replier_cost_ = c->cost_;
                        replier_dist_ = c->hop_cnt_;
                        replier_iface_ = h->iface();
                        if (h->iface() < 0)
                                {
                                replier_ = pkt2agent (pkt);
                                downstream_lms_.addr_ = LMS_NOADDR;
                                downstream_lms_.port_ = LMS_NOPORT;
                                }
                        else    {
                                replier_ = iface2link (h->iface());
                                downstream_lms_.addr_ = LMS_NOADDR;
                                downstream_lms_.port_ = LMS_NOPORT;
                                }
                        if (--lh->ttl_ > 0)
                                target_->recv (pkt);
                        else    Packet::free(pkt);
                        return;
                        }
                 break;
           }
           
         case LMS_REFRESH:
             {
                 struct lms_ctl *c = (struct lms_ctl *)pkt->accessdata ();
                 c->hop_cnt_++;
#ifdef LMS_DEBUG
printf("%s replier iface %d, h->iface %d, replier cost %d, c cost %d\n",
  uname_, replier_iface_, h->iface(), replier_cost_, c->cost_);
printf("replier_= %d, down_addr_= %d, down_port_ = %d\n\n",
  replier_, downstream_lms_.addr_, downstream_lms_.port_);
#endif     

                 // update from existing replier
                 // adjust cost and distance and send upstream
                 if (replier_iface_ == h->iface())
                 {
                     if (replier_cost_ != c->cost_)
                         replier_cost_ = c->cost_;
                     if (replier_dist_ != c->hop_cnt_)
                         replier_dist_ = c->hop_cnt_;
#ifdef LMS_DEBUG
printf ("%s REPLIER_UPDATE iface %d, cost %d, hops %d\n\n",
   uname_, replier_iface_, replier_cost_, replier_dist_);
#endif

    
                     if (h->iface() < 0)
                     {
                         replier_ = pkt2agent (pkt);
                         have_rcvr_ = 1;
                         downstream_lms_.addr_ = LMS_NOADDR;
                         downstream_lms_.port_ = LMS_NOPORT;
                     }
                     else
                     {  
                         // updates our entry for downstream_lms_ 
                         downstream_lms_ = c->downstream_lms_;
                         // places our own address in downstream_lms_
                         // before sending it upstream.
                         c->downstream_lms_.addr_ = addr();
                         c->downstream_lms_.port_ = port();
                         
                         replier_ = NULL;
                     }
                     send_upstream (pkt);
                 }

                 // found a better replier
                 else if (replier_cost_ > c->cost_)
                 {
                     replier_cost_ = c->cost_;
                     replier_dist_ = c->hop_cnt_;
                     replier_iface_ = h->iface();
                     
                     if (h->iface() < 0)
                     {
                         replier_ = pkt2agent (pkt);
                         have_rcvr_ = 1;
                         downstream_lms_.addr_ = LMS_NOADDR;
                         downstream_lms_.port_ = LMS_NOPORT;
                     }
                     else
                     {  
                         // updates our entry for downstream_lms_ 
                         downstream_lms_ = c->downstream_lms_;
                         // places our own address in downstream_lms_
                         // before sending it upstream.
                         c->downstream_lms_.addr_ = addr();
                         c->downstream_lms_.port_ = port();
                                
                         replier_ = NULL;                                    
                     }
#ifdef LMS_DEBUG
printf ("%s REPLIER iface %d, cost %d, hops %d\n\n",
  uname_, replier_iface_, replier_cost_, replier_dist_);
#endif
                     send_upstream (pkt);
                 }
                 else
                     Packet::free(pkt);
                 return;
             }

         case LMS_REQ:
             if (replier_iface_ == h->iface())
                 send_upstream (pkt);
             else {
                 // fill turning point info, but only
                 // if no-one else has done so already.
                 if (lh->tp_addr_ == LMS_NOADDR)
                        {
                                        lh->tp_addr_  = addr();
                                        lh->tp_port_  = port();
                                        lh->tp_iface_ = h->iface_;
                        }
                 send2replier (pkt);
                }
             break;

         case LMS_DMCAST:
             {
#ifdef LMS_DEBUG
a2 = lh->src_;
printf ("LMS-DMCAST at agent %s, iface %d src %d:%d group %d\n\n",
  uname_, lh->tp_iface_, a2>>8, a2&0xff, lh->group_);
#endif
                 hdr_ip* iph = HDR_IP(pkt);
                 NsObject* tgt = iface2link (lh->tp_iface_);

                 if (tgt)
                 {
                     packet_t t = h->ptype();
                     iph->saddr() = lh->src_;
                     iph->daddr() = lh->group_;
                     tgt->recv (pkt);
                 }
                 else   {
                     printf ("FATAL: %s no such iface %d\n", uname_, lh->tp_iface_);
                     abort ();
                 }
             }
             break;
             
         case LMS_LEAVE:
             {
                 
                 if (replier_iface_ == h->iface())
                 {
                     downstream_lms_.addr_ = LMS_NOADDR;
                     downstream_lms_.port_ = LMS_NOPORT;
                     
                     replier_ = NULL;
                     replier_cost_ = 1000000;
                     replier_iface_ = LMS_NOIFACE;
                     send_upstream (pkt);
                 }
                 else
                     Packet::free(pkt);
                 
                 return;
             }
             
         case LMS_SETUP:
             {
                 
                 if (h->iface() < 0)
                 {
                     // Source is attached - mark it as the replier
                     replier_ = pkt2agent (pkt);
                     replier_iface_ = h->iface();
                     downstream_lms_.addr_ = LMS_NOADDR;
                     downstream_lms_.port_ = LMS_NOPORT;
                     
                     replier_cost_ = 0;         // XXX
                     
#ifdef LMS_DEBUG
printf ("%s REPLIER iface %d, cost %d\n\n", uname_, replier_iface_, replier_cost_);
#endif

                 }
                 upstream_lms_ = lh->from_;
                 upstream_iface_ = h->iface();
                 
#ifdef LMS_DEBUG    
printf ("%s upstream %d\n\n", uname_, upstream_lms_);
#endif
 
                 lh->from_ = addr();
                 Tcl::instance().evalf("[%s set node_] agent %d", name(), port());
                 target_->recv (pkt);
                 return;
             }
             
         case LMS_SPM:
             {
                 if (upstream_lms_ < 0)
                 {
                     struct lms_spm *spm = (struct lms_spm *)pkt->accessdata ();
                     nsaddr_t adr = spm->spm_path_;
#ifdef LMS_DEBUG    
printf ("%s LMS_SPM seqno %d, upstream %d:%d\n\n",
  uname_, spm->spm_seqno_, adr>>8, adr&0xff);
#endif

                      if (upstream_lms_ != spm->spm_path_)
                          upstream_lms_ = spm->spm_path_;

                      spm->spm_path_ = addr();
                 }
                 break;
             }
             
         case LMS_LINKS:
             {
                 Tcl& tcl = Tcl::instance();
                 char wrk[64];
                 int  n1 = lh->from();
                 int  n2 = addr();
                 
                 if (n1 != n2 && !have_rcvr_)
                 {
                     sprintf (wrk, "lappend tree_links {%d %d}", n1, n2);
                     tcl.eval (wrk);
                 }
                 
                 lh->from_ = addr();
                 target_->recv (pkt);
                 return;
             }
         default:
             printf ("FATAL: %s uknown LMS packet type: %d\n", uname_, lh->type());
             abort ();
     }          
}

     

int LmsAgent::command (int argc, const char*const* argv)
{
        Tcl& tcl = Tcl::instance();

        if (argc == 6) {
                if (strcmp(argv[1], "send-lms") == 0) {
                        Packet* pkt = allocpkt();
                        hdr_lms* ph = HDR_LMS(pkt);

                        ph->type() = atoi (argv[2]);
                        ph->from() = atoi(argv[3]);
                        ph->src() = atoi(argv[4]);
                        ph->group() = atoi(argv[5]);

                        send(pkt, 0);
                        return (TCL_OK);
                }
        }
        
        return (Agent::command(argc, argv));
}


/*
 * Given an interface number,
 * return its outgoing link
 */
NsObject* LmsAgent::iface2link (int iface)
{
        Tcl&    tcl = Tcl::instance();
        char    wrk[64];

        sprintf (wrk, "[%s set node_] ifaceGetOutLink %d", name (), iface);
        tcl.evalc (wrk);
        char* result = tcl.result ();

#ifdef LMS_DEBUG
printf ("[iface2link] agent %s\n", result);
#endif

        NsObject* obj = (NsObject*)TclObject::lookup(result);
        return (obj);
}

/*
 * Given a packet, determine which agent sent it
 * Agent MUST be attached to this node
 */
NsObject* LmsAgent::pkt2agent (Packet *pkt)
{
        Tcl&            tcl = Tcl::instance();
        char            wrk[64];
        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 LMS_DEBUG
printf ("[pkt2agent] port %d, agent %s\n", port, result);
#endif

        agent = (NsObject*)TclObject::lookup (result);
        return (agent);
}

void LmsAgent::send_upstream (Packet *p)
{
        if (upstream_lms_ < 0)
                {
                printf ("FATAL: %s upstream_lms_ not set!\n", uname_);
                abort ();
                }
        hdr_ip* ih = HDR_IP(p);
        hdr_lms* lh = HDR_LMS(p);

        lh->from_ = addr();
        ih->daddr() = upstream_lms_;
        target_->recv(p);
}


void LmsAgent::send_downstream (Packet *p)
{
    if (downstream_lms_.addr_ < 0)
    {
        printf ("FATAL: %s downstream_lms_ not set!\n", uname_);
        abort ();
    }
    hdr_ip* ih = HDR_IP(p);
    hdr_lms* lh = HDR_LMS(p);
    
    lh->from_ = addr();
    ih->daddr() = downstream_lms_.addr_;
    ih->dport() = downstream_lms_.port_;
    
    target_->recv(p);
}

---