sendmanager.cc

COPE the first practical network coding scheme which is developped on click

#include <click/config.h>
#include <click/confparse.hh>
#include <click/error.hh>
#include <click/glue.hh>

#include <click/etheraddress.hh>
#include <click/straccum.hh>

#include "sendmanager.hh"

int
SendManager::do_events(bool find_rexmits)
{
  // look up _timer for upper_bound with current time (first element with eventtime > current time i.e. all elements lesser are good)

  // work backwards from this to begin, and execute all the events

  // if event is a rexmit event
  //       : look up the packet in _sentpacks
  //       : if packet is not acked
  //             queue up the packet in _ptorexmit
  //             in

  // if event is a timeout
  //       : check if packet is acked or has reached maxxmits
  //              if so, kill the packet from _sentpacks
  //              if not, set _timedout to 1 in PacketState

  // delete the event from _timer

  _rexmit_p1 = _rexmit_p2 = 0;

  int32_t now = Timestamp::now().msec1();

  // click_chatter("%s: do_events() called with find_rexmits %d at %u", id().cc(), find_rexmits, now);

  int still_looking = 2;
  uint32_t prev_hop, next_hop;
  prev_hop = next_hop = 0;

  int nfound = 0;

  EventMap::iterator bound = _eventmap.upper_bound(now);
  EventMap::iterator i = _eventmap.begin();

  while (i != bound) {

    EventList *el = (*i).second;

    EventList::iterator prev = el->begin();
    EventList::iterator next = prev;

    while (next != el->end()) {
      const Event &e = *next;
      bool acted_on = false;
      
      StringAccum sa;
      sa << ": do_events() found event of type " << (int) e._event_type << " with EA " << e._pid._dst << ", nseq " << e._pid._seq << ", still_looking is " << still_looking << ", nfound is " << nfound;
      click_chatter("%s: %s", id().cc(), sa.c_str());
      
      if ((e._event_type == EVENT_REXMIT) && (find_rexmits)) {
	if (still_looking == 2) {
	  acted_on = true;
	  PacketGID *pgid = _ngm.findp(e._pid);
	  PacketState *ps = _sentpacks.find(*pgid);
	  // this packet is not acked, if it was, the corresponding timer event would have been removed
	  _rexmit_p1 = ps->_p;
	  nfound = 1;
	  
	  // check if rexmitp1 is xor'able, if so, still_looking is 1, else 0
	  if (xor_worthy(_rexmit_p1, _myip, prev_hop, next_hop)) {
	    still_looking = 1;
	  } else {
	    still_looking = 0;
	  }

	  if (!_enable_coding) {
	    still_looking = 0;
	  }

	  click_chatter("%s: do_events() found packet 1 to rexmit, still_looking is %d", id().cc(), still_looking);

	} else if (still_looking == 1) {
	  PacketGID *pgid = _ngm.findp(e._pid);
	  PacketState *ps = _sentpacks.find(*pgid);
	  if (xor_candidate(ps->_p, _myip, prev_hop, next_hop)) {
	    acted_on = true;
	    _rexmit_p2 = ps->_p;
	    nfound = 2;
	    still_looking = 0;
	    click_chatter("%s: do_events() found packet 2 to rexmit, still_looking is %d", id().cc(), still_looking);
	  }
	}
      } else if (e._event_type == EVENT_EXPIRE) {
	acted_on = true;
	PacketGID *pgid = _ngm.findp(e._pid);
	PacketState *ps = _sentpacks.find(*pgid);
	if (ps->no_more_xmits(_max_xmits)) {
	  // this packet should be deleted
	  ps->_p->kill();
	  _sentpacks.remove(*pgid);
	  _ngm.remove(e._pid);
	  click_chatter("%s: do_events() killing timed out packet", id().cc());
	} else {
	  ps->_timedout = true;
	  click_chatter("%s: do_events() setting timedout to 1", id().cc());
	}
      }
      
      if (acted_on) {
	if (prev == next) {
	  el->pop_front();
	  prev = next = el->begin();
	} else {
	  ++next;
	  el->erase_after(prev);
	}
	click_chatter("%s: do_events() acted on event", id().cc());
      } else {
	prev = next++;
	click_chatter("%s: do_events() did not act on event", id().cc());
      }
    }

    if (el->empty()) {
      click_chatter("%s: do_events() deleting eventmap entry", id().cc());
      // delete the whole entry
      EventMap::iterator j = i++;
      _eventmap.erase(j);
    } else {
      ++i;
    }
  }

  return nfound;
}

