recvmanager.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 <clicknet/ether.h>
#include <elements/wifi/sr/srpacket.hh>

#include "recvmanager.hh"
#include "ackheader.hh"

RecvManager::RecvManager()
  : Element(0, 0),
    _aliases(0), _ack_timeout(1), _max_acks(16), _enable_acks(true)
{
}

RecvManager::~RecvManager()
{
}

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

int
RecvManager::configure(Vector<String> &conf, ErrorHandler *errh)
{
  int res;

  unsigned t_max_acks(0), t_ack_timeout(0);

  res = cp_va_parse(conf, this, errh,
		    cpKeywords, 
		    "ALIASTABLE", cpElement, "Alias Table for neighbors", &_aliases,
		    "MAXACKS", cpUnsigned, "Maximum number of acks", &t_max_acks,
		    "ACKTIMEOUT", cpUnsigned, "Ack timeout", &t_ack_timeout,
		    cpEnd);

  if (!_aliases) {
    return errh->error("ALIASTABLE not specified");
  }
  if (_aliases->cast("AliasTable") == 0) {
    return errh->error("ALIASTABLE element is not an AliasTable");
  }

  if (t_max_acks) {
    _max_acks = t_max_acks;
  }
  if (t_ack_timeout) {
    _ack_timeout = t_ack_timeout;
  }

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

int
RecvManager::enable_acks(const String &arg, Element *e,
		    void *, ErrorHandler *errh)
{
  RecvManager *rm = static_cast<RecvManager *>(e);

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

  rm->set_enable_acks(x);

  if (x)
    click_chatter("%{element}: acks enabled", rm);
  else
    click_chatter("%{element}: acks disabled", rm);
    
  return 0;
}

void
RecvManager::add_handlers()
{
  add_write_handler("enable_acks", enable_acks, 0);
  // add_default_handlers(true);
  //add_read_handler("mappings", static_print_mappings, 0);
  
}

bool
RecvManager::register_packet_to_ack(Packet *p, int now)
{

  if (!_enable_acks) {
    return 1; // packets are always new
  }

  // acks are enabled at this point
  struct click_ether *eth_p = (click_ether *)p->data();
  struct srpacket *srh = (struct srpacket *)(eth_p + 1);
  uint8_t alias = _aliases->lookup(EtherAddress(eth_p->ether_shost));
  click_chatter("%s register_packet_to_ack called alias %u, seq %u, timedout size is %u at %u", id().cc(), (uint16_t)alias, srh->nseq(), _timedout.size(), Timestamp::now().msec1());

  AckStateList *asl = _pending.find(alias, 0);
  if (!asl) {
    click_chatter("%s register_packet_to_ack empty ackstatelist for alias %u, timedout size is %u", id().cc(), alias, _timedout.size());
    asl = new AckStateList;
    _pending.insert(alias, asl);
  }

  // acks are stored in sorted order of sequence number 
  AckStateList::iterator prev = asl->begin();
  AckStateList::iterator next = prev;
  bool toinsert = 1;
  while (next != asl->end()) {
    if ((*next)->_seq >= srh->nseq()) {
      click_chatter("%s register_packet_to_ack found boundary element in AckStateList with prev %u, next %u, timedout size is %u", id().cc(), (*prev)->_seq, (*next)->_seq, _timedout.size());
      if ((*next)->_seq == srh->nseq()) {
	toinsert = 0;
      }
      break;
    }
    prev = next++;
  }

  // this is a duplicate, packet was already received
  if (toinsert == 0) {
    click_chatter("%s register_packet_to_ack skipping insert, duplicate, timedout size is %u", id().cc(), _timedout.size());
    return toinsert;
  }

  AckState *as = new AckState(srh->nseq(), now + _ack_timeout);

  if (prev == next) {
    asl->push_front(as);
  } else {
    asl->insert_after(prev, as);
  }
  
  // at this point, this is a new seq, set up a timeout for the ack
  // also insert into the timeout array if it is not already there
  click_chatter("%s register_packet_to_ack adding %u to timedout neighbors for time %u, timedout size is %u", id().cc(), alias, as->_timeout, _timedout.size());
  NbrStateList *nsl = _timedout[as->_timeout];
  click_chatter("%s register_packet_to_ack NbrStateList entry is %x, timedout size is %u", id().cc(), nsl, _timedout.size());
  if (!nsl) {
    nsl = new NbrStateList;
    _timedout[as->_timeout] = nsl;
    click_chatter("%s register_packet_to_ack had to insert new NbrStateList entry %x for %u, timedoutsize is %u", id().cc(), nsl, as->_timeout, _timedout.size());
  }


  // neighbors are stored in sorted order for easy searching
  NbrStateList::iterator prevn = nsl->begin();
  NbrStateList::iterator nextn = prevn;
  bool exists = 0;
  while (nextn != nsl->end()){
    if ((*nextn)->_alias >= alias) {
      click_chatter("%s register_packet_to_ack found boundary element in NbrStateList with alias %u, timedout size is %u", id().cc(), (*nextn)->_alias, _timedout.size());
      if ((*nextn)->_alias == alias) {
	exists = 1;
      }
      break;
    }
    prevn = nextn++;
  }
  
  if (exists) {
    (*nextn)->_nacks++;
  } else {
    NbrState *ns = new NbrState(alias);
    if (prevn == nextn) {
      nsl->push_front(ns);
    } else {
      nsl->insert_after(prevn, ns);
    }
  }

  return toinsert;
}

RecvManager::AckVector *
RecvManager::get_acks_timeout(uint32_t now, NbrList &nl)
{
  if (!_enable_acks) {
    // no acks timeout if no acks are sent
    return 0;
  }

  TimeoutList::iterator bound = _timedout.upper_bound(now);

  // find the unique set of neighbors that has timed out
  for (TimeoutList::iterator i = _timedout.begin(); ((i != bound) && (i != _timedout.end())); i++) {
    NbrStateList *nsl = (*i).second;
    for (NbrStateList::const_iterator j = nsl->begin(); j != nsl->end(); j++) {
      nl.insert((*j)->_alias, 1);
      click_chatter("%s get_acks_timeout adding alias %u to timedout list, timedout size is %u", id().cc(), (*j)->_alias, _timedout.size());
    }
  }

  AckVector *av = new AckVector;
  for (NbrList::iterator i = nl.begin(); i != nl.end(); i++) {
    uint8_t alias = i.key();
    click_chatter("%s get acks timeout finding ackstatelist for %u, timedout size is %u", id().cc(), alias, _timedout.size());
    AckStateList *asl = _pending.find(alias, 0);
    AckStateList::iterator j;
    for (j = asl->begin(); ((j != asl->end()) && (*j)->_acked); j++);
    if (j == asl->end()) {
      continue;
    }
    click_chatter("%s get acks timeout looking for first unacked element, timedout size is %u", id().cc(), _timedout.size());
    uint16_t start = (*j)->_seq;
    (*j)->_acked = 1;
    TimeoutList::iterator x = _timedout.find((*j)->_timeout);
    NbrStateList *nsl = (*x).second;
    NbrStateList::iterator prev = nsl->begin();
    NbrStateList::iterator next = prev;
    click_chatter("%s get_acks_timeout adding acks starting from %u for alias %u, timedout size is %u", id().cc(), start, alias, _timedout.size());
    while (next != nsl->end()) {
      if ((*next)->_alias == alias) {
	click_chatter("%s get_acks_timeout updating nacks for alias %u for start, nacks is %u, timedout size is %u", id().cc(), alias, (*next)->_nacks, _timedout.size());
	--(*next)->_nacks;
	if ((*next)->_nacks == 0) {
	  click_chatter("%s get_acks_timeout deleting NbrStateList entry for alias %u for start, timedout size is %u", id().cc(), alias, _timedout.size());
	  if (prev == next) {
	    nsl->pop_front();
	  } else {
	    nsl->erase_after(prev);
	  }
	}
	break;
      }
      prev = next++;
    }
    if (nsl->empty()) {
    // delete the entire timeout entry
      click_chatter("%s get_acks_timeout deleting entire timeout entry at %u for start, timedout size is %u", id().cc(), (*x).first, _timedout.size());
      _timedout.erase(x);
    }
      
    uint8_t bmp = 0;
    j++;
    while (j != asl->end()) {
      uint16_t seq = (*j)->_seq;
      if (seq > start + 8) {
	break;
      }
      click_chatter("%s get_acks_timeout considering bitmap entry for %u, acked is %u, timedout size is %u", id().cc(), seq, (*j)->_acked, _timedout.size());
      if ((*j)->_acked == 0) {
	(*j)->_acked = 1;
	x = _timedout.find((*j)->_timeout);
	NbrStateList *nsl = (*x).second;
	prev = nsl->begin();
	next = prev;
	while (next != nsl->end()) {
	  if ((*next)->_alias == alias) {
	    click_chatter("%s get_acks_timeout updating nacks for alias %u for seq, nacks is %u, timedout size is %u", id().cc(), alias, (*next)->_nacks, _timedout.size());
	    (*next)->_nacks--;
	    if ((*next)->_nacks == 0) {
	      click_chatter("%s get_acks_timeout deleting NbrStateList entry for alias %u for seq, timedout size is %u", id().cc(), alias, _timedout.size());
	      if (prev == next) {
		nsl->pop_front();
	      } else {
		nsl->erase_after(prev);
	      }
	    }
	    break;
	  }
	  prev = next++;
	}
	
	if (nsl->empty()) {
	  // delete the entire timeout entry
	  click_chatter("%s get_acks_timeout deleting entire timeout entry at %u for seq, timedout size is %u", id().cc(), (*x).first, _timedout.size());
	  _timedout.erase(x);
	}
      }
      
      int offset;
      if (seq > start) {
	offset = seq - start - 1;
      } else {
	offset = uint32_t(seq) + uint32_t(1 << 16) - uint32_t(start);
      }
      bmp = bmp | (0x01 << offset);

      j++;
    }
    click_chatter("%s: get_acks_timeout adding ackunit entry with alias %u, start %u, bitmap %#x, timedout size is %u at %u", id().cc(), alias, start, bmp, _timedout.size(), Timestamp::now().msec1());
    av->push_back(AckUnit(alias, start, bmp));
  }

  // got all the timed out acks here
  return av;
}
			 
WritablePacket *
RecvManager::get_acks(bool qempty, int now)
{

  // click_chatter("%s get_acks called with %u at time %u", id().cc(), qempty, now);

  if (!_enable_acks) {
    // no acks to send
    return 0;
  }

  if (_pending.begin() == _pending.end()) {
    return 0;
  }

  NbrList nl;
  AckVector* av = get_acks_timeout(now, nl);
  unsigned int num_acks = (unsigned)av->size();

  if (num_acks != 0) {
    click_chatter("%s get_acks returned from timeout with %u acks, timedout size is %u", id().cc(), num_acks, _timedout.size());
  }

  if ((qempty) || (num_acks >= _max_acks)) {
    return make_packet(av);
  }

  // now av wasn't big enough, look for general acks
  int hash_size  = _pending.size();

  PendingAcks::iterator i = _pending.begin();
  int jumps = random() % _pending.size();
  for (int x = 0; x < jumps; x++) {
    if (i == _pending.end()) {
      i = _pending.begin();
    }
    i++;
  }
  
  for (int j = 0; ((j < hash_size) && (num_acks < _max_acks)); j++) {
    if (i == _pending.end()) {
      i = _pending.begin();
    }
    uint8_t alias = i.key();
    // don't insert another entry for a neighbor that has already been inserted by get_acks_timeout
    if (nl.findp(alias)) {
      click_chatter("%s get_acks skipping alias %u, already populated by get_acks_timeout, timedout size is %u", id().cc(), alias, _timedout.size());
      i++;
      continue;
    }
    AckStateList *asl = i.value();
    AckStateList::iterator k;
    for (k = asl->begin(); (k != asl->end()) && (*k)->_acked; k++);
    if (k == asl->end()) {
      continue;
    }

    uint16_t start = (*k)->_seq;
    (*k)->_acked = 1;
    TimeoutList::iterator x = _timedout.find((*k)->_timeout);
    NbrStateList *nsl = (*x).second;
    NbrStateList::iterator prev = nsl->begin();
    NbrStateList::iterator next = prev;
    click_chatter("%s get_acks adding acks starting from %u for alias %u for start, timedout size is %u", id().cc(), start, alias, _timedout.size());
    while (next != nsl->end()) {
      if ((*next)->_alias == alias) {
	click_chatter("%s get_acks updating nacks for alias %u for start, nacks is %u, timedout size is %u", id().cc(), alias, (*next)->_nacks, _timedout.size());
	(*next)->_nacks--;
	if ((*next)->_nacks == 0) {
	  click_chatter("%s get_acks deleting NbrStateList entry for alias %u for start, timedout size is %u", id().cc(), alias, _timedout.size());
	  if (prev == next) {
	    nsl->pop_front();
	  } else {
	    nsl->erase_after(prev);
	  }
	}
	break;
      }
      prev = next++;
    }
    if (nsl->empty()) {
      // delete the entire timeout entry
      click_chatter("%s get_acks deleting entire timeout entry at %u for start, timedout size is %u", id().cc(), (*x).first, _timedout.size());
      _timedout.erase(x);
    }
      
    uint8_t bmp = 0;
    k++;
    while (k != asl->end()) {
      uint16_t seq = (*k)->_seq;
      if (seq > start + 8) {
	break;
      }
      click_chatter("%s get_acks considering bitmap entry for %u, acked is %u, timedout size is %u", id().cc(), seq, (*k)->_acked, _timedout.size());
      if ((*k)->_acked == 0) {
	(*k)->_acked = 1;

	// update the entry for this neighbor in the timedout hash
	x = _timedout.find((*k)->_timeout);
	nsl = (*x).second;
	prev = nsl->begin();
	next = prev;
	while (next != nsl->end()) {
	  if ((*next)->_alias == alias) {
	    click_chatter("%s get_acks updating nacks for alias %u for seq, nacks is %u, timedout size is %u", id().cc(), alias, (*next)->_nacks, _timedout.size());
	    (*next)->_nacks--;
	    if ((*next)->_nacks == 0) {
	      click_chatter("%s get_acks deleting NbrStateList entry for alias %u for start, timedout size is %u", id().cc(), alias, _timedout.size());
	      if (prev == next) {
		nsl->pop_front();
	      } else {
		nsl->erase_after(prev);
	      }
	    }
	    break;
	  }
	  prev = next++;
	}

	if (nsl->empty()) {
	  // delete the entire timeout entry
	  click_chatter("%s get_acks deleting entire timeout entry at %u for seq, timedout size is %u", id().cc(), (*x).first, _timedout.size());
	  _timedout.erase(x);
	}
      }
      
      int offset;
      if (seq > start) {
	offset = seq - start - 1;
      } else {
	offset = uint32_t(seq) + uint32_t(1 << 16) - uint32_t(start);
      }
      bmp = bmp | (0x01 << offset);

      k++; // move on to the next ack
    }
    click_chatter("%s get_acks adding ackunit entry with alias %u, start %u, bitmap %#x, timedout size is %u at %u", id().cc(), alias, start, bmp, _timedout.size(), Timestamp::now().msec1());
    av->push_back(AckUnit(alias, start, bmp));
    num_acks++;
    i++;
  }
    
  return make_packet(av);
}
    
WritablePacket *
RecvManager::make_packet(AckVector *av)
{
  int nentries = av->size();

  if (nentries == 0) {
    return 0;
  }

  struct click_enc_ack *ackh;
  uint8_t hlen = click_enc_ack::get_hlen(nentries);
  WritablePacket *p = Packet::make(hlen);
  ackh = (struct click_enc_ack *)p->data();
  memset(ackh, 0, hlen);

  ackh->set_nentries(nentries);
  int idx = 0;
  click_chatter("%s make_packet adding %u ackh entries, timedout size is %u", id().cc(), nentries, _timedout.size());
  for (AckVector::const_iterator i = av->begin(); i != av->end(); i++, idx++) {
    ackh->set_entry(idx, (*i)._alias, (*i)._seq, (*i)._bmp);
    click_chatter("%s make_packet adding entry %d - alias %u, seq %u, bmp %#x, timedout size is %u", id().cc(), idx, (*i)._alias, (*i)._seq, (*i)._bmp, _timedout.size());
  }

  delete av;
  return p;
}

uint32_t
RecvManager::get_next_ack_time()
{
  TimeoutList::iterator i = _timedout.begin();
  if (i != _timedout.end()) {
    return (*i).first;
  }
  return 0;
}
  
#include <click/bighashmap.cc>
#include <click/vector.cc>
#if EXPLICIT_TEMPLATE_INSTANCES
template class Vector<AckUnit>;
template class HashMap<uint8_t, AckStateList *>;
template class HashMap<uint8_t, uint8_t>;
#endif
CLICK_ENDDECLS
EXPORT_ELEMENT(RecvManager)