reorderer.cc

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

/*
 * reorderer.{cc,hh} -- reorders TCP packets before handing them to application
 * Wenjun Hu
 *
 * Copyright (c) 2005 University of Cambridge
 *
 * 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/ipaddress.hh>
#include <click/confparse.hh>
#include <click/error.hh>
#include <click/glue.hh>
#include <click/straccum.hh>
#include <click/timer.hh>
//#include <clicknet/ether.h>
#include <clicknet/ip.h>
//#include <clicknet/tcp.h>
#include "reorderer.hh"
//#include <elements/wifi/sr/srpacket.hh>

CLICK_DECLS

Reorderer::Reorderer()
  : Element(1, 2),
    _on(true),
    _timeout(200),
    _udp_seen(0),
    _tcp_seen(0),
    _ordered(0),
    _late(0),
    _reordered(0),
    _timedout(0),
    _dups(0),
    _rst_set(0),
    _timer(this)
{
}

Reorderer::~Reorderer()
{
}


int
Reorderer::configure(Vector<String> &conf, ErrorHandler *errh)
{
  bool on = true;
  uint16_t timeout = 200;

  int res = cp_va_parse(conf, this, errh,
			/* optional */
			cpKeywords,
			"ON", cpBool, "Whether to enable element", &on,
			"TIMEOUT", cpUnsigned, "timeout", &timeout,
			cpEnd);

  if (res < 0)
    return -1;

  _on = on;
  _timeout = timeout;

  return res;
}


int
Reorderer::initialize(ErrorHandler *)
{
  _timer.initialize(this);

  return 0;
}


void 
Reorderer::run_timer()
{
  //click_chatter("%s: timer up at %d", id().cc(), Timestamp::now().msec1());
  int32_t reg_time = _time_vec[0];
  int32_t time_to_go = Timestamp::now().msec1() - reg_time;

  // necessary?
  if (time_to_go > 0 && time_to_go < (int)_timeout) {
    _timer.schedule_after_ms(_timeout - time_to_go);
    click_chatter("%s: timer run too early, %d ms to go", id().cc(), time_to_go);
    return;
  }

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

  /*
  if (!pktreg || !(*pktreg))
    click_chatter("%s: no packets to timeout, error during registration", id().cc());
  else
    click_chatter("%s: %d packets to release", id().cc(), (*pktreg)->size());

  click_chatter("%s: expect %d iterations in for loop", id().cc(), (*pktreg)->end() - (*pktreg)->begin());
  */

  for (PktRegIter i = (*pktreg)->begin(); i; i++) {
    PacketEntry *pe = i.value();
    String keystring = i.key();
    Packet *pkt = pe->_p;

    //click_chatter("%s: timeout, releasing packet, key string is %s", id().cc(), keystring.cc());
    output(1).push(pkt);
    pe->_p = (Packet *)0;
    _timedout++;
  }

  (*pktreg)->clear();
  _timeout_reg.remove(reg_time);
  delete (*pktreg);
  _time_vec.erase(_time_vec.begin());

  schedule_timer();
}


void
Reorderer::reset()
{
  _time_vec.clear();

  // _pkt_table
  for (PktMap::iterator i = _pkt_table.begin(); i; i++) {
    FlowState *fs = i.value();
    if (fs)
      delete fs;
  }
  _pkt_table.clear();

  // _timeout_reg
  for (TimeoutReg::iterator i = _timeout_reg.begin(); i; i++) {
    PktReg *preg = i.value();
    if (!preg) // this shouldn't happen
      continue;

    for (PktRegIter j = preg->begin(); j; j++) {
      PacketEntry *pe = j.value();
      if (pe)  
	delete pe; 
    }
    preg->clear();
    delete preg;
  }

  _udp_seen = 0;
  _tcp_seen = 0;
  _ordered = 0;
  _late = 0;
  _reordered = 0;
  _timedout = 0;
  _dups = 0;
  _rst_set = 0;
}

