reorderer.cc

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

      
    } else if (seq_diff < 0) {  // in case we've already sent out a packet with a larger seq
      // seq_diff should be < 2^32, but this should be taken care of by start_offset      
      output(1).push(next_entry->_p);  // next_entry->_p shouldn't be NULL here
      _late++;
      _reordered++;
      _ordered--;

      click_chatter("%s: seen a late packet!!", id().cc());

      start_seq = next_entry->_tcpseq + next_entry->_pktlen;
      if (start_seq >= (*flowstate)->_skip_seq_start) {
	(*flowstate)->_skip_seq_start -= next_entry->_pktlen;
	start_seq = (*flowstate)->_skip_seq_end;
      }

      // ***note that the above is a little optimistic, 
      // assuming it's unlikely to get more than two packets 
      // that arrive in reverse order to their sent order

      // deregister timeout - may not need this when we only register timeout when there is backlog
      sa.clear();
      sa << sport << " " << dport << " " << next_entry->_tcpseq;
      deregister_timeout(next_entry->_timestamp, sa.take_string());

      pv->erase(pv->begin() + start_offset);  
      delete next_entry;

    } else {
      // *** maybe only register timeout here?

      if (!_timer.scheduled())
	schedule_timer();
      break;
    }

    if (start_offset == pv->size())
      start_offset = 0;
  }
   
  (*flowstate)->_next_seq = start_seq;

  // some cleanup 
  if (!pv->size()) {
    if (count_ordered) {
      _ordered++;
      _reordered--; // the packet we just released actually arrived in order
    }
      
    //click_chatter("%s: backlog cleared", id().cc());  // debugging  
    if (((*flowstate)->_num_pkt_post_fin == 0) && !((*flowstate)->has_hole())) {
      click_chatter("%s: flow completed, deleting flow state", id().cc());
      _pkt_table.remove(key);
      delete (*flowstate);
    }
  } else {
    
    //click_chatter("%s: packet still backlogged %d, starting vector offset %d, last seq %u", id().cc(), pv->size(), start_offset, start_seq);  // debugging  
  }

  return;
}
  

Reorderer::PacketEntry * 
Reorderer::insert(FlowState *fs, PktVector *pv, Packet *p, tcp_seq_t tcpseq, uint16_t pktlen)
{
  //click_chatter("%s: in insert(), current pkt vector size %d", id().cc(), pv->size());
  Vector<PacketEntry *>::iterator i = pv->begin(); 
  for (; i < pv->end(); i++) {
    tcp_seq_t temp_seq = (*i)->_tcpseq;

    // debugging
    //click_chatter("%s: seq in vector: %u", id().cc(), temp_seq);

    if (temp_seq == tcpseq) { // a duplicate, rtx timer must have fired too quickly or maybe TCP has timed out 
      _dups++;
      // debugging
      click_chatter("%s: duplicate seq %u", id().cc(), tcpseq);
      // *** kill the packet?
      drop(p);
      return 0; 
    }

    if (temp_seq > tcpseq) {
      PacketEntry *pe = new PacketEntry(p, tcpseq, pktlen);
      pv->insert(i, pe);

      // TODO*** indicate if the seq has wrapped around -- check this!!
      if (temp_seq - tcpseq > 268435456)  // 2 to power 28
	fs->_start_offset++;
      
      return pe;
    }
  } 

  if (i == (pv->end())) {
    PacketEntry *pe = new PacketEntry(p, tcpseq, pktlen);
    pv->push_back(pe);
    return pe;
  }

  return 0;  
}


void 
Reorderer::register_timeout(String key, PacketEntry *pe)
{
  int32_t time = pe->_timestamp;

  //click_chatter("%s: registering packet at %d, key is %s", id().cc(), time, key.cc());

  // times should be in increasing order, so we shouldn't need to worry about 'insertion-sort' the times
  // the latest time may well be the same as 'time', if packets arrive very quickly 
  if (_time_vec.size() && _time_vec[_time_vec.size()-1] == time) {
    // register the packet entry in the corresponding 
    PktReg **pktreg = _timeout_reg.findp(time);
    (*pktreg)->insert(key, pe);
  } else {
    // add the time, create a registry entry, and register the packet entry
    _time_vec.push_back(time);
    PktReg *pktreg = new PktReg;
    pktreg->insert(key, pe);
    _timeout_reg.insert(time, pktreg);
  }

  //click_chatter("%s: leaving register_timeout, _time_vec.size() is %d, first one is %d", id().cc(), _time_vec.size(), _time_vec[0]);
}


