almah.cc

为NS2仿真中的一个overlay多播协议的协议模块

/***************************************************************************
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 ***************************************************************************/

#include "almah.h"
#include <iostream.h>
#include <random.h>
#include <cmu-trace.h>
#include "connector.h"
#include "template.h"
#include <string.h>

#define NETWORK_DIAMETER        30
#define T_TX_MAX 0.01 //max time to send a packet

static class Overlay_Class : public TclClass {
public:
	Overlay_Class() : TclClass("Agent/ALMAH") {}
	TclObject* create(int, const char*const*) {
		return (new ALMAH_Agent());
	}
} class_overlay;

char ALMAH_Agent::nwrk_[256];

ALMAH_Agent* ALMAH_Agent::Agents[50];

ALMAH_Agent::ALMAH_Agent() : Agent(PT_ALMAH), namChan_(0),
ptimer(this),htimer(this), ttimer_2(this),pingtimer(this),dttimer(this), rttimer(this) ,topreqtimer(this),loopr_timer(this), htimer_2(this){
	
  found_neighbour = false;
  index = here_.addr_;
  LIST_INIT(&mbhead);
  LIST_INIT(&nbhead);
  LIST_INIT(&bhead);
  LIST_INIT(&dhead);
  bind("packetSize_", &size_);
  bind("I_TTL", &I_TTL);
  bind("INTERVAL_TTL",&INTERVAL_TTL);
  bind("MAX_TTL", &MAX_TTL);
  bind("TREE_INT",&TREE_INT);
  bind("M_cast",&M_cast);
  bind("ON_SendRequestTimer",&ON_SendRequestTimer);
  bind("ON_PingTimer",&ON_PingTimer);
  bind("ON_PRINT_NAM",&ON_PRINT_NAM);
  bind("TH_LEVEL_1",&threshold_level1);
  bind("TH_LEVEL_2",&threshold_level2); 
  bind("TH_LEVEL_3",&threshold_level3);
  bind("TH_LEVEL_4",&threshold_level4); 
  bind("DATA_SEQUENCE_NUMBER",&dsn);
  bind("DATA_RECEIVED_COUNTER",&drc);
  bind("MEAN_DATA_DELAY",&mdelay);
  bind("HOP_BASED_MODE",&hop_based_mode);
  req_last_ttl = I_TTL;
  req_ttl_ =  I_TTL;    
  DT_FILE=NULL;	
  parent_in = -1;
  ON_PingTimer = 0;
  parent_in_switch = -1;
  parent_in_distance = -1;
  topology_list = NULL;
  path_list = NULL;
  parent_in_rtt = 0;
  TH_DISTANCE_SETTED = 1;
  switch_time_on = false;
  SWITCHING = false;	
  NACKING = false;
  JOIN_ACKING = false;
  REQUEST_TOPOLOGY_STATUS = false;
  LOOP_RESET_ = false;
  expiry_interval=60;  
}

int ALMAH_Agent::command(int argc, const char*const* argv)
{
    
  index = here_.addr_;
  ns_addr_t ind =  here_;

    if (strcmp(argv[1], "start") == 0){
          
          htimer.handle((Event*) 0);	// una-tantum Join Req 

		  rttimer.handle((Event*) 0);  // RefreshTopology Timer
		  
		  pingtimer.handle((Event*) 0); // Ping Timer
		  
		  is_member=true;

         return (TCL_OK);  
    }
    if (strcmp(argv[1], "agent-attach")==0) {		// attacca il puntatore
        ALMAH_Agent* pippo = (ALMAH_Agent*)TclObject::lookup(argv[2]);

        this->Agents[pippo->index] = pippo;

			  return (TCL_OK);
		}
    if (strcmp(argv[1], "send_data") == 0){
         this->sendDATA(); 
         return (TCL_OK);
    } else if (strcmp(argv[1], "send_broadcast") == 0) {
		      //this->send_Broadcast();
      return (TCL_OK);
    } else if (strcmp(argv[1], "data_trace") == 0) {
		      DT_FILE=fopen(argv[2],"w");
      return (TCL_OK);
	}

  return (Agent::command(argc, argv));
}

