sentstore.cc

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

/*
 * sentstore.{cc,hh} -- stores all forwarded packets, 
 * resends sent packets if necessary
 * Wenjun Hu
 *
 * Copyright (c) 2005 University of Cambridge
 * Copyright (c) 2005 Massachusetts Institute of Technology
 *
 * Permission is hereby granted, free of charge, to any person obtaining a
 * copy of this software and associated documentation files (the "Software"),
 * to deal in the Software without restriction, subject to the conditions
 * listed in the Click LICENSE file. These conditions include: you must
 * preserve this copyright notice, and you cannot mention the copyright
 * holders in advertising related to the Software without their permission.
 * The Software is provided WITHOUT ANY WARRANTY, EXPRESS OR IMPLIED. This
 * notice is a summary of the Click LICENSE file; the license in that file is
 * legally binding.
 */

#include <click/config.h>
#include <click/confparse.hh>
#include <clicknet/ether.h>
#include <clicknet/ip.h>
#include <click/error.hh>
#include <click/glue.hh>
#include <click/straccum.hh>
#include <click/timestamp.hh>
#include <click/packet_anno.hh>
#include <elements/wifi/sr/srpacket.hh>
#include <elements/wifi/sr/path.hh>
#include "sentstore.hh"
#include "encodedheader.hh"

CLICK_DECLS

SentStore::SentStore()
  : Element(1,1), 
    _added(0), 
    _removed(0), 
    _expired(0),
    _unacked(0),
    _retxed(0),
    //_ip(), 
    _timeout(60000),
    _granularity(50),
    _timer(this)
{
}

SentStore::~SentStore()
{
}

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

int
SentStore::configure(Vector<String> &conf, ErrorHandler *errh)
{
    int res = cp_va_parse(conf, this, errh,
			  cpKeywords,
			  //"IP", cpIPAddress, "IP address", &_ip,
			  cpOptional,
			  cpKeywords,
			  "TIMEOUT", cpUnsigned, "timeout", &_timeout,
			  "GRAN", cpUnsigned, "granularity", &_granularity,
			  cpEnd);

    /*
    if (!_ip)
      return errh->error("IP not specified");
    */      
  
    return res;
}

int
SentStore::initialize(ErrorHandler *)
{
    // this is a silly implementation, but at least it works for now
    // a better implementation would use associative vectors 
    // to keep track of times and vectors of packet entries 
    for (int i = 0; i <= _timeout/_granularity * 3; i++) {
      PerUnitEntries* entries = new PerUnitEntries;
      _removal_entries.push_back(entries);
    }

    _timer.initialize(this);
    //_timer.schedule_after_ms(_timeout); -- wait till first packet is added to the storage

    reset();

    return 0;
}

void 
SentStore::run_timer()
{
    int now = Timestamp::now().msec1();

    if (now - _start_time >= _timeout) {
      PerUnitEntries* v = _removal_entries[0];

      clear_entries(v);
 
      _removal_entries.erase(_removal_entries.begin());
      _removal_entries.push_back(v); // to reuse the memory space

      _start_time += _granularity;

      //debugging
      //click_chatter("%s: start time changed to %d", id().cc(), _start_time);
    }      

    if (_added - _removed > 0)
      _timer.schedule_after_ms(_granularity);
}

/*
void 
SentStore::add_await_ack_entry(uint32_t nexthop, uint32_t next2hop, uint32_t seq)
{
    //mark the corresponding packet
    uint32_t packetkey = (next2hop)?next2hop:nexthop;
    Per2hopNbQ **nbq = _queues.findp(packetkey);

    if (!nbq || !(*nbq)) {
      click_chatter("%s: missing packet: nexthop %s, dataseq %d", id().cc(), IPAddress(nexthop).s().cc(), seq);
      return;
    }

    WritablePacket *mp = (WritablePacket *)((*nbq)->findp(seq));

    if (!mp) {
      click_chatter("%s: missing packet: nexthop %s, dataseq %d", id().cc(), IPAddress(nexthop).s().cc(), seq);
      return;
    }

    SET_SEND_ERR_ANNO(mp, 1); 

    // add an entry to keep track of acks too
    NbEntries** nbq = _await_acks.findp(nexthop);

    if (!nbq || !(*nbq)) {  ///***check again 
      NbEntries *new_q = new NbEntries;
      nbq = &new_q;
      _await_acks.insert(nexthop, new_q);
    } 

    PacketEntry *pe = new PacketEntry(next2hop, seq);
    (*nbq)->insert(seq, pe); 
 
}
*/

