lms-sender.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-sender.cc
 *
 * This implements the Sending LMS agent, "Agent/LMS/Sender".
 *
 * Christos Papadopoulos. 
 * christos@isi.edu
 */

#include "agent.h"
#include "tclcl.h"
#include "packet.h"
#include "ip.h"
#include "rtp.h"
#include "config.h"
#include "random.h"
#include "trafgen.h"
#include "lms.h"

static int snd_uid_ = 0;

class LmsSender;

class LmsSender: public Agent {
 public:
        LmsSender();
        LmsSender *next_;
        int  command (int argc, const char*const* argv);
        void recv (Packet* pkt, Handler*);

 protected:
        void handle_lms_pkt (Packet* pkt);
        virtual void sendmsg(int nbytes, const char *flags = 0);
        void send_lms_pkt (int, int);
        void send_dmcast (hdr_lms* lh, int seqno, int fid);
        int  add_req (Packet *rq);
        void send_spm ();
        void solicit_naks ();
        void print_stats ();
        void print_all_stats (int drops);
    
        char    uname_[16];     // unique name in the form r#
        int     spm_seqno_;     // spm sequence number
        int     lms_cost_;      // for sender this is usually zero
        int     lms_ttl_;       // solicit nacks this many hops away
        int     packetSize_;    // remember pkt size for retransmissions
        int     lmsPacketSize_; // packet size for lms packets
        Packet  *req_list_;     // request list to detect duplicate requests
        int     req_list_sz_;   // current size of above list

        // Stats
        int     req_rcvd_;      // number of requests received
        int     dup_reqs_;      // number of duplicate requests received
        int     dmcasts_sent_;  // number of dmcasts sent

 static int     max_dup_naks_;  // max num of duplicate naks    
    
        // int  maxpkts_;
        int     seqno_;         // XXX: RTP should handle this

        //LmsSenderTimer lms_sender_timer_;
        
        // int  off_lms_;
        // int  off_ip_;
        // int  off_rtp_;
        // int  off_cmn_;
};

static class LmsSenderClass : public TclClass {
public:
        LmsSenderClass() : TclClass("Agent/LMS/Sender") {}
        TclObject* create(int, const char*const*) {
                return (new LmsSender());
        }
} class_lms_sender;


// Obligatory definition of the class static
int LmsSender::max_dup_naks_ = 0;

LmsSender::LmsSender(): Agent(PT_LMS), seqno_(-1)
{
        sprintf (uname_, "sender%d", snd_uid_++);
        
        lms_cost_ = 0;
        spm_seqno_ = 0;
        req_list_ = NULL;
        req_rcvd_ = dup_reqs_ = max_dup_naks_ = 0;
        dmcasts_sent_ = 0;
        bind("lmsPacketSize_", &lmsPacketSize_);
        bind("packetSize_", &packetSize_);
        // bind("off_ip_", &off_ip_);
        // bind("off_lms_", &off_lms_);
        // bind("off_rtp_", &off_rtp_);
        // bind("off_cmn_", &off_cmn_);
        // bind("maxpkts_", &maxpkts_);
}

//
// Send a data packet
//
void LmsSender::sendmsg(int nbytes, const char* flags)
{
        Packet*   p = allocpkt ();
        hdr_ip* ih = HDR_IP(p);
        hdr_cmn* th = HDR_CMN(p);
        hdr_rtp* rh = HDR_RTP(p); // XXX

        bzero (HDR_LMS(p), sizeof (struct hdr_lms));
        packetSize_ = nbytes;   
        th->size_ = packetSize_;   
        rh->seqno() = ++seqno_; // XXX

        ih->flowid() = 1; //for coloring in NAM

#ifdef LMS_DEBUG
packet_t t = th->ptype();
const char* nname = packet_info.name(t);
double now = Scheduler::instance().clock();
printf("SNDR: at %f %s sendmsg pkt %d type is %s, size is %d\n\n",
  now, uname_, seqno_, nname, th->size_);
#endif

        target_->recv(p);
}

