aggcounter.cc

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

// CLEAR, REAGGREGATE

void
AggregateCounter::clear_node(Node *n)
{
    if (n->child[0]) {
	clear_node(n->child[0]);
	clear_node(n->child[1]);
    }
    free_node(n);
}

int
AggregateCounter::clear(ErrorHandler *errh)
{
    if (_root)
	clear_node(_root);
    
    if (!(_root = new_node())) {
	if (errh)
	    errh->error("out of memory!");
	return -1;
    }
    _root->aggregate = 0;
    _root->count = 0;
    _root->child[0] = _root->child[1] = 0;
    _num_nonzero = 0;
    _count = 0;
    return 0;
}


void
AggregateCounter::reaggregate_node(Node *n)
{
    Node *l = n->child[0], *r = n->child[1];
    uint32_t count = n->count;
    free_node(n);
    
    if ((n = find_node(count, false))) {
	if (!n->count)
	    _num_nonzero++;
	n->count++;
	_count++;
    }

    if (l) {
	reaggregate_node(l);
	reaggregate_node(r);
    }
}

void
AggregateCounter::reaggregate_counts()
{
    Node *old_root = _root;
    _root = 0;
    clear();
    reaggregate_node(old_root);
}


// HANDLERS

static void
write_batch(FILE *f, AggregateCounter::WriteFormat format,
	    uint32_t *buffer, int pos, double count, ErrorHandler *)
{
    if (format == AggregateCounter::WR_BINARY)
	fwrite(buffer, sizeof(uint32_t), pos, f);
    else if (format == AggregateCounter::WR_TEXT_IP)
	for (int i = 0; i < pos; i += 2)
	    fprintf(f, "%d.%d.%d.%d %u\n", (buffer[i] >> 24) & 255, (buffer[i] >> 16) & 255, (buffer[i] >> 8) & 255, buffer[i] & 255, buffer[i+1]);
    else if (format == AggregateCounter::WR_TEXT_PDF)
	for (int i = 0; i < pos; i += 2)
	    fprintf(f, "%u %.12g\n", buffer[i], buffer[i+1] / count);
    else if (format == AggregateCounter::WR_TEXT)
	for (int i = 0; i < pos; i += 2)
	    fprintf(f, "%u %u\n", buffer[i], buffer[i+1]);
}

void
AggregateCounter::write_nodes(Node *n, FILE *f, WriteFormat format,
			      uint32_t *buffer, int &pos, int len,
			      ErrorHandler *errh) const
{
    if (n->count > 0) {
	buffer[pos++] = n->aggregate;
	buffer[pos++] = n->count;
	if (pos == len) {
	    write_batch(f, format, buffer, pos, _count, errh);
	    pos = 0;
	}
    }

    if (n->child[0])
	write_nodes(n->child[0], f, format, buffer, pos, len, errh);
    if (n->child[1])
	write_nodes(n->child[1], f, format, buffer, pos, len, errh);
}

int
AggregateCounter::write_file(String where, WriteFormat format,
			     ErrorHandler *errh) const
{
    FILE *f;
    if (where == "-")
	f = stdout;
    else
	f = fopen(where.cc(), (format == WR_BINARY ? "wb" : "w"));
    if (!f)
	return errh->error("%s: %s", where.cc(), strerror(errno));

    fwrite(_output_banner.data(), 1, _output_banner.length(), f);
    if (_output_banner.length() && _output_banner.back() != '\n')
	fputc('\n', f);
    fprintf(f, "!num_nonzero %u\n", _num_nonzero);
    if (format == WR_BINARY) {
#if CLICK_BYTE_ORDER == CLICK_BIG_ENDIAN
	fprintf(f, "!packed_be\n");
#elif CLICK_BYTE_ORDER == CLICK_LITTLE_ENDIAN
	fprintf(f, "!packed_le\n");
#else
	format = WR_TEXT;
#endif
    } else if (format == WR_TEXT_IP)
	fprintf(f, "!ip\n");
    
    uint32_t buf[1024];
    int pos = 0;
    write_nodes(_root, f, format, buf, pos, 1024, errh);
    if (pos)
	write_batch(f, format, buf, pos, _count, errh);

    bool had_err = ferror(f);
    if (f != stdout)
	fclose(f);
    if (had_err)
	return errh->error("%s: file error", where.cc());
    else
	return 0;
}

int
AggregateCounter::write_file_handler(const String &data, Element *e, void *thunk, ErrorHandler *errh)
{
    AggregateCounter *ac = static_cast<AggregateCounter *>(e);
    String fn;
    if (!cp_filename(cp_uncomment(data), &fn))
	return errh->error("argument should be filename");
    int int_thunk = (int)thunk;
    return ac->write_file(fn, (WriteFormat)int_thunk, errh);
}

enum {
    AC_FROZEN, AC_ACTIVE, AC_BANNER, AC_STOP, AC_REAGGREGATE, AC_CLEAR,
    AC_AGGREGATE_CALL, AC_COUNT_CALL, AC_NAGG, AC_COUNT
};