void ALMAH_Agent::recv(Packet* pkt, Handler* h)
{
  //printf("now is %f\n", now);
  hdr_ip *ih = hdr_ip::access(pkt);
  hdr_almah *ol =  hdr_almah::access(pkt);
  hdr_cmn *ch = hdr_cmn::access(pkt);

  ch->size() -= IP_HDR_LEN;	


    bool is_neighbor = false;
    //i verify if it is a neighbor: if it is i block the tree create packets
    OBAMP_Overlay_Neighbor *nb;
    OBAMP_Overlay_Neighbor *nbn;
    nb = nbhead.lh_first;
    if(nb!=0)
      for(; nb; nb = nbn) {
        nbn = nb->nb_link.le_next;
        if(nb->nb_addr == ih->saddr()){
           is_neighbor = true;
			break;	
        }
      }
  
  
 /*
  * Incoming Packets.
 */
 if(this->is_member)
    switch(ol->Message_ID) {
  
    case 1:
      recvJOIN_REQ(pkt, h);
      break;            

    case 2:
		recvJOIN_ACK(pkt, h);
		break; 
   
    case 7:
      recvDATA(pkt);	// Andrea DATA
      break;

    case 8:
      pong_(pkt, h);  
      break;
    
    case 10:
	    recvJOINACKNotify(pkt,h);
	    break;
     
    case 12:      
          recv_NACK(pkt, h);
      break;   
	
	case 14:      
          recvParentIn(pkt, h);
      break;

	case 15:      
          recvRequestTopology(pkt, h);
      break; 

	case 16:      
          recvTopology(pkt, h);
      break;

	case 17:      
          recvRequestSwitch(pkt, h);
      break; 

	case 18:      
          recvAckSwitch(pkt, h);
      break; 

	case 20:      
          recvNACK_ACK(pkt, h);
      break;
	 
	case 21:      
          recvHello(pkt, h);
      break; 

	case 22:      
          recvHelloReply(pkt, h);
      break; 
    
    case 23:
    	  recvJOINACKConf(pkt);
    	  break;
	case 24:      
          recvLOOP_RESET(pkt);
      break; 
	case 25:      
          recvLOOP_RESET_ACK(pkt,h);
      break; 
   
    default:
      fprintf(stderr, "Invalid OVERLAY type (%x)\n", ol->Message_ID);
      exit(1);
    }
   else
    if(ih->dst_.addr_ == IP_BROADCAST)
   {
      this->recv_Broadcast(pkt, h);
   }else{
      drop(pkt,"CLAUDIO");
   }
 
 
}

void ALMAH_Agent::sendPING(int node, double delay){

	  Packet* pkt = allocpkt();     
      hdr_almah* hdrret = hdr_almah::access(pkt);
      hdr_cmn* ch = hdr_cmn::access(pkt);
      hdr_ip* ipret = hdr_ip::access(pkt);
      ch->size() = 8+8;
      hdrret->src_IP = here_.addr_;
      ipret->daddr() = node;
      ipret->dport() = ipret->sport();      
      hdrret->ret = 0;
      hdrret->send_time = Scheduler::instance().clock()+delay;
      hdrret->Message_ID = 8;
      send(pkt, 0);
	  return;
}


void ALMAH_Agent::sendDATA(){	

      Packet* pkt = allocpkt();
      hdr_almah* hdr = hdr_almah::access(pkt);
      hdr_cmn* ch = HDR_CMN(pkt);
      hdr_ip* ip = hdr_ip::access(pkt);
	  hdr->Message_ID = 7;
      hdr->src_IP = here_.addr_;
	  hdr->Seq_Number=dsn++;
	  hdr->send_time=Scheduler::instance().clock();
	  ip->dport()=ip->sport();
      ch->size() = size_+8;
      forwardData(pkt);
	  return;
}

void ALMAH_Agent::send_NACK(int node){
		   		
        Packet* pkt = allocpkt();
        hdr_almah* hdrret = hdr_almah::access(pkt);
        hdr_cmn* ch = HDR_CMN(pkt);
        hdr_ip* ipret = hdr_ip::access(pkt);
        hdrret->Message_ID = 12;
        hdrret->src_IP = here_.addr_;
        ipret->daddr() = node;
        ipret->dport() = ipret->sport();
		ch->size_=8+8;
		Scheduler::instance().schedule(&ttimer_2, pkt, 0);
    return;
}

