Queue

[Eddie Kohler]

Hi all,

I think I've figured out what the problem was with DelayUnqueue.

Basically, the combination of significant delay (like 0.5s) and high packet
rates produced bad behavior. Detailed explanation follows. The short
version is, Try the version of delayunqueue.cc currently checked in to the
CVS repository (and attached below).

Please let me know if you still have problems, and thanks for reporting
this.

---

Details: Click has both Tasks and Timers. Tasks run very frequently, and
should be used for frequent events; Timers run less frequently, and should
be used for less frequent events. DelayUnqueue used both a Task and a Timer
-- the Task for short delays (< 0.1s) and the Timer for long delays (>= 0.1s).

This plan makes sense when the rate of packet arrivals matches up with the
delay -- and in particular, when long delays correspond to infrequent
arrivals. But say you have a high packet rate (= frequent arrivals) and a
long delay. Here's what happened:

    Packet 1 arrives with timestamp 0.0s
		=> DelayUnqueue adds delay, packet 1 should be emitted
		   at timestamp 0.5s
		=> This is a long delay so it schedules a Timer
    0.5s later, DelayUnqueue's Timer goes off
		=> Packet 1 emitted
    Packet 2 arrives with timestamp 0.001s (it had been stored in a Queue)
		=> DelayUnqueue adds delay, packet 2 should be emitted
		   at timestamp 0.501s
		=> ***Because the delay is long (0.5s), DelayUnqueue
		   schedules a Timer, even though it might already be
		   time to emit the packet***
    Some time later, DelayUnqueue's Timer goes off
		=> Packet 2 emitted
		=> repeat

You see the problem.

The solution was to decide whether to set a Timer based on the actual time
that the packet should be emitted, rather than statically based on the
delay. This seems to work in practice. I used this simple user-level
configuration to test:

    InfiniteSource(LIMIT 300000, STOP true)
	-> q::Queue(200000)
	-> DelayUnqueue(0.5s)
	-> c::Counter
	-> Discard
    DriverManager(wait_stop, wait 0.5s, stop)

With the old DelayUnqueue, the Queue overflowed, and the output packet rate
was low (~6500 p/s). With the new DelayUnqueue, no Queue overflow and a
nicely higher packet rate (~17000 p/s, about the same as the rate without
Queue or DelayUnqueue).

Eddie

-------------- next part --------------
/*
 * delayunqueue.{cc,hh} -- element pulls packets from input, delays pushing
 * the packet to output port.
 *
 * Copyright (c) 1999-2001 Massachusetts Institute of Technology
 * Copyright (c) 2002 International Computer Science Institute
 *
 * 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/package.hh>
#include <click/error.hh>
#include <click/confparse.hh>
#include <click/glue.hh>
#include "delayunqueue.hh"
#include <click/standard/scheduleinfo.hh>
CLICK_DECLS

DelayUnqueue::DelayUnqueue()
  : Element(1, 1), _p(0), _task(this), _timer(&_task)
{
  MOD_INC_USE_COUNT;
}

DelayUnqueue::~DelayUnqueue()
{
  MOD_DEC_USE_COUNT;
}

int
DelayUnqueue::configure(Vector<String> &conf, ErrorHandler *errh)
{
  return cp_va_parse(conf, this, errh,
		     cpInterval, "delay", &_delay, 0);
}

int
DelayUnqueue::initialize(ErrorHandler *errh)
{
  ScheduleInfo::join_scheduler(this, &_task, errh);
  _timer.initialize(this);
  _signal = Notifier::upstream_pull_signal(this, 0, &_task);
  return 0;
}

void
DelayUnqueue::cleanup(CleanupStage)
{
  if (_p)
    _p->kill();
}

void
DelayUnqueue::run_scheduled()
{
 retry:
  // read a packet
  if (!_p && (_p = input(0).pull())) {
    if (!_p->timestamp_anno().tv_sec) // get timestamp if not set
      click_gettimeofday(&_p->timestamp_anno());
    _p->timestamp_anno() += _delay;
  }
  
  if (_p) {
    struct timeval now, diff;
    click_gettimeofday(&now);
    diff = _p->timestamp_anno() - now;

    if (diff.tv_sec < 0 || (diff.tv_sec == 0 && diff.tv_usec == 0)) {
      // packet ready for output
      _p->timestamp_anno() = now;
      output(0).push(_p);
      _p = 0;
      goto retry;
    } else if (diff.tv_sec == 0 && diff.tv_usec < 100000)
      // small delta, reschedule Task
      /* Task rescheduled below */;
    else {
      // large delta, schedule Timer
      _timer.schedule_at(_p->timestamp_anno());
      return;			// without rescheduling
    }
  } else {
    // no Packet available
    if (!_signal)
      return;			// without rescheduling
  }

  _task.fast_reschedule();
}

String
DelayUnqueue::read_param(Element *e, void *)
{
  DelayUnqueue *u = (DelayUnqueue *)e;
  return cp_unparse_interval(u->_delay) + "\n";
}

void
DelayUnqueue::add_handlers()
{
  add_read_handler("delay", read_param, (void *)0);
  add_task_handlers(&_task);
}

CLICK_ENDDECLS
EXPORT_ELEMENT(DelayUnqueue)
ELEMENT_MT_SAFE(DelayUnqueue)