tokenqueue.cc

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

	if (nfo->_congestion) {
	    pk->set_flag(FLAG_ECN);
	    struct timeval now;
	    click_gettimeofday(&now);
	    StringAccum sa;
	    sa << "TokenQueue " << now;
	    sa << " ECN";
	    sa << " pk->seq " << pk->data_seq();
	    sa << path_to_string(nfo->_p);
	    click_chatter("%s", sa.take_string().cc());
	}
	nfo->_congestion = false;
    }
    return 0;

}

void
TokenQueue::process_source(struct srpacket *pk) 
{
    Path p = pk->get_path();
    PathInfo *nfo = NULL;
    
    nfo = _paths.findp(p);
    if (!nfo) {
	nfo = _paths.findp(reverse_path(p));
    }
    if (!nfo) {
	if (_debug) {
	    StringAccum sa;
	    sa << id() << " " << Timestamp::now();
	    sa << " create:  new_path " << path_to_string(p);
	    click_chatter("%s", sa.take_string().cc());
	}
	_paths.insert(p, PathInfo(p, this));
	nfo = _paths.findp(p);

	nfo->_last_tx = nfo->_last_rx = nfo->_first_tx = nfo->_first_rx
	    = nfo->_last_real = Timestamp::now();
	nfo->_active = false;
    }

    /* we want the token if we're not active
     * or if we're new *
     */
    if (!nfo->_active) {
	nfo->_last_real = Timestamp::now();
	nfo->_active = true;
	nfo->_token = true;
	_tokens++;
	nfo->_towards = nfo->other_endpoint(_sr_forwarder->ip());
    }
}
void
TokenQueue::process_forward(struct srpacket *pk) 
{
    Path p = pk->get_path();
    PathInfo *nfo = find_path_info(p);
    Timestamp now = Timestamp::now();

    IPAddress towards = p[p.size()-1];

    if (pk->seq2() > nfo->_seq) {
	if (_debug) {
	    click_chatter("seq no reset");
	}
	if (nfo->_token) {
	    _tokens--;
	}
	 nfo->reset_rx(pk->seq2(), towards);
    } else if (nfo->_seq != pk->seq2()) {
	if (_debug) {
	    StringAccum sa;
	    sa << id() << " " << now;
	    sa << " old_seq";
	    sa << " towards " << p[p.size()-1];
	    sa << " expected " << nfo->_seq;
	    click_chatter("%s", sa.take_string().cc());
	}
	return;
    }

    if (nfo->_towards != towards) {
	if (nfo->_towards.addr() < towards.addr()) {
	if (_debug) {
	    click_chatter("towards reset");
	}
	if (nfo->_token) {
	    _tokens--;
	}
	 nfo->reset_rx(pk->seq2(), towards);
	} else {
	    if (_debug) {
		StringAccum sa;
		sa << id() << " " << now;
		sa << " dup_token";
		sa << " seq " << pk->seq2();
		sa << " towards " << p[p.size()-1];
		sa << " expected " << nfo->_seq;
		click_chatter("%s", sa.take_string().cc());
	    }
	    return;
	}
    }
    /* a packet that I'm forwarding */
    if (!pk->flag(FLAG_SCHEDULE_FAKE)) {
	nfo->_last_real = now;
    }
    nfo->_packets_rx++;
    if (nfo->_packets_rx == 1) {
	nfo->_first_rx = now;
	if (_debug) {
	    StringAccum sa;
	    sa << id() << " " << now;
	    sa << " first_rx";
	    sa << " seq " << pk->seq2();
	    sa << " since_tx " << now - nfo->_last_tx;
	    click_chatter("%s", sa.take_string().cc());
	}
    } 
    nfo->_last_rx = now;
    if (pk->flag(FLAG_SCHEDULE_TOKEN)) {
	if (nfo->_token) {
	    StringAccum sa;
	    sa << id() << " " << now;
	    sa << " dup_token";
	    sa << " seq " << pk->seq2();
	    click_chatter("%s", sa.take_string().cc());
	} else {
	    nfo->_expected_rx = pk->seq() + 1;
	    nfo->_rx_token = true;
	    if (_debug) {
		StringAccum sa;
		sa << id() << " " << now;
		sa << " token_rx";
		sa << " seq " << pk->seq2();
		sa << " expected " << nfo->_expected_rx;
		sa << " packets_rx " << nfo->_packets_rx;
		sa << " rx_time " << now - nfo->_first_rx;
		click_chatter("%s", sa.take_string().cc());
	    }
	}
    }

    if (!nfo->_token && nfo->_rx_token && nfo->_expected_rx >= nfo->_packets_rx) {
	/* I have now received all the packets */
	if (_debug) {
	    StringAccum sa;
	    sa << id() << " " << now;
	    sa << " final_rx";
	    sa << " seq " << pk->seq2();
	    sa << " rx_time " << now - nfo->_first_rx;
	    click_chatter("%s", sa.take_string().cc());
	}
	nfo->_token = true;
	_tokens++;
	if (nfo->is_endpoint(_sr_forwarder->ip())) {
	    nfo->_towards = nfo->other_endpoint(nfo->_towards);
	}
    }
}