//
// Send an LMS packet
// set LMS header and adjust size_
//
void LmsSender::send_lms_pkt (int type, int cost)
{
        Packet* p = allocpkt();

        (HDR_CMN(p))->ptype_ = PT_LMS;
        (HDR_IP(p))->flowid() = 7; //Coloring in NAM

        if (type != LMS_SETUP)
                (HDR_CMN(p))->size_ = lmsPacketSize_; 
                //(HDR_CMN(p))->size_ = sizeof(hdr_lms); 
        else 
                (HDR_CMN(p))->size_ = packetSize_; 
                //(HDR_CMN(p))->size_ = sizeof(hdr_lms); 

        hdr_lms* lh = HDR_LMS(p);
        lh->type_ = type;
        lh->cost_ = cost;
        lh->from_ = addr();
//      lh->from_ = port();
        lh->src_ = addr();
        lh->group_ = daddr();
        lh->tp_addr_  = LMS_NOADDR;
        lh->tp_port_ = -1;
        lh->tp_iface_ = LMS_NOIFACE;
        lh->lo_ = lh->hi_ = -1;
        lh->ts_ = Scheduler::instance().clock();

#ifdef LMS_DEBUG
printf ("SNDR: %s send_lms_pkt from %d src %d dst %x type %d cost %d size %d at %f\n\n",
  uname_, lh->from_, lh->src_, lh->group_, lh->type_, lh->cost_,
  (HDR_CMN(p))->size_, lh->ts_);
#endif 
        target_->recv(p);
}

//
// Receive a packet
//
void LmsSender::recv(Packet* pkt, Handler*)
{
        hdr_cmn* h = HDR_CMN(pkt);

        if (h->ptype_ == PT_LMS)
                handle_lms_pkt (pkt);
        else    {
                printf ("ERROR: %s received non LMS pkt type %d\n", uname_, h->ptype_);
                Packet::free(pkt);
                }
}

/*
 * Handle LMS packets
 */
void LmsSender::handle_lms_pkt (Packet* pkt)
{
        int st = 0;

        hdr_lms* lh = HDR_LMS(pkt);
        hdr_ip* iph = HDR_IP(pkt);
#ifdef LMS_DEBUG
int a1, a2;
#endif
        switch (lh->type())
                {
                case LMS_REQ:
                        req_rcvd_++;
                        if (lh->src_ != addr())
                                {
                                printf ("ERROR: %s REQ with wrong source addr %d:%d\n",
                                        uname_, lh->src_>>8, lh->src_&0xff);
                                abort ();
                                }
#ifdef LMS_DEBUG
a1 = lh->from(); a2 = lh->src();
printf ("SNDR: %s got LMS_REQ from %d:%d src %d:%d group 0x%x\n\n",
  uname_, a1>>8, a1&0xff, a2>>8, a2&0xff, lh->group());
printf ("SNDR: TP: (%d, %d)\n\n", lh->tp_addr_, lh->tp_iface_);
#endif
                        if ((st = add_req (pkt)) != 0)
                                for (int i = lh->lo_; i <= lh->hi_; i++)
                                        send_dmcast (lh, i, 3);
                        else dup_reqs_++;

                        break;
                case LMS_REFRESH:
                        break;
                default:
                        printf ("***ERROR: %s Unexpected LMS packet type: %d\n\n", uname_, lh->type());
                        abort ();
                }
        if (!st)
                Packet::free(pkt);
}

int LmsSender::command(int argc, const char*const* argv)
{
    if (argc == 2) {
        if (strcmp(argv[1], "lms-setup") == 0) {
            send_lms_pkt (LMS_SETUP, packetSize_);      // XXX
            return (TCL_OK);
        }
        if (strcmp(argv[1], "gather-links") == 0) {
            send_lms_pkt (LMS_LINKS, 0);
            return (TCL_OK);
        }
        if (strcmp(argv[1], "print-stats") == 0) {
            print_stats ();
            return (TCL_OK);
        }
    }
    if (argc == 3)
    {
        if (strcmp(argv[1], "solicit-naks") == 0) {
            lms_ttl_ = atoi (argv[2]);
            solicit_naks ();
            return (TCL_OK);
        }
        if (strcmp(argv[1], "set-replier-cost") == 0)
        {
            lms_cost_  = atoi (argv[2]);
            send_lms_pkt (LMS_REFRESH, lms_cost_);
            return (TCL_OK);
        }
        if (strcmp(argv[1], "print-all-stats") == 0) {
            print_all_stats ( atoi(argv[2]) );
            return (TCL_OK);
        }         
        if (strcmp(argv[1], "send") == 0) {
            sendmsg( atoi (argv[2]), 0);
            return (TCL_OK);
        }
    }
    return (Agent::command(argc, argv));
}

/*
 * Send a LMS_DMCAST in response to LMS_REQ packet
 */
