codingmanager.cc

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

    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;
  }
  return combine_actual_pkts(&coded_actual_pkts, num_bytes, 0);
}

WritablePacket*
CodingManager::combine_actual_pkts(CodedActualPkts *coded_actual_pkts, int *num_bytes, int rexmitted)
{
  bool coded = false;
  if (coded_actual_pkts->size()>1) {
    coded = true;
    click_chatter("codedpktcnt: %d", coded_actual_pkts->size());
  }

  // Get the header length for the XOR header
  int hlen = click_xorn_header::get_hlen(coded_actual_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;
  int offset = hlen + sizeof(struct click_ether);
  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);
    *num_bytes += (sr_h->hlen_with_data() - sr_h->hlen_wo_data());
    const struct click_ip *ip_h = p_t->ip_header();
    uint16_t sport(0), dport(0);
    tcp_seq_t seq(0), ack(0);
    uint8_t flags(0);
    if (ip_h->ip_p == IP_PROTO_TCP) {
      const struct click_tcp *tcp_h = p_t->tcp_header();
      sport = ntohs(tcp_h->th_sport);
      dport = ntohs(tcp_h->th_dport);
      seq = ntohl(tcp_h->th_seq);
      ack = ntohl(tcp_h->th_ack);
      flags = tcp_h->th_flags;
    } else if (ip_h->ip_p == IP_PROTO_UDP) {
      const struct click_udp *udp_h = p_t->udp_header();
      sport = ntohs(udp_h->uh_sport);
      dport = ntohs(udp_h->uh_dport);
    }
    uint8_t alias = _aliases->lookup(EtherAddress(eth_h->ether_dhost));
    // Important change: pkt len in xor header now only includes the length from the IP header not the whole packet
    xorh->set_entry(i, ntohs(eth_h->ether_type), alias, sr_h->num_links(), (uint32_t)IPAddress(ip_h->ip_src), ntohs(ip_h->ip_id), ip_h->ip_ttl, (uint16_t)(sr_h->hlen_with_data() - sr_h->hlen_wo_data()));
    xorh->set_entry_additional(i, (uint32_t)IPAddress(ip_h->ip_dst), ntohs(ip_h->ip_len), ip_h->ip_p, sport, dport, seq, ack, flags); // FOR SNOOPER
    offset += sr_h->hlen_wo_data();
    if ((sr_h->hlen_with_data() - sr_h->hlen_wo_data()) > max_sz) {
      max_sz = (sr_h->hlen_with_data() - sr_h->hlen_wo_data());
    }
    x++;
    i++;
  }

  click_chatter("offset %d xorlen %d", offset, hlen);

  //Build the encoded part of the packet
  data_p = Packet::make(max_sz);
  assert(data_p->length() == max_sz);
  memset((void*)data_p->data(), 0, data_p->length());
  x = coded_actual_pkts->begin();
  uint32_t now = Timestamp::now().msec1();
  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);
    if (!rexmitted) {
      _sendmgr->add_new_packet(p_t, now);
    }
    unsigned char *xored_data = (unsigned char *)data_p->data();
    unsigned char *data = (unsigned char *)(p_t->ip_header());
    int p_tlen = sr_h->hlen_with_data() - sr_h->hlen_wo_data();
    for (int i = 0; i < p_tlen; i++) {
      xored_data[i] ^= data[i];
    }
    x++;
  }
 
  click_chatter("finished building packet");
  //Push SR headers
  x = coded_actual_pkts->end();
  x--;
  while (x >= coded_actual_pkts->begin()) {
    Packet *p_t = (*x);
    struct click_ether *eth_h = (struct click_ether *)p_t->data();
    struct srpacket *sr_h = (struct srpacket *)(eth_h + 1);    
    data_p->push(sr_h->hlen_wo_data());
    memcpy((void*)data_p->data(), (void*)sr_h, sr_h->hlen_wo_data());
    x--;
  }
  //Combine the xor header and the data
  if (coded) {
    data_p->push(hlen);
    memcpy((void*)data_p->data(), (void*)xor_p->data(), hlen);
  }
 
  // Push ether header
  data_p->push(sizeof(struct click_ether));
  Packet *p_t = (*(coded_actual_pkts->begin()));
  struct click_ether *eth_h = (struct click_ether *)p_t->data();
  memcpy((void*)data_p->data(), (void*)eth_h, sizeof(struct click_ether));

  struct click_ether *eth_ch = (struct click_ether *)data_p->data();
  if (coded) {
    click_chatter("%s coding happened", id().cc());
    if (_fix_enc_dst) {
      memcpy(eth_ch->ether_dhost, _enc_dst.data(), 6);
    }
    eth_ch->ether_type = htons(_enc_ethtype);
//    check_coding(data_p, coded_actual_pkts);
  }
  EtherAddress pktdst = EtherAddress(eth_ch->ether_dhost);
  // strictly for logging
  x = coded_actual_pkts->begin();
  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();   
    StringAccum sa_t;
    sa_t << "sending packet with ethdst = " <<  EtherAddress(eth_h->ether_dhost) << " pktdst = " << pktdst << " ipsrc = " << IPAddress(ip_h->ip_src) << " ipid = " << ntohs(ip_h->ip_id) << " coded = 1" << " srheader len " << sr_h->hlen_wo_data();
    click_chatter("%s: %u %s", id().cc(), Timestamp::now().msec1(), sa_t.c_str());
    sa_t.clear();
    x++;
  }

  click_chatter("sending out packet of size %d", data_p->length());
    
  xor_p->kill();
  x = coded_actual_pkts->begin();
  while (x < coded_actual_pkts->end()) {
    Packet *p_t = (*x);
    p_t->kill();
    x++;
  }
  return data_p;
}


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++;
  }
}