void
TokenQueue::push(int port, Packet *p_in)
{

    WritablePacket *p_out = p_in->uniqueify();
    
    if (!p_out) {
	return;
    }
    click_ether *eh = (click_ether *) p_in->data();
    if (eh->ether_type != htons(_et)) {
	if (enq(p_in)) {
	    _normal++;
	}
	goto done;
    } else if (port == 2) {
	_retransmits++;
	if (_debug) {
	    struct timeval now;
	    click_gettimeofday(&now);
	    StringAccum sa;
	    sa << id() << " " << now;
	    sa << " retransmit";
	    click_chatter("%s", sa.take_string().cc());
	}

    } else {
	click_ether *eh = (click_ether *) p_out->data();
	struct srpacket *pk = (struct srpacket *) (eh+1);
	if (port == 1) {
	    process_forward(pk);
	    if (pk->flag(FLAG_SCHEDULE_FAKE)) {
		p_out->kill();
	    } else {
		output(1).push(p_out);
	    }
	    return;
	} else if (_sr_forwarder->ip() == pk->get_link_node(0)) {
	    process_source(pk);
	}
	if (pk->flag(FLAG_SCHEDULE_FAKE)) {
	    p_out->kill();
	    return;
	}  
	pk->set_seq(0);
	pk->unset_flag(FLAG_SCHEDULE_TOKEN);
    }

    bubble_up(p_out);

 done: 
    if ((_normal > 0 || _tokens > 0  || _retransmits > 0) && !_empty_note.signal_active()) {
	/* there is work to be done! */
	_empty_note.wake_listeners();
    }
}

String
TokenQueue::static_print_stats(Element *f, void *)
{
    TokenQueue *d = (TokenQueue *) f;
    return d->print_stats();
}

String
TokenQueue::print_stats()
{
  StringAccum sa;

  Timestamp now = Timestamp::now();

  sa << " tokens " << _tokens;
  sa << " retransmits " << _retransmits;
  sa << " normal " << _normal;
  sa << " signal " << _empty_note.signal_active();
  sa << "\n";

  for (PathIter iter = _paths.begin(); iter; iter++) {
      const PathInfo &nfo = iter.value();
      sa << "[ " << path_to_string(nfo._p) << "] :";
      sa << " seq " << nfo._seq;
      sa << " token " << nfo._token;
      sa << " last_rx " << now - nfo._last_rx;
      sa << " packets_rx " << nfo._packets_rx;
      sa << " expected_rx " << nfo._expected_rx;
      sa << " last_tx " << (now - nfo._last_tx);
      sa << " packets_tx " << nfo._packets_tx;
      sa << "\n";
  }

  return sa.take_string();
}
String
TokenQueue::static_print_debug(Element *f, void *)
{
  StringAccum sa;
  TokenQueue *d = (TokenQueue *) f;
  sa << d->_debug << "\n";
  return sa.take_string();
}

