codingmanager_bak

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), _ack_pk_len(200), _enable_coding(true), _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;
  }
  click_ether *eth_p;
  struct srpacket *srh;

  eth_p = (click_ether *)p->data();
  srh = (struct srpacket *)(eth_p + 1);

  EtherAddress dst = EtherAddress(eth_p->ether_dhost);
  
  uint32_t prev_hop, next_hop;

  prev_hop = next_hop = 0;
  srh->find_surrounding_hops(prev_hop, next_hop, _myip.addr());
 
  // 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);

  NbrQueue::iterator itr = _queue.begin();
  bool present = 0;
  while(itr!=_queue.end()) {
    if (((*itr)->_ipid==ipid) && ((*itr)->_src==src)) {
      present = 1;
      break;
    }
    itr++;
  }
  
  if (present) {
    click_chatter("%s packet already present in queue", id().cc());
    return;
  }

  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;
  
  _queue.push_back(ps);
  int idx = _queue.size()-1;
  if (vqueue->size()==0) {
    ps->_idx = idx;
    click_chatter("position in queue %d", idx);
  }
  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;
  }

  NbrQueue::iterator x = _queue.begin();
  if (x==_queue.end()) return NULL;

  // Find and delete the first packet from the virtual queue
  NbrList todeldt;
  NbrList todelack;
  NbrQueue *queue_t;
  if ((*x)->_data) {
    queue_t = _dtpk_map.find((*x)->_nbr);
    queue_t->erase(queue_t->begin());
    if (!queue_t->size()) todeldt.push_back((*x)->_nbr);
  }
  else {
    queue_t = _ackpk_map.find((*x)->_nbr);
    queue_t->erase(queue_t->begin());
    if (!queue_t->size()) todelack.push_back((*x)->_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((*x));
  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;

  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_pkts->size());
  CodedPktList::iterator x = coded_pkts->begin();
  int i = 0;
  uint16_t max_sz = 0;
  while (x < coded_pkts->end()) {
    PacketState *ps = (*x);
    Packet *p_t = SimpleQueue::yank1_peek(SrcIdfilter(ps->_src, ps->_ipid));
    struct click_ether *eth_h = (struct click_ether *)p_t->data();
    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), ntohs(ip_h->ip_id), (uint32_t)IPAddress(ip_h->ip_src), alias, (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);
    }
    x++;
    i++;
  }
 
  click_chatter("building data packet");
  //Build the encoded part of the packet
  data_p = Packet::make(max_sz);
  memset((void*)data_p->data(), 0, data_p->length());
  x = coded_pkts->begin();
  while (x < coded_pkts->end()) {
    PacketState *ps = (*x);
    Packet *p_t = SimpleQueue::yank1_peek(SrcIdfilter(ps->_src, ps->_ipid));
    _listenmgr->add_new_packet(p_t);
    unsigned char *xored_data = (unsigned char *)data_p->data();
    unsigned char *data = (unsigned char *)((struct click_ether *)p_t->data() + 1);
    for (int i = 0; i < (int)(p_t->length()-sizeof(struct click_ether)); i++) {
      xored_data[i] ^= data[i];
    }
    x++;
    click_chatter("finished xoring");
  }
 
  click_chatter("finished building packet");
  //Combine the header and the data
  if (coded) {
    data_p->push(hlen);
    memcpy((void*)data_p->data(), (void*)xor_p->data(), hlen);
  }
 
  data_p->push(sizeof(struct click_ether));
  PacketState *ps = (*(coded_pkts->begin()));
  Packet *p_t = SimpleQueue::yank1_peek(SrcIdfilter(ps->_src, ps->_ipid));
  struct click_ether *eth_h = (struct click_ether *)p_t->data();
  memcpy((void*)data_p->data(), (void*)eth_h, sizeof(struct click_ether));

  if (coded) {
    click_chatter("%s coding happened", id().cc());
    struct click_ether *eth_ch = (struct click_ether *)data_p->data();
    eth_ch->ether_type = htons(_enc_ethtype);
//    check_coding(data_p);
  }

  xor_p->kill();
  flush_queues(coded_pkts);
  return data_p;
}


void CodingManager::check_coding(Packet *p)
{
  struct click_ether *eth_h = (struct click_ether *)p->data();
  struct click_xorn_header *xorh = (struct click_xorn_header *)(eth_h + 1);

  click_chatter("%s codedethtype %04x", id().cc(), ntohs(eth_h->ether_type));
  for (int i = 0; i < xorh->nentries(); i++) {
    StringAccum sa_t;
    sa_t << "Ethtype " << xorh->get_entry_ethtype(i) << " IPID " << xorh->get_entry_ipid(i) << " Source " << IPAddress(xorh->get_entry_src(i)) << " Alias " << xorh->get_entry_alias(i);
    click_chatter("%s Coded packet %s", id().cc(), sa_t.c_str());
  }
}


void
CodingManager::flush_queues(CodedPktList *coded_pkts) 
{
  CodedPktList::iterator x = coded_pkts->begin();
  while (x < coded_pkts->end()) {
    SrcIdfilter filter_t = SrcIdfilter((*x)->_src, (*x)->_ipid);
    if (SimpleQueue::size()) {
      while (Packet *q = deq()) {
        if (!filter_t(q)) {
          _tbuf.lifo_enq(q);
        } else {
          q->kill();
          break;
        }
      }
    }
    
    if (_tbuf.size()) {
      while (Packet *q = _tbuf.deq()) {
        lifo_enq(q);
      }
    }
   
//     
//     Packet *p_t = yank1(SrcIdfilter((*x)->_src, (*x)->_ipid));
//     if (p_t!=NULL) {
//       click_chatter("found the packet to kill");
// //      p_t->kill();
//     }
    x++;
  }
}
    

CLICK_ENDDECLS


#include <click/bighashmap.cc>
#if EXPLICIT_TEMPLATE_INSTANCES
template class Vector<PacketState*>;
template class HashMap<EtherAddress, NbrQueue*>;
template class Vector<PacketState*>;
template class Vector<EtherAddress>;
#endif
EXPORT_ELEMENT(CodingManager)