void 
Reorderer::deregister_timeout(int32_t time, String key)
{
  //click_chatter("%s: deregistering key %s for time %d", id().cc(), key.cc(), time);

  PktReg **pktreg = _timeout_reg.findp(time);

  // in case it was never registered
  if (!pktreg || !(*pktreg)) {
    click_chatter("%s: no *pktreg found for time %d", id().cc(), time);
    return;
  }

  (*pktreg)->remove(key);  
  // note that the PacketEntry pointer will be deleted in the caller after this function returns
  
  // some cleanup
  if (!((*pktreg)->size())) {
    //click_chatter("%s: no more entries left for this time", id().cc());
    _timeout_reg.remove(time);
    delete (*pktreg);
    
    // shouldn't take too many iterations...
    for (TimeVecIter tvi = _time_vec.begin(); tvi < _time_vec.end(); tvi++) {
      if ((*tvi) == time) {
	//click_chatter("%s: offset from beginning of _time_vec is %d", id().cc(), tvi - _time_vec.begin());

	if (tvi == _time_vec.begin()) {
	  // need to unschedule timer
	  _timer.unschedule();
	  //click_chatter("%s: at %d, timer unscheduled", id().cc(), Timestamp::now().msec1());
	}
	_time_vec.erase(tvi);
	break;
      }
    }
  } else {
    //click_chatter("%s: %d packets left in the registry", id().cc(), (*pktreg)->size());
  }

}


void 
Reorderer::schedule_timer()
{
  if (!_time_vec.size())
    return;

  //click_chatter("%s: registered time is %d", id().cc(), _time_vec[index]);
  int32_t time_to_sched = _time_vec[0] + (int)_timeout - Timestamp::now().msec1();

  if (time_to_sched <= 0) {// shouldn't really happen, but just in case...
    //click_chatter("%s: time_to_sched is %d, next batch also timed out", id().cc(), time_to_sched);

    run_timer();
  } else {
    //click_chatter("%s: at %d, scheduling timer after %d", id().cc(), Timestamp::now().msec1(), time_to_sched);
    _timer.schedule_after_ms(time_to_sched);
  }
}


// implemented as a separate function just in case we want to do something about the dropped packets
void 
Reorderer::drop(Packet *p)
{
  /*
  if (noutputs() == 2)
    output(1).push(p);
  else
  */
  p->kill();
}

// ------------------ handler stuff ----------------------------------------

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

  StringAccum sa;
  Reorderer *r = (Reorderer *)(e);
  sa << "NUDPPkts: " << r->_udp_seen << "; NTCPPkts: " << r->_tcp_seen << "; NorderedPkts: " << r->_ordered << "; NlatePkts: " << r->_late << "; NreorderedPkts: " << r->_reordered << "; NtimedoutPkts: " << r->_timedout << "; NdupPkts: " << r->_dups << "; NotherPkts: " << r->_rst_set << "\n";
  click_chatter("%{element}: statistics ----------", r);
  if (r->_on)
    return sa.take_string();
  else
    click_chatter("%{element}: reordering turned off", r);

  return "";
}


int
Reorderer::static_clear(const String &arg, Element *e,
                        void *, ErrorHandler *errh)
{
  Reorderer *r = (Reorderer *)(e);

  bool b;

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

  if (b) {
    r->reset();
    click_chatter("%{element}: statistics cleared\n", r);
  }

  return 0;
}


int
Reorderer::enable(const String &arg, Element *e,
		  void *, ErrorHandler *errh)
{
  Reorderer *r = (Reorderer *)e;

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

  if (r->_on) {
    click_chatter("%{element}: reordering enabled", r);
    r->reset();
  } else
    click_chatter("%{element}: reordering disabled", r);
    
  return 0;
}


int
Reorderer::set_timeout(const String &arg, Element *e,
			     void *, ErrorHandler *errh)
{
  Reorderer *r = (Reorderer *)(e);

  unsigned int x;
  if (!cp_unsigned(arg, &x)) {
    return errh->error("`timeout' must be an unsigned");
  }
  r->_timeout = x;
  
  click_chatter("%{element}: timeout set to %u", r, x);
    
  return 0;
}



void 
Reorderer::add_handlers()
{
  add_read_handler("stats", stats, 0);
  add_write_handler("clear", static_clear, 0);
  add_write_handler("enable", enable, 0);
  add_write_handler("timeout", set_timeout, 0);
}


// generate Vector/HashMap template instance
#include <click/bighashmap.cc>
#include <click/vector.cc>
#if EXPLICIT_TEMPLATE_INSTANCES
template class Vector<PacketEntry *>;
template class HashMap<String, FlowState*>;
template class Vector<int32_t>;
template class HashMap<String, PacketEntry*>;
template class HashMap<int32_t, HashMap<String, PacketEntry*>>;
#endif

CLICK_ENDDECLS
EXPORT_ELEMENT(Reorderer)