transmittedqueue.cc

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

    newkey << IPAddress(src_ip) << seq;

    //(*nbq)->insert(seq, p_copy); 
    (*nbq)->insert(newkey.take_string(), p_copy); 

    // borrow a few bytes in user_anno as flags, 
    // these will be overwritten should the packets be retransmitted (within the magic field...)
    // so no worries 
    // clear the first two bytes of all_user_anno (may not be necessary)
    SET_PAINT_ANNO(p_copy, 0);     // the packet has not been used for decoding
    SET_SEND_ERR_ANNO(p_copy, 0);  // the packet has been acked and can be removed when timed out

    //debugging
    //click_chatter("\n%s: storage size %d", id().cc(), (*nbq)->size());

    /* moved to decoding part
    // we need to modify the sr header of the packet copy, 
    // because the index of the next node is incremented before the encoding
    // also recompute checksum
    srh->_next++;
    srh->set_checksum();
    */

    if (_added - _removed <= 0) {
      _start_time = Timestamp::now().msec1();
      _timer.schedule_after_ms(_timeout);
    }

    // also insert an entry for removal
    PacketEntry *pe = new PacketEntry(key, seq, src_ip);
    int current_time = Timestamp::now().msec1();
    int offset = (current_time - _start_time) / _granularity;
    
    //debugging
    if (_start_time < 0)
      click_chatter("%s: ***negative start time!!****", id().cc());

    if (current_time < 0)
      click_chatter("%s: ***negative current time!!****", id().cc());

    if (offset < 0 || offset >= _removal_entries.size()) {
      click_chatter("%s: wrong offset in _removal_entries %d", id().cc(), offset);
      click_chatter("current time is %d, start time is %d", current_time, _start_time);
    }

    _removal_entries[offset]->push_back(pe);

    _added++;

      // debugging
      //click_chatter("%s: packets in storage -- %d --", id().cc(), _added);

      // force storage clearance if necessary - in case timer is not properly scheduled
      if (offset >= _timeout / _granularity)
	run_timer();

      //debugging
      //click_chatter("%{element}: packet added, 2-hop neighbour is %d, seq is %d\n", this, next_2hop, seq);
      //click_chatter("length: %d, cksum: 0x%.4x", srh->hlen_with_data(), (unsigned long) ntohs(srh->_cksum));
      //click_chatter("%s: packets added/removed/stored - %d/%d/%d", id().cc(), _added, _removed, _added - _removed);
      
    

    output(0).push(p);
  } else {
    //debugging
    //click_chatter("\n%{element}: push port 1 - hope to decode a packet\n", this);

    // port = 1 -- decoding

    // see if that's for me
    click_ether *eh = (click_ether *) p->data();
    struct click_xored *extrah = (struct click_xored *)(eh + 1);
    uint32_t host0 = extrah->prev_hop[0];
    uint32_t host1 = extrah->prev_hop[1];
    uint32_t matching_host;
    uint32_t src_ip;
    uint16_t seq;
    uint16_t ethtype;

    if (_ip.addr() == host1) {
      matching_host = host0;
      seq = ntohs(extrah->seq[1]);
      src_ip = extrah->src[1];
      ethtype = extrah->ether_type[0];
    } 
    else if (_ip.addr() == host0) {
      matching_host = host1;
      seq = ntohs(extrah->seq[0]);
      src_ip = extrah->src[0];
      ethtype = extrah->ether_type[1];
    } else {
      //debugging
      click_chatter("%{element}: push - bailing, packet not for me\n", this);
      
      click_chatter("\nthis packet's protocol type: %d", ntohs(eh->ether_type));
      click_chatter("Encoded header: [protocol type, sequence number, source IP, previous hop]\n");
      click_chatter("Packet 1: [%d, %d, %d, %d]", ntohs(extrah->ether_type[0]), ntohs(extrah->seq[0]), extrah->src[0], host0);
      click_chatter("Packet 2: [%d, %d, %d, %d]", ntohs(extrah->ether_type[1]), ntohs(extrah->seq[1]), extrah->src[1], host1);
      click_chatter("total length %d, shorter packet length %d \n", extrah->hlen_with_data(), extrah->short_length);

      // the packet is not for me
      p->kill();  
      _overheard++;

      //debugging
      //click_chatter("%{element}: push - packet killed\n", this);
      return;
    }
      
    uint8_t *src = eh->ether_shost;
    uint8_t *dst = eh->ether_dhost;

    //click_chatter("%{element}: matching host is %d, seq number is %d", this, matching_host, seq); 


    // find the corresponding packet for decoding
    Per2hopNbQ **nbq = _queues.findp(matching_host);

    if (!nbq || !(*nbq)) { ///***check again
      //click_chatter("%{element}: unable to locate the matching packet's storage, dropping packet", this); 
      //***maybe print some packet header info
      p->kill(); //***?
      _undecodable++;
      return;
    }

    StringAccum newkey;
    newkey << IPAddress(src_ip) << seq;
    Packet *mp = *((*nbq)->findp(newkey.take_string()));

    if (!mp) {
      //click_chatter("%{element}: unable to find the matching packet, dropping packet\n", this);

      p->kill(); // ***for now; might implement proper handling later
      _undecodable++;
      return;
    }


    // see if this packet has been used, i.e., we received a duplicate encoded packet
    if (PAINT_ANNO(mp)) {
      _dupe++;
      return;
    }

    WritablePacket *matching_p = mp->uniqueify();

    // a few things to do to the matching packet before we decode...
    // 1. we need to decrement ttl 
    // because the actual (sent) packet's ttl is decremented before encoding is done
    // code copied from DecIPTTL::simple_action
    click_ip *ip_copy = matching_p->ip_header();
    assert(ip_copy);

    ip_copy->ip_ttl--;

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

    // 2. get rid of the ethernet header
    matching_p->pull(sizeof(struct click_ether));

    // 3. modify the sr header...
    struct srpacket *srh = (struct srpacket *)matching_p->data();
    srh->_next++;
    srh->set_checksum();


    // to the received encoded packet, strip off the ether header and the extra header
    WritablePacket *wp = p->uniqueify();
    wp->pull(sizeof(click_ether)+sizeof(struct click_xored));

    uint16_t len1 = matching_p->length();
    uint16_t len2 = 0;
    uint16_t slen = ntohs(extrah->short_length);

    if (len1 == slen) {
      len2 = extrah->data_len();
      //len1 -= 4;
    } else {
      len2 = slen;
      //len1 -= 4;
    }

    WritablePacket *original_p = xor_packet(matching_p, wp, len1, len2);
    _decoded++;
    SET_PAINT_ANNO(original_p, 1);

    // debugging
    //click_chatter("%s: len: received (len2), matching, decoded - %d (%d), %d, %d", id().cc(), wp->length(), len2, mp->length(), original_p->length());

    if (matching_p->length() == original_p->length()) {
      // the received packet should've been shorter, so get rid of the padding
      original_p->take(original_p->length() - slen);

      // debugging
      //click_chatter("%s: received too long, adjusted packet len - %d", id().cc(), original_p->length());

//***check length difference between uint32_t and uint16_t!
    } else {
      original_p->take(8); // these packets have travelled across two hops, so there are eight bytes extra...

      // debugging
      //click_chatter("%s: adjusted packet len - %d", id().cc(), original_p->length());
    }


    // put the ethernet header back
    original_p = original_p->push(sizeof(click_ether));
    click_ether *oeth = (click_ether *)original_p->data();
    memcpy(oeth->ether_dhost, dst, 6);
    memcpy(oeth->ether_shost, src, 6);
    oeth->ether_type = ethtype; 

    // clear up, free the memory if appropriate   ***check again
    /* -- now that packets will be removed anyway after timeout, we'll not do it here
       something is wrong if packets are removed here and searched for again after timeout
    (*nbq)->remove(seq); 
    matching_p->kill(); //***???
    _removed++;

    //debugging
    click_chatter("\n%s: storage size %d", id().cc(), (*nbq)->size());
    click_chatter("%s: packets added/removed/stored - %d/%d/%d", id().cc(), _added, _removed, _added - _removed);

    if ((*nbq)->size() == 0) {
      _queues.remove(matching_host);
      delete (*nbq);
    }
    */

    //debugging

    //click_chatter("%{element}: push - decoding finished\n", this);

    output(1).push(original_p);
  }
}