void LmsSender::send_dmcast (hdr_lms* lh, int seqno, int fid)
{
        Packet          *p = allocpkt();
        hdr_rtp         *rh = HDR_RTP(p);
        hdr_ip          *piph = HDR_IP(p);
        hdr_cmn         *ch = HDR_CMN(p);
        hdr_lms         *plh = HDR_LMS(p);

        dmcasts_sent_++;
        rh->seqno() = seqno;
        piph->daddr() = lh->tp_addr_;
        piph->dport() = lh->tp_port_;
        ch->size_= packetSize_;
        piph->flowid() = fid;           // Green for repair; to color packets in NAM

        plh->type_ = LMS_DMCAST;
        plh->from_ = addr();
//      plh->from_ = port();
        plh->src_ = addr();
        plh->group_ = daddr();
        plh->tp_addr_ = lh->tp_addr_;
        plh->tp_port_ = lh->tp_port_;
        plh->tp_iface_ = lh->tp_iface_;
        packet_t t = ch->ptype();
        const char* nname = packet_info.name(t);
        
        target_->recv(p);
}

// Add a new request to the request list.
// Return 1 if the request was added,
// 0 if this was a duplicate request and was not added.
// Requests are added to the head of the list
// and the list size is limited to 10 (XXX: should have time limit instead)
//
int LmsSender::add_req (Packet *rq)
{
        hdr_lms *lh = HDR_LMS(rq);
        Packet  *p = req_list_;
        int     i = 0;

        if (!p)
        {
            req_list_ = rq;
            rq->next_ = 0;
            req_list_sz_ = 1;
            return 1;
        }
        while (p)
        {
            if (i++ > 10)
                break;
            hdr_lms *plh = HDR_LMS(p);
            if ((plh->lo_ == lh->lo_) &&
                (plh->hi_ == lh->hi_) &&
                (plh->tp_addr_ == lh->tp_addr_) &&
                (plh->tp_iface_ == lh->tp_iface_))
            {
                struct lms_nak *nh = (struct lms_nak *)p->accessdata();
                // increments the dup_cnt_ for this NAK, and updates
                // max_dup_naks_ if appropriate
                ++nh->dup_cnt_;

#ifdef LMS_DEBUG
   printf ("SNDR: %s got %d dup reqs, max is %d\n", uname_, nh->dup_cnt_, max_dup_naks_);
#endif                
                
                if( nh->dup_cnt_ > max_dup_naks_)
                    max_dup_naks_ = nh->dup_cnt_;
                                
                return 0;                
            }
            
            p = p->next_;
        }
        if (i > 10 && p && p->next_)
        {
            Packet::free (p->next_);
            p->next_ = 0;
            req_list_sz_--;
        }
        rq->next_ = req_list_;
        req_list_ = rq;
        req_list_sz_++;
        return 1;
}


void LmsSender::send_spm ()
{
        Packet* p = allocpkt (sizeof (struct lms_spm));
        (HDR_CMN(p))->ptype_ = PT_LMS;
                (HDR_CMN(p))->size_ = sizeof(struct lms_spm) + sizeof(hdr_lms);
        (HDR_IP(p))->flowid() = 7;

        hdr_lms* lh = HDR_LMS(p);
        
        lh->type_ = LMS_SPM;

        struct lms_spm *spm = (struct lms_spm *)p->accessdata();
        spm->spm_seqno_ = spm_seqno_++;
        spm->spm_path_ = addr();
        spm->spm_ts_ = Scheduler::instance().clock();

        target_->recv(p);
}

void LmsSender::solicit_naks ()
{
        Packet* p = allocpkt(sizeof (struct lms_ctl));
        struct lms_ctl  *ctl = (struct lms_ctl*)p->accessdata ();
        (HDR_CMN(p))->size_ = sizeof(struct lms_ctl) + sizeof(hdr_lms); 
        (HDR_CMN(p))->ptype_ = PT_LMS;
    (HDR_IP(p))->flowid() = 7;

        hdr_lms* lh = HDR_LMS(p);
        lh->type_ = LMS_SRC_REFRESH;
        lh->ttl_  = lms_ttl_;
        lh->from_ = addr();
        lh->src_  = addr();
        lh->group_ = daddr();

        ctl->cost_ = lms_cost_;
        ctl->hop_cnt_ = 0;

        target_->recv(p);
}

void LmsSender::print_stats ()
{
        printf ("%s:\n", uname_);
        printf ("\tPackets sent:\t\t%d\n", seqno_);
        printf ("\tRequests received:\t%d\n", req_rcvd_);
        printf ("\tDuplicate Requests:\t%d\n", dup_reqs_);
        printf ("\tDMCASTs sent:\t\t%d\n", dmcasts_sent_);
        printf ("\n");
}

void LmsSender::print_all_stats (int drops)
{
    if (drops) 
        printf ("\t%.5lf\t  %d", float(dup_reqs_)/float(drops), max_dup_naks_);
    else 
        printf ("\t0.0\t  0");
}

---