Packet *
SendManager::take_packet(const PacketGID &pgid)
{
  PacketState *ps = _sentpacks.find(pgid);
  click_chatter("%s: take_packet() called with IP src %u, ipid %u", id().cc(), pgid._src.addr(), pgid._id);
  if (!ps) {
    click_chatter("%s: take_packet() did not find packet, returning null", id().cc());
    return 0;
  }

  click_chatter("%s: take_packet() returning packet", id().cc());
  return ps->_p->clone();
}

// you need to do 3 things
// add this packet to _sentpacks with _acked=0, _nxmits=1, _rexmit_time=now+_rtx_timeout, _timedout=0, _p=p
// add this packet to _eventmap with _event_type = EVENT_REXMIT with _rexmit_time
// add this packet to _eventmap with _event_type = EVENT_EXPIRE with now+_exp_timeout

void 
SendManager::add_new_packet(Packet *p, int now)
{
  PacketID pid = PacketID(p);
  PacketGID pgid = PacketGID(p);
  StringAccum sa;
  sa << "add_new_packet() called with PID (" << pid._dst << ", " << pid._seq << "), PGID ( " << pgid._src << ", " <<  pgid._id << ") ipid at";
  click_chatter("%s: %s %u", id().cc(), sa.c_str(), now);

  _ngm.insert(pid, pgid);

  PacketState *ps = new PacketState(p->clone(), now + _rtx_timeout);
  _sentpacks.insert(pgid, ps);
  
  // handle the case when _max_xmits == 1
  if (ps->_nxmits < _max_xmits) {
    Event e(EVENT_REXMIT, pid);
    EventMap::iterator i = _eventmap.find(ps->_rexmit_time);
    if (i == _eventmap.end()) {
      EventList *el = new EventList;
      el->push_front(e);
      _eventmap[ps->_rexmit_time] = el;
      click_chatter("%s: add_new_packet() created new EventList entry for rexmit", id().cc());
    } else {
      EventList *el = (*i).second;
      el->push_front(e);
      click_chatter("%s: add_new_packet() added event to existing EventList for rexmit", id().cc());
    }
  }

  Event e(EVENT_EXPIRE, pid);
  int32_t expire_time = now + _exp_timeout;
  EventMap::iterator i = _eventmap.find(expire_time);
  if (i == _eventmap.end()) {
    EventList *el = new EventList;
    el->push_front(e);
    _eventmap[expire_time] = el;
    click_chatter("%s: add_new_packet() created new EventList entry for timeout", id().cc());
  } else {
    EventList *el = (*i).second;
    el->push_front(e);
    click_chatter("%s: add_new_packet() added event to existing EventList for timeout", id().cc());
  }

  click_chatter("%s: add_new_packet returned", id().cc());

}

void 
SendManager::add_rexmitted_packet(Packet *p, int now)
{
  PacketGID pgid = PacketGID(p);
  PacketID pid = PacketID(p);

  StringAccum sa;
  sa << "add_rexmitted_packet() called with PID (" << pid._dst << ", " << pid._seq << "), PGID ( " << pgid._src << ", " <<  pgid._id << ") ipid at ";
  click_chatter("%s: %s %u", id().cc(), sa.c_str(), now);

  PacketState *ps = _sentpacks.find(pgid);

  ps->_nxmits++;

  if (ps->_nxmits == _max_xmits) {
    click_chatter("%s: nxmits reached max_xmits", id().cc());
    if (ps->_timedout) {
      click_chatter("%s: and packet timedout, removing", id().cc());
      p->kill();
      _sentpacks.remove(pgid);
      _ngm.remove(pid);
    }
  } else {
    ps->_rexmit_time = now + _rtx_timeout;
    Event e(EVENT_REXMIT, pid);
    click_chatter("%s: adding rexmit event, nxmits is %u", id().cc(), ps->_nxmits);
    EventMap::iterator i = _eventmap.find(ps->_rexmit_time);
    if (i == _eventmap.end()) {
      EventList *el = new EventList;
      el->push_front(e);
      _eventmap[ps->_rexmit_time] = el;
    } else {
      EventList *el = (*i).second;
      el->push_front(e);
    }
  }
}

