rsvp.cc

rsvp and wfq patch for Netowrk Simulator 2

	  }
	  if ((rsearch2->p->phop->get_hop() == phop) &&
	      (rsearch2->sender->get_addr() == sender)) {
	    if (rsearch2->fspec->get_rate() > lub_rate) {
	      lub_rate = rsearch2->fspec->get_rate();
	    }
	    if (rsearch2->fspec->get_size() > lub_size) {
	      lub_size = rsearch2->fspec->get_size();
	    }
	    rsearch2->processed = 1;
	  }
	  rsearch2 = rsearch2->next;
	}
	t = s->tcsb_list;
	while ((t != NULL) && ((t->sender != sender) || 
			       (t->oifhop != rsearch->oifhop))) {
	  t = t->next;
	}
	if (t == NULL) {
	  t = new tcsb();
	  t->fl = new FLOWSPEC(0, 0);
	  t->ofl = new FLOWSPEC(0, 0);
	  t->oifhop = rsearch->oifhop;
	  t->sender = sender;
	  t->next = s->tcsb_list;
	  s->tcsb_list = t;
	}
	if ((lub_rate != t->ofl->get_rate()) || 
	    (lub_size != t->ofl->get_size()) ||
	    forced) {
	  msg->add_object(new FLOWSPEC(lub_rate, lub_size));
	  msg->add_object(new FILTER_SPEC(sender, ip6_));
	  snd = 1;
	  delete t->ofl;
	  t->ofl = new FLOWSPEC(lub_rate, lub_size);
	}
      }
      rsearch = rsearch->next;
    }
    if (snd) {
      if (ip6_) {
	size_ = msg->get_length() + 40; // Size for standard IPv6 header
      } else {
	size_ = msg->get_length() + 24;
      }
      // dst_ = r->p->phop->get_hop() << Address::instance().NodeShift_[1];
      dst_.addr_ = r->p->phop->get_hop() ;
      fid_ = 46;
      Packet *pkt = allocpkt();
      pkt->allocdata(msg->get_conlength());
      memcpy(pkt->accessdata(), msg->get_contents(), msg->get_conlength());
      send(pkt, 0);
      delete msg;
    }
  }
  return snd;
}


/* Create and send a resv tear message for a list of RSBs in a session */
void RSVPAgent::send_resv_tear_messages(session *s, rsb *r_list) {
  if (nam_) {
    type_ = PT_RESV_TEAR;
  }
  rsb *rsearch = r_list;
  rsb *rsearch2, *rsearch3;
  while (rsearch != NULL) {
    if ((rsearch->p->phop->get_hop() != -1) && !rsearch->processed) {
      RSVPmessage *msg = new RSVPmessage(RESVTEAR, 32);
      msg->add_object(new SESSION(s->s->get_contents()));
      // msg->add_object(new RSVP_HOP(addr_ >> Address::instance().NodeShift_[1],
      msg->add_object(new RSVP_HOP(here_.addr_ , 
				   ip6_));
      msg->add_object(new STYLE(rsearch->st->get_contents()));
      nsaddr_t phop = rsearch->p->phop->get_hop();
      nsaddr_t sender;
      rsearch2 = rsearch;
      while (rsearch2 != NULL) {
	sender = rsearch2->sender->get_addr();
	if (!rsearch2->processed && (rsearch2->p->phop->get_hop() == phop)) 
	  {
	    sender = rsearch2->sender->get_addr();
	    rsearch3 = rsearch;
	    while (rsearch3 != NULL) {
	      if ((rsearch3->p->phop->get_hop() == phop) &&
		  (rsearch3->sender->get_addr() == sender)) {
		/* The 'processed' flags will not have to be deleted from
		   the RSBs later, since the r_list will always be deleted
		   after being processed by this function. */
		rsearch3->processed = 1;
	      }
	      rsearch3 = rsearch3->next;
	  }
	  msg->add_object(new FILTER_SPEC(sender, ip6_));
	}
	rsearch2 = rsearch2->next;
      }
      if (ip6_) {
	size_ = msg->get_length() + 40; // Size for standard IPv6 header
      } else {
	size_ = msg->get_length() + 24;
      }
      // dst_ = rsearch->p->phop->get_hop() << Address::instance().NodeShift_[1];
      dst_.addr_ = rsearch->p->phop->get_hop() ;
      fid_ = 46;
      Packet *pkt = allocpkt();
      // hdr_cmn* hdr = (hdr_cmn*)pkt->access(off_cmn_); - unused
      pkt->allocdata(msg->get_conlength());
      memcpy(pkt->accessdata(), msg->get_contents(), msg->get_conlength());
      send(pkt, 0);
      delete msg;
    }
    rsearch = rsearch->next;
  }
}