void ALMAH_Agent::sendNACK_ACK(int node){

        Packet* pkt = allocpkt();
        hdr_almah* hdrret = hdr_almah::access(pkt);
        hdr_cmn* ch = HDR_CMN(pkt);
        hdr_ip* ipret = hdr_ip::access(pkt);
        hdrret->Message_ID = 20;//identificativo del tipo di messaggio
        hdrret->src_IP = here_.addr_;//imposto IP sorgente del TREE CREATE NACK
        ipret->daddr() = node;
        ipret->dport() = ipret->sport();
		ch->size_=8+8;
        send(pkt, 0);
    return;
}

void ALMAH_Agent::recvNACK_ACK(Packet* pkt, Handler*){

		hdr_ip* ipret = hdr_ip::access(pkt);
		int node;
		node = ipret->saddr();
		nb_delete(node);
		NACKING = false;
		parent_in_switch = -1;
        Packet::free(pkt);
    return;
}


void ALMAH_Agent::recv_Broadcast(Packet* pkt, Handler*){
    forwardBroadcast(pkt);
    return;
}

void ALMAH_Agent::pong_(Packet* pkt, Handler*){
  
  hdr_ip* hdrip = hdr_ip::access(pkt);
  hdr_almah* hdr = hdr_almah::access(pkt);
  int src_ip = hdrip->src().addr_;
  double now=Scheduler::instance().clock();
  if (hdr->ret == 0) {
        double stime = hdr->send_time;   
        Packet::free(pkt);
        Packet* pktret = allocpkt();
        hdr_almah* hdr = hdr_almah::access(pktret);
        hdr_cmn* ch = HDR_CMN(pktret);
        hdr_ip* ipret = hdr_ip::access(pktret);
        hdr->Message_ID = 8;
        ch->size() = 8+8;
        hdr->src_IP = here_.addr_;
        ipret->daddr() = src_ip;
        ipret->dport() = ipret->sport();
        hdr->ret = 1;
        hdr->send_time = stime;
        send(pktret, 0);
   } else {
   		OBAMP_Overlay_Neighbor *mb;
		mb = this->mb_lookup(src_ip);
		if(mb !=NULL){
			mb->distance = NETWORK_DIAMETER - hdrip->ttl()+1;
			mb->nb_expire = now + 2;
			mb->last_receiving_time = now-hdr->send_time;	// Note that the variable last_receiving_time is used to store the RTT
		}
        Packet::free(pkt);
    }
    return;
} 

void ALMAH_Agent::sendHello(int node){

        Packet* pkt = allocpkt();
        hdr_almah* hdrret = hdr_almah::access(pkt);
        hdr_cmn* ch = HDR_CMN(pkt);
        hdr_ip* ipret = hdr_ip::access(pkt);
        hdrret->Message_ID = 21;
        hdrret->src_IP = here_.addr_;
        hdrret->send_time=Scheduler::instance().clock();
        ipret->daddr() = node;
        ipret->dport() = ipret->sport();
		ch->size_=8+8;
        send(pkt, 0);
		return;
}

void ALMAH_Agent::sendHelloReply(int node, double send_time){

        Packet* pkt = allocpkt();
        hdr_almah* hdrret = hdr_almah::access(pkt);
        hdr_cmn* ch = HDR_CMN(pkt);
        hdr_ip* ipret = hdr_ip::access(pkt);
        hdrret->Message_ID = 22;
        hdrret->src_IP = here_.addr_;
        hdrret->send_time=send_time;
        ipret->daddr() = node;
        ipret->dport() = ipret->sport();
		ch->size_=8+8;
		if(this->path_list!=NULL)
			hdrret->PathList = new lista(this->path_list);
		else hdrret->PathList = NULL;
			//hdrret->PathList = new lista();
        send(pkt, 0);
		return;
}

