scramblequeue.cc

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

/*
   void
   ScrambleQueue::extra_push(int, Packet *p)
   {
// Code taken from FullNoteQueue::push().
int next = next_i(_tail);

if (next != _head) {
_q[_tail] = p;
_tail = next;

int s = size();
if (s > _highwater_length)
_highwater_length = s;

#if !NOTIFIERQUEUE_LOCK
// This can leave a single packet in the queue indefinitely in
// multithreaded Click, because of a race condition with pull().
if (!signal_active())
wake_listeners();
#else
if (s == 1) {
_lock.acquire();
if (!signal_active())
wake_listeners();
_lock.release();
}
#endif

if (s == capacity())
_full_note.sleep_listeners();

} else {
if (_drops == 0)
click_chatter("%{element}: overflow", this);
_drops++;
p->kill();
}
}
*/

  String
ScrambleQueue::stats(Element *e, void *)
{
  StringAccum sa;
  ScrambleQueue *sq = static_cast<ScrambleQueue *>(e);

  sa << "Pushes: " << sq->_pushes << "; Pulls: " << sq->_pulls << "; Onlyack: " << sq->_onlyack << "; Onlydata: " << sq->_onlydata << "; Qsize: " << sq->size() << "\n";
  sa << "Pkts:  Encoded: " << sq->_totencodedpkts << "; Unencoded: " << sq->_totunencodedpkts << "; Rexmits: " << sq->_totrexmitpkts << "\n";
  sa << "Bytes: Encoded: " << sq->_totencodedbytes << "; Unencoded: " << sq->_totunencodedbytes << "; Rexmits: " << sq->_totrexmitbytes << "\n";
  sa << "Missed Coding Opportunities: " << sq->_missed_coding_opp << "\n";
  if (sq->_enable_coding) {
    sa << "encoding enabled, ";
    /*
       if (sq->_use_bcast) 
       sa << "using broadcast; ";
       else
       sa << "using pseudo-broadcast\n";
       */

    if (sq->_nxorattempts > 0) {
      sa << "AvDepth : " << int(sq->_avdepth*1.0/sq->_nxorattempts + 0.5) << "\n";
    } else {
      sa << "AvDepth : -1\n";
    }

    // now spit out the packets by class
    sa << "Npackets:";
    for (FlowCounter::iterator i = (sq->_npackets).begin(); i != (sq->_npackets).end(); i++) {
      sa << " " << i.key() << ": " << i.value();
    }
    sa << "\n";

    sa << "Nrexmits:";
    for (FlowCounter::iterator i = (sq->_nrexmits).begin(); i != (sq->_nrexmits).end(); i++) {
      sa << " " << i.key() << ": " << i.value();
    }
    sa << "\n";

    sa << "Nencoded:";
    for (FlowCounter::iterator i = (sq->_nencoded).begin(); i != (sq->_nencoded).end(); i++) {
      sa << " " << i.key() << ": " << i.value();
    }
    sa << "\n";

    sa << "Nunencoded:";
    for (FlowCounter::iterator i = (sq->_nunencoded).begin(); i != (sq->_nunencoded).end(); i++) {
      sa << " " << i.key() << ": " << i.value();
    }

  }


  click_chatter("%{element}: statistics ----------", sq);
  return sa.take_string();
}

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

  bool b;

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

  if (b) {
    sq->_pulls = sq->_avdepth = sq->_pushes = sq->_onlyack = sq->_onlydata = sq->_nxorattempts = 0;
    sq->_totencodedpkts = sq->_totunencodedpkts = sq->_totrexmitpkts = 0;
    sq->_totencodedbytes = sq->_totunencodedbytes = sq->_totrexmitbytes = 0;

    while ((sq->_npackets).begin() != (sq->_npackets).end()) {
      (sq->_npackets).remove(((sq->_npackets).begin()).key());
    }
    while ((sq->_nrexmits).begin() != (sq->_nrexmits).end()) {
      (sq->_nrexmits).remove(((sq->_nrexmits).begin()).key());
    }
    while ((sq->_nencoded).begin() != (sq->_nencoded).end()) {
      (sq->_nencoded).remove(((sq->_nencoded).begin()).key());
    }
    while ((sq->_nunencoded).begin() != (sq->_nunencoded).end()) {
      (sq->_nunencoded).remove(((sq->_nunencoded).begin()).key());
    }
    click_chatter("%{element}: statistics cleared\n", sq);
  }

  return 0;
}

int
ScrambleQueue::enable_coding(const String &arg, Element *e,
		    void *, ErrorHandler *errh)
{
  ScrambleQueue *sq = static_cast<ScrambleQueue *>(e);

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

  sq->set_enable_coding(x);

  if (x) {
    click_chatter("%{element}: coding enabled", sq);
  } else {
    click_chatter("%{element}: coding disabled", sq);
  }
    
  return 0;
}

/*
int
ScrambleQueue::enable(const String &arg, Element *e,
    void *, ErrorHandler *errh)
{
  ScrambleQueue *sq = static_cast<ScrambleQueue *>(e);

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

  if (sq->_scramble_enabled)
    click_chatter("%{element}: encoding enabled", sq);
  else
    click_chatter("%{element}: encoding disabled", sq);

  return 0;
}
*/