/* Update an existing PSB with new values. */
void RSVPAgent::update_psb(session *s, psb *p, double rate, long bucket, 
			   char ttl, nsaddr_t sender, nsaddr_t phop, 
			   double refresh, int iface, RSVPChecker *check) {
  /* The following is slightly different from what the RSVP Processing
     Rules suggest for updated PSBs */
  char diff = ((p->sender->get_addr() != sender) || 
	       (p->s_tspec->get_rate() != rate) ||
	       (p->s_tspec->get_size() != bucket) ||
	       (p->ttl != ttl));
  clear_psb(p);
  p->sender = new SENDER_TEMPLATE(sender, ip6_);
  p->s_tspec = new SENDER_TSPEC(rate, bucket);
  p->phop = new RSVP_HOP(phop, ip6_);
  p->ttl = ttl;
  p->iif = iface;
  p->check = check;
  if (phop > -1) { // Doesn't make sense when call came from API
    p->timeout = (lifetime_factor_ + 0.5) * 1.5 * refresh + 
      Scheduler::instance().clock();
  }
  /* Send a path message for this PSB to inform downstream nodes
     of the updated path state. */
  if ((diff) || 
      ((is_mcast(s->s->get_dest())) && 
	(is_leaf(p->sender->get_addr(), s->s->get_dest())))) {
    send_path_message(s, p);
    s->path_ref = Scheduler::instance().clock() + s->path_tv->get_r();
    reschedule_session(s);
    /* Make an upcall to the API if necessary */
    if ((noisy_ & UPC_PATH) && (diff)) {
      Tcl& tcl = Tcl::instance();
      tcl.evalf("%s upcall-path %d %f %d %d", name(),  s->sid, rate, 
		bucket, sender);
    }
  }
}


/* Update an existing RSB with new values. */
void RSVPAgent::update_rsb(session *s, rsb *r, psb *p, nsaddr_t sender, 
			   FLOWSPEC *fl, const char *style, nsaddr_t nhop, 
			   double refresh, nsaddr_t fromhop) {
  long nstyle;
  if (strcasecmp(style, "FF") == 0) {
    nstyle = STYLE_FF;
  }
  r->modified = ((r->fspec->get_rate() != fl->get_rate()) ||
		 (r->fspec->get_size() != fl->get_size()) ||
		 (r->st->get_style() != nstyle) ||
		 (r->p != p) ||
		 (r->oifhop != fromhop));
  if (nhop > -1) { // Doesn't make sense when call came from API
    r->timeout = (lifetime_factor_ + 0.5) * 1.5 * refresh + 
      Scheduler::instance().clock();
  }
  /* Store the old values, so in case the reservation is rejected by
     admission control, it can be restored. */
  r->old = new rsb();
  r->old->nhop = r->nhop;
  r->old->sender = r->sender;
  r->old->fspec = r->fspec;
  r->old->st = r->st;
  r->old->oifhop = r->oifhop;
  r->old->p = r->p;
  if (r->modified) {
    r->nhop = new RSVP_HOP(nhop, ip6_);
    r->sender = new FILTER_SPEC(sender, ip6_);
    r->fspec = new FLOWSPEC(fl->get_contents());
    r->st = new STYLE(nstyle);
    r->oifhop = fromhop;
    r->p = p;
  }
}