void ALMAH_Agent::recvHello(Packet* pkt, Handler*){

    double now = CURRENT_TIME;
    hdr_ip* hdrip = hdr_ip::access(pkt);
    hdr_almah* hdrret = hdr_almah::access(pkt);
	OBAMP_Overlay_Neighbor* nb;
	nb=nb_lookup(hdrip->saddr());	
	if(nb!=NULL){
		nb->nb_expire = now + 1.5*2*TREE_INT;
	}else{
		nb_insert(hdrip->saddr());
		nb=nb_lookup(hdrip->saddr());		
	}
	this->sendHelloReply(hdrip->saddr(),hdrret->send_time);	
    Packet::free(pkt);
    return;
}

void ALMAH_Agent::recvHelloReply(Packet* pkt, Handler*){

    hdr_ip* hdrip = hdr_ip::access(pkt);
    hdr_almah* hdr = hdr_almah::access(pkt);
	OBAMP_Overlay_Neighbor* nb;	
	nb=nb_lookup(hdrip->saddr());		
	if(parent_in == hdrip->saddr()){
			if(path_list!= NULL)
				delete path_list;
			if(hdr->PathList!=NULL)
				path_list = new lista(hdr->PathList);
			else
				path_list = new lista();
			path_list->inserisci(index, parent_in);
			
			if(path_list->trova_ordine(index,2) != NULL)
				if(parent_in != -1){
					NACKING = true;
					send_NACK(parent_in);					
					parent_in = -1;
					parent_in_switch = -1;
					SWITCHING = false;
					ON_PingTimer = 0;
					htimer.count = 0;
					if(!htimer.is_busy)
						Scheduler::instance().schedule(&htimer,(new Event),0);
					printf("loop detect at node :   %d\n", index);
					sendLOOP_RESET();
					Packet::free(pkt);
					return;
				}
			parent_in_distance = NETWORK_DIAMETER - hdrip->ttl()+1;
			parent_in_rtt = Scheduler::instance().clock()-hdr->send_time;
			if(index!=0)
				this->RefreshParent();
	}
    Packet::free(pkt);
    return;
}

void ALMAH_Agent::recv_NACK(Packet* pkt, Handler*){
    hdr_ip* hdrip = hdr_ip::access(pkt);     
    nb_delete(hdrip->src_.addr_);
    sendNACK_ACK(hdrip->src_.addr_);
    Packet::free(pkt);
    return;
}

void ALMAH_Agent::recvDATA(Packet* pkt){	// Andrea Data

    hdr_cmn* ch = hdr_cmn::access(pkt);
	hdr_ip* ip = hdr_ip::access(pkt);
	hdr_almah* ov = hdr_almah::access(pkt);
	int id_ = ch->uid();
	if(d_lookup(id_) == NULL){
		drc++;
		mdelay=1.0*(mdelay*(drc-1)+(Scheduler::instance().clock()-ov->send_time))/drc;		
		if (DT_FILE!=NULL) fprintf(DT_FILE,"%f\t%f\t%d\t%d\t%d\t%u\t%d\n",(float)Scheduler::instance().clock(),(float)(Scheduler::instance().clock()-ov->send_time),(int)ip->saddr(),(int)ov->Seq_Number,(int)(NETWORK_DIAMETER-ip->ttl()+1),(unsigned int)ch->uid_,(int)ov->src_IP);  
	}
	forwardData(pkt);
}
void ALMAH_Agent::forwardBroadcast(Packet* pkt){

        hdr_cmn* ch = HDR_CMN(pkt);
        hdr_ip* ip = hdr_ip::access(pkt);
        int id_ = ch->uid();
        int src_ = ip->saddr();
        OBAMP_Overlay_Broadcast_Cache *b;
		ch->direction() = hdr_cmn::DOWN;
		b_purge();      	
		b = this->b_lookup(id_); 
        if(b == NULL){
		   if (src_!=index) ch->size() += IP_HDR_LEN;
		   this->b_insert(src_,ip->ttl_,id_);
		   target_->recv(pkt);		   
           return ;
        } else {
              if(b->TTL < ip->ttl_){
              this->b_delete(id_);
              this->b_insert(src_,ip->ttl_,id_);
			  if (src_!=index) ch->size() += IP_HDR_LEN;
			  target_->recv(pkt);	
              return ;
            } else {
              drop(pkt, "Broadcast Packet dropped - just in cache");
              return ;
            }
        } 
         
}

void ALMAH_Agent::forwardData(Packet* pkt){