void
Reorderer::push(int, Packet *p)
{
  if (!_on) {
    output(0).push(p);
    return;
  }

  // send out non-tcp packets as they arrive
  const click_ip *iph = p->ip_header();
  //assert(iph);
  if (iph->ip_p != IP_PROTO_TCP) {
    click_chatter("%s: non-TCP packet, sending off straight away", id().cc());
    output(0).push(p);
    _udp_seen++;
    return;
  }

  click_chatter("%s: a TCP packet, inserting to reordering buffer", id().cc());
  _tcp_seen++;

  const struct click_tcp *tcph = p->tcp_header();
  uint16_t sport = tcph->th_sport;
  uint16_t dport = tcph->th_dport;
  tcp_seq_t tcpseq = ntohl(tcph->th_seq);
  unsigned tcpflag = tcph->th_flags;
  
  // debugging
  click_chatter("%s: tcp info ------------- time %d -------------", id().cc(), Timestamp::now().msec1());
  click_chatter("%s: tcp info - pkt len %u, ip hdr len %u, total len %u", id().cc(), p->length(), iph->ip_hl * 4, ntohs(iph->ip_len));
  click_chatter("%s: tcp info [sport, dport, seq, ack seq, data offset, flag, win]", id().cc());
  click_chatter("%s: tcp info [%u, %u, %u, %u, %u, 0x%x, %u]", id().cc(), sport, dport, tcpseq, ntohl(tcph->th_ack), tcph->th_off * 4, tcph->th_flags, ntohs(tcph->th_win));
  

  if (tcpflag & 0x4) { // there's some error, so better not holding the packet? or just restrict to 0x14?
    // seq doesn't get incremented for such a packet
    click_chatter("%s: rst flag set; sending off", id().cc());
    output(1).push(p);  // or shall we use port 0?
    _rst_set++;
    return;
  } 

  StringAccum sa;
  sa << sport << " " << dport;
  String key = sa.take_string();

  // find the right vector, create one if necessary...
  // debugging
  //click_chatter("%s: about to insert, src port %u, dest port %u, seq %u", id().cc(), sport, dport, tcpseq);

  FlowState **flowstate = _pkt_table.findp(key);
  if (!flowstate || !(*flowstate)) {
    // debugging
    click_chatter("%s: creating new flowstate for (src port %u, dest port %u)", id().cc(), sport, dport);

    FlowState *new_fs = new FlowState;
    flowstate = &new_fs;
    _pkt_table.insert(key, new_fs);
  }
  PktVector *pv = (*flowstate)->_pktv;

  // insert the packet entry into the vector
  uint16_t seglen = ntohs(iph->ip_len) - iph->ip_hl * 4 - tcph->th_off * 4;

  // watch out for certain TCP flags
  if (tcpflag & 0x2) {
    seglen++;
    //click_chatter("%s: seen syn flag", id().cc());

    if (!(tcpflag & 0x10)) { // syn, but not syn ack - i'm the receiver
      (*flowstate)->_num_pkt_post_fin--;  // see FlowState class definition in reorderer.hh
      //click_chatter("%s: seen syn", id().cc());
    }
  } else if (tcpflag & 0x1) { // fin or fin ack
    (*flowstate)->_num_pkt_post_fin++;
    (*flowstate)->_seen_fin = true;
    seglen++;
    //click_chatter("%s: fin/fin ack seen", id().cc());

    if (tcpflag & 0xe) { // the fin is piggybacked on something else, so we expect more than 1 ack to come
      (*flowstate)->_num_pkt_post_fin--;
      //click_chatter("%s: seen other flag as well as fin", id().cc());
    }
  } else if ((tcpflag & 0x10) && (*flowstate)->_seen_fin) {
    (*flowstate)->_num_pkt_post_fin++;
    //click_chatter("%s: seen an ack for fin", id().cc());
  }

  // possibly redundant check
  if ((*flowstate)->_num_pkt_post_fin > 0)
    click_chatter("%s: for flow %s, seen %d post-fin packets!", id().cc(), key.cc(), (*flowstate)->_num_pkt_post_fin);

  PacketEntry *pe = insert((*flowstate), pv, p, tcpseq, seglen);  

  if (!pe) // must have received a duplicate packet
    return;

  sa.clear();
  sa << sport << " " << dport << " " << pe->_tcpseq;
  // register a timeout event
  register_timeout(sa.take_string(), pe);

  // now clear any backlog
  // run through the vector until a gap is reached in the sequence numbers
  int start_offset = (*flowstate)->_start_offset; 
  tcp_seq_t start_seq = (*flowstate)->_next_seq;

  bool count_ordered = false;
  if (pv->size() == 1) 
    count_ordered = true;

  //click_chatter("%s: about to clear backlog", id().cc());
  //click_chatter("%s: %d packet backlogged, starting vector offset %d, last seq %u", id().cc(), pv->size(), start_offset, start_seq);

  while (pv->size()) {
    PacketEntry *next_entry = (*pv)[start_offset];
    if (!start_seq) { // the first ever packet we're getting in a new flow
      start_seq = next_entry->_tcpseq;
      (*flowstate)->_skip_seq_start = start_seq;
    }

    int seq_diff = (int)next_entry->_tcpseq - (int)start_seq;

    if (seq_diff == 0) {
      Packet *pkt = next_entry->_p;
      if (pkt) { // ie, it hasn't been timed out
	// debugging
	//click_chatter("%s: releasing packet, src port %u, dest port %u, tcp seq %u", id().cc(), sport, dport, start_seq);
	output(1).push(pkt);
	_reordered++;

	// deregister timeout - always need this!
	sa.clear();
	sa << sport << " " << dport << " " << next_entry->_tcpseq;
	deregister_timeout(next_entry->_timestamp, sa.take_string());
      } else {
	//click_chatter("%s: packet probably timed out, src port %u, dest port %u, tcp seq %u", id().cc(), sport, dport, start_seq);
      }

      start_seq += next_entry->_pktlen;  
      (*flowstate)->_skip_seq_end = start_seq;

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