// note that this function might be called multiple times
// for the same packet, so make sure that the packet is actually 
// there before updating it

void
SendManager::register_ack(const PacketID &pid)
{

  StringAccum sa;
  sa << "register_ack() register_ack() called with PID (" << pid._dst << ", " << pid._seq << ")";
  click_chatter("%s: %s", id().cc(), sa.c_str());

  _nacksrecd++;

  PacketGID *pgid = _ngm.findp(pid);

  if (!pgid) {
    // it might be a dupack for a packet that has timed out
    // in that case, you won't find it in ngm
    click_chatter("%s: register_ack() cannot find ngm state for this packet", id().cc());
    return;
  }

  sa.clear();
  sa << "register_ack() Matched packet gid IP src " << pgid->_src << " for ipid " << pgid->_id;
  click_chatter("%s: %s", id().cc(), sa.c_str());

  // if it is there in ngm, it should be there in _sentpacks
  PacketState *ps = _sentpacks.find(*pgid);

  // if the packet doesn't have any more rexmits lined up for whatever
  // reason, then move on. there's nothing left to do
  if (ps->no_more_xmits(_max_xmits)) {
    click_chatter("%s: register_ack() has nothing more to do, no more xmits", id().cc());
    return;
  }

  // at this point, more rexmit events might be possible
  // we should register the ack, and also cancel the pending rexmit event

  ps->_acked = 1;
  sa.clear();
  // sa << "register_ack() Found packet state for gid ";
  // click_chatter("%s: %s", id().cc(), sa.c_str());
  
  // delete the timer entry
  EventMap::iterator i = _eventmap.find(ps->_rexmit_time);
    
  EventList *el = (*i).second;
  EventList::iterator prev = el->begin();
  EventList::iterator next = prev;
  
  while (next != el->end()) {
    const Event &e = *next;
    if ((e._event_type == EVENT_REXMIT) && (e._pid == pid)) {
      sa.clear();
      sa << "register_ack() Deleting retransmit event ";
      click_chatter("%s: %s scheduled for %u", id().cc(), sa.c_str(), ps->_rexmit_time);
      if (prev == next) {
	el->pop_front();
      } else {
	el->erase_after(prev);
      }
      break;
    }
    prev = next;
    ++next;
  }

  if (el->empty()) {
    click_chatter("%s: register_ack() all events for this time deleted, deleting eventmap entry", id().cc());
    _eventmap.erase(i);
  }
  
  // if timed out, delete the packet completely
  if (ps->_timedout) {
    ps->_p->kill();
    _sentpacks.remove(*pgid);
    _ngm.remove(pid);
    click_chatter("%s register_ack() packet timed out deleting from sentpacks", id().cc());
  }
  click_chatter("%s register_acks finished registering", id().cc());
}

uint32_t
SendManager::get_next_rexmit_time()
{
  for (EventMap::const_iterator i = _eventmap.begin(); i != _eventmap.end(); i++) {
    int32_t now = (*i).first;
    const EventList *el = (*i).second;
    for (EventList::const_iterator j = el->begin(); j != el->end(); j++) {
      const Event &e = *j;
      if (e._event_type != EVENT_REXMIT) {
	continue;
      }
      const PacketGID &pgid = _ngm.find(e._pid);
      const PacketState *ps = _sentpacks.find(pgid);
      if (!ps->no_more_xmits(_max_xmits)) {

	StringAccum sa;
	sa << ": get_next_rexmit_time() found event of type " << (int) e._event_type << " with EA " << e._pid._dst << ", nseq " << e._pid._seq << ", IP " << pgid._src << ", ipid "  << pgid._id;
	click_chatter("%s: %s", id().cc(), sa.c_str());
      
	return (unsigned)now;
      }
    }
  }

  return 0;
}