/*
   int
   ScrambleQueue::use_bcast(const String &arg, Element *e,
   void *, ErrorHandler *errh)
   {
   ScrambleQueue *sq = static_cast<ScrambleQueue *>(e);

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

   if (sq->_use_bcast)
   click_chatter("%{element}: using broadcast", sq);
   else
   click_chatter("%{element}: using pseudo-broadcast", sq);

   return 0;
   }

*/
int 
ScrambleQueue::set_enc_dst(const String &arg, Element *e, void *, ErrorHandler *errh)
{
  ScrambleQueue *sq = static_cast<ScrambleQueue *>(e);
  
  EtherAddress eth;
  String s = cp_uncomment(arg);
  
  if (!cp_ethernet_address(s, &eth))
    return errh->error("enc_dst parameter must be ethernet address");
  sq->_enc_dst = eth;
  sq->_fix_enc_dst = true;
  
  return 0;
}

void
ScrambleQueue::wakeup_if_needed()
{
  if (!_empty_note.signal_active()) {
    click_chatter("%s register next ack wakeup called at %u", id().cc(), Timestamp::now().msec1());
    update_timer();
  } else {
    click_chatter("%s register next ack wakeup has nothing to do, scrambleq listeners awake", id().cc());
  }
}

/*
   void
   ScrambleQueue::register_next_rexmit_wakeup(uint32_t when)
   {
   _next_rexmit_time = when;
   click_chatter("%s register next rexmit wakeup called with %u", id().cc(), when);
   register_wakeup();
   }
*/

void
ScrambleQueue::register_wakeup()
{
  // return;
  uint32_t future = _next_ack_time;
  if (_next_rexmit_time) {
    if (_next_ack_time) {
      if (_next_recp_time) {
        future = (_next_recp_time < _next_ack_time) ? _next_recp_time : _next_ack_time;
        future = (future < _next_rexmit_time) ? future : _next_rexmit_time;
      }
      else {
        future = (_next_ack_time < _next_rexmit_time) ? _next_ack_time : _next_rexmit_time;
      }
    }
    else {
      if (_next_recp_time) {
        future = (_next_rexmit_time < _next_recp_time) ? _next_rexmit_time : _next_recp_time;
      }
      else {
        future = _next_rexmit_time;
      }
    }
  }
  else {
    if (_next_ack_time) {
      if (_next_recp_time) {
        future = (_next_recp_time < _next_ack_time) ? _next_recp_time : _next_ack_time;
      }
      else {
        future = _next_ack_time;
      }
    }
    else {
      if (_next_recp_time) {
        future = _next_recp_time;
      }
      else {
        future = 0;
      }
    }
  }
   
  if (future == _next_time_to_fire) {
    // optimization to avoid rescheduling and unscheduling
    click_chatter("%s register wakeup optimising for next time %u", id().cc(), _next_time_to_fire);
    return;
  }

  _timer.unschedule();

  if (!future) {
    _next_time_to_fire = 0;
    click_chatter("%s register wakeup cancelling timer", id().cc());
    return;
  }

  // here future is non zero and unique
  if (!_last_time_fired) {
    _next_time_to_fire = future;
  } else if (_last_time_fired + _min_timer_interval < future) {
    _next_time_to_fire = future;
  } else {
    _next_time_to_fire = _last_time_fired + _min_timer_interval;
  }
  _timer.schedule_after_ms(_next_time_to_fire - Timestamp::now().msec1());
  click_chatter("%s register wakeup rescheduling timer to %u, last fired at %u", id().cc(), _next_time_to_fire, _last_time_fired);

}

  void
ScrambleQueue::update_timer()
{
  _next_ack_time = _recvmgr->get_next_ack_time();
  _next_recp_time = _listenmgr->get_next_recp_time();
  _next_rexmit_time = _sendmgr->get_next_rexmit_time();
  click_chatter("%s update timer called, next ack time is %u, next recp time %u next rexmit time is %u", id().cc(), _next_ack_time, _next_recp_time, _next_rexmit_time);
  register_wakeup();
}


  void
ScrambleQueue::run_timer()
{
  _last_time_fired = Timestamp::now().msec1();
  _next_time_to_fire = 0;
  if (!_empty_note.signal_active()) {
    click_chatter("%s run_timer executed at %u, waking up listeners", id().cc(), _last_time_fired);
    _empty_note.wake_listeners();
  } else {
    click_chatter("%s run_timer executed at %u, nothing to do", id().cc(), _last_time_fired);
  }
}


  void 
ScrambleQueue::add_handlers()
{
  add_read_handler("stats", stats, (void *)0);
  add_write_handler("clear", static_clear, 0);
  add_write_handler("enable", enable_coding, 0); 
  add_write_handler("enable_coding", enable_coding, 0); 
  //  add_write_handler("use_broadcast", use_bcast, 0); 
  add_write_handler("set_enc_dst", set_enc_dst, 0); 
  SimpleQueue::add_handlers();
}

CLICK_ENDDECLS


#include <click/bighashmap.cc>
#if EXPLICIT_TEMPLATE_INSTANCES
template class HashMap<EtherAddress, uint16_t>;
#endif

ELEMENT_REQUIRES(CodingManager)
EXPORT_ELEMENT(ScrambleQueue)