/* The following function has not been tested thoroughly for
   measurement based admission control algorithms. It will probably be
   replaced sooner or later by algorithm-specific functions. */
int modify_flow(ADC *adc, int cl, double r_old, int b_old, double r, int b) {
  if (adc->peak_rate(cl, r, b) <= adc->peak_rate(cl, r_old, b_old)) {
    /* The new reservation is smaller than the old one =>
       notify admission control of the change and return 1 */
    adc->teardown_action(cl, r_old, b_old);
    adc->admit_flow(cl, r, b);
    return 1;
  } else {
    /* Check if the new flow can be admitted. */
    adc->teardown_action(cl, r_old, b_old);
    if (adc->admit_flow(cl, r, b)) {
      /* Yes => return 1 */
      return 1;
    } else {
      /* No => re-install the old flow */
      adc->admit_flow(cl, r_old, b_old);
      return 0;
    }
  }
}



/* Find all RSBs that match the (session, sender, nhop) triple for the
   'active' RSB, compute the LUB of their flowspecs then query admission 
   control and perform a reservation if possible. This function only works 
   for the FF style. */
int RSVPAgent::update_traffic_control(session *s, rsb *r) {
  rsb *rsearch = s->rsb_list;
  nsaddr_t sender = r->sender->get_addr();
  nsaddr_t nhop = r->nhop->get_hop();
  nsaddr_t oifhop = r->oifhop;
  double lub_rate = 0;
  long lub_size = 0;
  tcsb *t = s->tcsb_list;
  tcsb *ts;
  ADC *adc;
  char admitted;
  char refresh_needed = 0;
  char is_biggest = 1;
  Tcl& tcl = Tcl::instance();
  if (nhop != -1) {
    /* First find the matching TCSB for the (session, sender, oif) triple */
    while ((t != NULL) && ((t->sender != sender) || (t->oifhop != oifhop))) {
      t = t->next;
    }
    /* Now compute the LUB for the flowspecs from all matching RSBs. */
    while (rsearch != NULL) {
      if ((rsearch->sender->get_addr() == sender) && 
	  (rsearch->oifhop == oifhop)) {
	if ((r != rsearch) &&
	    (rsearch->fspec->get_rate() >= r->fspec->get_rate()) &&
	    (rsearch->fspec->get_size() >= r->fspec->get_size())) {
	  is_biggest = 0;
	}
	if (rsearch->fspec->get_rate() > lub_rate) {
	  lub_rate = rsearch->fspec->get_rate();
	}
	if (rsearch->fspec->get_size() > lub_size) {
	  lub_size = rsearch->fspec->get_size();
	}
      }
      rsearch = rsearch->next;
    }
    /* Check first if the LUB flowspec is zero and a tcspec exists.
       In that case, remove the existing reservation, because this
       can only happen when an RSB was removed (e.g. when a teardown
       message arrived). */
    if ((t != NULL) && (lub_rate == 0)) {
      /* Notify admission control of the change. */
      tcl.evalf("%s get-adc %d %d", name(),
		// addr_ >> Address::instance().NodeShift_[1], r->oifhop);
		here_.addr_ , r->oifhop);
      adc = (ADC *)TclObject::lookup(tcl.result());
      modify_flow(adc, 1, t->fl->get_rate(), t->fl->get_size(), 0, 0);
      tcl.evalf("[Simulator instance] delete-filter-fid %d %d %d %d",
		// addr_ >> Address::instance().NodeShift_[1], r->oifhop,
		here_.addr_ , r->oifhop,
		sender, s->s->get_fid());
      num_flows_--;
      tcl.evalf("[Simulator instance] delete-reservation %d %d %s",
		// addr_ >> Address::instance().NodeShift_[1], r->oifhop,
		here_.addr_ , r->oifhop,
		t->handle->name());
      /* Delete t */
      if (t == s->tcsb_list) {
	s->tcsb_list = t->next;
      } else {
	ts = s->tcsb_list;
	while (ts->next != t) {
	  ts = ts->next;
	}
	ts->next = t->next;
      }
      delete t->fl;
      delete t->ofl;
      delete t;
    } else {
      /* Now query admission control for lub_rate and lub_size. */
      tcl.evalf("%s get-adc %d %d", name(),
		// addr_ >> Address::instance().NodeShift_[1], r->oifhop);
		here_.addr_ , r->oifhop);
      adc = (ADC *)TclObject::lookup(tcl.result());
      if (t == NULL) { // New flow
	admitted = adc->admit_flow(1, lub_rate, lub_size);
      } else { // Modification of existing flow
	admitted = modify_flow(adc, 1, t->fl->get_rate(),
				    t->fl->get_size(), lub_rate, lub_size);
      }
      if (!admitted) {
	send_resv_err_message(s, r);
	if (r->confirm != NULL) {
	  delete r->confirm;
	  r->confirm = NULL;
	}
	/* Delete the active rsb if it is new, or return it to the old state
	   if it was modified. */
	if (r->is_new) {
	  if (s->rsb_list == r) {
	    s->rsb_list = r->next;
	    delete r;
	  } else {
	    rsearch = s->rsb_list;
	    while ((rsearch != NULL) && (rsearch->next != r)) {
	      rsearch = rsearch->next;
	    }
	    rsearch->next = r->next;
	    clear_rsb(r);
	    delete r;
	    num_rsb_--;
	  }
	} else {
	  clear_rsb(r);
	  r->nhop = r->old->nhop;
	  r->sender = r->old->sender;
	  r->fspec = r->old->fspec;
	  r->st = r->old->st;
	  r->p = r->old->p;
	  r->oifhop = r->old->oifhop;
	  delete r->old;
	  r->old = NULL;
	}
      } else {
	/* Delete all the pointers to the former RSB values in the
	   active RSB if it was modified. They are no longer needed. */
	if (!r->is_new) {
	  delete r->old;
	  r->old = NULL;
	}
	/* Does the RSB have a RESV_CONFIRM object and was it merged into
	   an existing reservation, or is this the leaf node for this RSB? 
	   If so, send a ResvConf message and remove the RESV_CONFIRM 
	   object. */
	if ((r->confirm != NULL) && 
	    (!is_biggest || (r->p->phop->get_hop() == -1))) {
	  send_resv_conf_message(s, r);
	  delete r->confirm;
	  r->confirm = NULL;
	}
	/* Perform the actual reservation. If there is no TCSB yet, 
	   create one and get a new handle to a WFQClass from Tcl. Otherwise
	   modify the reservation with the existing handle. */
	if (t == NULL) {
	  t = new tcsb();
	  t->fl = new FLOWSPEC(lub_rate, lub_size);
	  t->ofl = new FLOWSPEC(0, 0);
	  t->oifhop = oifhop;
	  t->sender = sender;
	  t->next = s->tcsb_list;
	  s->tcsb_list = t;
	  tcl.evalf("[Simulator instance] add-reservation %d %d %f %d",
		    // addr_ >> Address::instance().NodeShift_[1],
		    here_.addr_ ,
		    r->oifhop, lub_rate, lub_size);
	  num_flows_++;
	  t->handle = TclObject::lookup(tcl.result());
	  tcl.evalf("[Simulator instance] add-filter-fid %d %d %s %d %d",
		    // addr_ >> Address::instance().NodeShift_[1], r->oifhop,
		    here_.addr_ , r->oifhop,
		    t->handle->name(), sender, s->s->get_fid());
	  refresh_needed = 1;
	} else {
	  if ((t->fl->get_rate() != lub_rate) || (t->fl->get_size() != lub_size)) {
	    delete t->fl;
	    t->fl = new FLOWSPEC(lub_rate, lub_size);
	    tcl.evalf("[Simulator instance] modify-reservation %d %d %s %f %d",
		      // addr_ >> Address::instance().NodeShift_[1],
		      here_.addr_ ,
		      r->oifhop, t->handle->name(), lub_rate, lub_size);
	    refresh_needed = 1;
	  }
	}
      }
    }
  }
  return refresh_needed;
}