String
TransmittedQueue::stats(Element *e, void *)
{

  StringAccum sa;
  TransmittedQueue *tq = static_cast<TransmittedQueue *>(e);
  sa << "packets added: " << tq->_added << "; packets timed out: " << tq->_expired;
  sa << "; total in storage: " << (tq->_added - tq->_removed) << "\n";
  sa << "packets decoded: " << tq->_decoded << "; packets undecodable: " << tq->_undecodable;
  sa << "; packets overheard: " << tq->_overheard << "\n";
  sa << "duplicate encoded packets: " << tq->_dupe << "; not acked: " << tq->_unacked << "; retx'ed" << tq->_retxed << "\n";
  return sa.take_string();
}

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

  bool b;

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

  if (b) {
    tq->reset();

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

  return 0;
}

void 
TransmittedQueue::add_handlers()
{
  add_read_handler("stats", stats, 0);
  add_write_handler("clear", static_clear, 0);
}

// adapted from checksrheader.cc, not sure if necessary
#include <click/hashmap.cc>
#include <click/vector.cc>
#if EXPLICIT_TEMPLATE_INSTANCES
template class HashMap<String, Packet*>; 
template class HashMap<IPAddress, HashMap<String, Packet*>>; 
template class HashMap<String, PacketEntry*>;
template class HashMap<IPAddress, HashMap<String, PacketEntry*>>; 
template class Vector<TransmittedQueue::PacketEntry *>;
template class Vector<Vector<TransmittedQueue::PacketEntry *> *>;
template class Vector<int>;
#endif


CLICK_ENDDECLS
EXPORT_ELEMENT(TransmittedQueue)