String
AggregateCounter::read_handler(Element *e, void *thunk)
{
    AggregateCounter *ac = static_cast<AggregateCounter *>(e);
    switch ((int)thunk) {
      case AC_FROZEN:
	return cp_unparse_bool(ac->_frozen) + "\n";
      case AC_ACTIVE:
	return cp_unparse_bool(ac->_active) + "\n";
      case AC_BANNER:
	return ac->_output_banner;
      case AC_AGGREGATE_CALL:
	if (ac->_call_nnz == (uint32_t)(-1))
	    return "";
	else
	    return String(ac->_call_nnz) + " " + ac->_call_nnz_h->unparse();
      case AC_COUNT_CALL:
	if (ac->_call_count == (uint64_t)(-1))
	    return "";
	else
	    return String(ac->_call_count) + " " + ac->_call_count_h->unparse();
      case AC_COUNT:
	return String(ac->_count) + "\n";
      case AC_NAGG:
	return String(ac->_num_nonzero) + "\n";
      default:
	return "<error>";
    }
}

int
AggregateCounter::write_handler(const String &data, Element *e, void *thunk, ErrorHandler *errh)
{
    AggregateCounter *ac = static_cast<AggregateCounter *>(e);
    String s = cp_uncomment(data);
    switch ((int)thunk) {
      case AC_FROZEN: {
	  bool val;
	  if (!cp_bool(s, &val))
	      return errh->error("argument to 'frozen' should be bool");
	  ac->_frozen = val;
	  return 0;
      }
      case AC_ACTIVE: {
	  bool val;
	  if (!cp_bool(s, &val))
	      return errh->error("argument to 'active' should be bool");
	  ac->_active = val;
	  return 0;
      }
      case AC_STOP:
	ac->_active = false;
	ac->router()->please_stop_driver();
	return 0;
      case AC_REAGGREGATE:
	ac->reaggregate_counts();
	return 0;
      case AC_BANNER:
	ac->_output_banner = data;
	if (data && data.back() != '\n')
	    ac->_output_banner += '\n';
	else if (data && data.length() == 1)
	    ac->_output_banner = "";
	return 0;
      case AC_CLEAR:
	return ac->clear(errh);
      case AC_AGGREGATE_CALL: {
	  uint32_t new_nnz = (uint32_t)(-1);
	  if (s) {
	      if (!cp_unsigned(cp_pop_spacevec(s), &new_nnz))
		  return errh->error("argument to 'aggregate_call' should be 'N HANDLER [VALUE]'");
	      else if (HandlerCall::reset_write(ac->_call_nnz_h, s, ac, errh) < 0)
		  return -1;
	  }
	  ac->_call_nnz = new_nnz;
	  return 0;
      }
      case AC_COUNT_CALL: {
	  uint64_t new_count = (uint64_t)(-1);
	  if (s) {
	      if (!cp_unsigned64(cp_pop_spacevec(s), &new_count))
		  return errh->error("argument to 'count_call' should be 'N HANDLER [VALUE]'");
	      else if (HandlerCall::reset_write(ac->_call_count_h, s, ac, errh) < 0)
		  return -1;
	  }
	  ac->_call_count = new_count;
	  return 0;
      }
      default:
	return errh->error("internal error");
    }
}

void
AggregateCounter::add_handlers()
{
    add_write_handler("write_text_file", write_file_handler, (void *)WR_TEXT);
    add_write_handler("write_ascii_file", write_file_handler, (void *)WR_TEXT);
    add_write_handler("write_file", write_file_handler, (void *)WR_BINARY);
    add_write_handler("write_ip_file", write_file_handler, (void *)WR_TEXT_IP);
    add_write_handler("write_pdf_file", write_file_handler, (void *)WR_TEXT_PDF);
    add_read_handler("freeze", read_handler, (void *)AC_FROZEN);
    add_write_handler("freeze", write_handler, (void *)AC_FROZEN);
    add_read_handler("active", read_handler, (void *)AC_ACTIVE);
    add_write_handler("active", write_handler, (void *)AC_ACTIVE);
    add_write_handler("stop", write_handler, (void *)AC_STOP);
    add_write_handler("reaggregate_counts", write_handler, (void *)AC_REAGGREGATE);
    add_write_handler("counts_pdf", write_handler, (void *)AC_REAGGREGATE);
    add_read_handler("banner", read_handler, (void *)AC_BANNER);
    add_write_handler("banner", write_handler, (void *)AC_BANNER);
    add_write_handler("clear", write_handler, (void *)AC_CLEAR);
    add_read_handler("aggregate_call", read_handler, (void *)AC_AGGREGATE_CALL);
    add_write_handler("aggregate_call", write_handler, (void *)AC_AGGREGATE_CALL);
    add_read_handler("count_call", read_handler, (void *)AC_COUNT_CALL);
    add_write_handler("count_call", write_handler, (void *)AC_COUNT_CALL);
    add_read_handler("count", read_handler, (void *)AC_COUNT);
    add_read_handler("nagg", read_handler, (void *)AC_NAGG);
}

ELEMENT_REQUIRES(userlevel)
EXPORT_ELEMENT(AggregateCounter)
CLICK_ENDDECLS