codingmanager_test

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

#include <click/config.h>
#include <click/confparse.hh>
#include <clicknet/ether.h>
#include <clicknet/ip.h>
#include <clicknet/tcp.h>
#include <click/ipaddress.hh>
#include <click/etheraddress.hh>
#include <click/error.hh>
#include <click/straccum.hh>
#include <click/timestamp.hh>
#include <click/string.hh>
#include <elements/wifi/sr/srpacket.hh>
#include <elements/wifi/sr/path.hh>

#include "codingmanager.hh"


CLICK_DECLS

CodingManager::CodingManager()
  : SimpleQueue(), 
    _enc_ethtype(0), 
    _myip(), 
    _aliases(0), 
    _guess_mgr(0), 
    _listenmgr(0), 
    _enable_coding(true),
    _ack_pk_len(200), 
    _max_coded_pkts(5)
{
  click_chatter("init coding manager");
}

CodingManager::~CodingManager()
{
}

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

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

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

  cm->set_enable_coding(x);

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

*/

void
CodingManager::add_virtual(Packet *p)
{
  if (!_enable_coding) {
    return;
  }

  struct click_ether *eth_p = (click_ether *)p->data();

  EtherAddress dst = EtherAddress(eth_p->ether_dhost);
  
  // Get IP header information
  const struct click_ip *ip_h = p->ip_header();
  IPAddress src = IPAddress(ip_h->ip_src);
  uint16_t ipid = ntohs(ip_h->ip_id);


  StringAccum sa;
  sa << "Adding packet to virtual queue with src " << src << " ipid " << ipid << " headed to " << dst;
  click_chatter("%s %s", id().cc(), sa.c_str());
  // now decide which queue to go to
  NbrQueue *vqueue;
  uint32_t  pklen = p->length();
  if (pklen < _ack_pk_len) {
    click_chatter("is an ack packet");
    vqueue = _ackpk_map.find(dst);
    if (vqueue==NULL) {
      click_chatter("first time to destination creating queue");
      vqueue = new NbrQueue;
      _ackpk_map.insert(dst, vqueue);
    }
  }
  else {
    click_chatter("is a data packet");
    vqueue = _dtpk_map.find(dst);
    if (vqueue==NULL) {
      click_chatter("first time data packet");
      vqueue = new NbrQueue;
      _dtpk_map.insert(dst, vqueue);
    }
  }



  PacketState *ps = new PacketState(dst, src, ipid);
  if (pklen < _ack_pk_len) ps->_data = false;
  
  vqueue->push_back(ps);

  uint32_t now = Timestamp::now().usec1();
  click_chatter("%u push queue size %d", now, SimpleQueue::size());
  return;

}

// Find the position of a particular packet in the queue
int 
CodingManager::find_position(PacketState *ps)
{
  NbrQueue::iterator x = _queue.begin();
  if (!x) return -1;

  int idx = 0;
  while (x < _queue.end()) {
    if ((*(*x))==(*ps)) return idx;
    idx++;
    x++;
  }

  return -1;
}

WritablePacket*
CodingManager::copy_pkt(Packet *p)
{
  WritablePacket *new_p = Packet::make(p->length());
  memcpy((void *)new_p->data(), (void *)p->data(), p->length());
  return new_p;
}
  
  
//Codes normal packets
WritablePacket*
CodingManager::find_coding_candidates(int port)
{
  if (!_enable_coding) {
    Packet *p = SimpleQueue::pull(port);
    if (p) {
      uint32_t now = Timestamp::now().usec1();
      click_chatter("%u pull queue size %d", now, (SimpleQueue::size()+1));
      WritablePacket *new_p = p->uniqueify();
      return new_p;
    }
    return NULL;
  }

  if (SimpleQueue::size() == 0) return NULL;

  // Find and delete the first packet from the virtual queue
  Packet *p_head = SimpleQueue::deq();
  struct click_ether *eth_p_head = (click_ether *)p_head->data();

  EtherAddress dst_head = EtherAddress(eth_p_head->ether_dhost);
  
  // Get IP header information
  const struct click_ip *ip_h_head = p_head->ip_header();
  IPAddress src_head = IPAddress(ip_h_head->ip_src);
  uint16_t ipid_head = ntohs(ip_h_head->ip_id);
  PacketState *ps_head = new PacketState(dst_head, src_head, ipid_head);
  if (p_head->length() < _ack_pk_len) ps_head->_data = false;
  SimpleQueue::lifo_enq(p_head);

  NbrList todeldt;
  NbrList todelack;
  NbrQueue *queue_t;
  if (ps_head->_data) {
    queue_t = _dtpk_map.find(ps_head->_nbr);
    queue_t->erase(queue_t->begin());
    if (!queue_t->size()) todeldt.push_back(ps_head->_nbr);
  }
  else {
    queue_t = _ackpk_map.find(ps_head->_nbr);
    queue_t->erase(queue_t->begin());
    if (!queue_t->size()) todelack.push_back(ps_head->_nbr);
  }


  //Now look for other packets to code with
  NbrQueueMap::iterator dt_x = _dtpk_map.begin();
  NbrQueueMap::iterator ack_x = _ackpk_map.begin();

  CodedPktList *coded_pkts = new CodedPktList;
  coded_pkts->push_back(ps_head);
  int coded_pkt_cnt = 1;

  //First data packets
  NbrQueue *queue;
  NbrQueue::iterator head;
  while (dt_x != _dtpk_map.end()) {
    StringAccum sa_t;
    sa_t << "Looking in data Nbrqueue for " << (dt_x.key());
    click_chatter("coding data packets: %s", sa_t.c_str());
    queue = dt_x.value();
    head = queue->begin();
    if (is_coding_worthy(coded_pkts, (*head))) {
      click_chatter("found a coding worthy data packet");
      coded_pkts->push_back((*head));
      coded_pkt_cnt++;
      head = queue->erase(head);
      if (head >= queue->end()) {
        todeldt.push_back(dt_x.key());
      }
      if (coded_pkt_cnt >= _max_coded_pkts) break;
    }
    dt_x++;
  }
 
  //Ack packets
  if (coded_pkt_cnt < _max_coded_pkts) {
    while(ack_x != _ackpk_map.end()) {
      StringAccum sa_t;
      sa_t << "Looking in ack Nbrqueue for " << (ack_x.key());
      click_chatter("coding ack packets: %s", sa_t.c_str());
      queue = ack_x.value();
      head = queue->begin();
      if (is_coding_worthy(coded_pkts, (*head))) {
        coded_pkts->push_back((*head));
        coded_pkt_cnt++;
        head = queue->erase(head);
        if (head >= queue->end()) {
          todelack.push_back(ack_x.key());
        }
        if (coded_pkt_cnt >= _max_coded_pkts) break;
      }
      ack_x++;
    }
  }


  //delete queues both data and ack which have become empty
  NbrList::iterator todeldtx = todeldt.begin();
  while (todeldtx < todeldt.end()) {
    _dtpk_map.remove((*todeldtx));
    StringAccum sa_t;
    click_chatter("%s", sa_t.c_str()); 
    todeldtx++;
  }
 
  NbrList::iterator todelackx = todelack.begin();
  while (todelackx < todelack.end()) {
    _ackpk_map.remove((*todelackx));
    StringAccum sa_t;
    click_chatter("%s", sa_t.c_str()); 
    todelackx++;
  }
   
  /*
  //delete packets from the main queue
  CodedPktList::iterator i = coded_pkts->begin();
  i = coded_pkts->begin();
  while (i < coded_pkts->end()) {
    PacketState *ps_t = (*i);
    del_pkt(ps_t);
    i++;
  }
  */
    
  //Make the packet
  WritablePacket *new_p = combine_pkts(coded_pkts);
  return new_p;

}