String
TokenQueue::static_print_packet_timeout(Element *f, void *)
{
  StringAccum sa;
  TokenQueue *d = (TokenQueue *) f;
  sa << d->_max_tx_packet_ms << "\n";
  return sa.take_string();
}

String
TokenQueue::static_print_threshold(Element *f, void *)
{
  StringAccum sa;
  TokenQueue *d = (TokenQueue *) f;
  sa << d->_threshold << "\n";
  return sa.take_string();
}



int
TokenQueue::static_clear(const String &arg, Element *e,
			void *, ErrorHandler *errh) 
{
  TokenQueue *n = (TokenQueue *) e;
  bool b;

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

  if (b) {
    n->clear();
  }
  return 0;
}

void
TokenQueue::clear() 
{
  
  struct timeval now;
  click_gettimeofday(&now);
  Vector<Path> to_clear;
  for (PathIter iter = _paths.begin(); iter; iter++) {
      PathInfo nfo = iter.value();
      to_clear.push_back(nfo._p);
  }

  for (int x = 0; x < to_clear.size(); x++) {
      click_chatter("TokenQueue %s: removing %s\n", 
		    id().cc(),
		    path_to_string(to_clear[x]).cc());
      _paths.remove(to_clear[x]);
  }
}

int
TokenQueue::static_write_debug(const String &arg, Element *e,
			void *, ErrorHandler *errh) 
{
  TokenQueue *n = (TokenQueue *) e;
  bool b;

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

  n->_debug = b;
  return 0;
}


int
TokenQueue::static_write_packet_timeout(const String &arg, Element *e,
			void *, ErrorHandler *errh) 
{
  TokenQueue *n = (TokenQueue *) e;
  unsigned int b;

  if (!cp_unsigned(arg, &b))
    return errh->error("`packet_timeout' must be a unsigned int");

  return n->set_packet_timeout(errh, b);
}

int
TokenQueue::set_packet_timeout(ErrorHandler *errh, unsigned int x) 
{

  if (!x) {
    return errh->error("PACKET_TIMEOUT must not be 0");
  }
  _max_tx_packet_ms = x;
  return 0;
}

int
TokenQueue::static_write_threshold(const String &arg, Element *e,
			void *, ErrorHandler *errh) 
{
  TokenQueue *n = (TokenQueue *) e;
  unsigned int b;

  if (!cp_unsigned(arg, &b))
    return errh->error("`threshold' must be a unsigned int");

  return n->set_threshold(errh, b);
}

int
TokenQueue::set_threshold(ErrorHandler *errh, int x) 
{

  if (x < 0) {
    return errh->error("THRESHOLD must be > 0");
  }
  _threshold = x;
  return 0;
}




void 
TokenQueue::add_handlers()
{
    add_write_handler("clear", static_clear, 0);
    add_read_handler("stats", static_print_stats, 0);
    add_write_handler("debug", static_write_debug, 0);
    add_read_handler("debug", static_print_debug, 0);
    
    add_write_handler("threshold", static_write_threshold, 0);
    add_read_handler("threshold", static_print_threshold, 0);
    
    add_write_handler("packet_timeout", static_write_packet_timeout, 0);
    add_read_handler("packet_timeout", static_print_packet_timeout, 0);
    
    NotifierQueue::add_handlers();
}
// generate template instances
#include <click/bighashmap.cc>
#include <click/vector.cc>
#if EXPLICIT_TEMPLATE_INSTANCES
template class HashMap<Path, PathInfo>;
template class Vector< Vector<IPAddress> >;
#endif

CLICK_ENDDECLS
ELEMENT_REQUIRES(NotifierQueue)
EXPORT_ELEMENT(TokenQueue)