String
SendManager::stats(Element *e, void *)
{
  StringAccum sa;
  SendManager *sm = static_cast<SendManager *>(e);
  sa << "Nacksrecd: " << sm->_nacksrecd << "\n";
  click_chatter("%{element}: statistics ----------", sm);
  return sa.take_string();
}

int
SendManager::static_clear(const String &arg, Element *e,
                        void *, ErrorHandler *errh)
{
  SendManager *sm = static_cast<SendManager *>(e);

  bool b;

  if (!cp_bool(arg, &b))
    return errh->error("`clear' must be a boolean");

  if (b) {
    sm->_nacksrecd = 0;

    click_chatter("%{element}: statistics cleared\n", sm);
  }

  return 0;
}

int
SendManager::enable_coding(const String &arg, Element *e,
                        void *, ErrorHandler *errh)
{
  SendManager *sm = static_cast<SendManager *>(e);

  if (!cp_bool(arg, &(sm->_enable_coding)))
    return errh->error("`enable' must be a boolean");

  if (sm->_enable_coding)
    click_chatter("%{element}: encoding enabled", sm);
  else
    click_chatter("%{element}: encoding disabled", sm);
    
  return 0;
}

int
SendManager::rtx_timeout(const String &arg, Element *e,
			     void *, ErrorHandler *errh)
{
  SendManager *sm = static_cast<SendManager *>(e);

  unsigned int x;
  if (!cp_unsigned(arg, &x)) {
    return errh->error("`rtx_timeout' must be an unsigned");
  }
  sm->set_rtx_timeout(x);
  
  click_chatter("rtx_timeout set to %u", x);
    
  return 0;
}

int
SendManager::max_xmits(const String &arg, Element *e,
			     void *, ErrorHandler *errh)
{
  SendManager *sm = static_cast<SendManager *>(e);

  unsigned int x;
  if (!cp_unsigned(arg, &x)) {
    return errh->error("`max_xmits' must be an unsigned");
  }
  sm->set_max_xmits(x);
  
  click_chatter("max_xmits set to %u", x);
    
  return 0;
}

void  
SendManager::add_handlers()
{
  add_read_handler("stats", stats, 0);
  add_write_handler("clear", static_clear, 0);
  add_write_handler("enable", enable_coding, 0);
  add_write_handler("enable_coding", enable_coding, 0);
  add_write_handler("rtx_timeout", rtx_timeout, 0);
  add_write_handler("max_xmits", max_xmits, 0);
}

SendManager::SendManager()
  : Element(0,0), _enable_coding(true), _nacksrecd(0), _max_xmits(4), _rtx_timeout(5), _exp_timeout(200000)
{
}

SendManager::~SendManager()
{
}

void *
SendManager::cast(const char *n)
{
  if (strcmp(n, "SendManager") == 0)
    return (SendManager *) this;
  else
    return 0;
}

int 
SendManager::configure(Vector<String> &conf, ErrorHandler *errh)
{
  int res;
  unsigned t_max_xmits(0), t_rtx_timeout(0), t_exp_timeout(0); 
  
  res = cp_va_parse(conf, this, errh,
		    cpKeywords, 
		    "IP", cpIPAddress, "IP Address", &_myip,
		    "MAX_XMITS", cpUnsigned, "Maximum transmits", &t_max_xmits,
		    "RTX_TIMEOUT", cpUnsigned, "Retransmit Timeout", &t_rtx_timeout,
		    "EXP_TIMEOUT", cpUnsigned, "Expiry Timeout", &t_exp_timeout,
		    cpEnd);

  if (t_max_xmits) _max_xmits = t_max_xmits;
  if (t_rtx_timeout) _rtx_timeout = t_rtx_timeout;
  if (t_exp_timeout) _exp_timeout = t_exp_timeout;

  click_chatter("send manager done");
  return res;
}

/*
void 
SendManager::add_handlers()
{
  add_default_handlers(true);
  //add_read_handler("mappings", static_print_mappings, 0);
  
}
*/

#include <click/bighashmap.cc>
#if EXPLICIT_TEMPLATE_INSTANCES
template class HashMap<PacketID, PacketGID>;
template class HashMap<PacketID, PacketState *>;
#endif

EXPORT_ELEMENT(SendManager)