void
CodingManager::del_pkt(PacketState *ps) {
  NbrQueue::iterator x = _queue.begin();
  if (!x) return;

  while (x < _queue.end()) {
    if ((*(*x))==(*ps)) {
      click_chatter("erasing a packet from virtual queue");
      x = _queue.erase(x);
      break;
    }
    x++;
  }
}
 

bool
CodingManager::is_coding_worthy(CodedPktList *coded_pkts, PacketState *ps)
{
  CodedPktList::iterator x = coded_pkts->begin();
  while (x < coded_pkts->end()) {
    if (!_guess_mgr->check_presence((*x), ps)) return 0;
    x++;
  }
  return 1;
}

WritablePacket*
CodingManager::combine_pkts(CodedPktList *coded_pkts)
{
  bool coded = false;

  if (coded_pkts->size()>1) coded = true;

  CodedActualPkts coded_actual_pkts;
  CodedPktList::iterator itr = coded_pkts->begin();
  while (itr < coded_pkts->end()) {
    PacketState *ps = (*itr);
    Packet *p_t = SimpleQueue::yank1(SrcIdfilter(ps->_src, ps->_ipid));
    coded_actual_pkts.push_back(p_t);
    itr++;
  }
  
 
  if (coded_actual_pkts.size() != coded_pkts->size()) {
    click_chatter("%s alert did not find the same number of coded packets", id().cc());
    return NULL;
  }
  
  int hlen = click_xorn_header::get_hlen(coded_pkts->size());
  WritablePacket *xor_p = Packet::make(hlen);
  WritablePacket *data_p = NULL;
 

  // Build XOR header
  struct click_xorn_header *xorh = (struct click_xorn_header *)xor_p->data();
  xorh->set_nentries(coded_actual_pkts.size());
  CodedActualPkts::iterator x = coded_actual_pkts.begin();
  int i = 0;
  uint16_t max_sz = 0;
  while (x < coded_actual_pkts.end()) {
    Packet *p_t = (*x);
    struct click_ether *eth_h = (struct click_ether *)p_t->data();
    struct srpacket *sr_h = (struct srpacket *)(eth_h + 1);
    const struct click_ip *ip_h = p_t->ip_header();
    uint8_t alias = _aliases->lookup(EtherAddress(eth_h->ether_dhost));
    xorh->set_entry(i, ntohs(eth_h->ether_type), alias, sr_h->nseq(), sr_h->next(), (uint32_t)IPAddress(ip_h->ip_src), ntohs(ip_h->ip_id), ip_h->ip_ttl, (uint16_t)p_t->length());
    if (coded) {
      StringAccum sa_t;
      sa_t << "Coding packet to " << EtherAddress(eth_h->ether_dhost) << " IP source " << IPAddress(ip_h->ip_src) << " ipid " << ntohs(ip_h->ip_id);
      click_chatter("%s %s", id().cc(), sa_t.c_str());
    }
    if ((p_t->length() - sizeof(struct click_ether)) > max_sz) {
      max_sz = p_t->length() - sizeof(struct click_ether);