void 
SentStore::reset()
{
    for (int i = 0; i < _removal_entries.size(); i++) 
      clear_entries(_removal_entries[i]);
 
    _start_time = Timestamp::now().msec1();

    _added = 0;
    _removed = 0;
    _expired = 0;
    _unacked = 0;
    _retxed = 0;
}

void 
SentStore::clear_entries(PerUnitEntries* v)
{
    for (int i = 0; i < v->size(); i++) {
      PacketEntry *pe = (*v)[i];
      //uint32_t pe_nb = pe->_nb;
      uint16_t pe_seq = pe->_seq;
      uint32_t pe_src = pe->_src;

      // clear the packets
      StringAccum newkey;
      newkey << IPAddress(pe_src) << pe_seq;
      String keystring = newkey.take_string();
      Packet **p = _storage.findp(keystring);

      if (p && (*p)) {
	if (!SEND_ERR_ANNO(*p)) {
	  // only delete the packet if there's no need to retransmit it
	  // otherwise the ack/retransmit manager will take care of it 
	  _storage.remove(keystring); 
	  (*p)->kill(); 
	  _expired++;
	  _removed++;

	} else {
	  _unacked++;
	}
      }
	//}
      
      delete pe; // the index info to the packet is deleted irrespective of the packet
    }

    v->clear();
}

Packet *
SentStore::getPacket(uint32_t src_ip, uint16_t seq)
{
  StringAccum newkey;
  newkey << IPAddress(src_ip) << seq;
  Packet **p = _storage.findp(newkey.take_string());
  if (p)
    return (*p);
  else
    return 0;
}

void
SentStore::push(int port, Packet* p)
{
  // adds a copy of the packet to the queue
  WritablePacket *p_copy = p->clone()->uniqueify();  

  if (!p_copy) {
    click_chatter("%{element}: might have run out of memory\n", this);
    return;
  }

  // extract the expected next hop and hop after the next
  struct click_ether *eth = (struct click_ether *)p_copy->data();
  struct srpacket *srh = (struct srpacket *)(eth + 1);  
  Path route_p = srh->get_path();
  uint32_t src_ip = route_p[0].addr();
  /*
  uint32_t next_2hop = 0;
  uint32_t nexthop = 0;
  for (int i = 0; i<route_p.size(); i++) {
    if (route_p[i].addr()==_ip.addr()) {
      nexthop = route_p[i+1]; // if the packet ends up here, there's at least a next hop along the path
      next_2hop = ((i+2)<route_p.size())?route_p[i+2].addr():0;
      break;
    }
  }

  uint32_t key = (next_2hop)?next_2hop:nexthop;
    
  //if (key) {
  //debugging
  //click_chatter("%{element}: looking for the 2-hop neighbour storage for %d\n", this, next_2hop);

  Per2hopNbQ** nbq = _queues.findp(key);

  if (!nbq || !(*nbq)) {  ///***check again 
    Per2hopNbQ *new_q = new Per2hopNbQ;
    nbq = &new_q;
    _queues.insert(key, new_q);
  } 
  */

  //uint32_t seq = srh->data_seq();
  click_ip *ip_copy = p_copy->ip_header();
  uint16_t seq = ntohs(ip_copy->ip_id);
  StringAccum newkey;
  newkey << IPAddress(src_ip) << seq;

  _storage.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());

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

  // also insert an entry for removal
  PacketEntry *pe = new PacketEntry(src_ip, seq);
  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);
}


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

  StringAccum sa;
  SentStore *tq = static_cast<SentStore *>(e);
  sa << "packets added: " << tq->_added << "; packets timed out: " << tq->_expired;
  sa << "; total in storage: " << (tq->_added - tq->_removed) << "\n";
  sa << "not acked: " << tq->_unacked << "; retx'ed " << tq->_retxed << "\n";
  return sa.take_string();
}

int
SentStore::static_clear(const String &arg, Element *e,
                        void *, ErrorHandler *errh)
{
  SentStore *tq = static_cast<SentStore *>(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 
SentStore::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<SentStore::PacketEntry *>;
template class Vector<Vector<SentStore::PacketEntry *> *>;
template class Vector<int>;
#endif


CLICK_ENDDECLS
EXPORT_ELEMENT(SentStore)