/*
//Testing the coding part

void CodingManager::check_coding(Packet *p, CodedActualPkts *coded_actual_pkts)
{
  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 nentries %d", id().cc(), ntohs(eth_h->ether_type), xorh->nentries());
  for (unsigned 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());
    int res = do_decoding(p, xorh->get_entry_alias(i), coded_actual_pkts);
    if (res) {
      click_chatter("%s decoding worked for alias %u", id().cc(), xorh->get_entry_alias(i));
    }
  }

}

bool
CodingManager::do_decoding(Packet *p_orig, uint8_t alias, CodedActualPkts *coded_actual_pkts)
{
  Packet *p = p_orig->clone();
  
  click_ether *eh = (click_ether *) p->data();
  struct click_xorn_header *xorh = (struct click_xorn_header *)(eh + 1);
  int nentries = xorh->nentries();

  click_chatter("%s: push() got packet of length %u", id().cc(), p->length());

  int idx = -1;
  for (int i = 0; i < nentries; i++) {
    if (xorh->get_entry_alias(i) == alias) {
      idx = i;
      break;
    }
  }
  if (idx == -1) {
    p->kill();
    return false;
  }

  
  click_chatter("%s: push() packet matched me, idx is %d", id().cc(), idx);


  StringAccum sa;
  WritablePacket *p_enc = p->uniqueify();
  Packet *cmp_pkt;
  CodedActualPkts::iterator x = coded_actual_pkts->begin();
  for (int i = 0; i < nentries; i++) {
    if (i==idx) {
      cmp_pkt = (*x);
      x++;
      continue;
    }

    IPAddress ip_t = IPAddress(xorh->get_entry_src(i));
    uint16_t ipid_t = xorh->get_entry_ipid(i);
    PacketRID pid(ip_t, ipid_t);

    sa << "push() needs to xor with PID ( " << pid._src << ", " <<  pid._id << ")";
    click_chatter("%s %s", id().cc(), sa.c_str());
    Packet *pkt_ovhd = _listenmgr->get_packet(pid);
//    Packet *pkt_flist = (*x);
//    Packet *pkt_ovhd = pkt_flist->clone();
    if (!pkt_ovhd) {
      click_chatter("%s: %s not found in cache", id().cc(), sa.c_str());
      p_enc->kill();
      return false;
    }

    xor_decode(p_enc, pkt_ovhd, i);
    x++;
  }
  
  Packet *decoded = finish_decode(p_enc, idx);
 
  // compare it with the packet you sent
  
  IPAddress ip_t = IPAddress(xorh->get_entry_src(idx));
  uint16_t ipid_t = xorh->get_entry_ipid(idx);
  PacketRID pid(ip_t, ipid_t);
//  Packet *pkt = _listenmgr->get_packet(pid);
  

  Packet *pkt = cmp_pkt;
  
  WritablePacket *pkt_t = pkt->uniqueify();
  click_ip *iph = pkt_t->ip_header();
  StringAccum sa_t;
  sa_t << "comparing with Src " << IPAddress(iph->ip_src) << " ipid " << ntohs(iph->ip_id) << " length " << pkt_t->length();
  click_chatter("%s %s", id().cc(), sa_t.c_str());
 
  iph->ip_ttl--;
  unsigned long sum = (~ntohs(iph->ip_sum) & 0xFFFF) + 0xFEFF;
  iph->ip_sum = ~htons(sum + (sum >> 16));

  struct click_ether *eth = (struct click_ether *)pkt_t->data();
  struct srpacket *srh = (struct srpacket *)(eth+1);
  srh->set_next(srh->next()+1);
  sum = (~ntohs(srh->_cksum) & 0xFFFF) - 0xFFFE;
  srh->_cksum = ~htons(sum + (sum >> 16));
*/

  /*
  uint16_t pktlen = xorh->get_entry_len(idx) - sizeof(struct click_ether);
  unsigned char *dec_data = (unsigned char *)decoded->data() + sizeof(struct click_ether);
  unsigned char *orig_data = (unsigned char *)((struct click_ether *)(pkt_t->data()) + 1);
  int res = memcmp(dec_data, orig_data, pktlen);
  if (res == 0) {
    click_chatter("%s packet matched perfectly", id().cc());
    return true;
  }
  else {
    click_chatter("%s packet did not match writing out byte by byte now", id().cc());
    uint8_t *mod_data = (uint8_t*)pkt_t->data() + sizeof(struct click_ether);
    uint8_t *sent_data = (uint8_t*)decoded->data() + sizeof(struct click_ether);   
    StringAccum sa_t1;
    sa_t1 << "Unmatched bytes ";
    for (uint16_t i = 0; i < pktlen; i++) {
      if (mod_data[i] != sent_data[i]) {
        sa_t1 << i << " : ";
      }
    }
    click_chatter("%s %s", id().cc(), sa_t1.c_str());
    return false;
  }
}

WritablePacket*
CodingManager::finish_decode(WritablePacket *dec, int idx)
{
  click_ether *eth_h = (struct click_ether *)(dec->data());
  EtherAddress src = EtherAddress(eth_h->ether_shost);
  EtherAddress dst = EtherAddress(eth_h->ether_dhost);

  struct click_xorn_header *xorh = (struct click_xorn_header *)(eth_h + 1);
  uint16_t et = xorh->get_entry_ethtype(idx);
  int xorh_len = xorh->get_hlen(xorh->nentries());
  int dec_len = dec->length() - xorh_len;
  int pkt_len = xorh->get_entry_len(idx);
  

  dec->pull(sizeof(struct click_ether));
  dec->pull(xorh_len);

  dec->push(sizeof(struct click_ether));
  click_ether *eh = (struct click_ether *)dec->data();
  memcpy(eh->ether_shost, src.data(), 6);
  memcpy(eh->ether_dhost, dst.data(), 6);
  eh->ether_type = htons(et);

  click_chatter("%s decoded packet ethtype %04x", id().cc(), et);
  if (pkt_len < dec_len) {
    dec->take(dec_len-pkt_len);
  }
  return dec;
}

void
CodingManager::xor_decode(WritablePacket *enc, Packet *pkt, int i)
{
  // Headers of the received encoded packet
  click_ether *eth_enc = (struct click_ether *)(enc->data());
  struct click_xorn_header *xorh = (struct click_xorn_header *)(eth_enc + 1);

  // First modify the packet we have to make sure we xor with the right version
  WritablePacket *pkt_t = pkt->uniqueify();
  struct click_ether *eh = (struct click_ether *)pkt_t->data();
  struct srpacket *srh = (struct srpacket *)(eh+1);
  click_ip *iph = pkt_t->ip_header();

  // First the IP header
  iph->ip_ttl = xorh->get_entry_ipttl(i);
  set_ip_cksum(pkt_t);

  // Now the SR header
  srh->set_next(xorh->get_entry_next(i));
  srh->set_nseq(xorh->get_entry_nseq(i));
  srh->set_checksum();

  StringAccum sa_t;
  sa_t << "decoding with Src " << IPAddress(iph->ip_src) << " ipid " << ntohs(iph->ip_id) << " length " << pkt_t->length();
  click_chatter("%s %s", id().cc(), sa_t.c_str());

  // First modify the packet we have to make sure we xor with the right version
  WritablePacket *pkt_t = pkt->uniqueify();
  click_ip *iph = pkt_t->ip_header();
  StringAccum sa_t;
  sa_t << "decoding with Src " << IPAddress(iph->ip_src) << " ipid " << ntohs(iph->ip_id) << " length " << pkt_t->length();
  click_chatter("%s %s", id().cc(), sa_t.c_str());

  
  iph->ip_ttl--;
  unsigned long sum = (~ntohs(iph->ip_sum) & 0xFFFF) + 0xFEFF;
  iph->ip_sum = ~htons(sum + (sum >> 16));

  struct click_ether *eh = (struct click_ether *)pkt_t->data();
  struct srpacket *srh = (struct srpacket *)(eh+1);
  srh->set_next(srh->next()+1);
  sum = (~ntohs(srh->_cksum) & 0xFFFF) - 0xFFFE;
  srh->_cksum = ~htons(sum + (sum >> 16));


  // Now do the decoding with the appropriate packet
  click_ether *eth_enc = (struct click_ether *)(enc->data());
  struct click_xorn_header *xorh = (struct click_xorn_header *)(eth_enc + 1);
*/
  /*
  int orig_plen = pkt_t->length() - sizeof(struct click_ether);
  int recd_plen = xorh->get_entry_len(i) - sizeof(struct click_ether);

  if (orig_plen != recd_plen) {
    click_chatter("%s lengths do not match o:%d r:%d", id().cc(), orig_plen, recd_plen);
  }

  int min_len = ((orig_plen < recd_plen) ? orig_plen : recd_plen);
  unsigned char *enc_data = (unsigned char *)enc->data() + sizeof(struct click_ether) + xorh->get_hlen(xorh->nentries());
  const unsigned char *orig_data = (const unsigned char *)((struct click_ether *)(pkt_t->data()) + 1);

  for (int j = 0; j < min_len; j++) {
    enc_data[j] ^= orig_data[j];
  }  

  pkt_t->kill();
  return;
}

void
CodingManager::set_ip_cksum(WritablePacket *p)
{
  click_ip *ip = p->ip_header();
  unsigned plen = p->length() - p->ip_header_offset();
  unsigned hlen;
  
  if (!ip || plen < sizeof(click_ip))
    goto bad;

  hlen = ip->ip_hl << 2;
  if (hlen < sizeof(click_ip) || hlen > plen)
    goto bad;

  ip->ip_sum = 0;
  ip->ip_sum = click_in_cksum((unsigned char *)ip, hlen);
  return;

 bad:
  click_chatter("SetIPChecksum: bad lengths");
  return;
}
*/
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

ELEMENT_PROVIDES(CodingManager)
// EXPORT_ELEMENT(CodingManager)